#
# Copyright © 2022-2023 University of Strasbourg. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import mimiqlink
import hashlib
import tempfile
import json
import os
import shutil
from mimiqcircuits.circuit import Circuit
from mimiqcircuits.qcsresults import QCSResults
from mimiqcircuits.__version__ import __version__
import numpy as np
from time import sleep
# maximum nbumber of samples allowed
MAX_SAMPLES = 2**16
# default value for the number of samples
DEFAULT_SAMPLES = 1000
# minimum and maximum bond dimension allowed
MIN_BONDDIM = 1
MAX_BONDDIM = 2**12
# minimum and maximum entanglement dimension allowed
MIN_ENTDIM = 4
MAX_ENTDIM = 64
# default bond dimension
DEFAULT_BONDDIM = 256
# default entanglement dimension
DEFAULT_ENTDIM = 16
# default time limit
DEFAULT_TIME_LIMIT = 30
# default algorithm
DEFAULT_ALGORITHM = "auto"
RESULTSPB_FILE = "results.pb"
CIRCUITPB_FILE = "circuit.pb"
CIRCUITQASM_FILE = "circuit.qasm"
def _hash_file(filename):
"""
Calculate the SHA-256 hash of a file.
Args:
filename (str): The name of the file.
Returns:
str: The hexadecimal representation of the file's SHA-256 hash.
"""
sha256_hash = hashlib.sha256()
hash = ""
with open(filename, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
hash = sha256_hash.hexdigest()
return hash
[docs]
class MimiqConnection(mimiqlink.MimiqConnection):
"""Represents a connection to the Mimiq Server.
Inherits from: mimiqlink.MimiqConnection python.
"""
def __get_timelimit(self):
"""Fetch the maximum time limit for execution from the server."""
limits = self.user_limits
if limits and limits.get("enabledMaxTimeout"):
return limits.get("maxTimeout", DEFAULT_TIME_LIMIT)
return DEFAULT_TIME_LIMIT
[docs]
def execute(
self,
circuit,
label="pyapi_v" + __version__,
algorithm=DEFAULT_ALGORITHM,
nsamples=DEFAULT_SAMPLES,
bitstrings=None,
timelimit=None,
bonddim=None,
entdim=None,
seed=None,
qasmincludes=None,
):
"""
Execute a circuit simulation using the MIMIQ cloud services.
Args:
circuit (Circuit): The quantum circuit to be executed.
label (str): The label for the execution (default: "circuitsimu").
algorithm (str): The algorithm to be used for execution (default: "auto").
nsamples (int): The number of samples to generate (default: 1000).
bitstrings (list): List of bitstrings for conditional execution (default: None).
timelimit (int): The time limit for execution in seconds (default: 5 * 60).
bonddim (int): The bond dimension for the MPS algorithm (default: None).
entdim (int): The entangling dimension for the MPS algorithm (default: None).
seed (int): The seed for generating random numbers (default: randomly generated). If provided,
uses the specified seed.
qasmincludes (list): List of OPENQASM files to include in the execution (default: None).
Returns:
str: The execution identifier.
Raises:
ValueError: If bonddim, nsamples, or timelimit exceeds the allowed limits.
Examples:
...
>>> from mimiqcircuits import *
>>> conn = MimiqConnection(url = "https://mimiq.qperfect.io/api")
>>> conn.connect()
Starting authentication server on port 1444 (http://localhost:1444)
>>> c = Circuit()
>>> c.push(GateH(),range(10))
10-qubit circuit with 10 instructions:
├── H @ q[0]
├── H @ q[1]
├── H @ q[2]
├── H @ q[3]
├── H @ q[4]
├── H @ q[5]
├── H @ q[6]
├── H @ q[7]
├── H @ q[8]
└── H @ q[9]
<BLANKLINE>
>>> job=conn.execute(c,algorithm="auto")
>>> res=conn.get_results(job)
>>> res
QCSResults:
├── simulator: MIMIQ-StateVector 0.12.1
├── amplitudes time: 1.16e-07s
├── total time: 0.001320324s
├── compression time: 1.717e-05s
├── sample time: 0.000999485s
├── apply time: 0.000115216s
├── fidelity estimate (min,max): (1.000, 1.000)
├── average ≥2-qubit gate error (min,max): (0.000, 0.000)
├── 1 executions
├── 0 amplitudes
└── 1000 samples
Connecting Using Credentials
----------------------------
.. code-block:: python
conn = MimiqConnection(url="https://mimiq.qperfect.io/api")
conn.connectUser("Email_address", "Password")
Saving and Loading Tokens
-------------------------
.. code-block:: python
conn.savetoken("qperfect.json")
conn = MimiqConnection.loadtoken("qperfect.json")
Closing a Connection and Checking Connection Status
---------------------------------------------------
.. code-block:: python
conn.close()
conn.isOpen()
"""
if isinstance(circuit, Circuit) and circuit.is_symbolic():
raise ValueError(
"The circuit contains unevaluated symbolic parameters and cannot be processed until all parameters are fully evaluated."
)
if timelimit is None:
timelimit = self.__get_timelimit()
if timelimit > self.__get_timelimit():
raise ValueError(
f"Timelimit cannot be set more than {self.get_time_limit} minutes."
)
if bitstrings is None:
bitstrings = []
else:
nq = circuit.num_qubits()
for b in bitstrings:
if len(b) != nq:
raise ValueError(
"The number of qubits in the bitstring is not equal to the number of qubits in the circuit."
)
# if seed is none default it to a random int32 seed
if seed is None:
seed = int(np.random.randint(0, np.iinfo(np.int_).max, dtype=np.int_))
if (algorithm == "auto" or algorithm == "mps") and bonddim is None:
bonddim = DEFAULT_BONDDIM
if (algorithm == "auto" or algorithm == "mps") and entdim is None:
entdim = DEFAULT_ENTDIM
if bonddim is not None and (bonddim < MIN_BONDDIM or bonddim > MAX_BONDDIM):
raise ValueError(f"bonddim must be between {MIN_BONDDIM} and {MAX_BONDDIM}")
if entdim is not None and (entdim < MIN_ENTDIM or entdim > MAX_ENTDIM):
raise ValueError(f"entdim must be between {MIN_ENTDIM} and {MAX_ENTDIM}")
if nsamples > MAX_SAMPLES:
raise ValueError(f"nsamples must be less than {MAX_SAMPLES}")
with tempfile.TemporaryDirectory() as tmpdir:
files = []
allfiles = []
if isinstance(circuit, Circuit):
circuit_filename = os.path.join(tmpdir, CIRCUITPB_FILE)
circuit.saveproto(circuit_filename)
elif isinstance(circuit, str):
if not os.path.isfile(circuit):
raise FileNotFoundError(f"File {circuit} not found.")
circuit_filename = os.path.join(tmpdir, CIRCUITQASM_FILE)
shutil.copyfile(circuit, circuit_filename)
if qasmincludes is None:
qasmincludes = []
for file in qasmincludes:
base = os.path.basename(file)
if not os.path.isfile(file):
raise FileNotFoundError(f"File {file} not found.")
shutil.copyfile(file, os.path.join(tmpdir, base))
files.append({"name": base, "hash": _hash_file(file)})
allfiles.append(os.path.join(tmpdir, base))
else:
raise TypeError("circuit must be a Circuit object or a OPENQASM file")
circuit_hash = _hash_file(circuit_filename)
files.append(
{"name": os.path.basename(circuit_filename), "hash": circuit_hash}
)
allfiles.append(circuit_filename)
jsonbitstrings = ["bs" + o.to01() for o in bitstrings]
pars = {
"algorithm": algorithm,
"bitstrings": jsonbitstrings,
"samples": nsamples,
"seed": seed,
}
if bonddim is not None:
pars["bondDimension"] = bonddim
req = {
"executor": "Circuits",
"timelimit": timelimit,
"files": files,
"parameters": pars,
"apilang: ": "python",
"apiversion": __version__,
"circuitsapiversion": __version__,
}
req_filename = os.path.join(tmpdir, "parameters.json")
allfiles.append(req_filename)
with open(req_filename, "w") as f:
json.dump(req, f)
emutype = "CIRC"
return self.request(
emutype,
algorithm,
label,
timelimit,
allfiles,
)
[docs]
def get_results(self, execution, interval=10):
"""Retrieve the results of a completed execution.
Args:
execution (str): The execution identifier.
interval (int): The interval (in seconds) for checking job status (default: 10).
Returns:
Results: An instance of the QCSResults class.
Raises:
RuntimeError: If the remote job encounters an error.
"""
while not self.isJobDone(execution):
sleep(interval)
if self.isJobCanceled(execution):
raise RuntimeError("Remote job canceled.")
infos = self.requestInfo(execution)
if infos["status"] == "ERROR":
if "errorMessage" in infos:
msg = infos["errorMessage"]
raise RuntimeError(f"Remote job errored: {msg}")
else:
raise RuntimeError("Remote job errored.")
with tempfile.TemporaryDirectory() as tmpdir:
names = self.downloadResults(execution, destdir=tmpdir)
if RESULTSPB_FILE not in names:
raise RuntimeError("File not found in results. Update Your library.")
results = QCSResults.loadproto(os.path.join(tmpdir, RESULTSPB_FILE))
return results
_all__ = ["MimiqConnection"]
