diff --git a/.github/workflows/mypy.yml b/.github/workflows/mypy.yml new file mode 100644 index 000000000..2fb845d1a --- /dev/null +++ b/.github/workflows/mypy.yml @@ -0,0 +1,25 @@ +name: mypy + +on: + pull_request: + push: + branches: + - main + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }} + cancel-in-progress: true + +jobs: + lint: + runs-on: ubuntu-latest + name: Run my[py] linter + steps: + - uses: actions/checkout@v3 + - uses: actions/setup-python@v4 + with: + python-version: "3.10" + - uses: pre-commit/action@v3.0.0 + with: + extra_args: mypy --all-files diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ac2d2f82b..2247f5f4d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -10,6 +10,7 @@ ci: autoupdate_commit_msg: "⬆️🪝 update pre-commit hooks" autofix_commit_msg: "🎨 pre-commit fixes" + skip: [mypy] repos: # Standard hooks @@ -73,3 +74,19 @@ repos: hooks: - id: ruff args: ["--fix"] + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.0.1 + hooks: + - id: mypy + files: ^(src|tests|setup.py) + args: [] + additional_dependencies: + - importlib_resources + - pytest + - types-setuptools + - types-requests + - types-tqdm + - types-flask + - pytket-qiskit + - qiskit[all] diff --git a/README.md b/README.md index 52959820f..fbebe8ce2 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,7 @@ def get_benchmark( circuit_size: int = None, benchmark_instance_name: str = None, compiler: str = "qiskit", - compiler_settings: dict[str, dict[str, any]] = None, + compiler_settings: mqt.bench.CompilerSettings = None, gate_set_name: str = "ibm", device_name: str = "ibm_washington", ): @@ -195,17 +195,12 @@ The available parameters are: - `compiler_settings`: Optimization level for `"qiskit"` (`0`-`3`), placement for `"tket"` (`lineplacement` or `graphplacement`), exemplary shown: ```python -compiler_settings = { - "qiskit": {"optimization_level": 1}, -} -``` - -or +from mqt.bench import CompilerSettings -```python -compiler_settings = { - "tket": {"placement": "lineplacement"}, -} +compiler_settings = CompilerSettings( + qiskit={"optimization_level": 1}, + tket={"placement": "lineplacement"}, +) ``` - `gate_set_name`: `"ibm"`, `"rigetti"`, `"ionq"`, or `"oqc"` diff --git a/pyproject.toml b/pyproject.toml index a3bf06cf5..70d6198ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,6 +83,25 @@ precision = 1 [tool.black] line-length = 120 +[tool.mypy] +mypy_path = "$MYPY_CONFIG_FILE_DIR/src" +files = ["src", "tests", "setup.py"] +python_version = "3.9" +strict = true +show_error_codes = true +enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] +warn_unreachable = true +explicit_package_bases = true +pretty = true + +[[tool.mypy.overrides]] +module = ["pytket.*"] +implicit_reexport = true + +[[tool.mypy.overrides]] +module = ["qiskit.*", "qiskit_nature.*", "qiskit_machine_learning.*", "qiskit_finance.*","qiskit_optimization.*", "joblib.*", "networkx.*", "pandas.*"] +ignore_missing_imports = true + [tool.ruff] select = [ "E", "F", "W", # flake8 diff --git a/src/mqt/bench/__init__.py b/src/mqt/bench/__init__.py index 5c1adee87..9c88365a7 100644 --- a/src/mqt/bench/__init__.py +++ b/src/mqt/bench/__init__.py @@ -3,5 +3,21 @@ get_benchmark, BenchmarkGenerator, timeout_watcher, + CompilerSettings, + QiskitSettings, + TKETSettings, ) from mqt.bench import qiskit_helper, tket_helper, utils + +__all__ = [ + "generate", + "get_benchmark", + "BenchmarkGenerator", + "timeout_watcher", + "qiskit_helper", + "tket_helper", + "utils", + "CompilerSettings", + "QiskitSettings", + "TKETSettings", +] diff --git a/src/mqt/bench/benchmark_generator.py b/src/mqt/bench/benchmark_generator.py index 0357c88fd..b8abff484 100644 --- a/src/mqt/bench/benchmark_generator.py +++ b/src/mqt/bench/benchmark_generator.py @@ -5,22 +5,58 @@ import signal import sys from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, Callable, TypedDict + +from joblib import Parallel, delayed +from mqt.bench import qiskit_helper, tket_helper, utils +from qiskit import QuantumCircuit if TYPE_CHECKING: # pragma: no cover - from qiskit import QuantumCircuit + from pytket.circuit import Circuit if TYPE_CHECKING or sys.version_info >= (3, 10, 0): # pragma: no cover from importlib import resources else: import importlib_resources as resources -from joblib import Parallel, delayed -from mqt.bench import qiskit_helper, tket_helper, utils +from dataclasses import dataclass + + +class Benchmark(TypedDict, total=False): + name: str + include: bool + min_qubits: int + max_qubits: int + min_nodes: int + max_nodes: int + min_index: int + max_index: int + min_uncertainty: int + max_uncertainty: int + instances: list[str] + ancillary_mode: list[str] + stepsize: int + precheck_possible: bool + + +@dataclass +class QiskitSettings: + optimization_level: int = 1 + + +@dataclass +class TKETSettings: + placement: str = "lineplacement" + + +@dataclass +class CompilerSettings: + qiskit: QiskitSettings | None = None + tket: TKETSettings | None = None class BenchmarkGenerator: - def __init__(self, cfg_path: str = "./config.json", qasm_output_path: str | None = None): + def __init__(self, cfg_path: str = "./config.json", qasm_output_path: str | None = None) -> None: with Path(cfg_path).open() as jsonfile: self.cfg = json.load(jsonfile) print("Read config successful") @@ -33,12 +69,11 @@ def __init__(self, cfg_path: str = "./config.json", qasm_output_path: str | None Path(self.qasm_output_path).mkdir(exist_ok=True, parents=True) def create_benchmarks_from_config(self, num_jobs: int = -1) -> bool: - Parallel(n_jobs=num_jobs, verbose=100)( - delayed(self.generate_benchmark)(benchmark) for benchmark in self.cfg["benchmarks"] - ) + benchmarks = [Benchmark(benchmark) for benchmark in self.cfg["benchmarks"]] # type: ignore[misc] + Parallel(n_jobs=num_jobs, verbose=100)(delayed(self.generate_benchmark)(benchmark) for benchmark in benchmarks) return True - def generate_benchmark(self, benchmark): # noqa: PLR0912 + def generate_benchmark(self, benchmark: Benchmark) -> None: # noqa: PLR0912 lib = utils.get_module_for_benchmark(benchmark["name"]) file_precheck = benchmark["precheck_possible"] if benchmark["include"]: @@ -101,7 +136,7 @@ def generate_benchmark(self, benchmark): # noqa: PLR0912 if not success_flag: break - def generate_circuits_on_all_levels(self, qc, num_qubits, file_precheck): + def generate_circuits_on_all_levels(self, qc: QuantumCircuit, num_qubits: int, file_precheck: bool) -> bool: success_generated_circuits_t_indep = self.generate_target_indep_level_circuit(qc, num_qubits, file_precheck) if not success_generated_circuits_t_indep: @@ -110,7 +145,7 @@ def generate_circuits_on_all_levels(self, qc, num_qubits, file_precheck): self.generate_target_dep_level_circuit(qc, num_qubits, file_precheck) return True - def generate_target_indep_level_circuit(self, qc: QuantumCircuit, num_qubits: int, file_precheck): + def generate_target_indep_level_circuit(self, qc: QuantumCircuit, num_qubits: int, file_precheck: bool) -> bool: num_generated_circuits = 0 res_indep_qiskit = timeout_watcher( qiskit_helper.get_indep_level, @@ -130,8 +165,8 @@ def generate_target_indep_level_circuit(self, qc: QuantumCircuit, num_qubits: in return num_generated_circuits != 0 - def generate_target_dep_level_circuit(self, qc: QuantumCircuit, num_qubits: int, file_precheck): - compilation_paths = [ + def generate_target_dep_level_circuit(self, qc: QuantumCircuit, num_qubits: int, file_precheck: bool) -> bool: + compilation_paths: list[tuple[str, list[tuple[str, int]]]] = [ ("ibm", [("ibm_washington", 127), ("ibm_montreal", 27)]), ("rigetti", [("rigetti_aspen_m2", 80)]), ("ionq", [("ionq11", 11)]), @@ -220,35 +255,36 @@ def generate_target_dep_level_circuit(self, qc: QuantumCircuit, num_qubits: int, num_generated_benchmarks += 1 return num_generated_benchmarks != 0 - def start_benchmark_generation(self, create_circuit_function, parameters, file_precheck) -> bool: + def start_benchmark_generation( + self, create_circuit_function: Callable[..., QuantumCircuit], parameters: list[int | str], file_precheck: bool + ) -> bool: res_qc_creation = timeout_watcher(create_circuit_function, self.timeout, parameters) if not res_qc_creation: return False + assert isinstance(res_qc_creation, QuantumCircuit) return self.generate_circuits_on_all_levels(res_qc_creation, res_qc_creation.num_qubits, file_precheck) def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 benchmark_name: str, level: str | int, - circuit_size: int = None, - benchmark_instance_name: str = None, + circuit_size: int | None = None, + benchmark_instance_name: str | None = None, compiler: str | None = "qiskit", - compiler_settings: dict[str, dict[str, any]] | None = None, + compiler_settings: CompilerSettings | None = None, gate_set_name: str | None = "ibm", device_name: str | None = "ibm_washington", -): +) -> QuantumCircuit | Circuit: """Returns one benchmark as a Qiskit::QuantumCircuit Object. - Keyword arguments: benchmark_name -- name of the to be generated benchmark level -- Choice of level, either as a string ("alg", "indep", "nativegates" or "mapped") or as a number between 0-3 where 0 corresponds to "alg" level and 3 to "mapped" level circuit_size -- Input for the benchmark creation, in most cases this is equal to the qubit number benchmark_instance_name -- Input selection for some benchmarks, namely "groundstate" and "shor" compiler -- "qiskit" or "tket" - compiler_settings -- Dictionary containing the respective compiler settings for the specified compiler (e.g., optimization level for Qiskit or placement for TKET) + CompilerSettings -- Data class containing the respective compiler settings for the specified compiler (e.g., optimization level for Qiskit or placement for TKET) gate_set_name -- "ibm", "rigetti", "ionq", or "oqc" device_name -- "ibm_washington", "ibm_montreal", "rigetti_aspen_m2", "ionq11", ""oqc_lucy"" - Return values: Quantum Circuit Object -- Representing the benchmark with the selected options, either as Qiskit::QuantumCircuit or Pytket::Circuit object (depending on the chosen compiler---while the algorithm level is always provided using Qiskit) """ @@ -269,16 +305,12 @@ def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 msg = "benchmark_instance_name must be defined for this benchmark." raise ValueError(msg) - if benchmark_instance_name is not None and not isinstance(benchmark_instance_name, str): - msg = "benchmark_instance_name must be None or str." - raise ValueError(msg) - if compiler is not None and compiler.lower() not in utils.get_supported_compilers(): msg = f"Selected compiler must be in {utils.get_supported_compilers()}." raise ValueError(msg) - if compiler_settings is not None and not isinstance(compiler_settings, dict): - msg = "compiler_settings must be None or dict[str, dict[str, any]]." + if compiler_settings is not None and not isinstance(compiler_settings, CompilerSettings): + msg = "compiler_settings must be of type CompilerSettings or None." # type:ignore[unreachable] raise ValueError(msg) if gate_set_name is not None and gate_set_name not in utils.get_supported_gatesets(): @@ -328,10 +360,8 @@ def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 raise ValueError(msg) if compiler_settings is None: - compiler_settings = { - "qiskit": {"optimization_level": 1}, - "tket": {"placement": "lineplacement"}, - } + compiler_settings = CompilerSettings(QiskitSettings(), TKETSettings()) + assert (compiler_settings.tket is not None) or (compiler_settings.qiskit is not None) independent_level = 1 if level == "indep" or level == independent_level: @@ -342,16 +372,22 @@ def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 native_gates_level = 2 if level == "nativegates" or level == native_gates_level: + assert gate_set_name is not None if compiler == "qiskit": - opt_level = compiler_settings["qiskit"]["optimization_level"] + assert compiler_settings.qiskit is not None + opt_level = compiler_settings.qiskit.optimization_level return qiskit_helper.get_native_gates_level(qc, gate_set_name, circuit_size, opt_level, False, True) if compiler == "tket": return tket_helper.get_native_gates_level(qc, gate_set_name, circuit_size, False, True) mapped_level = 3 if level == "mapped" or level == mapped_level: + assert gate_set_name is not None + assert device_name is not None if compiler == "qiskit": - opt_level = compiler_settings["qiskit"]["optimization_level"] + assert compiler_settings.qiskit is not None + opt_level = compiler_settings.qiskit.optimization_level + assert isinstance(opt_level, int) return qiskit_helper.get_mapped_level( qc, gate_set_name, @@ -362,7 +398,8 @@ def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 True, ) if compiler == "tket": - placement = compiler_settings["tket"]["placement"].lower() + assert compiler_settings.tket is not None + placement = compiler_settings.tket.placement.lower() lineplacement = placement == "lineplacement" return tket_helper.get_mapped_level( qc, @@ -378,7 +415,7 @@ def get_benchmark( # noqa: PLR0911, PLR0912, PLR0915 raise ValueError(msg) -def generate(num_jobs: int = -1): +def generate(num_jobs: int = -1) -> None: parser = argparse.ArgumentParser(description="Create Configuration") parser.add_argument("--file-name", type=str, help="optional filename", default="./config.json") args = parser.parse_args() @@ -386,11 +423,13 @@ def generate(num_jobs: int = -1): benchmark_generator.create_benchmarks_from_config(num_jobs) -def timeout_watcher(func, timeout, args): +def timeout_watcher( + func: Callable[..., bool | QuantumCircuit], timeout: int, args: list[Any] +) -> bool | QuantumCircuit | Circuit: class TimeoutException(Exception): # Custom exception class pass - def timeout_handler(_signum, _frame): # Custom signal handler + def timeout_handler(_signum: Any, _frame: Any) -> None: # Custom signal handler raise TimeoutException() # Change the behavior of SIGALRM diff --git a/src/mqt/bench/benchmarks/__init__.py b/src/mqt/bench/benchmarks/__init__.py index c3de5ffc7..ed1904c3c 100644 --- a/src/mqt/bench/benchmarks/__init__.py +++ b/src/mqt/bench/benchmarks/__init__.py @@ -2,3 +2,14 @@ from mqt.bench.benchmarks.qiskit_application_nature import groundstate from mqt.bench.benchmarks.qiskit_application_optimization import routing, tsp from mqt.bench.benchmarks.qiskit_application_ml import qgan + +__all__ = [ + "pricingput", + "pricingcall", + "portfolioqaoa", + "portfoliovqe", + "groundstate", + "routing", + "tsp", + "qgan", +] diff --git a/src/mqt/bench/benchmarks/ae.py b/src/mqt/bench/benchmarks/ae.py index 8573cd07c..59e8942a8 100644 --- a/src/mqt/bench/benchmarks/ae.py +++ b/src/mqt/bench/benchmarks/ae.py @@ -2,11 +2,12 @@ from __future__ import annotations -from mqt.bench.utils import get_estimation_problem -from qiskit.algorithms import AmplitudeEstimation +import numpy as np +from qiskit import QuantumCircuit +from qiskit.algorithms import AmplitudeEstimation, EstimationProblem -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing Quantum Amplitude Estimation. Keyword arguments: @@ -23,3 +24,42 @@ def create_circuit(num_qubits: int): qc.measure_all() return qc + + +class BernoulliQ(QuantumCircuit): # type: ignore[misc] + """A circuit representing the Bernoulli Q operator.""" + + def __init__(self, probability: float) -> None: + super().__init__(1) # circuit on 1 qubit + + self._theta_p = 2 * np.arcsin(np.sqrt(probability)) + self.ry(2 * self._theta_p, 0) + + def __eq__(self, other: object) -> bool: + return isinstance(other, BernoulliQ) and self._theta_p == other._theta_p + + def power(self, power: float, _matrix_power: bool = True) -> QuantumCircuit: + # implement the efficient power of Q + q_k = QuantumCircuit(1) + q_k.ry(2 * power * self._theta_p, 0) + return q_k + + +def get_estimation_problem() -> EstimationProblem: + """Returns a estimation problem instance for a fixed p value.""" + + p = 0.2 + + """A circuit representing the Bernoulli A operator.""" + a = QuantumCircuit(1) + theta_p = 2 * np.arcsin(np.sqrt(p)) + a.ry(theta_p, 0) + + """A circuit representing the Bernoulli Q operator.""" + q = BernoulliQ(p) + + return EstimationProblem( + state_preparation=a, # A operator + grover_operator=q, # Q operator + objective_qubits=[0], # the "good" state Psi1 is identified as measuring |1> in qubit 0 + ) diff --git a/src/mqt/bench/benchmarks/dj.py b/src/mqt/bench/benchmarks/dj.py index a6b8d0e04..9befc87eb 100644 --- a/src/mqt/bench/benchmarks/dj.py +++ b/src/mqt/bench/benchmarks/dj.py @@ -6,12 +6,12 @@ from qiskit import QuantumCircuit -def dj_oracle(case, n): +def dj_oracle(case: str, n: int) -> QuantumCircuit: # plus one output qubit oracle_qc = QuantumCircuit(n + 1) if case == "balanced": - np.random.seed = 10 + np.random.seed(10) b_str = "" for _ in range(n): b = np.random.randint(0, 2) @@ -38,7 +38,7 @@ def dj_oracle(case, n): return oracle_gate -def dj_algorithm(oracle, n): +def dj_algorithm(oracle: QuantumCircuit, n: int) -> QuantumCircuit: dj_circuit = QuantumCircuit(n + 1, n) dj_circuit.x(n) @@ -59,7 +59,7 @@ def dj_algorithm(oracle, n): return dj_circuit -def create_circuit(n: int, balanced: bool = True): +def create_circuit(n: int, balanced: bool = True) -> QuantumCircuit: """Returns a quantum circuit implementing the Deutsch-Josza algorithm. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/ghz.py b/src/mqt/bench/benchmarks/ghz.py index 0cc02bf5b..0cdbf8d87 100644 --- a/src/mqt/bench/benchmarks/ghz.py +++ b/src/mqt/bench/benchmarks/ghz.py @@ -3,7 +3,7 @@ from qiskit import QuantumCircuit, QuantumRegister -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the GHZ state. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/graphstate.py b/src/mqt/bench/benchmarks/graphstate.py index 972679ff1..20f9c21a2 100644 --- a/src/mqt/bench/benchmarks/graphstate.py +++ b/src/mqt/bench/benchmarks/graphstate.py @@ -5,7 +5,7 @@ from qiskit.circuit.library import GraphState -def create_circuit(num_qubits: int, degree: int = 2): +def create_circuit(num_qubits: int, degree: int = 2) -> QuantumCircuit: """Returns a quantum circuit implementing a graph state. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/grover.py b/src/mqt/bench/benchmarks/grover.py index c06e976b5..31262ff95 100644 --- a/src/mqt/bench/benchmarks/grover.py +++ b/src/mqt/bench/benchmarks/grover.py @@ -6,7 +6,7 @@ from qiskit.circuit.library import GroverOperator -def create_circuit(num_qubits: int, ancillary_mode: str = "noancilla"): +def create_circuit(num_qubits: int, ancillary_mode: str = "noancilla") -> QuantumCircuit: """Returns a quantum circuit implementing Grover's algorithm. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/hhl.py b/src/mqt/bench/benchmarks/hhl.py index 735c3b897..33e44774a 100644 --- a/src/mqt/bench/benchmarks/hhl.py +++ b/src/mqt/bench/benchmarks/hhl.py @@ -2,14 +2,19 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.algorithms.linear_solvers.hhl import HHL from qiskit.algorithms.linear_solvers.matrices.tridiagonal_toeplitz import ( TridiagonalToeplitz, ) +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the HHL algorithm for a specific example matrix. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qaoa.py b/src/mqt/bench/benchmarks/qaoa.py index 82f2cce95..d02ef0da9 100644 --- a/src/mqt/bench/benchmarks/qaoa.py +++ b/src/mqt/bench/benchmarks/qaoa.py @@ -2,13 +2,19 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from mqt.bench.utils import get_examplary_max_cut_qp from qiskit.algorithms.minimum_eigensolvers import QAOA from qiskit.algorithms.optimizers import SLSQP from qiskit.primitives import Sampler +from qiskit_optimization import QuadraticProgram + +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Approximation Optimization Algorithm for a specific max-cut example. @@ -17,6 +23,7 @@ def create_circuit(num_qubits: int): """ qp = get_examplary_max_cut_qp(num_qubits) + assert isinstance(qp, QuadraticProgram) qaoa = QAOA(sampler=Sampler(), reps=2, optimizer=SLSQP(maxiter=25)) qaoa_result = qaoa.compute_minimum_eigenvalue(qp.to_ising()[0]) diff --git a/src/mqt/bench/benchmarks/qft.py b/src/mqt/bench/benchmarks/qft.py index 36a27d969..d26dcf728 100644 --- a/src/mqt/bench/benchmarks/qft.py +++ b/src/mqt/bench/benchmarks/qft.py @@ -4,7 +4,7 @@ from qiskit.circuit.library import QFT -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Fourier Transform algorithm. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qftentangled.py b/src/mqt/bench/benchmarks/qftentangled.py index ce2079534..c9ed84bca 100644 --- a/src/mqt/bench/benchmarks/qftentangled.py +++ b/src/mqt/bench/benchmarks/qftentangled.py @@ -4,7 +4,7 @@ from qiskit.circuit.library import QFT -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Fourier Transform algorithm using entangled qubits. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qiskit_application_finance/portfolioqaoa.py b/src/mqt/bench/benchmarks/qiskit_application_finance/portfolioqaoa.py index 4f7e3a42e..d523ca4dc 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_finance/portfolioqaoa.py +++ b/src/mqt/bench/benchmarks/qiskit_application_finance/portfolioqaoa.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +from typing import TYPE_CHECKING from qiskit.algorithms.minimum_eigensolvers import QAOA from qiskit.algorithms.optimizers import COBYLA @@ -11,8 +12,11 @@ from qiskit_finance.data_providers import RandomDataProvider from qiskit_optimization.converters import QuadraticProgramToQubo +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit -def create_circuit(num_qubits: int): + +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit of QAOA applied to a specific portfolio optimization task. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qiskit_application_finance/portfoliovqe.py b/src/mqt/bench/benchmarks/qiskit_application_finance/portfoliovqe.py index b9c1708d8..a50d03f5a 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_finance/portfoliovqe.py +++ b/src/mqt/bench/benchmarks/qiskit_application_finance/portfoliovqe.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +from typing import TYPE_CHECKING from qiskit.algorithms.minimum_eigensolvers import VQE from qiskit.algorithms.optimizers import COBYLA @@ -12,8 +13,11 @@ from qiskit_finance.data_providers import RandomDataProvider from qiskit_optimization.converters import QuadraticProgramToQubo +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit -def create_circuit(num_qubits: int): + +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit of VQE applied to a specific portfolio optimization task. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qiskit_application_finance/pricingcall.py b/src/mqt/bench/benchmarks/qiskit_application_finance/pricingcall.py index 092d667ea..f4b571b48 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_finance/pricingcall.py +++ b/src/mqt/bench/benchmarks/qiskit_application_finance/pricingcall.py @@ -2,13 +2,18 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.algorithms import IterativeAmplitudeEstimation from qiskit_finance.applications.estimation import EuropeanCallPricing from qiskit_finance.circuit.library import LogNormalDistribution +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_uncertainty_qubits: int = 5): +def create_circuit(num_uncertainty_qubits: int = 5) -> QuantumCircuit: """Returns a quantum circuit of Iterative Amplitude Estimation applied to a problem instance of pricing call options. diff --git a/src/mqt/bench/benchmarks/qiskit_application_finance/pricingput.py b/src/mqt/bench/benchmarks/qiskit_application_finance/pricingput.py index df5e26252..cfd13646d 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_finance/pricingput.py +++ b/src/mqt/bench/benchmarks/qiskit_application_finance/pricingput.py @@ -2,13 +2,18 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.algorithms import EstimationProblem, IterativeAmplitudeEstimation from qiskit.circuit.library import LinearAmplitudeFunction from qiskit_finance.circuit.library import LogNormalDistribution +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_uncertainty_qubits: int = 5): +def create_circuit(num_uncertainty_qubits: int = 5) -> QuantumCircuit: """Returns a quantum circuit of Iterative Amplitude Estimation applied to a problem instance of pricing put options. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qiskit_application_ml/qgan.py b/src/mqt/bench/benchmarks/qiskit_application_ml/qgan.py index 09bf76f74..893435b1a 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_ml/qgan.py +++ b/src/mqt/bench/benchmarks/qiskit_application_ml/qgan.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import TYPE_CHECKING, cast + import numpy as np from qiskit import Aer from qiskit.circuit.library import TwoLocal @@ -9,8 +11,12 @@ from qiskit_finance.circuit.library import UniformDistribution from qiskit_machine_learning.algorithms import QGAN, NumPyDiscriminator +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + from qiskit.circuit.parameter import Parameter + -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing Quantum Generative Adversarial Networks algorithm. Keyword arguments: @@ -18,7 +24,7 @@ def create_circuit(num_qubits: int): """ seed = 10 - np.random.seed = seed + np.random.seed(seed) algorithm_globals.random_seed = seed # Number training data samples @@ -34,8 +40,6 @@ def create_circuit(num_qubits: int): upper_bound_value = 2**num_qubits - 1 bounds = np.array([0.0, upper_bound_value]) # Set number of qubits per data dimension as list of k qubit values[#q_0,...,#q_k-1] - num_qubits = [num_qubits] - # Set number of training epochs # Note: The algorithm's runtime can be shortened by reducing the number of training epochs. num_epochs = 5 @@ -43,7 +47,7 @@ def create_circuit(num_qubits: int): batch_size = 10 # Initialize qGAN - qgan = QGAN(real_data, bounds, num_qubits, batch_size, num_epochs, snapshot_dir=None) + qgan = QGAN(real_data, bounds, [num_qubits], batch_size, num_epochs, snapshot_dir=None) qgan.seed = 10 # Set quantum instance to run the quantum generator quantum_instance = QuantumInstance( @@ -54,10 +58,10 @@ def create_circuit(num_qubits: int): ) # Set an initial state for the generator circuit - init_dist = UniformDistribution(sum(num_qubits)) + init_dist = UniformDistribution(num_qubits) # Set the ansatz circuit - ansatz = TwoLocal(int(np.sum(num_qubits)), "ry", "cz", reps=1) # entanglement=entangler_map, + ansatz = TwoLocal(num_qubits, "ry", "cz", reps=1) # entanglement=entangler_map, # You can increase the number of training epochs and use random initial parameters. init_params = np.random.rand(ansatz.num_parameters_settable) * 2 * np.pi @@ -68,9 +72,13 @@ def create_circuit(num_qubits: int): # Set quantum generator qgan.set_generator(generator_circuit=g_circuit, generator_init_params=init_params) # The parameters have an order issue that following is a temp. workaround - qgan._generator._free_parameters = sorted(g_circuit.parameters, key=lambda p: p.name) + + def fct(p: Parameter) -> str: + return cast(str, p.name) + + qgan._generator._free_parameters = sorted(g_circuit.parameters, key=fct) # Set classical discriminator neural network - discriminator = NumPyDiscriminator(len(num_qubits)) + discriminator = NumPyDiscriminator() qgan.set_discriminator(discriminator) qgan.run(quantum_instance) diff --git a/src/mqt/bench/benchmarks/qiskit_application_nature/groundstate.py b/src/mqt/bench/benchmarks/qiskit_application_nature/groundstate.py index 341c570d5..dc147b5d1 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_nature/groundstate.py +++ b/src/mqt/bench/benchmarks/qiskit_application_nature/groundstate.py @@ -3,6 +3,8 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from qiskit.algorithms.minimum_eigensolvers import VQE from qiskit.algorithms.optimizers import COBYLA from qiskit.circuit.library import TwoLocal @@ -10,8 +12,11 @@ from qiskit_nature.second_q.drivers import PySCFDriver from qiskit_nature.second_q.mappers import JordanWignerMapper, QubitConverter +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(choice): +def create_circuit(choice: str) -> QuantumCircuit: """Returns a quantum circuit implementing Ground State Estimation. Keyword arguments: @@ -21,7 +26,7 @@ def create_circuit(choice): driver = PySCFDriver(atom=molecule) es_problem = driver.run() - converter = QubitConverter(JordanWignerMapper()) + converter = QubitConverter(JordanWignerMapper()) # type: ignore[no-untyped-call] second_q_op = es_problem.second_q_ops() operator = converter.convert_only(second_q_op[0]) @@ -45,7 +50,7 @@ def create_circuit(choice): return qc -def get_molecule(benchmark_instance_name: str): +def get_molecule(benchmark_instance_name: str) -> list[str]: """Returns a Molecule object depending on the parameter value.""" m_1 = ["H 0.0 0.0 0.0", "H 0.0 0.0 0.735"] m_2 = ["Li 0.0 0.0 0.0", "H 0.0 0.0 2.5"] diff --git a/src/mqt/bench/benchmarks/qiskit_application_optimization/routing.py b/src/mqt/bench/benchmarks/qiskit_application_optimization/routing.py index d76c36763..783daee31 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_optimization/routing.py +++ b/src/mqt/bench/benchmarks/qiskit_application_optimization/routing.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import TYPE_CHECKING, cast + import numpy as np from qiskit.algorithms.minimum_eigensolvers import VQE from qiskit.algorithms.optimizers import SLSQP @@ -10,14 +12,20 @@ from qiskit.utils import algorithm_globals from qiskit_optimization import QuadraticProgram +if TYPE_CHECKING: # pragma: no cover + from numpy.typing import NDArray + from qiskit import QuantumCircuit + class Initializer: - def __init__(self, n): + def __init__(self, n: int) -> None: self.n = n - def generate_instance(self): + def generate_instance( + self, + ) -> tuple[NDArray[np.float_], NDArray[np.float_], NDArray[np.float_],]: n = self.n - np.random.seed = 10 + np.random.seed(10) xc = (np.random.rand(n) - 0.5) * 10 yc = (np.random.rand(n) - 0.5) * 10 @@ -32,12 +40,14 @@ def generate_instance(self): class QuantumOptimizer: - def __init__(self, instance, n, K): + def __init__(self, instance: NDArray[np.float_], n: int, K: int) -> None: self.instance = instance self.n = n self.K = K - def binary_representation(self, x_sol=0): + def binary_representation( + self, x_sol: NDArray[np.float_] + ) -> tuple[NDArray[np.float_], NDArray[np.float_], float, float]: instance = self.instance n = self.n K = self.K @@ -81,28 +91,27 @@ def binary_representation(self, x_sol=0): c = 2 * A * (n - 1) + 2 * A * (K**2) try: - max(x_sol) - # Evaluates the cost distance from a binary representation of a path - def fun(x): - return np.dot(np.around(x), np.dot(Q, np.around(x))) + np.dot(g, np.around(x)) + c + def fun(x: NDArray[np.float_]) -> float: + return cast(float, np.dot(np.around(x), np.dot(Q, np.around(x))) + np.dot(g, np.around(x)) + c) cost = fun(x_sol) except Exception: cost = 0 - return Q, g, c, cost + return Q, g, cast(float, c), cost - def construct_problem(self, Q, g, c) -> QuadraticProgram: + def construct_problem(self, Q: NDArray[np.float_], g: NDArray[np.float_], c: float) -> QuadraticProgram: qp = QuadraticProgram() for i in range(self.n * (self.n - 1)): qp.binary_var(str(i)) - qp.objective.quadratic = Q - qp.objective.linear = g + + qp.objective.quadratic = Q # type: ignore[assignment] + qp.objective.linear = g # type: ignore[assignment] qp.objective.constant = c return qp - def solve_problem(self, qp): + def solve_problem(self, qp: QuadraticProgram) -> QuantumCircuit: algorithm_globals.random_seed = 10 ansatz = RealAmplitudes(self.n) @@ -111,7 +120,7 @@ def solve_problem(self, qp): return vqe.ansatz.bind_parameters(vqe_result.optimal_point) -def create_circuit(num_nodes: int = 3, num_vehs: int = 2): +def create_circuit(num_nodes: int = 3, num_vehs: int = 2) -> QuantumCircuit: """Returns a quantum circuit solving a routing problem. Keyword arguments: @@ -127,7 +136,7 @@ def create_circuit(num_nodes: int = 3, num_vehs: int = 2): xc, yc, instance = initializer.generate_instance() quantum_optimizer = QuantumOptimizer(instance, n, k) - q, g, c, binary_cost = quantum_optimizer.binary_representation() + q, g, c, binary_cost = quantum_optimizer.binary_representation(x_sol=np.array(0.0, dtype=float)) qp = quantum_optimizer.construct_problem(q, g, c) # Instantiate the quantum optimizer class with parameters: qc = quantum_optimizer.solve_problem(qp) diff --git a/src/mqt/bench/benchmarks/qiskit_application_optimization/tsp.py b/src/mqt/bench/benchmarks/qiskit_application_optimization/tsp.py index 5d4feac28..64b5857d3 100644 --- a/src/mqt/bench/benchmarks/qiskit_application_optimization/tsp.py +++ b/src/mqt/bench/benchmarks/qiskit_application_optimization/tsp.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from qiskit.algorithms.minimum_eigensolvers import VQE from qiskit.algorithms.optimizers import SPSA from qiskit.circuit.library import TwoLocal @@ -10,8 +12,11 @@ from qiskit_optimization.applications import Tsp from qiskit_optimization.converters import QuadraticProgramToQubo +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_nodes: int): +def create_circuit(num_nodes: int) -> QuantumCircuit: """Returns a quantum circuit solving the Travelling Salesman Problem (TSP). Keyword arguments: diff --git a/src/mqt/bench/benchmarks/qpeexact.py b/src/mqt/bench/benchmarks/qpeexact.py index 74399270b..c3290a750 100644 --- a/src/mqt/bench/benchmarks/qpeexact.py +++ b/src/mqt/bench/benchmarks/qpeexact.py @@ -8,7 +8,7 @@ from qiskit.circuit.library import QFT -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Phase Estimation algorithm for a phase which can be exactly estimated. diff --git a/src/mqt/bench/benchmarks/qpeinexact.py b/src/mqt/bench/benchmarks/qpeinexact.py index 190b6ff31..a59fbdde1 100644 --- a/src/mqt/bench/benchmarks/qpeinexact.py +++ b/src/mqt/bench/benchmarks/qpeinexact.py @@ -8,7 +8,7 @@ from qiskit.circuit.library import QFT -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Phase Estimation algorithm for a phase which cannot be exactly estimated. diff --git a/src/mqt/bench/benchmarks/qwalk.py b/src/mqt/bench/benchmarks/qwalk.py index 631ed332c..3034f02cc 100644 --- a/src/mqt/bench/benchmarks/qwalk.py +++ b/src/mqt/bench/benchmarks/qwalk.py @@ -8,7 +8,7 @@ def create_circuit( depth: int = 3, coin_state_preparation: QuantumCircuit = None, ancillary_mode: str = "noancilla", -): +) -> QuantumCircuit: """Returns a quantum circuit implementing the Quantum Walk algorithm. Keyword arguments: diff --git a/src/mqt/bench/benchmarks/realamprandom.py b/src/mqt/bench/benchmarks/realamprandom.py index d99276900..e476b7f65 100644 --- a/src/mqt/bench/benchmarks/realamprandom.py +++ b/src/mqt/bench/benchmarks/realamprandom.py @@ -1,10 +1,15 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.circuit.library import RealAmplitudes +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the RealAmplitudes ansatz with random parameter values. @@ -12,7 +17,7 @@ def create_circuit(num_qubits: int): num_qubits -- number of qubits of the returned quantum circuit """ - np.random.seed = 10 + np.random.seed(10) qc = RealAmplitudes(num_qubits, entanglement="full", reps=3) num_params = qc.num_parameters qc = qc.bind_parameters(np.random.rand(num_params)) diff --git a/src/mqt/bench/benchmarks/shor.py b/src/mqt/bench/benchmarks/shor.py index 511b2f61d..23d63d65c 100644 --- a/src/mqt/bench/benchmarks/shor.py +++ b/src/mqt/bench/benchmarks/shor.py @@ -1,9 +1,14 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from qiskit.algorithms.factorizers import Shor +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_to_be_factorized: int, a: int = 2): +def create_circuit(num_to_be_factorized: int, a: int = 2) -> QuantumCircuit: """Returns a quantum circuit implementing the Shor's algorithm. Keyword arguments: @@ -18,7 +23,7 @@ def create_circuit(num_to_be_factorized: int, a: int = 2): return qc -def get_instance(choice: str): +def get_instance(choice: str) -> list[int]: instances = { "xsmall": [9, 4], # 18 qubits "small": [15, 4], # 18 qubits diff --git a/src/mqt/bench/benchmarks/su2random.py b/src/mqt/bench/benchmarks/su2random.py index 23e2bbe69..94eff01a0 100644 --- a/src/mqt/bench/benchmarks/su2random.py +++ b/src/mqt/bench/benchmarks/su2random.py @@ -1,10 +1,15 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.circuit.library import EfficientSU2 +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing EfficientSU2 ansatz with random parameter values. @@ -12,7 +17,7 @@ def create_circuit(num_qubits: int): num_qubits -- number of qubits of the returned quantum circuit """ - np.random.seed = 10 + np.random.seed(10) qc = EfficientSU2(num_qubits, entanglement="full", reps=3) num_params = qc.num_parameters qc = qc.bind_parameters(np.random.rand(num_params)) diff --git a/src/mqt/bench/benchmarks/twolocalrandom.py b/src/mqt/bench/benchmarks/twolocalrandom.py index e7c251517..d932db805 100644 --- a/src/mqt/bench/benchmarks/twolocalrandom.py +++ b/src/mqt/bench/benchmarks/twolocalrandom.py @@ -1,10 +1,15 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import numpy as np from qiskit.circuit.library import TwoLocal +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit + -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the TwoLocal ansatz with random parameter values. @@ -12,7 +17,7 @@ def create_circuit(num_qubits: int): num_qubits -- number of qubits of the returned quantum circuit """ - np.random.seed = 10 + np.random.seed(10) qc = TwoLocal(num_qubits, "ry", "cx", entanglement="full", reps=3) num_params = qc.num_parameters qc = qc.bind_parameters(np.random.rand(num_params)) diff --git a/src/mqt/bench/benchmarks/vqe.py b/src/mqt/bench/benchmarks/vqe.py index e024b400d..741b39aba 100644 --- a/src/mqt/bench/benchmarks/vqe.py +++ b/src/mqt/bench/benchmarks/vqe.py @@ -2,14 +2,20 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from mqt.bench.utils import get_examplary_max_cut_qp from qiskit.algorithms.minimum_eigensolvers import VQE from qiskit.algorithms.optimizers import SLSQP from qiskit.circuit.library import RealAmplitudes from qiskit.primitives import Estimator +from qiskit_optimization import QuadraticProgram + +if TYPE_CHECKING: # pragma: no cover + from qiskit import QuantumCircuit -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the Variational Quantum Eigensolver Algorithm for a specific max-cut example. @@ -18,6 +24,7 @@ def create_circuit(num_qubits: int): """ qp = get_examplary_max_cut_qp(num_qubits) + assert isinstance(qp, QuadraticProgram) ansatz = RealAmplitudes(num_qubits, reps=2) vqe = VQE(ansatz=ansatz, optimizer=SLSQP(maxiter=25), estimator=Estimator()) diff --git a/src/mqt/bench/benchmarks/wstate.py b/src/mqt/bench/benchmarks/wstate.py index ff2670b38..08822d4a3 100644 --- a/src/mqt/bench/benchmarks/wstate.py +++ b/src/mqt/bench/benchmarks/wstate.py @@ -4,7 +4,7 @@ from qiskit import QuantumCircuit, QuantumRegister -def create_circuit(num_qubits: int): +def create_circuit(num_qubits: int) -> QuantumCircuit: """Returns a quantum circuit implementing the W state. Keyword arguments: @@ -14,7 +14,7 @@ def create_circuit(num_qubits: int): q = QuantumRegister(num_qubits, "q") qc = QuantumCircuit(q, name="wstate") - def f_gate(qc: QuantumCircuit, q: QuantumRegister, i: int, j: int, n: int, k: int): + def f_gate(qc: QuantumCircuit, q: QuantumRegister, i: int, j: int, n: int, k: int) -> None: theta = np.arccos(np.sqrt(1 / (n - k + 1))) qc.ry(-theta, q[j]) qc.cz(q[i], q[j]) diff --git a/src/mqt/bench/evaluation/__init__.py b/src/mqt/bench/evaluation/__init__.py index de20c3751..e013429c4 100644 --- a/src/mqt/bench/evaluation/__init__.py +++ b/src/mqt/bench/evaluation/__init__.py @@ -5,3 +5,11 @@ count_occurrences, count_qubit_numbers_per_compiler, ) + +__all__ = [ + "evaluate_qasm_file", + "create_statistics", + "EvaluationResult", + "count_occurrences", + "count_qubit_numbers_per_compiler", +] diff --git a/src/mqt/bench/py.typed b/src/mqt/bench/py.typed new file mode 100644 index 000000000..5f3ea3d91 --- /dev/null +++ b/src/mqt/bench/py.typed @@ -0,0 +1,2 @@ +# Instruct type checkers to look for inline type annotations in this package. +# See PEP 561. diff --git a/src/mqt/bench/qiskit_helper.py b/src/mqt/bench/qiskit_helper.py index c983182e6..35a31f4b2 100644 --- a/src/mqt/bench/qiskit_helper.py +++ b/src/mqt/bench/qiskit_helper.py @@ -1,16 +1,16 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal, overload -if TYPE_CHECKING: +if TYPE_CHECKING: # pragma: no cover from qiskit import QuantumCircuit from mqt.bench import utils from qiskit.compiler import transpile -def get_native_gates(gate_set_name: str): +def get_native_gates(gate_set_name: str) -> list[str]: if gate_set_name == "ionq": return get_ionq_native_gates() if gate_set_name == "oqc": @@ -22,30 +22,54 @@ def get_native_gates(gate_set_name: str): raise ValueError("Unknown gate set name: " + gate_set_name) -def get_ibm_native_gates(): +def get_ibm_native_gates() -> list[str]: return ["rz", "sx", "x", "cx", "measure"] -def get_rigetti_native_gates(): +def get_rigetti_native_gates() -> list[str]: return ["rx", "rz", "cz", "measure"] -def get_ionq_native_gates(): +def get_ionq_native_gates() -> list[str]: return ["rxx", "rz", "ry", "rx", "measure"] -def get_oqc_native_gates(): +def get_oqc_native_gates() -> list[str]: return ["rz", "sx", "x", "ecr", "measure"] +@overload def get_indep_level( qc: QuantumCircuit, - num_qubits: int, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> QuantumCircuit: + ... + + +@overload +def get_indep_level( + qc: QuantumCircuit, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + +def get_indep_level( + qc: QuantumCircuit, + num_qubits: int | None, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | QuantumCircuit: """Handles the creation of the benchmark on the target-independent level. Keyword arguments: @@ -78,16 +102,44 @@ def get_indep_level( ) +@overload +def get_native_gates_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + opt_level: int, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> QuantumCircuit: + ... + + +@overload +def get_native_gates_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + opt_level: int, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + def get_native_gates_level( qc: QuantumCircuit, gate_set_name: str, - num_qubits: int, + num_qubits: int | None, opt_level: int, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | QuantumCircuit: """Handles the creation of the benchmark on the target-dependent native gates level. Keyword arguments: @@ -130,17 +182,47 @@ def get_native_gates_level( ) +@overload +def get_mapped_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + device_name: str, + opt_level: int, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> QuantumCircuit: + ... + + +@overload +def get_mapped_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + device_name: str, + opt_level: int, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + def get_mapped_level( qc: QuantumCircuit, gate_set_name: str, - num_qubits: int, + num_qubits: int | None, device_name: str, opt_level: int, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | QuantumCircuit: """Handles the creation of the benchmark on the target-dependent mapped level. Keyword arguments: diff --git a/src/mqt/bench/tket_helper.py b/src/mqt/bench/tket_helper.py index 239f5c9b3..d2e54846e 100644 --- a/src/mqt/bench/tket_helper.py +++ b/src/mqt/bench/tket_helper.py @@ -1,23 +1,29 @@ from __future__ import annotations from pathlib import Path +from typing import TYPE_CHECKING, Literal, overload from mqt.bench import utils -from pytket import OpType, architecture, circuit +from pytket import OpType +from pytket.architecture import Architecture # type: ignore[attr-defined] from pytket.extensions.qiskit import qiskit_to_tk -from pytket.passes import ( +from pytket.passes import ( # type: ignore[attr-defined] CXMappingPass, FullPeepholeOptimise, PlacementPass, RoutingPass, auto_rebase_pass, ) -from pytket.placement import GraphPlacement, LinePlacement +from pytket.placement import GraphPlacement, LinePlacement # type: ignore[attr-defined] from pytket.qasm import circuit_to_qasm_str from qiskit import QuantumCircuit, transpile +if TYPE_CHECKING: # pragma: no cover + from pytket.circuit import Circuit + from pytket.passes import RebaseCustom # type: ignore[attr-defined] -def get_rebase(gate_set_name: str, get_gatenames: bool = False): + +def get_rebase(gate_set_name: str, get_gatenames: bool = False) -> RebaseCustom: if gate_set_name == "ionq": return get_ionq_rebase(get_gatenames) if gate_set_name == "oqc": @@ -29,38 +35,62 @@ def get_rebase(gate_set_name: str, get_gatenames: bool = False): raise ValueError("Unknown gate set name: " + gate_set_name) -def get_ionq_rebase(get_gatenames: bool = False): +def get_ionq_rebase(get_gatenames: bool = False) -> RebaseCustom: if get_gatenames: return ["rz", "ry", "rx", "rxx", "measure"] return auto_rebase_pass({OpType.Rz, OpType.Ry, OpType.Rx, OpType.XXPhase, OpType.Measure}) -def get_oqc_rebase(get_gatenames: bool = False): +def get_oqc_rebase(get_gatenames: bool = False) -> RebaseCustom: if get_gatenames: return ["rz", "sx", "x", "ecr", "measure"] return auto_rebase_pass({OpType.Rz, OpType.SX, OpType.X, OpType.ECR, OpType.Measure}) -def get_rigetti_rebase(get_gatenames: bool = False): +def get_rigetti_rebase(get_gatenames: bool = False) -> RebaseCustom: if get_gatenames: return ["rz", "rx", "cz", "measure"] return auto_rebase_pass({OpType.Rz, OpType.Rx, OpType.CZ, OpType.Measure}) -def get_ibm_rebase(get_gatenames: bool = False): +def get_ibm_rebase(get_gatenames: bool = False) -> RebaseCustom: if get_gatenames: return ["rz", "sx", "x", "cx", "measure"] return auto_rebase_pass({OpType.Rz, OpType.SX, OpType.X, OpType.CX, OpType.Measure}) +@overload +def get_indep_level( + qc: QuantumCircuit, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> Circuit: + ... + + +@overload def get_indep_level( qc: QuantumCircuit, - num_qubits: int, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + +def get_indep_level( + qc: QuantumCircuit, + num_qubits: int | None, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | Circuit: """Handles the creation of the benchmark on the target-independent level. Keyword arguments: @@ -102,15 +132,41 @@ def get_indep_level( ) +@overload def get_native_gates_level( - qc: circuit, + qc: QuantumCircuit, gate_set_name: str, - num_qubits: int, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> Circuit: + ... + + +@overload +def get_native_gates_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + +def get_native_gates_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | Circuit: """Handles the creation of the benchmark on the target-dependent native gates level. Keyword arguments: @@ -165,17 +221,47 @@ def get_native_gates_level( ) +@overload +def get_mapped_level( + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + device_name: str, + lineplacement: bool, + file_precheck: bool, + return_qc: Literal[True], + target_directory: str = "./", + target_filename: str = "", +) -> Circuit: + ... + + +@overload def get_mapped_level( - qc: circuit, + qc: QuantumCircuit, + gate_set_name: str, + num_qubits: int | None, + device_name: str, + lineplacement: bool, + file_precheck: bool, + return_qc: Literal[False], + target_directory: str = "./", + target_filename: str = "", +) -> bool: + ... + + +def get_mapped_level( + qc: QuantumCircuit, gate_set_name: str, - num_qubits: int, + num_qubits: int | None, device_name: str, lineplacement: bool, file_precheck: bool, return_qc: bool = False, target_directory: str = "./", target_filename: str = "", -): +) -> bool | Circuit: """Handles the creation of the benchmark on the target-dependent mapped level. Keyword arguments: @@ -222,7 +308,7 @@ def get_mapped_level( native_gatenames = get_rebase(gate_set_name, True) native_gate_set_rebase = get_rebase(gate_set_name) - arch = architecture.Architecture(cmap) + arch = Architecture(cmap) native_gate_set_rebase.apply(qc_tket) FullPeepholeOptimise(target_2qb_gate=OpType.TK2).apply(qc_tket) diff --git a/src/mqt/bench/utils.py b/src/mqt/bench/utils.py index 3804b5b3f..fcd378dc8 100644 --- a/src/mqt/bench/utils.py +++ b/src/mqt/bench/utils.py @@ -4,7 +4,7 @@ import sys from datetime import date from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast if TYPE_CHECKING: # pragma: no cover from types import ModuleType @@ -15,9 +15,9 @@ import numpy as np from pytket import __version__ as __tket_version__ from qiskit import QuantumCircuit, __qiskit_version__ -from qiskit.algorithms import EstimationProblem from qiskit.providers.fake_provider import FakeMontreal, FakeWashington from qiskit.transpiler.passes import RemoveBarriers +from qiskit_optimization.applications import Maxcut if TYPE_CHECKING or sys.version_info >= (3, 10, 0): # pragma: no cover from importlib import metadata, resources @@ -25,8 +25,9 @@ import importlib_metadata as metadata import importlib_resources as resources -if TYPE_CHECKING: +if TYPE_CHECKING: # pragma: no cover from qiskit.circuit import QuantumRegister, Qubit + from qiskit_optimization import QuadraticProgram from dataclasses import dataclass @@ -43,7 +44,7 @@ class SupermarqFeatures: qasm_path = str(resources.files("mqt.benchviewer") / "static/files/qasm_output/") -def get_supported_benchmarks(): +def get_supported_benchmarks() -> list[str]: return [ "ae", "dj", @@ -76,94 +77,45 @@ def get_supported_benchmarks(): ] -def get_supported_levels(): +def get_supported_levels() -> list[str | int]: return ["alg", "indep", "nativegates", "mapped", 0, 1, 2, 3] -def get_supported_compilers(): +def get_supported_compilers() -> list[str]: return ["qiskit", "tket"] -def get_supported_gatesets(): +def get_supported_gatesets() -> list[str]: return ["ibm", "rigetti", "ionq", "oqc"] -def get_supported_devices(): +def get_supported_devices() -> list[str]: return ["ibm_washington", "ibm_montreal", "rigetti_aspen_m2", "ionq11", "oqc_lucy"] -def get_default_qasm_output_path(): +def get_default_qasm_output_path() -> str: """Returns the path where all .qasm files are stored.""" return str(resources.files("mqt.benchviewer") / "static" / "files" / "qasm_output") -def get_zip_file_path(): +def get_zip_file_path() -> str: """Returns the path where the zip file is stored.""" return str(resources.files("mqt.benchviewer") / "static/files/MQTBench_all.zip") -def get_examplary_max_cut_qp(n_nodes: int, degree: int = 2): +def get_examplary_max_cut_qp(n_nodes: int, degree: int = 2) -> QuadraticProgram: """Returns a quadratic problem formulation of a max cut problem of a random graph. Keyword arguments: n_nodes -- number of graph nodes (and also number of qubits) degree -- edges per node """ - try: - from qiskit_optimization.applications import Maxcut - except Exception: - print("Please install qiskit_optimization.") - return None graph = nx.random_regular_graph(d=degree, n=n_nodes, seed=111) maxcut = Maxcut(graph) return maxcut.to_quadratic_program() -class BernoulliA(QuantumCircuit): - """A circuit representing the Bernoulli A operator.""" - - def __init__(self, probability): - super().__init__(1) # circuit on 1 qubit - - theta_p = 2 * np.arcsin(np.sqrt(probability)) - self.ry(theta_p, 0) - - -class BernoulliQ(QuantumCircuit): - """A circuit representing the Bernoulli Q operator.""" - - def __init__(self, probability): - super().__init__(1) # circuit on 1 qubit - - self._theta_p = 2 * np.arcsin(np.sqrt(probability)) - self.ry(2 * self._theta_p, 0) - - def __eq__(self, other): - return isinstance(other, BernoulliQ) and self._theta_p == other._theta_p - - def power(self, power: float, _matrix_power: bool = True): - # implement the efficient power of Q - q_k = QuantumCircuit(1) - q_k.ry(2 * power * self._theta_p, 0) - return q_k - - -def get_estimation_problem(): - """Returns a estimation problem instance for a fixed p value.""" - - p = 0.2 - - a = BernoulliA(p) - q = BernoulliQ(p) - - return EstimationProblem( - state_preparation=a, # A operator - grover_operator=q, # Q operator - objective_qubits=[0], # the "good" state Psi1 is identified as measuring |1> in qubit 0 - ) - - -def get_rigetti_aspen_m2_map(): +def get_rigetti_aspen_m2_map() -> list[list[int]]: """Returns a coupling map of Rigetti Aspen M2 chip.""" c_map_rigetti = [] for j in range(5): @@ -199,7 +151,7 @@ def get_rigetti_aspen_m2_map(): return c_map_rigetti -def get_ionq11_c_map(): +def get_ionq11_c_map() -> list[list[int]]: ionq11_c_map = [] for i in range(0, 11): for j in range(0, 11): @@ -208,7 +160,7 @@ def get_ionq11_c_map(): return ionq11_c_map -def get_openqasm_gates(): +def get_openqasm_gates() -> list[str]: """Returns a list of all quantum gates within the openQASM 2.0 standard header.""" # according to https://github.com/Qiskit/qiskit-terra/blob/main/qiskit/qasm/libs/qelib1.inc return [ @@ -260,11 +212,11 @@ def get_openqasm_gates(): def save_as_qasm( qc_str: str, filename: str, - gate_set: list = None, + gate_set: list[str] | None = None, mapped: bool = False, - c_map=None, + c_map: list[list[int]] | None = None, target_directory: str = "", -): +) -> bool: """Saves a quantum circuit as a qasm file. Keyword arguments: @@ -308,7 +260,7 @@ def save_as_qasm( return True -def get_cmap_oqc_lucy(): +def get_cmap_oqc_lucy() -> list[list[int]]: """Returns the coupling map of the OQC Lucy quantum computer.""" # source: https://github.com/aws/amazon-braket-examples/blob/main/examples/braket_features/Verbatim_Compilation.ipynb @@ -316,21 +268,22 @@ def get_cmap_oqc_lucy(): return [[0, 1], [0, 7], [1, 2], [2, 3], [7, 6], [6, 5], [4, 3], [4, 5]] -def get_cmap_from_devicename(device: str): +def get_cmap_from_devicename(device: str) -> list[list[int]]: if device == "ibm_washington": - return FakeWashington().configuration().coupling_map + return cast(list[list[int]], FakeWashington().configuration().coupling_map) if device == "ibm_montreal": - return FakeMontreal().configuration().coupling_map + return cast(list[list[int]], FakeMontreal().configuration().coupling_map) if device == "rigetti_aspen_m2": return get_rigetti_aspen_m2_map() if device == "oqc_lucy": return get_cmap_oqc_lucy() if device == "ionq11": return get_ionq11_c_map() - return False + error_msg = f"Device {device} is not supported." + raise ValueError(error_msg) -def postprocess_single_oqc_file(filename: str): +def postprocess_single_oqc_file(filename: str) -> None: with Path(filename).open() as f: lines = f.readlines() with Path(filename).open("w") as f: @@ -341,7 +294,7 @@ def postprocess_single_oqc_file(filename: str): f.write("opaque ecr q0,q1;\n") -def create_zip_file(): +def create_zip_file() -> int: return subprocess.call(f"zip -rj {get_zip_file_path()} {get_default_qasm_output_path()}", shell=True) @@ -351,7 +304,7 @@ def calc_qubit_index(qargs: list[Qubit], qregs: list[QuantumRegister], index: in if qargs[index] not in reg: offset += reg.size else: - qubit_index = offset + reg.index(qargs[index]) + qubit_index: int = offset + reg.index(qargs[index]) return qubit_index error_msg = f"Global qubit index for local qubit {index} index not found." raise ValueError(error_msg) @@ -359,7 +312,7 @@ def calc_qubit_index(qargs: list[Qubit], qregs: list[QuantumRegister], index: in def calc_supermarq_features( qc: QuantumCircuit, -) -> tuple[float, float, float, float, float]: +) -> SupermarqFeatures: qc.remove_final_measurements(inplace=True) qc = RemoveBarriers()(qc) connectivity_collection: list[list[int]] = [] @@ -415,7 +368,7 @@ def calc_supermarq_features( ) -def get_module_for_benchmark(benchmark_name) -> ModuleType: +def get_module_for_benchmark(benchmark_name: str) -> ModuleType: if benchmark_name in ["portfolioqaoa", "portfoliovqe", "pricingcall", "pricingput"]: return import_module("mqt.bench.benchmarks.qiskit_application_finance." + benchmark_name) if benchmark_name == "qgan": diff --git a/src/mqt/benchviewer/__init__.py b/src/mqt/benchviewer/__init__.py index d03f8d988..4c3784f09 100644 --- a/src/mqt/benchviewer/__init__.py +++ b/src/mqt/benchviewer/__init__.py @@ -1,2 +1,8 @@ +from mqt.benchviewer.main import start_server + + from mqt.benchviewer.main import Server -from mqt.benchviewer.backend import Backend +from mqt.benchviewer.backend import Backend, BenchmarkConfiguration + + +__all__ = ["start_server", "Server", "Backend", "BenchmarkConfiguration"] diff --git a/src/mqt/benchviewer/backend.py b/src/mqt/benchviewer/backend.py index 9e04baec4..66d24ae3c 100644 --- a/src/mqt/benchviewer/backend.py +++ b/src/mqt/benchviewer/backend.py @@ -4,7 +4,9 @@ import os import re import sys +from dataclasses import dataclass from pathlib import Path +from typing import TYPE_CHECKING, cast from zipfile import ZIP_DEFLATED, ZipFile import pandas as pd @@ -12,19 +14,50 @@ from packaging import version from tqdm import tqdm -if sys.version_info < (3, 10, 0): - import importlib_metadata as metadata -else: +if TYPE_CHECKING or sys.version_info >= (3, 10, 0): # pragma: no cover from importlib import metadata +else: + import importlib_metadata as metadata -from typing import TYPE_CHECKING -if TYPE_CHECKING: +if TYPE_CHECKING: # pragma: no cover from collections.abc import Iterable +@dataclass +class BenchmarkConfiguration: + min_qubits: int + max_qubits: int + indices_benchmarks: list[int] + indep_qiskit_compiler: bool + indep_tket_compiler: bool + nativegates_qiskit_compiler: bool + nativegates_tket_compiler: bool + mapped_qiskit_compiler: bool + mapped_tket_compiler: bool + native_qiskit_opt_lvls: list[int] | None = None + native_gatesets: list[str] | None = None + mapped_qiskit_opt_lvls: list[int] | None = None + mapped_tket_placements: list[str] | None = None + mapped_devices: list[str] | None = None + + +@dataclass +class ParsedBenchmarkName: + benchmark: str + num_qubits: int + indep_flag: bool + nativegates_flag: bool + mapped_flag: bool + compiler: str | int + compiler_settings: str | int | None + gate_set: str | None + target_device: str | None + filename: str + + class Backend: - def __init__(self): + def __init__(self) -> None: self.benchmarks = [ {"name": "Amplitude Estimation (AE)", "id": "1", "filename": "ae"}, {"name": "Deutsch-Jozsa", "id": "2", "filename": "dj"}, @@ -90,84 +123,61 @@ def __init__(self): self.database: pd.DataFrame | None = None self.mqtbench_all_zip: ZipFile | None = None - def filter_database(self, filter_criteria: tuple): # noqa: PLR0912 + def filter_database(self, benchmark_config: BenchmarkConfiguration) -> list[str]: # noqa: PLR0912 """Filters the database according to the filter criteria. Keyword arguments: filterCriteria -- list of all filter criteria database -- database containing all available benchmarks + Return values: db_filtered["path"].to_list() -- list of all file paths of the selected benchmark files """ - if len(self.database) == 0: - return [] - - colnames = [ - "benchmark", - "num_qubits", - "indep_flag", - "nativegates_flag", - "mapped_flag", - "compiler", - "compiler_settings", - "gate_set", - "target_device", - "path", - ] + colnames = list(ParsedBenchmarkName.__annotations__.keys()) db_filtered = pd.DataFrame(columns=colnames) db_filtered["indep_flag"] = db_filtered["indep_flag"].astype(bool) db_filtered["nativegates_flag"] = db_filtered["nativegates_flag"].astype(bool) db_filtered["mapped_flag"] = db_filtered["mapped_flag"].astype(bool) - - ( - (min_qubits, max_qubits), - indices_benchmarks, - (indep_qiskit_compiler, indep_tket_compiler), - ( - (nativegates_qiskit_compiler, nativegates_tket_compiler), - native_qiskit_opt_lvls, - native_gatesets, - ), - ( - (mapped_qiskit_compiler, mapped_tket_compiler), - (mapped_qiskit_opt_lvls, mapped_tket_placements), - mapped_devices, - ), - ) = filter_criteria + if self.database is None or self.database.empty: + return [] selected_scalable_benchmarks = [] selected_nonscalable_benchmarks = [] - for identifier in indices_benchmarks: - if 0 < int(identifier) <= len(self.benchmarks): - name = self.benchmarks[int(identifier) - 1]["filename"] + for identifier in benchmark_config.indices_benchmarks: + if 0 < identifier <= len(self.benchmarks): + name = self.benchmarks[identifier - 1]["filename"] selected_scalable_benchmarks.append(name) - elif 0 < int(identifier) <= len(self.benchmarks) + len(self.nonscalable_benchmarks): - name = self.nonscalable_benchmarks[int(identifier) - 1 - len(self.benchmarks)]["filename"] + elif 0 < identifier <= len(self.benchmarks) + len(self.nonscalable_benchmarks): + name = self.nonscalable_benchmarks[identifier - 1 - len(self.benchmarks)]["filename"] selected_nonscalable_benchmarks.append(name) db_tmp = self.database.loc[ ( - (self.database["num_qubits"] >= min_qubits) - & (self.database["num_qubits"] <= max_qubits) + (self.database["num_qubits"] >= benchmark_config.min_qubits) + & (self.database["num_qubits"] <= benchmark_config.max_qubits) & (self.database["benchmark"].isin(selected_scalable_benchmarks)) ) | (self.database["benchmark"].isin(selected_nonscalable_benchmarks)) ] - if indep_qiskit_compiler: + if benchmark_config.indep_qiskit_compiler: db_tmp1 = db_tmp.loc[(db_tmp["indep_flag"]) & (db_tmp["compiler"] == "qiskit")] db_filtered = pd.concat([db_filtered, db_tmp1]) - if indep_tket_compiler: + if benchmark_config.indep_tket_compiler: db_tmp2 = db_tmp.loc[(db_tmp["indep_flag"]) & (db_tmp["compiler"] == "tket")] db_filtered = pd.concat([db_filtered, db_tmp2]) - if nativegates_qiskit_compiler: - for gate_set in native_gatesets: - for opt_lvl in native_qiskit_opt_lvls: + if ( + benchmark_config.nativegates_qiskit_compiler + and benchmark_config.native_gatesets + and benchmark_config.native_qiskit_opt_lvls + ): + for gate_set in benchmark_config.native_gatesets: + for opt_lvl in benchmark_config.native_qiskit_opt_lvls: db_tmp3 = db_tmp.loc[ (db_tmp["nativegates_flag"]) & (db_tmp["gate_set"] == gate_set) @@ -176,16 +186,20 @@ def filter_database(self, filter_criteria: tuple): # noqa: PLR0912 ] db_filtered = pd.concat([db_filtered, db_tmp3]) - if nativegates_tket_compiler: - for gate_set in native_gatesets: + if benchmark_config.nativegates_tket_compiler and benchmark_config.native_gatesets: + for gate_set in benchmark_config.native_gatesets: db_tmp4 = db_tmp.loc[ (db_tmp["nativegates_flag"]) & (db_tmp["gate_set"] == gate_set) & (db_tmp["compiler"] == "tket") ] db_filtered = pd.concat([db_filtered, db_tmp4]) - if mapped_qiskit_compiler: - for opt_lvl in mapped_qiskit_opt_lvls: - for device in mapped_devices: + if ( + benchmark_config.mapped_qiskit_compiler + and benchmark_config.mapped_qiskit_opt_lvls + and benchmark_config.mapped_devices + ): + for opt_lvl in benchmark_config.mapped_qiskit_opt_lvls: + for device in benchmark_config.mapped_devices: db_tmp5 = db_tmp.loc[ (db_tmp["mapped_flag"]) & (db_tmp["target_device"] == device) @@ -194,9 +208,13 @@ def filter_database(self, filter_criteria: tuple): # noqa: PLR0912 ] db_filtered = pd.concat([db_filtered, db_tmp5]) - if mapped_tket_compiler: - for placement in mapped_tket_placements: - for device in mapped_devices: + if ( + benchmark_config.mapped_tket_compiler + and benchmark_config.mapped_tket_placements + and benchmark_config.mapped_devices + ): + for placement in benchmark_config.mapped_tket_placements: + for device in benchmark_config.mapped_devices: db_tmp6 = db_tmp.loc[ (db_tmp["mapped_flag"]) & (db_tmp["target_device"] == device) @@ -205,44 +223,7 @@ def filter_database(self, filter_criteria: tuple): # noqa: PLR0912 ] db_filtered = pd.concat([db_filtered, db_tmp6]) - return db_filtered["path"].to_list() - - class NoSeekBytesIO: - def __init__(self, fp: io.BytesIO): - self.fp = fp - self.deleted_offset = 0 - - def write(self, b): - return self.fp.write(b) - - def tell(self): - return self.deleted_offset + self.fp.tell() - - def hidden_tell(self): - return self.fp.tell() - - def seekable(self): - return False - - def hidden_seek(self, offset, start_point=io.SEEK_SET): - return self.fp.seek(offset, start_point) - - def truncate_and_remember_offset(self, size): - self.deleted_offset += self.fp.tell() - self.fp.seek(0) - return self.fp.truncate(size) - - def get_value(self): - return self.fp.getvalue() - - def close(self): - return self.fp.close() - - def read(self): - return self.fp.read() - - def flush(self): - return self.fp.flush() + return cast(list[str], db_filtered["filename"].to_list()) def generate_zip_ephemeral_chunks( self, @@ -256,14 +237,15 @@ def generate_zip_ephemeral_chunks( Return values: Generator of bytes to send to the browser """ - fileobj = self.NoSeekBytesIO(io.BytesIO()) + fileobj = NoSeekBytesIO(io.BytesIO()) - with ZipFile(fileobj, mode="w") as zf: + with ZipFile(fileobj, mode="w") as zf: # type: ignore[arg-type] for individual_file in filenames: individual_file_as_path = Path(individual_file) + assert self.mqtbench_all_zip is not None zf.writestr( individual_file_as_path.name, - data=self.mqtbench_all_zip.read(individual_file_as_path.name), + data=self.mqtbench_all_zip.read(individual_file), compress_type=ZIP_DEFLATED, compresslevel=3, ) @@ -275,7 +257,7 @@ def generate_zip_ephemeral_chunks( yield fileobj.read() fileobj.close() - def get_selected_file_paths(self, prepared_data: tuple): + def get_selected_file_paths(self, prepared_data: BenchmarkConfiguration) -> list[str]: """Extracts all file paths according to the prepared user's filter criteria. Keyword arguments: @@ -284,20 +266,15 @@ def get_selected_file_paths(self, prepared_data: tuple): Return values: file_paths -- list of filter criteria for each selected benchmark """ + return self.filter_database(prepared_data) - if prepared_data: - return self.filter_database(prepared_data) - return False - - def init_database( - self, - ): - """Generates the database and saves it into a class variable.""" + def init_database(self) -> bool: + """Generates the database and saves it into a global variable.""" assert self.mqtbench_all_zip is not None print("Initiating database...") - self.database = self.create_database(self.mqtbench_all_zip) + self.database = create_database(self.mqtbench_all_zip) print(f"... done: {len(self.database)} benchmarks.") if not self.database.empty: @@ -306,11 +283,11 @@ def init_database( print("Database initialization failed.") return False - def prepare_form_input(self, form_data: dict): # noqa: PLR0912, PLR0915 + def prepare_form_input(self, form_data: dict[str, str]) -> BenchmarkConfiguration: """Formats the formData extracted from the user's inputs.""" - - min_qubits = -1 - max_qubits = -1 + min_qubits = 2 + max_qubits = 130 + indices_benchmarks = [] indep_qiskit_compiler = False indep_tket_compiler = False nativegates_qiskit_compiler = False @@ -323,154 +300,64 @@ def prepare_form_input(self, form_data: dict): # noqa: PLR0912, PLR0915 mapped_tket_placements = [] mapped_devices = [] - pat = re.compile(r"_\d+") - num_benchmarks = [] for k, v in form_data.items(): - m = pat.search(k) - if m: - num = m.group()[1:] - num_benchmarks.append(num) - - if "minQubits" in k: - min_qubits = v - if min_qubits == "": - min_qubits = 2 - - if "maxQubits" in k: - max_qubits = v - if max_qubits == "": - max_qubits = 130 - if "indep_qiskit_compiler" in k: - indep_qiskit_compiler = True - if "indep_tket_compiler" in k: - indep_tket_compiler = True - - if "nativegates_qiskit_compiler" in k: - nativegates_qiskit_compiler = True - if "nativegates_tket_compiler" in k: - nativegates_tket_compiler = True - if "nativegates_qiskit_compiler_opt0" in k: - native_qiskit_opt_lvls.append(0) - if "nativegates_qiskit_compiler_opt1" in k: - native_qiskit_opt_lvls.append(1) - if "nativegates_qiskit_compiler_opt2" in k: - native_qiskit_opt_lvls.append(2) - if "nativegates_qiskit_compiler_opt3" in k: - native_qiskit_opt_lvls.append(3) - - if "nativegates_ibm" in k: - native_gatesets.append("ibm") - if "nativegates_rigetti" in k: - native_gatesets.append("rigetti") - if "nativegates_oqc" in k: - native_gatesets.append("oqc") - if "nativegates_ionq" in k: - native_gatesets.append("ionq") - - if "mapped_qiskit_compiler" in k: - mapped_qiskit_compiler = True - if "mapped_tket_compiler" in k: - mapped_tket_compiler = True - if "mapped_qiskit_compiler_opt0" in k: - mapped_qiskit_opt_lvls.append(0) - if "mapped_qiskit_compiler_opt1" in k: - mapped_qiskit_opt_lvls.append(1) - if "mapped_qiskit_compiler_opt2" in k: - mapped_qiskit_opt_lvls.append(2) - if "mapped_qiskit_compiler_opt3" in k: - mapped_qiskit_opt_lvls.append(3) - - if "mapped_tket_compiler_graph" in k: - mapped_tket_placements.append("graph") - if "mapped_tket_compiler_line" in k: - mapped_tket_placements.append("line") - - if "device_ibm_montreal" in k: - mapped_devices.append("ibm_montreal") - if "device_ibm_washington" in k: - mapped_devices.append("ibm_washington") - if "device_rigetti_aspen" in k: - mapped_devices.append("rigetti_aspen") - if "device_oqc_lucy" in k: - mapped_devices.append("oqc_lucy") - if "device_ionq_ionq11" in k: - mapped_devices.append("ionq11") - - return ( - (int(min_qubits), int(max_qubits)), - num_benchmarks, - (indep_qiskit_compiler, indep_tket_compiler), - ( - (nativegates_qiskit_compiler, nativegates_tket_compiler), - native_qiskit_opt_lvls, - native_gatesets, - ), - ( - (mapped_qiskit_compiler, mapped_tket_compiler), - (mapped_qiskit_opt_lvls, mapped_tket_placements), - mapped_devices, - ), - ) - - def create_database(self, zip_file: ZipFile): - """Creates the database based on the provided directories. - - Keyword arguments: - qasm_path -- zip containing all .qasm files - - Return values: - database -- database containing all available benchmarks - """ - rows_list = [] - - for filename in zip_file.namelist(): - if filename.endswith(".qasm"): - parsed_data = parse_data(filename) - rows_list.append(parsed_data) - - colnames = [ - "benchmark", - "num_qubits", - "indep_flag", - "nativegates_flag", - "mapped_flag", - "compiler", - "compiler_settings", - "gate_set", - "target_device", - "path", - ] - - database = pd.DataFrame(rows_list, columns=colnames) - database["num_qubits"] = database["num_qubits"].astype(int) - database["benchmark"] = database["benchmark"].astype(str) - return database - - def handle_github_api_request(self, repo_url: str) -> requests.Response: - # If the environment variable GITHUB_TOKEN is set, use it to authenticate to the GitHub API - # to increase the rate limit from 60 to 5000 requests per hour per IP address. - headers = None - if "GITHUB_TOKEN" in os.environ: - headers = {"Authorization": f"token {os.environ['GITHUB_TOKEN']}"} - - response = requests.get(f"https://api.github.com/repos/cda-tum/mqtbench/{repo_url}", headers=headers) - success_code = 200 - if response.status_code == success_code: - return response - - msg = ( - f"Request to GitHub API failed with status code {response.status_code}!\n" - f"One reasons could be that the limit of 60 API calls per hour and IP address is exceeded.\n" - f"If you want to increase the limit, set the environment variable GITHUB_TOKEN to a GitHub personal access token.\n" - f"See https://docs.github.com/en/github/authenticating-to-github/creating-a-personal-access-token for more information." + if "select" in k: + found_benchmark_id = parse_benchmark_id_from_form_key(k) + if found_benchmark_id: + indices_benchmarks.append(found_benchmark_id) + min_qubits = int(v) if "minQubits" in k and v != "" else min_qubits + max_qubits = int(v) if "maxQubits" in k and v != "" else max_qubits + + indep_qiskit_compiler = "indep_qiskit_compiler" in k or indep_qiskit_compiler + indep_tket_compiler = "indep_tket_compiler" in k or indep_tket_compiler + + nativegates_qiskit_compiler = "nativegates_qiskit_compiler" in k or nativegates_qiskit_compiler + nativegates_tket_compiler = "nativegates_tket_compiler" in k or nativegates_tket_compiler + native_qiskit_opt_lvls.append(0) if "nativegates_qiskit_compiler_opt0" in k else None + native_qiskit_opt_lvls.append(1) if "nativegates_qiskit_compiler_opt1" in k else None + native_qiskit_opt_lvls.append(2) if "nativegates_qiskit_compiler_opt2" in k else None + native_qiskit_opt_lvls.append(3) if "nativegates_qiskit_compiler_opt3" in k else None + native_gatesets.append("ibm") if "nativegates_ibm" in k else None + native_gatesets.append("rigetti") if "nativegates_rigetti" in k else None + native_gatesets.append("oqc") if "nativegates_oqc" in k else None + native_gatesets.append("ionq") if "nativegates_ionq" in k else None + + mapped_qiskit_compiler = "mapped_qiskit_compiler" in k or mapped_qiskit_compiler + mapped_tket_compiler = "mapped_tket_compiler" in k or mapped_tket_compiler + mapped_qiskit_opt_lvls.append(0) if "mapped_qiskit_compiler_opt0" in k else None + mapped_qiskit_opt_lvls.append(1) if "mapped_qiskit_compiler_opt1" in k else None + mapped_qiskit_opt_lvls.append(2) if "mapped_qiskit_compiler_opt2" in k else None + mapped_qiskit_opt_lvls.append(3) if "mapped_qiskit_compiler_opt3" in k else None + mapped_tket_placements.append("graph") if "mapped_tket_compiler_graph" in k else None + mapped_tket_placements.append("line") if "mapped_tket_compiler_line" in k else None + mapped_devices.append("ibm_montreal") if "device_ibm_montreal" in k else None + mapped_devices.append("ibm_washington") if "device_ibm_washington" in k else None + mapped_devices.append("rigetti_aspen") if "device_rigetti_aspen" in k else None + mapped_devices.append("oqc_lucy") if "device_oqc_lucy" in k else None + mapped_devices.append("ionq11") if "device_ionq_ionq11" in k else None + + return BenchmarkConfiguration( + min_qubits=min_qubits, + max_qubits=max_qubits, + indices_benchmarks=indices_benchmarks, + indep_qiskit_compiler=indep_qiskit_compiler, + indep_tket_compiler=indep_tket_compiler, + nativegates_qiskit_compiler=nativegates_qiskit_compiler, + nativegates_tket_compiler=nativegates_tket_compiler, + native_qiskit_opt_lvls=native_qiskit_opt_lvls, + native_gatesets=native_gatesets, + mapped_qiskit_compiler=mapped_qiskit_compiler, + mapped_tket_compiler=mapped_tket_compiler, + mapped_qiskit_opt_lvls=mapped_qiskit_opt_lvls, + mapped_tket_placements=mapped_tket_placements, + mapped_devices=mapped_devices, ) - raise RuntimeError(msg) def read_mqtbench_all_zip( # noqa: PLR0912 self, + target_location: str, skip_question: bool = False, - target_location: str = None, - ): + ) -> bool: huge_zip_path = Path(target_location) / "MQTBench_all.zip" try: @@ -487,12 +374,12 @@ def read_mqtbench_all_zip( # noqa: PLR0912 version_found = False available_versions = [] - for elem in self.handle_github_api_request("tags").json(): + for elem in handle_github_api_request("tags").json(): available_versions.append(elem["name"]) for possible_version in available_versions: if version.parse(mqtbench_module_version) >= version.parse(possible_version): - response_json = self.handle_github_api_request(f"releases/tags/{possible_version}").json() + response_json = handle_github_api_request(f"releases/tags/{possible_version}").json() if "assets" in response_json: assets = response_json["assets"] elif "asset" in response_json: @@ -531,13 +418,14 @@ def read_mqtbench_all_zip( # noqa: PLR0912 self.mqtbench_all_zip = ZipFile(zip_bytes, mode="r") return True - def handle_downloading_benchmarks(self, target_location: str, download_url: str): + def handle_downloading_benchmarks(self, target_location: str, download_url: str) -> None: print("Start downloading benchmarks...") - r = requests.get(download_url) - # r = self.handle_github_api_request(download_url) - total_length = r.headers.get("content-length") - total_length = int(total_length) + r = requests.get(download_url, stream=True) + + content_length_response = r.headers.get("content-length") + assert content_length_response is not None + total_length = int(content_length_response) fname = target_location + "/MQTBench_all.zip" Path(target_location).mkdir(parents=True, exist_ok=True) @@ -553,10 +441,114 @@ def handle_downloading_benchmarks(self, target_location: str, download_url: str) bar.update(size) print(f"Download completed to {fname}. Server is starting now.") - return +def parse_data(filename: str) -> ParsedBenchmarkName: + """Extracts the necessary information from a given filename. + + Keyword arguments: + filename -- name of file + + Return values: + parsed_data -- parsed data extracted from filename + """ + benchmark = filename.split("_")[0].lower() + num_qubits = get_num_qubits(filename) + indep_flag = "indep" in filename + nativegates_flag = "nativegates" in filename + mapped_flag = "mapped" in filename + compiler, compiler_settings = get_compiler_and_settings(filename) + gate_set = get_gate_set(filename) if nativegates_flag or mapped_flag else None + target_device = get_target_device(filename) if mapped_flag else None + + return ParsedBenchmarkName( + benchmark=benchmark, + num_qubits=num_qubits, + indep_flag=indep_flag, + nativegates_flag=nativegates_flag, + mapped_flag=mapped_flag, + compiler=compiler, + compiler_settings=compiler_settings, + gate_set=gate_set, + target_device=target_device, + filename=filename, + ) + + +class NoSeekBytesIO: + def __init__(self, fp: io.BytesIO) -> None: + self.fp = fp + self.deleted_offset = 0 + + def write(self, b: bytes) -> int: + return self.fp.write(b) + + def tell(self) -> int: + return self.deleted_offset + self.fp.tell() + + def hidden_tell(self) -> int: + return self.fp.tell() + + def seekable(self) -> bool: + return False + + def hidden_seek(self, offset: int, start_point: int = io.SEEK_SET) -> int: + return self.fp.seek(offset, start_point) + + def truncate_and_remember_offset(self, size: int | None) -> int: + self.deleted_offset += self.fp.tell() + self.fp.seek(0) + return self.fp.truncate(size) + + def get_value(self) -> bytes: + return self.fp.getvalue() -def get_tket_settings(filename: str): + def close(self) -> None: + return self.fp.close() + + def read(self) -> bytes: + return self.fp.read() + + def flush(self) -> None: + return self.fp.flush() + + +def parse_benchmark_id_from_form_key(k: str) -> int | bool: + pat = re.compile(r"_\d+") + m = pat.search(k) + if m: + return int(m.group()[1:]) + return False + + +def get_opt_level(filename: str) -> int: + """Extracts the optimization level based on a filename. + Keyword arguments: + filename -- filename of a benchmark + Return values: + num -- optimization level + """ + + pat = re.compile(r"opt\d") + m = pat.search(filename) + num = m.group()[-1:] if m else -1 + return int(cast(str, num)) + + +def get_num_qubits(filename: str) -> int: + """Extracts the number of qubits based on a filename. + Keyword arguments: + filename -- filename of a benchmark + Return values: + num -- number of qubits + """ + + pat = re.compile(r"(\d+)\.") + m = pat.search(filename) + num = m.group()[0:-1] if m else -1 + return int(cast(str, num)) + + +def get_tket_settings(filename: str) -> str | None: if "mapped" not in filename: return None if "line" in filename: @@ -567,7 +559,7 @@ def get_tket_settings(filename: str): raise ValueError(error_msg) -def get_gate_set(filename: str): +def get_gate_set(filename: str) -> str: if "oqc" in filename: return "oqc" if "ionq" in filename: @@ -579,7 +571,7 @@ def get_gate_set(filename: str): raise ValueError("Unknown gate set: " + filename) -def get_target_device(filename: str): +def get_target_device(filename: str) -> str: if "ibm_washington" in filename: return "ibm_washington" if "ibm_montreal" in filename: @@ -593,7 +585,7 @@ def get_target_device(filename: str): raise ValueError("Unknown target device: " + filename) -def get_compiler_and_settings(filename: str): +def get_compiler_and_settings(filename: str) -> tuple[str, str | int | None]: if "qiskit" in filename: return "qiskit", get_opt_level(filename) if "tket" in filename: @@ -601,65 +593,66 @@ def get_compiler_and_settings(filename: str): raise ValueError("Unknown compiler: " + filename) -def parse_data(filename: str): - """Extracts the necessary information from a given filename. - - Keyword arguments: - filename -- name of file - - Return values: - parsed_data -- parsed data extracted from filename - """ - benchmark = filename.split("_")[0].lower() - num_qubits = get_num_qubits(filename) - indep_flag = "indep" in filename - nativegates_flag = "nativegates" in filename - mapped_flag = "mapped" in filename - compiler, compiler_settings = get_compiler_and_settings(filename) - gate_set = get_gate_set(filename) if nativegates_flag or mapped_flag else None - target_device = get_target_device(filename) if mapped_flag else None - - return [ - benchmark, - num_qubits, - indep_flag, - nativegates_flag, - mapped_flag, - compiler, - compiler_settings, - gate_set, - target_device, - filename, - ] - - -def get_opt_level(filename: str): - """Extracts the optimization level based on a filename. - +def create_database(zip_file: ZipFile) -> pd.DataFrame: + """Creates the database based on the provided directories. Keyword arguments: - filename -- filename of a benchmark - + qasm_path -- zip containing all .qasm files Return values: - num -- optimization level + database -- database containing all available benchmarks """ - - pat = re.compile(r"opt\d") - m = pat.search(filename) - num = m.group()[-1:] if m else -1 - return int(num) - - -def get_num_qubits(filename: str): - """Extracts the number of qubits based on a filename. - - Keyword arguments: - filename -- filename of a benchmark - - Return values: - num -- number of qubits - """ - - pat = re.compile(r"(\d+)\.") - m = pat.search(filename) - num = m.group()[0:-1] if m else -1 - return int(num) + rows_list = [] + + for filename in zip_file.namelist(): + if filename.endswith(".qasm"): + parsed_data = parse_data(filename) + rows_list.append(parsed_data) + + colnames = list(ParsedBenchmarkName.__annotations__.keys()) + + database = pd.DataFrame(rows_list, columns=colnames) + database["num_qubits"] = database["num_qubits"].astype(int) + database["benchmark"] = database["benchmark"].astype(str) + return database + + +def handle_downloading_benchmarks(target_location: str, download_url: str) -> None: + print("Start downloading benchmarks...") + + r = requests.get(download_url) + total_length = int(r.headers["content-length"]) + + fname = target_location + "/MQTBench_all.zip" + + Path(target_location).mkdir(parents=True, exist_ok=True) + with Path(fname).open("wb") as f, tqdm( + desc=fname, + total=total_length, + unit="iB", + unit_scale=True, + unit_divisor=1024, + ) as bar: + for data in r.iter_content(chunk_size=1024): + size = f.write(data) + bar.update(size) + print(f"Download completed to {fname}. Server is starting now.") + + +def handle_github_api_request(repo_url: str) -> requests.Response: + # If the environment variable GITHUB_TOKEN is set, use it to authenticate to the GitHub API + # to increase the rate limit from 60 to 5000 requests per hour per IP address. + headers = None + if "GITHUB_TOKEN" in os.environ: + headers = {"Authorization": f"token {os.environ['GITHUB_TOKEN']}"} + + response = requests.get(f"https://api.github.com/repos/cda-tum/mqtbench/{repo_url}", headers=headers) + success_code = 200 + if response.status_code == success_code: + return response + + msg = ( + f"Request to GitHub API failed with status code {response.status_code}!\n" + f"One reasons could be that the limit of 60 API calls per hour and IP address is exceeded.\n" + f"If you want to increase the limit, set the environment variable GITHUB_TOKEN to a GitHub personal access token.\n" + f"See https://docs.github.com/en/github/authenticating-to-github/creating-a-personal-access-token for more information." + ) + raise RuntimeError(msg) diff --git a/src/mqt/benchviewer/main.py b/src/mqt/benchviewer/main.py index bceb093e8..8bcc87a40 100644 --- a/src/mqt/benchviewer/main.py +++ b/src/mqt/benchviewer/main.py @@ -4,22 +4,26 @@ import os import sys from datetime import datetime +from typing import TYPE_CHECKING from flask import Flask, cli, jsonify, render_template, request, send_from_directory from mqt.benchviewer.backend import Backend -if sys.version_info < (3, 10, 0): # pragma: no cover +if TYPE_CHECKING or sys.version_info < (3, 10, 0): # pragma: no cover import importlib_resources as resources else: from importlib import resources +if TYPE_CHECKING: # pragma: no cover + from flask import Response + class Server: def __init__( self, + target_location: str, skip_question: bool = False, activate_logging: bool = False, - target_location: str = None, ): self.backend = Backend() @@ -28,13 +32,12 @@ def __init__( msg = "target_location is not writable. Please specify a different path." raise RuntimeError(msg) - res_zip = self.backend.read_mqtbench_all_zip(skip_question, self.target_location) + res_zip = self.backend.read_mqtbench_all_zip(self.target_location, skip_question) if not res_zip: msg = "Error while reading the MQTBench_all.zip file." raise RuntimeError(msg) - - res_db = self.backend.init_database() - if not res_db: + self.backend.init_database() + if self.backend.database is None: msg = "Error while initializing the database." raise RuntimeError(msg) @@ -47,15 +50,14 @@ def __init__( app = Flask(__name__, static_url_path="/mqtbench") -SERVER = None +SERVER: Server = None # type: ignore[assignment] PREFIX = "/mqtbench/" @app.route(f"{PREFIX}/", methods=["POST", "GET"]) @app.route(f"{PREFIX}/index", methods=["POST", "GET"]) -def index(): +def index() -> str: """Return the index.html file together with the benchmarks and nonscalable benchmarks.""" - return render_template( "index.html", benchmarks=SERVER.backend.benchmarks, @@ -64,7 +66,7 @@ def index(): @app.route(f"{PREFIX}/get_pre_gen", methods=["POST", "GET"]) -def download_pre_gen_zip(): +def download_pre_gen_zip() -> Response: filename = "MQTBench_all.zip" if SERVER.activate_logging: @@ -78,8 +80,8 @@ def download_pre_gen_zip(): app.logger.info("###### End ######") return send_from_directory( - directory=SERVER.target_location, - path=filename, + SERVER.target_location, + filename, as_attachment=True, mimetype="application/zip", download_name="MQTBench_all.zip", @@ -87,7 +89,7 @@ def download_pre_gen_zip(): @app.route(f"{PREFIX}/download", methods=["POST", "GET"]) -def download_data(): +def download_data() -> str | Response: """Triggers the downloading process of all benchmarks according to the user's input.""" if request.method == "POST": data = request.form @@ -106,7 +108,7 @@ def download_data(): app.logger.info("###### End ######") if file_paths: - return app.response_class( + return app.response_class( # type: ignore[no-any-return] SERVER.backend.generate_zip_ephemeral_chunks(file_paths), mimetype="application/zip", headers={"Content-Disposition": f'attachment; filename="MQTBench_{timestamp}.zip"'}, @@ -121,21 +123,21 @@ def download_data(): @app.route(f"{PREFIX}/legal") -def legal(): +def legal() -> str: """Return the legal.html file.""" return render_template("legal.html") @app.route(f"{PREFIX}/description") -def description(): +def description() -> str: """Return the description.html file in which the file formats are described.""" return render_template("description.html") @app.route(f"{PREFIX}/benchmark_description") -def benchmark_description(): +def benchmark_description() -> str: """Return the benchmark_description.html file together in which all benchmark algorithms are described in detail. """ @@ -144,28 +146,28 @@ def benchmark_description(): @app.route(f"{PREFIX}/get_num_benchmarks", methods=["POST"]) -def get_num_benchmarks(): +def get_num_benchmarks() -> Response: if request.method == "POST": data = request.form prepared_data = SERVER.backend.prepare_form_input(data) file_paths = SERVER.backend.get_selected_file_paths(prepared_data) - return jsonify({"num_selected": len(file_paths)}) - return jsonify({"num_selected": 0}) + return jsonify({"num_selected": len(file_paths)}) # type: ignore[no-any-return] + return jsonify({"num_selected": 0}) # type: ignore[no-any-return] def start_server( skip_question: bool = False, activate_logging: bool = False, - target_location: str = None, + target_location: str | None = None, debug_flag: bool = False, -): +) -> None: if not target_location: target_location = str(resources.files("mqt.benchviewer") / "static" / "files") Server( + target_location=target_location, skip_question=skip_question, activate_logging=activate_logging, - target_location=target_location, ) print( "Server is hosted at: http://127.0.0.1:5000" + PREFIX + ".", diff --git a/src/mqt/benchviewer/py.typed b/src/mqt/benchviewer/py.typed new file mode 100644 index 000000000..5f3ea3d91 --- /dev/null +++ b/src/mqt/benchviewer/py.typed @@ -0,0 +1,2 @@ +# Instruct type checkers to look for inline type annotations in this package. +# See PEP 561. diff --git a/tests/test_bench.py b/tests/test_bench.py index b6fb8a1b3..2fd0ac662 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -2,10 +2,17 @@ import json from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + import types import pytest from mqt.bench import ( BenchmarkGenerator, + CompilerSettings, + QiskitSettings, + TKETSettings, evaluation, get_benchmark, qiskit_helper, @@ -46,12 +53,29 @@ @pytest.fixture() -def output_path(): +def output_path() -> str: output_path = Path("./tests/test_output/") output_path.mkdir(parents=True, exist_ok=True) return str(output_path) +@pytest.fixture() +def sample_filenames() -> list[str]: + return [ + "ae_indep_qiskit_10.qasm", + "ghz_nativegates_rigetti_qiskit_opt3_54.qasm", + "ae_indep_tket_93.qasm", + "wstate_nativegates_rigetti_qiskit_opt0_79.qasm", + "ae_mapped_ibm_montreal_qiskit_opt1_9.qasm", + "ae_mapped_ibm_washington_qiskit_opt0_38.qasm", + "ae_mapped_oqc_lucy_qiskit_opt0_5.qasm", + "ae_mapped_rigetti_aspen_m2_qiskit_opt1_61.qasm", + "ae_mapped_ibm_washington_qiskit_opt2_88.qasm", + "qgan_mapped_ionq11_qiskit_opt3_3.qasm", + "qgan_mapped_oqc_lucy_tket_line_2.qasm", + ] + + @pytest.mark.parametrize( ("benchmark", "input_value", "scalable"), [ @@ -81,7 +105,9 @@ def output_path(): (qgan, 5, True), ], ) -def test_quantumcircuit_indep_level(benchmark, input_value, scalable, output_path): +def test_quantumcircuit_indep_level( + benchmark: types.ModuleType, input_value: int, scalable: bool, output_path: str +) -> None: if benchmark in (grover, qwalk): qc = benchmark.create_circuit(input_value, ancillary_mode="noancilla") else: @@ -151,7 +177,9 @@ def test_quantumcircuit_indep_level(benchmark, input_value, scalable, output_pat (qgan, 5, True), ], ) -def test_quantumcircuit_native_and_mapped_levels(benchmark, input_value, scalable, output_path): +def test_quantumcircuit_native_and_mapped_levels( + benchmark: types.ModuleType, input_value: int, scalable: bool, output_path: str +) -> None: if benchmark in (grover, qwalk): qc = benchmark.create_circuit(input_value, ancillary_mode="noancilla") else: @@ -261,41 +289,41 @@ def test_quantumcircuit_native_and_mapped_levels(benchmark, input_value, scalabl assert res -def test_openqasm_gates(): +def test_openqasm_gates() -> None: openqasm_gates = utils.get_openqasm_gates() num_openqasm_gates = 42 assert len(openqasm_gates) == num_openqasm_gates -def test_rigetti_cmap_generator(): +def test_rigetti_cmap_generator() -> None: num_edges = 212 assert len(utils.get_rigetti_aspen_m2_map()) == num_edges -def test_dj_constant_oracle(): +def test_dj_constant_oracle() -> None: qc = dj.create_circuit(5, False) assert qc.depth() > 0 -def test_groundstate(): +def test_groundstate() -> None: qc = groundstate.create_circuit("small") assert qc.depth() > 0 -def test_routing(): +def test_routing() -> None: qc = routing.create_circuit(4, 2) assert qc.depth() > 0 -def test_unidirectional_coupling_map(): - from pytket.architecture import Architecture +def test_unidirectional_coupling_map() -> None: + from pytket.architecture import Architecture # type: ignore[attr-defined] qc = get_benchmark( benchmark_name="dj", level="mapped", circuit_size=3, compiler="tket", - compiler_settings={"tket": {"placement": "graphplacement"}}, + compiler_settings=CompilerSettings(tket=TKETSettings(placement="graphplacement")), gate_set_name="oqc", device_name="oqc_lucy", ) @@ -381,9 +409,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 2}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=2)), "ionq", None, ), @@ -393,9 +419,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 2}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=2)), "ibm", None, ), @@ -405,9 +429,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 2}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=2)), "rigetti", None, ), @@ -417,9 +439,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 2}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=2)), "oqc", None, ), @@ -429,9 +449,7 @@ def test_unidirectional_coupling_map(): 6, None, "qiskit", - { - "qiskit": {"optimization_level": 3}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=3)), "ionq", None, ), @@ -441,9 +459,7 @@ def test_unidirectional_coupling_map(): 6, None, "qiskit", - { - "qiskit": {"optimization_level": 3}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=3)), "ibm", None, ), @@ -473,9 +489,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "ibm", "ibm_washington", ), @@ -485,9 +499,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "ibm", "ibm_montreal", ), @@ -497,9 +509,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "rigetti_aspen_m2", ), @@ -509,9 +519,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "ionq", "ionq11", ), @@ -521,9 +529,7 @@ def test_unidirectional_coupling_map(): 5, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "oqc", "oqc_lucy", ), @@ -533,9 +539,7 @@ def test_unidirectional_coupling_map(): 4, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "ibm", "ibm_washington", ), @@ -545,9 +549,7 @@ def test_unidirectional_coupling_map(): 4, None, "tket", - { - "tket": {"placement": "lineplacement"}, - }, + CompilerSettings(tket=TKETSettings(placement="lineplacement")), "ibm", "ibm_washington", ), @@ -557,9 +559,7 @@ def test_unidirectional_coupling_map(): 4, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "ibm", "ibm_montreal", ), @@ -569,9 +569,7 @@ def test_unidirectional_coupling_map(): 4, None, "tket", - { - "tket": {"placement": "graphplacement"}, - }, + CompilerSettings(tket=TKETSettings(placement="graphplacement")), "ibm", "ibm_montreal", ), @@ -581,9 +579,7 @@ def test_unidirectional_coupling_map(): 4, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "rigetti_aspen_m2", ), @@ -593,9 +589,7 @@ def test_unidirectional_coupling_map(): 4, None, "tket", - { - "tket": {"placement": "lineplacement"}, - }, + CompilerSettings(tket=TKETSettings(placement="lineplacement")), "rigetti", "rigetti_aspen_m2", ), @@ -605,9 +599,7 @@ def test_unidirectional_coupling_map(): 4, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "oqc", "oqc_lucy", ), @@ -617,24 +609,22 @@ def test_unidirectional_coupling_map(): 4, None, "tket", - { - "tket": {"placement": "graphplacement"}, - }, + CompilerSettings(tket=TKETSettings(placement="graphplacement")), "oqc", "oqc_lucy", ), ], ) def test_get_benchmark( - benchmark_name, - level, - circuit_size, - benchmark_instance_name, - compiler, - compiler_settings, - gate_set_name, - device_name, -): + benchmark_name: str, + level: str | int, + circuit_size: int, + benchmark_instance_name: str | None, + compiler: str, + compiler_settings: CompilerSettings | None, + gate_set_name: str | None, + device_name: str | None, +) -> None: qc = get_benchmark( benchmark_name, level, @@ -654,7 +644,7 @@ def test_get_benchmark( assert gate_type in qiskit_helper.get_native_gates(gate_set_name) or gate_type == "barrier" -def test_get_benchmark_faulty_parameters(): +def test_get_benchmark_faulty_parameters() -> None: match = "Selected benchmark is not supported. Valid benchmarks are" with pytest.raises(ValueError, match=match): get_benchmark("wrong_name", 2, 6) @@ -663,12 +653,10 @@ def test_get_benchmark_faulty_parameters(): get_benchmark( "qpeexact", 8, - "wrong_size", + "wrong_size", # type: ignore[arg-type] None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "rigetti_aspen_m2", ) @@ -680,12 +668,24 @@ def test_get_benchmark_faulty_parameters(): -1, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), + "rigetti", + "rigetti_aspen_m2", + ) + + match = "benchmark_instance_name must be defined for this benchmark." + with pytest.raises(ValueError, match=match): + get_benchmark( + "shor", + 1, + 3, + 2, # type: ignore[arg-type] + "qiskit", + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "rigetti_aspen_m2", ) + match = "Selected compiler must be in" with pytest.raises(ValueError, match=match): get_benchmark( @@ -694,13 +694,11 @@ def test_get_benchmark_faulty_parameters(): 3, None, "wrong_compiler", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "rigetti_aspen_m2", ) - match = "compiler_settings must be None" + match = "compiler_settings must be of type CompilerSettings or None" with pytest.raises(ValueError, match=match): get_benchmark( "qpeexact", @@ -708,7 +706,7 @@ def test_get_benchmark_faulty_parameters(): 3, None, "qiskit", - "wrong_compiler_settings", + "wrong_compiler_settings", # type: ignore[arg-type] "rigetti", "rigetti_aspen_m2", ) @@ -720,9 +718,7 @@ def test_get_benchmark_faulty_parameters(): 3, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "wrong_gateset", "rigetti_aspen_m2", ) @@ -734,15 +730,13 @@ def test_get_benchmark_faulty_parameters(): 3, None, "qiskit", - { - "qiskit": {"optimization_level": 1}, - }, + CompilerSettings(qiskit=QiskitSettings(optimization_level=1)), "rigetti", "wrong_device", ) -def test_create_benchmarks_from_config(output_path): +def test_create_benchmarks_from_config(output_path: str) -> None: config = { "timeout": 120, "benchmarks": [ @@ -765,7 +759,7 @@ def test_create_benchmarks_from_config(output_path): file.unlink() -def test_configure_end(output_path): +def test_configure_end(output_path: str) -> None: # delete all files in the test directory and the directory itself for f in Path(output_path).iterdir(): f.unlink() @@ -792,7 +786,7 @@ def test_zip_creation() -> None: ) def test_saving_qasm_to_alternative_location_with_alternative_filename( abstraction_level: int, -): +) -> None: directory = "." filename = "ae_test_qiskit" qc = get_benchmark("ae", abstraction_level, 5) @@ -825,7 +819,7 @@ def test_saving_qasm_to_alternative_location_with_alternative_filename( path.unlink() -def test_oqc_postprocessing(): +def test_oqc_postprocessing() -> None: qc = get_benchmark("ghz", 1, 5) assert qc directory = "." @@ -886,7 +880,7 @@ def test_oqc_postprocessing(): path.unlink() -def test_evaluate_qasm_file(): +def test_evaluate_qasm_file() -> None: qc = get_benchmark("dj", 1, 5) filename = "test_5.qasm" qc.qasm(filename=filename) @@ -897,21 +891,6 @@ def test_evaluate_qasm_file(): path.unlink() -FILENAMES = [ - "ae_indep_qiskit_10.qasm", - "ghz_nativegates_rigetti_qiskit_opt3_54.qasm", - "ae_indep_tket_93.qasm", - "wstate_nativegates_rigetti_qiskit_opt0_79.qasm", - "ae_mapped_ibm_montreal_qiskit_opt1_9.qasm", - "ae_mapped_ibm_washington_qiskit_opt0_38.qasm", - "ae_mapped_oqc_lucy_qiskit_opt0_5.qasm", - "ae_mapped_rigetti_aspen_m2_qiskit_opt1_61.qasm", - "ae_mapped_ibm_washington_qiskit_opt2_88.qasm", - "qgan_mapped_ionq11_qiskit_opt3_3.qasm", - "qgan_mapped_oqc_lucy_tket_line_2.qasm", -] - - @pytest.mark.parametrize( ("search_str", "expected_val"), [ @@ -927,8 +906,8 @@ def test_evaluate_qasm_file(): ("mapped_ionq11", 1), ], ) -def test_count_occurrences(search_str: str, expected_val: int) -> None: - assert evaluation.count_occurrences(FILENAMES, search_str) == expected_val +def test_count_occurrences(search_str: str, expected_val: int, sample_filenames: list[str]) -> None: + assert evaluation.count_occurrences(sample_filenames, search_str) == expected_val @pytest.mark.parametrize( @@ -938,8 +917,8 @@ def test_count_occurrences(search_str: str, expected_val: int) -> None: ("tket", [93, 2]), ], ) -def test_count_qubit_numbers_per_compiler(compiler: str, expected_val: list[int]) -> None: - assert evaluation.count_qubit_numbers_per_compiler(FILENAMES, compiler) == expected_val +def test_count_qubit_numbers_per_compiler(compiler: str, expected_val: list[int], sample_filenames: list[str]) -> None: + assert evaluation.count_qubit_numbers_per_compiler(sample_filenames, compiler) == expected_val def test_calc_supermarq_features() -> None: @@ -956,7 +935,7 @@ def test_BenchmarkGenerator() -> None: # This function is used to test the timeout watchers and needs two parameters since those values are logged when a timeout occurs. -def endless_loop(arg1: TestObject, run_forever: bool) -> None: # noqa: ARG001 +def endless_loop(arg1: TestObject, run_forever: bool) -> bool: # noqa: ARG001 while run_forever: pass return True @@ -981,10 +960,14 @@ def test_get_module_for_benchmark() -> None: def test_benchmark_helper() -> None: shor_instances = ["xsmall", "small", "medium", "large", "xlarge"] for elem in shor_instances: - res = shor.get_instance(elem) - assert res is not None - + res_shor = shor.get_instance(elem) + assert res_shor groundstate_instances = ["small", "medium", "large"] for elem in groundstate_instances: - res = groundstate.get_molecule(elem) - assert res is not None + res_groundstate = groundstate.get_molecule(elem) + assert res_groundstate + + +def test_get_cmap_from_devicename() -> None: + with pytest.raises(ValueError, match="Device wrong_name is not supported"): + utils.get_cmap_from_devicename("wrong_name") diff --git a/tests/test_benchviewer.py b/tests/test_benchviewer.py index 28d6ce81b..aaa6cb0f2 100644 --- a/tests/test_benchviewer.py +++ b/tests/test_benchviewer.py @@ -4,10 +4,10 @@ from typing import TYPE_CHECKING import pytest -from mqt.benchviewer import Backend, Server, backend +from mqt.benchviewer import Backend, BenchmarkConfiguration, Server, backend from mqt.benchviewer.main import app -if TYPE_CHECKING or sys.version_info >= (3, 10, 0): +if TYPE_CHECKING or sys.version_info >= (3, 10, 0): # pragma: no cover from importlib import resources else: import importlib_resources as resources @@ -23,7 +23,7 @@ ("HHL_indep_5.qasm", -1), ], ) -def test_get_opt_level(filename, expected_res): +def test_get_opt_level(filename: str, expected_res: int) -> None: assert int(backend.get_opt_level(filename)) == expected_res @@ -37,7 +37,7 @@ def test_get_opt_level(filename, expected_res): ("HHL_indep_5.qasm", 5), ], ) -def test_get_num_qubits(filename, expected_res): +def test_get_num_qubits(filename: str, expected_res: int) -> None: assert int(backend.get_num_qubits(filename)) == expected_res @@ -46,86 +46,86 @@ def test_get_num_qubits(filename, expected_res): [ ( "shor_15_4_nativegates_rigetti_qiskit_opt0_18.qasm", - [ - "shor", - 18, - False, - True, - False, - "qiskit", - 0, - "rigetti", - None, - "shor_15_4_nativegates_rigetti_qiskit_opt0_18.qasm", - ], + backend.ParsedBenchmarkName( + benchmark="shor", + num_qubits=18, + indep_flag=False, + nativegates_flag=True, + mapped_flag=False, + compiler="qiskit", + compiler_settings=0, + gate_set="rigetti", + target_device=None, + filename="shor_15_4_nativegates_rigetti_qiskit_opt0_18.qasm", + ), ), ( "dj_mapped_ibm_washington_qiskit_opt3_103.qasm", - [ - "dj", - 103, - False, - False, - True, - "qiskit", - 3, - "ibm", - "ibm_washington", - "dj_mapped_ibm_washington_qiskit_opt3_103.qasm", - ], + backend.ParsedBenchmarkName( + benchmark="dj", + num_qubits=103, + indep_flag=False, + nativegates_flag=False, + mapped_flag=True, + compiler="qiskit", + compiler_settings=3, + gate_set="ibm", + target_device="ibm_washington", + filename="dj_mapped_ibm_washington_qiskit_opt3_103.qasm", + ), ), ( "pricingcall_mapped_oqc_lucy_tket_line_5.qasm", - [ - "pricingcall", - 5, - False, - False, - True, - "tket", - "line", - "oqc", - "oqc_lucy", - "pricingcall_mapped_oqc_lucy_tket_line_5.qasm", - ], + backend.ParsedBenchmarkName( + benchmark="pricingcall", + num_qubits=5, + indep_flag=False, + nativegates_flag=False, + mapped_flag=True, + compiler="tket", + compiler_settings="line", + gate_set="oqc", + target_device="oqc_lucy", + filename="pricingcall_mapped_oqc_lucy_tket_line_5.qasm", + ), ), ( "portfoliovqe_nativegates_ionq_qiskit_opt1_3.qasm", - [ - "portfoliovqe", - 3, - False, - True, - False, - "qiskit", - 1, - "ionq", - None, - "portfoliovqe_nativegates_ionq_qiskit_opt1_3.qasm", - ], + backend.ParsedBenchmarkName( + benchmark="portfoliovqe", + num_qubits=3, + indep_flag=False, + nativegates_flag=True, + mapped_flag=False, + compiler="qiskit", + compiler_settings=1, + gate_set="ionq", + target_device=None, + filename="portfoliovqe_nativegates_ionq_qiskit_opt1_3.qasm", + ), ), ( "HHL_indep_qiskit_5.qasm", - [ - "hhl", - 5, - True, - False, - False, - "qiskit", - -1, - None, - None, - "HHL_indep_qiskit_5.qasm", - ], + backend.ParsedBenchmarkName( + benchmark="hhl", + num_qubits=5, + indep_flag=True, + nativegates_flag=False, + mapped_flag=False, + compiler="qiskit", + compiler_settings=-1, + gate_set=None, + target_device=None, + filename="HHL_indep_qiskit_5.qasm", + ), ), ], ) -def test_parse_data(filename, expected_res): +def test_parse_data(filename: str, expected_res: backend.ParsedBenchmarkName) -> None: assert backend.parse_data(filename) == expected_res -def test_prepare_form_input(): +def test_prepare_form_input() -> None: form_data = { "all_benchmarks": "true", "minQubits": "75", @@ -185,51 +185,70 @@ def test_prepare_form_input(): "device_ionq_ionq11": "true", } - expected_res = ( - (75, 110), - [ - "1", - "2", - "3", - "4", - "5", - "6", - "7", - "8", - "9", - "10", - "11", - "12", - "13", - "14", - "15", - "16", - "17", - "18", - "19", - "20", - "21", - "22", - "23", - "24", - "25", - "26", - "27", - "28", - ], - (True, True), - ((True, True), [0, 1, 2, 3], ["ibm", "rigetti", "oqc", "ionq"]), - ( - (True, True), - ([0, 1, 2, 3], ["graph", "line"]), - [ - "ibm_washington", - "ibm_montreal", - "rigetti_aspen", - "oqc_lucy", - "ionq11", - ], - ), + expected_res = BenchmarkConfiguration( + min_qubits=75, + max_qubits=110, + indices_benchmarks=list(range(1, 29)), + indep_qiskit_compiler=True, + indep_tket_compiler=True, + nativegates_qiskit_compiler=True, + native_qiskit_opt_lvls=[0, 1, 2, 3], + nativegates_tket_compiler=True, + native_gatesets=["ibm", "rigetti", "oqc", "ionq"], + mapped_qiskit_compiler=True, + mapped_qiskit_opt_lvls=[0, 1, 2, 3], + mapped_tket_compiler=True, + mapped_tket_placements=["graph", "line"], + mapped_devices=["ibm_washington", "ibm_montreal", "rigetti_aspen", "oqc_lucy", "ionq11"], + ) + backend = Backend() + assert backend.prepare_form_input(form_data) == expected_res + + form_data = { + "all_benchmarks": "true", + "minQubits": "75", + "maxQubits": "110", + "indep_qiskit_compiler": "true", + "indep_tket_compiler": "true", + "nativegates_qiskit_compiler": "true", + "nativegates_qiskit_compiler_opt0": "true", + "nativegates_qiskit_compiler_opt1": "true", + "nativegates_qiskit_compiler_opt2": "true", + "nativegates_qiskit_compiler_opt3": "true", + "nativegates_tket_compiler value=": "on", + "nativegates_ibm": "true", + "nativegates_rigetti": "true", + "nativegates_oqc": "true", + "nativegates_ionq": "true", + "mapped_qiskit_compiler": "true", + "mapped_qiskit_compiler_opt0": "true", + "mapped_qiskit_compiler_opt1": "true", + "mapped_qiskit_compiler_opt2": "true", + "mapped_qiskit_compiler_opt3": "true", + "mapped_tket_compiler": "true", + "mapped_tket_compiler_graph": "true", + "mapped_tket_compiler_line": "true", + "device_ibm_washington": "true", + "device_ibm_montreal": "true", + "device_rigetti_aspen_m2": "true", + "device_oqc_lucy": "true", + "device_ionq_ionq11": "true", + } + expected_res = BenchmarkConfiguration( + min_qubits=75, + max_qubits=110, + indices_benchmarks=[], + indep_qiskit_compiler=True, + indep_tket_compiler=True, + nativegates_qiskit_compiler=True, + native_qiskit_opt_lvls=[0, 1, 2, 3], + nativegates_tket_compiler=True, + native_gatesets=["ibm", "rigetti", "oqc", "ionq"], + mapped_qiskit_compiler=True, + mapped_qiskit_opt_lvls=[0, 1, 2, 3], + mapped_tket_compiler=True, + mapped_tket_placements=["graph", "line"], + mapped_devices=["ibm_washington", "ibm_montreal", "rigetti_aspen", "oqc_lucy", "ionq11"], ) backend = Backend() assert backend.prepare_form_input(form_data) == expected_res @@ -238,14 +257,14 @@ def test_prepare_form_input(): benchviewer = resources.files("mqt.benchviewer") -def test_read_mqtbench_all_zip(): +def test_read_mqtbench_all_zip() -> None: backend = Backend() with resources.as_file(benchviewer) as benchviewer_path: target_location = str(benchviewer_path / "static/files") assert backend.read_mqtbench_all_zip(skip_question=True, target_location=target_location) -def test_create_database(): +def test_create_database() -> None: backend = Backend() res_zip = backend.read_mqtbench_all_zip( @@ -253,65 +272,96 @@ def test_create_database(): ) assert res_zip - database = backend.init_database() - assert database - - input_data = ( - (2, 5), - ["4"], - (True, False), - ((False, False), [], []), - ((False, False), ([], []), []), + assert backend.database is None + backend.init_database() + + input_data = BenchmarkConfiguration( + min_qubits=2, + max_qubits=5, + indices_benchmarks=[4], + indep_qiskit_compiler=True, + indep_tket_compiler=False, + nativegates_qiskit_compiler=False, + nativegates_tket_compiler=False, + mapped_qiskit_compiler=False, + mapped_tket_compiler=False, ) + res = backend.get_selected_file_paths(input_data) + assert isinstance(res, list) assert len(res) > 3 - input_data = ( - (110, 120), - ["3"], - (False, False), - ((False, True), [], ["rigetti", "ionq"]), - ((False, False), ([], []), []), + input_data = BenchmarkConfiguration( + min_qubits=110, + max_qubits=120, + indices_benchmarks=[3], + indep_qiskit_compiler=False, + indep_tket_compiler=False, + nativegates_qiskit_compiler=False, + nativegates_tket_compiler=True, + mapped_qiskit_compiler=False, + mapped_tket_compiler=False, + native_gatesets=["rigetti", "ionq"], ) res = backend.get_selected_file_paths(input_data) + assert isinstance(res, list) assert len(res) > 15 - input_data = ( - (75, 110), - ["2"], - (False, False), - ((False, False), [], ["rigetti", "ionq"]), - ((True, True), ([1, 3], ["graph"]), ["ibm_washington", "rigetti_aspen_m2"]), + input_data = BenchmarkConfiguration( + min_qubits=75, + max_qubits=110, + indices_benchmarks=[2], + indep_qiskit_compiler=False, + indep_tket_compiler=False, + nativegates_qiskit_compiler=False, + nativegates_tket_compiler=False, + mapped_qiskit_compiler=True, + mapped_tket_compiler=True, + native_gatesets=["rigetti", "ionq"], + mapped_devices=["ibm_washington", "rigetti_aspen_m2"], + mapped_tket_placements=["graph"], ) res = backend.get_selected_file_paths(input_data) + assert isinstance(res, list) assert len(res) > 20 - input_data = ( - (2, 5), - ["23"], - (True, True), - ((True, False), [1, 3], ["rigetti", "ionq", "oqc", "ibm"]), - ( - (True, True), - ([1, 3], ["graph", "line"]), - ["ibm_montreal", "rigetti_aspen", "ionq11", "ocq_lucy"], - ), + input_data = BenchmarkConfiguration( + min_qubits=2, + max_qubits=5, + indices_benchmarks=[23], + indep_qiskit_compiler=True, + indep_tket_compiler=True, + nativegates_qiskit_compiler=True, + nativegates_tket_compiler=False, + mapped_qiskit_compiler=True, + mapped_tket_compiler=True, + native_gatesets=["rigetti", "ionq", "oqc", "ibm"], + mapped_devices=["ibm_montreal", "rigetti_aspen", "ionq11", "ocq_lucy"], + mapped_tket_placements=["graph", "line"], + native_qiskit_opt_lvls=[0, 3], + mapped_qiskit_opt_lvls=[0, 3], ) res = backend.get_selected_file_paths(input_data) + assert isinstance(res, list) assert len(res) > 20 - input_data = ( - (2, 130), - ["1"], - (False, False), - ((True, True), [], []), - ((True, True), ([], []), []), + input_data = BenchmarkConfiguration( + min_qubits=2, + max_qubits=130, + indices_benchmarks=[1], + indep_qiskit_compiler=False, + indep_tket_compiler=False, + nativegates_qiskit_compiler=True, + nativegates_tket_compiler=True, + mapped_qiskit_compiler=True, + mapped_tket_compiler=True, ) res = backend.get_selected_file_paths(input_data) + assert isinstance(res, list) assert res == [] -def test_streaming_zip(): +def test_streaming_zip() -> None: backend = Backend() backend.read_mqtbench_all_zip( skip_question=True, target_location=str(resources.files("mqt.benchviewer") / "static" / "files") @@ -323,12 +373,12 @@ def test_streaming_zip(): assert not list(backend.generate_zip_ephemeral_chunks(filenames=["not_existing_file.qasm"])) -def test_flask_server(): +def test_flask_server() -> None: with resources.as_file(benchviewer) as benchviewer_path: benchviewer_location = benchviewer_path target_location = str(benchviewer_location / "static/files") - assert Server( + Server( skip_question=True, activate_logging=False, target_location=target_location,