diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c23d6add..2e430847a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,9 +12,19 @@ releases may include breaking changes. ### Added +- ✨ Add opt-in per-pass timeouts for RL training and inference ([#789]) + ([**@flowerthrower**]) +- ✨ Add Qiskit's `TrivialLayout`, `ElidePermutations`, `SabreSwap`, + `BasicSwap`, `LookaheadSwap`, `RemoveIdentityEquivalent`, and + `Optimize1qGatesSimpleCommutation` passes to the RL actions ([#785]) + ([**@flowerthrower**]) +- ✨ Encode the RL qubit-count and depth observations as normalized + one-element `float32` arrays ([#784]) ([**@flowerthrower**]) - ✨ Expand the RL observation with normalized OpenQASM operation frequencies and include measurements in the shared ML feature schema ([#758]) ([**@flowerthrower**]) +- ✨ Evaluate configurable repeated candidates for stochastic Qiskit RL actions + and retain the highest-scoring result ([#757]) ([**@flowerthrower**]) - 👷 Enable testing on Python 3.14 ([#488]) ([**@denialhaag**]) - ✨ Add selectable `v2` and `v3` RL MDP strategies, make `v3` the default, and use the selected strategy in compilation traces and model artifact names @@ -22,6 +32,8 @@ releases may include breaking changes. ### Changed +- 🐛 Make the `OptimizeCliffords` RL action collect standard Clifford gates + before optimizing them ([#785]) ([**@flowerthrower**]) - 🔥 Drop support for Python 3.10 ([#773]) ([**@denialhaag**]) - ♻️ Split RL actions package into `base` and `registry` modules ([#769]) ([**@denialhaag**]) @@ -90,9 +102,13 @@ for previous changelogs._ +[#789]: https://github.com/munich-quantum-toolkit/predictor/pull/789 [#773]: https://github.com/munich-quantum-toolkit/predictor/pull/771 [#769]: https://github.com/munich-quantum-toolkit/predictor/pull/769 +[#784]: https://github.com/munich-quantum-toolkit/predictor/pull/784 +[#785]: https://github.com/munich-quantum-toolkit/predictor/pull/785 [#758]: https://github.com/munich-quantum-toolkit/predictor/pull/758 +[#757]: https://github.com/munich-quantum-toolkit/predictor/pull/757 [#755]: https://github.com/munich-quantum-toolkit/predictor/pull/755 [#731]: https://github.com/munich-quantum-toolkit/predictor/pull/731 [#714]: https://github.com/munich-quantum-toolkit/predictor/pull/714 diff --git a/UPGRADING.md b/UPGRADING.md index 6bd5fc3fb..c5c35338d 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -6,6 +6,32 @@ of changes including minor and patch releases, please refer to the ## [Unreleased] +### Expanded Qiskit action set + +The RL action space now includes the following Qiskit passes: + +- the `TrivialLayout` and `ElidePermutations` layout actions; +- the `SabreSwap`, `BasicSwap`, and `LookaheadSwap` routing actions; and +- the `RemoveIdentityEquivalent` and `Optimize1qGatesSimpleCommutation` + optimization actions. + +`ElidePermutations` establishes a trivial layout in the same action so its +output permutation remains part of the canonical layout. `OptimizeCliffords` now +collects standard Clifford gates before optimizing and decomposes the result for +subsequent passes. + +Existing RL models must be retrained because the action-space size and the +indices of later actions have changed. Code that persists or selects actions by +numeric index must be updated. + +### Compact scalar RL observations + +The `num_qubits` and `depth` entries in RL observations are now one-element +`float32` arrays in `[0, 1]` instead of discrete integers. The qubit count is +linearly scaled by the target device's qubit count; the depth is `log1p`-scaled +and capped at 999,999. Existing RL models must be retrained, and code that +consumes `PredictorEnv` observations directly must handle the new array values. + ### RL operation features The RL observation now includes normalized frequencies for supported OpenQASM @@ -18,6 +44,14 @@ Starting with this release, MQT Predictor no longer supports Python 3.10. As a result, MQT Predictor is no longer tested under Python 3.10 and requires Python 3.11 or later. +### Repeated stochastic Qiskit actions + +The stochastic `QiskitSabreMapping` action now evaluates 20 candidates by +default and retains the candidate with the best configured figure of merit. This +can increase compilation time and change the selected circuit. Set +`stochastic_action_trials=1` when constructing `PredictorEnv` to retain the +previous single-attempt behavior. + ### Atomic BQSKit compilation actions The composite actions `BQSKitO2`, `BQSKitSynthesis`, and `BQSKitMapping` are no diff --git a/docs/setup.md b/docs/setup.md index 1582cbb5e..63ee2bdea 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -45,9 +45,14 @@ from mqt.bench.targets import get_device device = get_device("ibm_falcon_27") rl_pred = RL_Predictor(device=device, figure_of_merit="expected_fidelity") -rl_pred.train_model(timesteps=100000) +rl_pred.train_model(timesteps=100000, pass_timeout=600) ``` +`pass_timeout` is optional and limits each compilation pass during training. If +it is omitted, pass execution is unbounded. Pass timeouts require POSIX signals +and execution on the main thread; unsupported environments emit a warning and +continue without a timeout. + Currently, the following figures of merit are supported: ```{code-cell} ipython3 @@ -121,9 +126,16 @@ from mqt.predictor import qcompile from mqt.bench import get_benchmark, BenchmarkLevel uncompiled_qc = get_benchmark("ghz", level=BenchmarkLevel.ALG, circuit_size=5) -compiled_qc, compilation_info, selected_device = qcompile(uncompiled_qc, figure_of_merit="expected_fidelity") +compiled_qc, compilation_info, selected_device = qcompile( + uncompiled_qc, + figure_of_merit="expected_fidelity", + pass_timeout=60, +) ``` +Inference has its own optional `pass_timeout`, so it can use a different limit +than training. + This returns: - the compiled quantum circuit, diff --git a/src/mqt/predictor/qcompile.py b/src/mqt/predictor/qcompile.py index a1f8240f0..6807ed6c5 100644 --- a/src/mqt/predictor/qcompile.py +++ b/src/mqt/predictor/qcompile.py @@ -27,6 +27,7 @@ def qcompile( qc: QuantumCircuit, figure_of_merit: figure_of_merit = "expected_fidelity", tracer_output_path: str | Path | None = None, + pass_timeout: float | None = None, ) -> tuple[QuantumCircuit, list[str], str]: """Compiles a given quantum circuit to a device with the highest predicted figure of merit. @@ -34,12 +35,21 @@ def qcompile( qc: The quantum circuit to be compiled. figure_of_merit: The figure of merit to be used for compilation. Defaults to "expected_fidelity". tracer_output_path: If provided, enables compiler tracing and exports the JSON log to this path/directory. + pass_timeout: Maximum duration in seconds for one compilation pass. + Defaults to None, which disables pass timeouts. Returns: A tuple containing the compiled quantum circuit, the compilation information, and the name of the device used for compilation. + + Raises: + ValueError: If ``pass_timeout`` is not positive. """ predicted_device = predict_device_for_figure_of_merit(qc, figure_of_merit=figure_of_merit) res = rl_compile( - qc, device=predicted_device, figure_of_merit=figure_of_merit, tracer_output_path=tracer_output_path + qc, + device=predicted_device, + figure_of_merit=figure_of_merit, + tracer_output_path=tracer_output_path, + pass_timeout=pass_timeout, ) return *res, predicted_device diff --git a/src/mqt/predictor/rl/actions/base.py b/src/mqt/predictor/rl/actions/base.py index 7d36bda02..20b09458c 100644 --- a/src/mqt/predictor/rl/actions/base.py +++ b/src/mqt/predictor/rl/actions/base.py @@ -52,6 +52,7 @@ class Action: preserves_layout: Whether action preserves existing layout. preserves_routing: Whether action preserves existing routing. preserves_synthesis: Whether action preserves synthesis state. + stochastic: Whether repeated execution can yield different results. """ name: str @@ -61,6 +62,7 @@ class Action: preserves_layout: bool = False preserves_routing: bool = False preserves_synthesis: bool = False + stochastic: bool = False @dataclass diff --git a/src/mqt/predictor/rl/actions/bqskit_actions.py b/src/mqt/predictor/rl/actions/bqskit_actions.py index 0d78632cc..35314d6e1 100644 --- a/src/mqt/predictor/rl/actions/bqskit_actions.py +++ b/src/mqt/predictor/rl/actions/bqskit_actions.py @@ -39,6 +39,8 @@ IfThenElsePass, LEAPSynthesisPass, ManyQuditGatesPredicate, + MGDPass, + QSDPass, QSearchSynthesisPass, RestoreMeasurements, SetModelPass, @@ -309,6 +311,18 @@ def bqskit_synthesis_actions() -> list[Action]: IfThenElsePass(DiagonalPredicate(1e-9), WalshDiagonalSynthesisPass()), ), ), + DeferredDeviceAction( + "QSDPass", + CompilationOrigin.BQSKIT, + PassType.SYNTHESIS, + transpile_pass=lambda device: _bqskit_partitioned_synthesis_factory(device, QSDPass()), + ), + DeferredDeviceAction( + "MGDPass", + CompilationOrigin.BQSKIT, + PassType.SYNTHESIS, + transpile_pass=lambda device: _bqskit_partitioned_synthesis_factory(device, MGDPass()), + ), DeferredDeviceAction( "FullQSDPass", CompilationOrigin.BQSKIT, diff --git a/src/mqt/predictor/rl/actions/qiskit_actions.py b/src/mqt/predictor/rl/actions/qiskit_actions.py index 173d168cd..5c4393279 100644 --- a/src/mqt/predictor/rl/actions/qiskit_actions.py +++ b/src/mqt/predictor/rl/actions/qiskit_actions.py @@ -11,6 +11,7 @@ from __future__ import annotations import logging +from copy import deepcopy from typing import TYPE_CHECKING, cast from qiskit.circuit import StandardEquivalenceLibrary @@ -37,24 +38,32 @@ from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout from qiskit.transpiler.passes import ( ApplyLayout, + BasicSwap, BasisTranslator, Collect2qBlocks, + CollectCliffords, CommutativeCancellation, CommutativeInverseCancellation, ConsolidateBlocks, DenseLayout, Depth, + ElidePermutations, EnlargeWithAncilla, FixedPoint, FullAncillaAllocation, GatesInBasis, InverseCancellation, + LookaheadSwap, MinimumPoint, Optimize1qGatesDecomposition, + Optimize1qGatesSimpleCommutation, OptimizeCliffords, RemoveDiagonalGatesBeforeMeasure, + RemoveIdentityEquivalent, SabreLayout, + SabreSwap, Size, + TrivialLayout, UnitarySynthesis, VF2Layout, VF2PostLayout, @@ -149,7 +158,7 @@ def qiskit_optimization_actions() -> list[Action]: "OptimizeCliffords", CompilationOrigin.QISKIT, PassType.OPT, - [OptimizeCliffords()], + [CollectCliffords(), OptimizeCliffords()], preserves_layout=True, preserves_routing=False, preserves_synthesis=False, @@ -163,6 +172,32 @@ def qiskit_optimization_actions() -> list[Action]: preserves_routing=True, preserves_synthesis=False, ), + DeviceIndependentAction( + "RemoveIdentityEquivalent", + CompilationOrigin.QISKIT, + PassType.OPT, + [RemoveIdentityEquivalent()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + DeferredDeviceAction( + "Optimize1qGatesSimpleCommutation", + CompilationOrigin.QISKIT, + PassType.OPT, + transpile_pass=lambda device: cast( + "list[Task]", + [ + Optimize1qGatesSimpleCommutation( + basis=device.operation_names, + run_to_completion=True, + ) + ], + ), + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), ] @@ -249,6 +284,73 @@ def qiskit_layout_actions() -> list[Action]: ], ), ), + DeferredDeviceAction( + "TrivialLayout", + CompilationOrigin.QISKIT, + PassType.LAYOUT, + transpile_pass=lambda device: cast( + "list[Task]", + [ + TrivialLayout(coupling_map=CouplingMap(device.build_coupling_map())), + FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ], + ), + ), + DeferredDeviceAction( + "ElidePermutations", + CompilationOrigin.QISKIT, + PassType.LAYOUT, + transpile_pass=lambda device: cast( + "list[Task]", + [ + ElidePermutations(), + TrivialLayout(coupling_map=CouplingMap(device.build_coupling_map())), + FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ], + ), + ), + ] + + +def qiskit_routing_actions() -> list[Action]: + """Return the Qiskit routing actions.""" + return [ + DeferredDeviceAction( + "SabreSwap", + CompilationOrigin.QISKIT, + PassType.ROUTING, + stochastic=True, + transpile_pass=lambda device: cast( + "list[Task]", [SabreSwap(coupling_map=CouplingMap(device.build_coupling_map()), heuristic="decay")] + ), + ), + DeferredDeviceAction( + "BasicSwap", + CompilationOrigin.QISKIT, + PassType.ROUTING, + transpile_pass=lambda device: cast( + "list[Task]", [BasicSwap(coupling_map=CouplingMap(device.build_coupling_map()))] + ), + ), + DeferredDeviceAction( + "LookaheadSwap", + CompilationOrigin.QISKIT, + PassType.ROUTING, + transpile_pass=lambda device: cast( + "list[Task]", + [ + LookaheadSwap( + coupling_map=CouplingMap(device.build_coupling_map()), + search_depth=1, + search_width=1, + ) + ], + ), + ), ] @@ -258,6 +360,7 @@ def qiskit_mapping_action() -> Action: "QiskitSabreMapping", CompilationOrigin.QISKIT, PassType.MAPPING, + stochastic=True, transpile_pass=lambda device: cast( "list[Task]", [SabreLayout(coupling_map=CouplingMap(device.build_coupling_map()), skip_routing=False)] ), @@ -342,9 +445,63 @@ def run_qiskit_action( device: Target, layout: TranspileLayout | None, input_qubit_count: int | None = None, + stochastic_action_trials: int = 1, + score: Callable[[QuantumCircuit], float] | None = None, +) -> tuple[QuantumCircuit, TranspileLayout | None]: + """Apply a Qiskit action and return the updated circuit and layout metadata. + + Stochastic actions are evaluated repeatedly when a score function is + supplied. The highest-scoring result is retained together with its layout. + """ + stochastic_run = action.stochastic and score is not None + attempts = stochastic_action_trials if stochastic_run else 1 + best_result: tuple[QuantumCircuit, TranspileLayout | None] | None = None + best_score: float | None = None + + for attempt in range(max(1, attempts)): + try: + altered_qc, candidate_layout = _run_qiskit_action_once( + action, + circuit, + device, + deepcopy(layout) if stochastic_run else layout, + input_qubit_count, + ) + except TimeoutError: + raise + except Exception: + if not stochastic_run: + raise + logger.exception("Stochastic action %s failed on attempt %d.", action.name, attempt + 1) + continue + if score is None: + return altered_qc, candidate_layout + + try: + candidate_score = score(altered_qc) + except TimeoutError: + raise + except Exception: # ruff:ignore[blind-except] + logger.warning("Could not evaluate stochastic action %s; using swap count instead.", action.name) + candidate_score = -float(altered_qc.count_ops().get("swap", 0)) + if best_score is None or candidate_score > best_score: + best_result = altered_qc, candidate_layout + best_score = candidate_score + + if best_result is not None: + return best_result + logger.error("All attempts for stochastic action %s failed.", action.name) + return circuit, layout + + +def _run_qiskit_action_once( + action: Action, + circuit: QuantumCircuit, + device: Target, + layout: TranspileLayout | None, + input_qubit_count: int | None, ) -> tuple[QuantumCircuit, TranspileLayout | None]: - """Apply a Qiskit action and return the updated circuit and layout metadata.""" - # Build the concrete Qiskit pass list for given action. + """Run one Qiskit action attempt and update its layout metadata.""" if action.name == "QiskitO3" and isinstance(action, DeferredDeviceAction): factory = cast("Callable[[list[str], CouplingMap | None], list[Task]]", action.transpile_pass) passes = factory(device.operation_names, CouplingMap(device.build_coupling_map()) if layout else None) @@ -370,6 +527,8 @@ def run_qiskit_action( if altered_qc.count_ops().get("unitary"): # Custom "unitary" gates can not be processed further by other passes altered_qc = altered_qc.decompose(gates_to_decompose="unitary") + if altered_qc.count_ops().get("clifford"): + altered_qc = altered_qc.decompose(gates_to_decompose="clifford") return altered_qc, layout diff --git a/src/mqt/predictor/rl/actions/registry.py b/src/mqt/predictor/rl/actions/registry.py index 5aa6c73ad..c398a33fd 100644 --- a/src/mqt/predictor/rl/actions/registry.py +++ b/src/mqt/predictor/rl/actions/registry.py @@ -51,11 +51,13 @@ def get_actions_by_pass_type() -> dict[PassType, list[Action]]: for _action in ( *qiskit_actions.qiskit_layout_actions(), + *qiskit_actions.qiskit_routing_actions(), qiskit_actions.qiskit_mapping_action(), qiskit_actions.qiskit_synthesis_action(), qiskit_actions.qiskit_o3_action(), *qiskit_actions.qiskit_optimization_actions(), qiskit_actions.qiskit_final_optimization_action(), + *tket_actions.tket_layout_actions(), tket_actions.tket_routing_action(), *tket_actions.tket_optimization_actions(), *bqskit_actions.bqskit_layout_actions(), diff --git a/src/mqt/predictor/rl/actions/tket_actions.py b/src/mqt/predictor/rl/actions/tket_actions.py index d5fc3cd03..7579fe535 100644 --- a/src/mqt/predictor/rl/actions/tket_actions.py +++ b/src/mqt/predictor/rl/actions/tket_actions.py @@ -10,7 +10,9 @@ from __future__ import annotations +import logging import operator +from functools import cache from typing import TYPE_CHECKING, cast from pytket import Qubit @@ -18,9 +20,17 @@ from pytket.architecture import Architecture from pytket.circuit import Node from pytket.extensions.qiskit import qiskit_to_tk, tk_to_qiskit -from pytket.passes import CliffordSimp, FullPeepholeOptimise, PeepholeOptimise2Q, RemoveRedundancies, RoutingPass -from pytket.placement import place_with_map -from qiskit.transpiler import Layout +from pytket.passes import ( + CliffordSimp, + FullPeepholeOptimise, + KAKDecomposition, + PeepholeOptimise2Q, + RemoveRedundancies, + RoutingPass, +) +from pytket.placement import GraphPlacement, NoiseAwarePlacement, Placement, place_with_map +from qiskit.transpiler import CouplingMap, Layout, PassManager, TranspileLayout +from qiskit.transpiler.passes import ApplyLayout, EnlargeWithAncilla, FullAncillaAllocation, SetLayout from mqt.predictor.rl.actions.base import CompilationOrigin, DeferredDeviceAction, DeviceIndependentAction, PassType @@ -29,11 +39,13 @@ from pytket import Circuit from qiskit import QuantumCircuit - from qiskit.passmanager.base_tasks import Task - from qiskit.transpiler import Target, TranspileLayout + from qiskit.circuit import Qubit as QiskitQubit + from qiskit.transpiler import Target from mqt.predictor.rl.actions.base import Action +logger = logging.getLogger("mqt-predictor") + class PreProcessTKETRoutingAfterQiskitLayout: """Pre-process TKET routing for circuits that already carry a Qiskit layout. @@ -49,6 +61,86 @@ def apply(self, circuit: Circuit) -> None: place_with_map(circuit=circuit, qmap=mapping) +@cache +def _prepare_noise_data(device: Target) -> tuple[dict[Node, float], dict[tuple[Node, Node], float], dict[Node, float]]: + """Extract calibration errors for TKET's noise-aware placement.""" + node_errors: dict[Node, float] = {} + link_errors: dict[tuple[Node, Node], float] = {} + readout_errors: dict[Node, float] = {} + + for operation_name in device.operation_names: + for qubits, properties in device[operation_name].items(): + if qubits is None or properties is None or properties.error is None: + continue + if len(qubits) == 1: + node_errors[Node(qubits[0])] = properties.error + elif len(qubits) == 2: + link_errors[Node(qubits[0]), Node(qubits[1])] = properties.error + + if "measure" in device: + for qubits, properties in device["measure"].items(): + if qubits is not None and len(qubits) == 1 and properties is not None and properties.error is not None: + readout_errors[Node(qubits[0])] = properties.error + + return node_errors, link_errors, readout_errors + + +def _noise_aware_placement(device: Target) -> list[Placement]: + node_errors, link_errors, readout_errors = _prepare_noise_data(device) + return [ + NoiseAwarePlacement( + Architecture(list(device.build_coupling_map())), + node_errors=node_errors, + link_errors=link_errors, + readout_errors=readout_errors, + timeout=5000, + maximum_matches=5000, + ) + ] + + +def _translate_placement( + circuit: QuantumCircuit, + placement: dict[Qubit, Node], + action_name: str, + num_device_qubits: int, +) -> Layout | None: + qiskit_qubits: dict[tuple[str, tuple[int, ...]], QiskitQubit] = {} + for qubit in circuit.qubits: + location = circuit.find_bit(qubit) + if location.registers: + register, register_index = location.registers[0] + qiskit_qubits[register.name, (register_index,)] = qubit + else: + qiskit_qubits["q", (location.index,)] = qubit + + qiskit_mapping: dict[QiskitQubit, int] = {} + unassigned_qubits: list[QiskitQubit] = [] + used_physical_indices: set[int] = set() + for tket_qubit, target_node in placement.items(): + qiskit_qubit = qiskit_qubits.get((str(tket_qubit.reg_name), tuple(int(i) for i in tket_qubit.index))) + if qiskit_qubit is None: + logger.warning("Placement failed (%s): unknown logical qubit %s.", action_name, tket_qubit) + return None + + if target_node.reg_name == "node" and target_node.index: + physical_index = int(target_node.index[0]) + qiskit_mapping[qiskit_qubit] = physical_index + used_physical_indices.add(physical_index) + else: + unassigned_qubits.append(qiskit_qubit) + + unassigned_qubits.extend(qubit for qubit in circuit.qubits if qubit not in qiskit_mapping) + unassigned_qubits = list(dict.fromkeys(unassigned_qubits)) + remaining_indices = [index for index in range(num_device_qubits) if index not in used_physical_indices] + if len(remaining_indices) < len(unassigned_qubits): + logger.warning("Placement failed (%s): insufficient free physical qubits.", action_name) + return None + + qiskit_mapping.update(zip(unassigned_qubits, remaining_indices, strict=False)) + return Layout(qiskit_mapping) + + def tket_optimization_actions() -> list[Action]: """Returns the TKET optimization actions.""" return [ @@ -70,6 +162,15 @@ def tket_optimization_actions() -> list[Action]: preserves_routing=False, preserves_synthesis=False, ), + DeviceIndependentAction( + "KAKDecomposition", + CompilationOrigin.TKET, + PassType.OPT, + [KAKDecomposition(allow_swaps=False)], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=False, + ), DeviceIndependentAction( "FullPeepholeOptimiseCX", CompilationOrigin.TKET, @@ -91,19 +192,40 @@ def tket_optimization_actions() -> list[Action]: ] +def tket_layout_actions() -> list[Action]: + """Return the TKET layout actions.""" + return [ + DeferredDeviceAction( + "GraphPlacement", + CompilationOrigin.TKET, + PassType.LAYOUT, + transpile_pass=lambda device: [ + GraphPlacement( + Architecture(list(device.build_coupling_map())), + timeout=5000, + maximum_matches=5000, + ) + ], + ), + DeferredDeviceAction( + "NoiseAwarePlacement", + CompilationOrigin.TKET, + PassType.LAYOUT, + transpile_pass=_noise_aware_placement, + ), + ] + + def tket_routing_action() -> Action: """Returns the TKET routing action.""" return DeferredDeviceAction( "RoutingPass", CompilationOrigin.TKET, PassType.ROUTING, - transpile_pass=lambda device: cast( - "list[Task]", - [ - PreProcessTKETRoutingAfterQiskitLayout(), - RoutingPass(Architecture(list(device.build_coupling_map()))), - ], - ), + transpile_pass=lambda device: [ + PreProcessTKETRoutingAfterQiskitLayout(), + RoutingPass(Architecture(list(device.build_coupling_map()))), + ], ) @@ -135,10 +257,43 @@ def run_tket_action( """Apply a TKET action and return the updated circuit and layout metadata.""" tket_qc = qiskit_to_tk(circuit, preserve_param_uuid=True) if callable(action.transpile_pass): - factory = cast("Callable[[Target], list[Task]]", action.transpile_pass) + factory = cast( + "Callable[[Target], list[TketBasePass | PreProcessTKETRoutingAfterQiskitLayout | Placement]]", + action.transpile_pass, + ) passes = factory(device) else: - passes = cast("list[Task]", action.transpile_pass) + passes = cast("list[TketBasePass | PreProcessTKETRoutingAfterQiskitLayout | Placement]", action.transpile_pass) + + if action.pass_type == PassType.LAYOUT: + if not passes or not isinstance(passes[0], Placement): + msg = f"TKET layout action {action.name} did not provide a placement pass." + raise TypeError(msg) + try: + placement = passes[0].get_placement_map(tket_qc) + except (RuntimeError, TypeError, ValueError) as error: + logger.warning("Placement failed (%s): %s.", action.name, error) + return circuit, layout + + qiskit_layout = _translate_placement(circuit, placement, action.name, device.num_qubits) + if qiskit_layout is None: + return circuit, layout + pass_manager = PassManager([ + SetLayout(qiskit_layout), + FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ]) + altered_qc = pass_manager.run(circuit) + applied_layout = cast("Layout", pass_manager.property_set["layout"]) + return altered_qc, TranspileLayout( + initial_layout=applied_layout, + input_qubit_mapping=pass_manager.property_set["original_qubit_indices"], + final_layout=pass_manager.property_set["final_layout"], + _output_qubit_list=altered_qc.qubits, + _input_qubit_count=circuit.num_qubits, + ) + for pass_ in passes: assert isinstance(pass_, TketBasePass | PreProcessTKETRoutingAfterQiskitLayout) pass_.apply(tket_qc) @@ -154,8 +309,10 @@ def run_tket_action( return altered_qc, layout -def is_tket_action_available(*, action: Action, has_layout: bool) -> bool: +def is_tket_action_available(*, action: Action, has_layout: bool, has_wide_operations: bool) -> bool: """Return whether a TKET action is available for the current layout state.""" + if has_wide_operations and action.pass_type in {PassType.LAYOUT, PassType.ROUTING}: + return False # TKET layout/optimization actions must not run after a Qiskit layout has been set # (it is not clear how tket will handle the layout). TKET routing actions, however, are # designed to work after a Qiskit layout via PreProcessTKETRoutingAfterQiskitLayout. diff --git a/src/mqt/predictor/rl/helper.py b/src/mqt/predictor/rl/helper.py index 44921e6d8..66cd7737b 100644 --- a/src/mqt/predictor/rl/helper.py +++ b/src/mqt/predictor/rl/helper.py @@ -11,6 +11,7 @@ from __future__ import annotations import logging +from math import log1p from pathlib import Path from typing import TYPE_CHECKING @@ -28,6 +29,7 @@ logger = logging.getLogger("mqt-predictor") +MAX_CIRCUIT_DEPTH = 999_999 OBSERVATION_OPERATIONS = tuple(get_openqasm_gates()) @@ -45,14 +47,14 @@ def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generat Raises: RuntimeError: If no quantum circuit could be read from the training circuits folder. """ - file_list = list(path_training_circuits.glob("*.qasm")) + file_list = sorted(path_training_circuits.glob("*.qasm")) path_zip = path_training_circuits / "training_data_compilation.zip" if len(file_list) == 0 and path_zip.exists(): with zipfile.ZipFile(str(path_zip), "r") as zip_ref: zip_ref.extractall(path_training_circuits) - file_list = list(path_training_circuits.glob("*.qasm")) + file_list = sorted(path_training_circuits.glob("*.qasm")) assert len(file_list) > 0 found_suitable_qc = False @@ -72,11 +74,13 @@ def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generat return qc, str(file_list[random_index]) -def create_feature_dict(qc: QuantumCircuit) -> dict[str, int | NDArray[np.float32]]: +def create_feature_dict(qc: QuantumCircuit, max_num_qubits: int) -> dict[str, NDArray[np.float32]]: """Create a normalized feature dictionary for a quantum circuit.""" operation_counts = dict(qc.count_ops()) total_operations = sum(value for gate, value in operation_counts.items() if gate != "barrier") - feature_dict: dict[str, int | NDArray[np.float32]] = { + normalized_num_qubits = min(qc.num_qubits, max_num_qubits) / max_num_qubits + normalized_depth = log1p(min(qc.depth(), MAX_CIRCUIT_DEPTH)) / log1p(MAX_CIRCUIT_DEPTH) + feature_dict = { **{ operation: np.array( [operation_counts.get(operation, 0) / total_operations if total_operations else 0.0], @@ -84,8 +88,8 @@ def create_feature_dict(qc: QuantumCircuit) -> dict[str, int | NDArray[np.float3 ) for operation in OBSERVATION_OPERATIONS }, - "num_qubits": qc.num_qubits, - "depth": qc.depth(), + "num_qubits": np.array([normalized_num_qubits], dtype=np.float32), + "depth": np.array([normalized_depth], dtype=np.float32), } supermarq_features = calc_supermarq_features(qc) diff --git a/src/mqt/predictor/rl/predictor.py b/src/mqt/predictor/rl/predictor.py index 4ff261a9f..44e315c64 100644 --- a/src/mqt/predictor/rl/predictor.py +++ b/src/mqt/predictor/rl/predictor.py @@ -72,26 +72,32 @@ def compile_as_predicted( self, qc: QuantumCircuit | str, tracer_output_path: str | Path | None = None, + pass_timeout: float | None = None, ) -> tuple[QuantumCircuit, list[str]]: """Compiles a given quantum circuit such that the given figure of merit is maximized by using the respectively trained optimized compiler. Arguments: qc: The quantum circuit to be compiled or the path to a qasm file containing the quantum circuit. tracer_output_path: Optional temporary path to export the compilation trace for this specific run. + pass_timeout: Maximum duration in seconds for one compilation pass. + Defaults to None, which disables pass timeouts. Returns: A tuple containing the compiled quantum circuit and the compilation information. If compilation fails, False is returned. Raises: RuntimeError: If an error occurs during compilation. + ValueError: If ``pass_timeout`` is not positive. """ original_tracer_output_path = self.env.tracer_output_path - - # Temporarily override singleton if a new path is explicitly provided - if tracer_output_path is not None: - self.env.tracer_output_path = tracer_output_path + original_pass_timeout = self.env.pass_timeout try: + # Temporarily override singleton settings for this compilation. + if tracer_output_path is not None: + self.env.tracer_output_path = tracer_output_path + self.env.pass_timeout = pass_timeout + trained_rl_model = load_model(self.model_name) obs, _ = self.env.reset(qc, seed=0) @@ -114,8 +120,8 @@ def compile_as_predicted( raise RuntimeError(msg) finally: - # Restore original singleton path self.env.tracer_output_path = original_tracer_output_path + self.env.pass_timeout = original_pass_timeout def train_model( self, @@ -123,6 +129,7 @@ def train_model( verbose: int = 2, test: bool = False, seed: int | None = None, + pass_timeout: float | None = None, ) -> None: """Trains all models for the given reward functions and device. @@ -132,6 +139,11 @@ def train_model( test: Whether to train the model for testing purposes. Defaults to False. seed: The random seed to use for reproducible training. Set to None to use true randomness. Defaults to None. + pass_timeout: Maximum duration in seconds for one compilation pass. + Defaults to None, which disables pass timeouts. + + Raises: + ValueError: If ``pass_timeout`` is not positive. """ if seed is not None: set_random_seed(seed) @@ -148,22 +160,27 @@ def train_model( batch_size = 64 progress_bar = True - logger.debug("Start training for: " + self.figure_of_merit + " on " + self.device_name) - model = MaskablePPO( - MaskableMultiInputActorCriticPolicy, - self.env, - verbose=verbose, - tensorboard_log=f"./{self.model_name}", - gamma=0.98, - n_steps=n_steps, - batch_size=batch_size, - n_epochs=n_epochs, - seed=seed, - ) - # Training Loop: In each iteration, the agent collects n_steps steps (rollout), - # updates the policy for n_epochs, and then repeats the process until total_timesteps steps have been taken. - model.learn(total_timesteps=timesteps, progress_bar=progress_bar) - model.save(get_path_trained_model() / self.model_name) + original_pass_timeout = self.env.pass_timeout + self.env.pass_timeout = pass_timeout + try: + logger.debug("Start training for: " + self.figure_of_merit + " on " + self.device_name) + model = MaskablePPO( + MaskableMultiInputActorCriticPolicy, + self.env, + verbose=verbose, + tensorboard_log=f"./{self.model_name}", + gamma=0.98, + n_steps=n_steps, + batch_size=batch_size, + n_epochs=n_epochs, + seed=seed, + ) + # Training Loop: In each iteration, the agent collects n_steps steps (rollout), + # updates the policy for n_epochs, and then repeats the process until total_timesteps steps have been taken. + model.learn(total_timesteps=timesteps, progress_bar=progress_bar) + model.save(get_path_trained_model() / self.model_name) + finally: + self.env.pass_timeout = original_pass_timeout def load_model(model_name: str) -> MaskablePPO: @@ -194,6 +211,7 @@ def rl_compile( predictor_singleton: Predictor | None = None, tracer_output_path: str | Path | None = None, mdp: MDPPolicy = "v3", + pass_timeout: float | None = None, ) -> tuple[QuantumCircuit, list[str]]: """Compiles a given quantum circuit to a device optimizing for the given figure of merit. @@ -206,12 +224,15 @@ def rl_compile( mdp: The MDP transition policy used when constructing a predictor. ``v2`` is the original strategy and ``v3`` is the default. When ``predictor_singleton`` is provided, its configured policy is used instead. + pass_timeout: Maximum duration in seconds for one compilation pass. + Defaults to None, which disables pass timeouts. Returns: A tuple containing the compiled quantum circuit and the compilation information. If compilation fails, False is returned. Raises: - ValueError: If figure_of_merit or device is None and predictor_singleton is also None. + ValueError: If figure_of_merit or device is None and predictor_singleton is also None, + or if ``pass_timeout`` is not positive. """ if predictor_singleton is None: if figure_of_merit is None: @@ -226,6 +247,8 @@ def rl_compile( tracer_output_path=tracer_output_path, mdp=mdp, ) - return predictor.compile_as_predicted(qc) + return predictor.compile_as_predicted(qc, pass_timeout=pass_timeout) - return predictor_singleton.compile_as_predicted(qc, tracer_output_path=tracer_output_path) + return predictor_singleton.compile_as_predicted( + qc, tracer_output_path=tracer_output_path, pass_timeout=pass_timeout + ) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 123f1b3f9..9d6280aaf 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -10,14 +10,20 @@ from __future__ import annotations +import contextlib import logging import re +import signal +import threading import time import warnings from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, get_args if TYPE_CHECKING: + from collections.abc import Iterator + from types import FrameType + from gymnasium.spaces import Space from qiskit.transpiler import Layout, Target @@ -28,7 +34,9 @@ from gymnasium.spaces import Box, Dict, Discrete from joblib import load from qiskit import QuantumCircuit -from qiskit.transpiler import CouplingMap, TranspileLayout +from qiskit.circuit import StandardEquivalenceLibrary +from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout +from qiskit.transpiler.passes import BasisTranslator from mqt.predictor.hellinger import get_hellinger_model_path from mqt.predictor.reward import ( @@ -60,6 +68,43 @@ MDP_POLICIES: frozenset[str] = frozenset(get_args(MDPPolicy)) +@contextlib.contextmanager +def _enforce_pass_timeout(pass_timeout: float | None) -> Iterator[None]: + if pass_timeout is None: + yield + return + + if not all(hasattr(signal, attribute) for attribute in ("SIGALRM", "ITIMER_REAL", "getitimer", "setitimer")): + warnings.warn("Pass timeouts are not supported on this platform.", RuntimeWarning, stacklevel=2) + yield + return + if threading.current_thread() is not threading.main_thread(): + warnings.warn("Pass timeouts are only supported on the main thread.", RuntimeWarning, stacklevel=2) + yield + return + + def timeout_handler(_signum: int, _frame: FrameType | None) -> None: + msg = f"Compilation pass exceeded the timeout of {pass_timeout:g} seconds." + raise TimeoutError(msg) + + previous_delay, previous_interval = signal.getitimer(signal.ITIMER_REAL) + if 0 < previous_delay <= pass_timeout: + yield + return + + previous_handler = signal.signal(signal.SIGALRM, timeout_handler) + start_time = time.monotonic() + try: + signal.setitimer(signal.ITIMER_REAL, pass_timeout) + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, previous_handler) + if previous_delay > 0: + remaining_delay = max(previous_delay - (time.monotonic() - start_time), 1e-6) + signal.setitimer(signal.ITIMER_REAL, remaining_delay, previous_interval) + + class PredictorEnv(Env): """Predictor environment for reinforcement learning.""" @@ -71,6 +116,8 @@ def __init__( max_steps: int | None = None, tracer_output_path: str | Path | None = None, mdp: MDPPolicy = "v3", + stochastic_action_trials: int = 20, + pass_timeout: float | None = None, ) -> None: """Initializes the PredictorEnv object. @@ -83,9 +130,14 @@ def __init__( mdp: The MDP transition policy. ``v2`` is the original MQT Predictor strategy. ``v3`` is the default and is permissive before layout while preserving the established compilation structure afterwards. + stochastic_action_trials: Number of attempts used to select the best + result from a stochastic action. + pass_timeout: Maximum duration in seconds for one compilation pass. + Defaults to None, which disables pass timeouts. Raises: - ValueError: If ``mdp`` is unsupported, if the reward function is + ValueError: If ``mdp`` is unsupported, ``pass_timeout`` is not positive, + if the reward function is "estimated_success_probability" and no calibration data is available for the device, or if the reward function is "estimated_hellinger_distance" and no trained model is available for @@ -96,10 +148,15 @@ def __init__( if mdp not in MDP_POLICIES: msg = f"Unsupported MDP policy: {mdp}." raise ValueError(msg) + if stochastic_action_trials < 1: + msg = "stochastic_action_trials must be at least one." + raise ValueError(msg) self.path_training_circuits = path_training_circuits or get_path_training_circuits() self.max_steps = max_steps self.mdp = mdp + self.stochastic_action_trials = stochastic_action_trials + self.pass_timeout = pass_timeout self.action_set = {} self.actions_synthesis_indices = [] @@ -177,14 +234,13 @@ def __init__( self.layout: TranspileLayout | None = None self.has_parameterized_gates = False - self.rng = np.random.default_rng(10) operation_spaces = { operation: Box(low=0, high=1, shape=(1,), dtype=np.float32) for operation in OBSERVATION_OPERATIONS } spaces: dict[str, Space] = { - "num_qubits": Discrete(128), - "depth": Discrete(1000000), + "num_qubits": Box(low=0, high=1, shape=(1,), dtype=np.float32), + "depth": Box(low=0, high=1, shape=(1,), dtype=np.float32), **operation_spaces, "program_communication": Box(low=0, high=1, shape=(1,), dtype=np.float32), "critical_depth": Box(low=0, high=1, shape=(1,), dtype=np.float32), @@ -202,6 +258,18 @@ def __init__( self.observation_space = Dict(spaces) self.filename = "" + @property + def pass_timeout(self) -> float | None: + """The current per-pass timeout in seconds.""" + return self._pass_timeout + + @pass_timeout.setter + def pass_timeout(self, pass_timeout: float | None) -> None: + if pass_timeout is not None and pass_timeout <= 0: + msg = "pass_timeout must be positive." + raise ValueError(msg) + self._pass_timeout = pass_timeout + def _collect_tracer_data( self, step_index: int, @@ -291,13 +359,14 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any start_time = time.perf_counter() try: self.used_actions.append(action_name) - altered_qc = self.apply_action(action) + with _enforce_pass_timeout(self.pass_timeout): + altered_qc = self.apply_action(action) action_duration = time.perf_counter() - start_time except Exception as exc: # ruff:ignore[blind-except] action_duration = time.perf_counter() - start_time # Different passes may fail for various reasons (e.g., found no routing solution). self.error_occurred = True - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Trace the error before aborting self._collect_tracer_data( @@ -337,7 +406,7 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any reward_val = 0 done = False - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Trace+truncate if step limit is reached if not done and self.max_steps is not None and self.num_steps >= self.max_steps: @@ -419,7 +488,11 @@ def reset( self.filename = str(qc) current_circuit_name = Path(str(qc)).stem else: - self.state, self.filename = get_state_sample(self.device.num_qubits, self.path_training_circuits, self.rng) + self.state, self.filename = get_state_sample( + self.device.num_qubits, + self.path_training_circuits, + self.np_random, + ) current_circuit_name = Path(self.filename).stem self.action_space = Discrete(len(self.action_set.keys())) @@ -447,7 +520,7 @@ def reset( self.num_qubits_uncompiled_circuit = self.state.num_qubits self.has_parameterized_gates = len(self.state.parameters) > 0 - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Setup tracer for the new episode if self.tracer_output_path is not None: @@ -493,6 +566,7 @@ def action_masks(self) -> list[bool]: A dense boolean mask ordered like ``self.action_set``. """ has_layout = self.layout is not None + has_wide_operations = any(len(instruction.qubits) > 2 for instruction in self.state.data) valid_action_indices = set(self.valid_actions) action_mask: list[bool] = [] @@ -508,7 +582,13 @@ def action_masks(self) -> list[bool]: if action.origin == CompilationOrigin.QISKIT: action_mask.append(is_qiskit_action_available(action, self.device)) elif action.origin == CompilationOrigin.TKET: - action_mask.append(is_tket_action_available(action=action, has_layout=has_layout)) + action_mask.append( + is_tket_action_available( + action=action, + has_layout=has_layout, + has_wide_operations=has_wide_operations, + ) + ) elif action.origin == CompilationOrigin.BQSKIT: action_mask.append( is_bqskit_action_available( @@ -549,6 +629,8 @@ def apply_action(self, action_index: int) -> QuantumCircuit: device=self.device, layout=self.layout, input_qubit_count=self.num_qubits_uncompiled_circuit, + stochastic_action_trials=self.stochastic_action_trials, + score=self._score_circuit if action.stochastic else None, ) elif action.origin == CompilationOrigin.TKET: altered_qc, self.layout = run_tket_action( @@ -570,6 +652,22 @@ def apply_action(self, action_index: int) -> QuantumCircuit: return altered_qc + def _score_circuit(self, circuit: QuantumCircuit) -> float: + """Calculate the configured figure of merit for an action candidate.""" + scoring_circuit = PassManager([ + BasisTranslator(StandardEquivalenceLibrary, target_basis=self.device.operation_names) + ]).run(circuit.copy()) + if self.reward_function == "expected_fidelity": + return expected_fidelity(scoring_circuit, self.device) + if self.reward_function == "estimated_success_probability": + return estimated_success_probability(scoring_circuit, self.device) + if self.reward_function == "estimated_hellinger_distance": + return estimated_hellinger_distance(scoring_circuit, self.device, self.hellinger_model) + if self.reward_function == "critical_depth": + return crit_depth(scoring_circuit) + msg = f"No implementation for reward function {self.reward_function}." + raise NotImplementedError(msg) + def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | Layout) -> bool: """True if every logical qubit in the circuit has a physical assignment.""" if isinstance(layout, TranspileLayout): diff --git a/src/mqt/predictor/rl/tracer.py b/src/mqt/predictor/rl/tracer.py index 171b1c323..868899fd6 100644 --- a/src/mqt/predictor/rl/tracer.py +++ b/src/mqt/predictor/rl/tracer.py @@ -209,7 +209,7 @@ def record_step( reward: float, current_qc: QuantumCircuit, figures_of_merit: FigureOfMeritMetrics, - features: dict[str, int | NDArray[np.float32]], + features: dict[str, NDArray[np.float32]], synthesized: bool, laid_out: bool, routed: bool, @@ -337,8 +337,6 @@ def _extract_device_metadata(device: Target) -> DeviceMetadata: ) @staticmethod - def _extract_float(val: int | NDArray[np.float32]) -> float: - """Extract a float from a scalar or one-element feature array.""" - if isinstance(val, int): - return float(val) + def _extract_float(val: NDArray[np.float32]) -> float: + """Extract a float from a one-element feature array.""" return float(val.item()) diff --git a/tests/compilation/test_helper_rl.py b/tests/compilation/test_helper_rl.py index 8a87031bf..0749f0830 100644 --- a/tests/compilation/test_helper_rl.py +++ b/tests/compilation/test_helper_rl.py @@ -45,22 +45,20 @@ def test_create_feature_dict() -> None: qc.ccx(0, 1, 2) qc.measure_all() - features = create_feature_dict(qc) + max_num_qubits = 20 + features = create_feature_dict(qc, max_num_qubits) - for name, feature in features.items(): - if name in {"num_qubits", "depth"}: - assert isinstance(feature, int) - else: - assert isinstance(feature, np.ndarray) - assert feature.dtype == np.float32 + for feature in features.values(): + assert isinstance(feature, np.ndarray) + assert feature.dtype == np.float32 np.testing.assert_allclose(features["h"], [1 / 6]) np.testing.assert_allclose(features["cx"], [1 / 6]) np.testing.assert_allclose(features["ccx"], [1 / 6]) np.testing.assert_allclose(features["measure"], [0.5]) np.testing.assert_allclose(features["x"], [0]) - assert features["num_qubits"] == qc.num_qubits - assert features["depth"] == qc.depth() + np.testing.assert_allclose(features["num_qubits"], [3 / max_num_qubits]) + np.testing.assert_allclose(features["depth"], [np.log1p(qc.depth()) / np.log1p(999_999)]) def test_get_path_trained_model() -> None: diff --git a/tests/compilation/test_predictor_rl.py b/tests/compilation/test_predictor_rl.py index af2825d89..326ef0e89 100644 --- a/tests/compilation/test_predictor_rl.py +++ b/tests/compilation/test_predictor_rl.py @@ -11,6 +11,9 @@ from __future__ import annotations import re +import signal +import time +from importlib import import_module from pathlib import Path from typing import TYPE_CHECKING, cast @@ -24,6 +27,7 @@ from qiskit.transpiler.passes import GatesInBasis from mqt.predictor.rl import Predictor, rl_compile +from mqt.predictor.rl import predictor as predictor_module from mqt.predictor.rl import predictorenv as predictorenv_module from mqt.predictor.rl.actions import ( CompilationOrigin, @@ -50,8 +54,8 @@ def test_predictor_env_reset_from_string() -> None: dump(qc, f) observation, _ = predictor.env.reset(qc=qasm_path) - assert observation == create_feature_dict(qc) - assert observation["num_qubits"] == qc.num_qubits + assert observation == create_feature_dict(qc, device.num_qubits) + assert observation["num_qubits"][0] == pytest.approx(qc.num_qubits / device.num_qubits) def test_predictor_env_esp_error() -> None: @@ -80,6 +84,122 @@ def test_predictor_env_rejects_unsupported_mdp() -> None: predictorenv_module.PredictorEnv(device=get_device("ibm_falcon_27"), mdp=invalid_mdp) +@pytest.mark.parametrize("pass_timeout", [0, -1]) +def test_predictor_env_rejects_nonpositive_pass_timeout(pass_timeout: float) -> None: + """Test that pass timeouts must be positive when enabled.""" + with pytest.raises(ValueError, match=re.escape("pass_timeout must be positive.")): + predictorenv_module.PredictorEnv(device=get_device("ibm_falcon_27"), pass_timeout=pass_timeout) + + +@pytest.mark.skipif(not hasattr(signal, "SIGALRM"), reason="SIGALRM is unavailable") +def test_predictor_env_truncates_timed_out_pass(monkeypatch: pytest.MonkeyPatch) -> None: + """Test that a pass timeout truncates the current episode.""" + env = predictorenv_module.PredictorEnv(device=get_device("ibm_falcon_27"), pass_timeout=0.01) + qc = QuantumCircuit(1) + env.reset(qc) + monkeypatch.setattr(env, "apply_action", lambda _action: time.sleep(1)) + + _, reward_val, terminated, truncated, info = env.step(env.actions_opt_indices[0]) + + assert reward_val == 0 + assert not terminated + assert truncated + assert info == { + "Truncated because of error": "TimeoutError: Compilation pass exceeded the timeout of 0.01 seconds." + } + + +def test_training_uses_temporary_pass_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """Test that training applies and restores its pass timeout.""" + observed_timeouts: list[float | None] = [] + + class FakeModel: + """Minimal MaskablePPO replacement that records the environment timeout.""" + + def __init__( + self, + _policy: object, + env: predictorenv_module.PredictorEnv, + **_kwargs: object, + ) -> None: + self.env = env + + def learn(self, *, total_timesteps: int, progress_bar: bool) -> None: + assert total_timesteps == 2 + assert not progress_bar + observed_timeouts.append(self.env.pass_timeout) + + def save(self, _path: Path) -> None: + pass + + monkeypatch.setattr(predictor_module, "MaskablePPO", FakeModel) + predictor = Predictor(figure_of_merit="expected_fidelity", device=get_device("ibm_falcon_27")) + + predictor.train_model(timesteps=2, test=True, pass_timeout=2) + + assert observed_timeouts == [2] + assert predictor.env.pass_timeout is None + + +def test_inference_uses_temporary_pass_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """Test that inference applies and restores its independent pass timeout.""" + observed_timeouts: list[float | None] = [] + qc = QuantumCircuit(1) + predictor = Predictor(figure_of_merit="expected_fidelity", device=get_device("ibm_falcon_27")) + + class FakeModel: + """Minimal trained model replacement.""" + + def predict(self, _obs: object, *, action_masks: object) -> tuple[int, None]: + assert action_masks == [] + return predictor.env.actions_opt_indices[0], None + + def fake_step(_action: int) -> tuple[dict[str, object], float, bool, bool, dict[str, object]]: + observed_timeouts.append(predictor.env.pass_timeout) + predictor.env.state = qc + return {}, 0, True, False, {} + + def fake_reset(_qc: QuantumCircuit | str, seed: int) -> tuple[dict[str, object], dict[str, object]]: + assert seed == 0 + predictor.env.error_occurred = False + return {}, {} + + monkeypatch.setattr(predictor_module, "load_model", lambda _model_name: FakeModel()) + monkeypatch.setattr(predictor_module, "get_action_masks", lambda _env: []) + monkeypatch.setattr(predictor.env, "reset", fake_reset) + monkeypatch.setattr(predictor.env, "step", fake_step) + + compiled_qc, _passes = predictor.compile_as_predicted(qc, pass_timeout=0.5) + + assert compiled_qc is qc + assert observed_timeouts == [0.5] + assert predictor.env.pass_timeout is None + + +def test_qcompile_forwards_inference_pass_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """Test that the top-level inference API forwards its pass timeout.""" + qcompile_module = import_module("mqt.predictor.qcompile") + device = get_device("ibm_falcon_27") + qc = QuantumCircuit(1) + + def fake_predict_device(_qc: QuantumCircuit, figure_of_merit: str) -> Target: + assert figure_of_merit == "expected_fidelity" + return device + + def fake_rl_compile(circuit: QuantumCircuit, **kwargs: object) -> tuple[QuantumCircuit, list[str]]: + assert kwargs["pass_timeout"] == 3 + return circuit, [] + + monkeypatch.setattr(qcompile_module, "predict_device_for_figure_of_merit", fake_predict_device) + monkeypatch.setattr(qcompile_module, "rl_compile", fake_rl_compile) + + compiled_qc, compilation_info, selected_device = qcompile_module.qcompile(qc, pass_timeout=3) + + assert compiled_qc is qc + assert compilation_info == [] + assert selected_device is device + + @pytest.mark.parametrize( ("mdp", "expected_action_groups"), [ diff --git a/tests/compilation/test_tracer.py b/tests/compilation/test_tracer.py index 6911a8657..772eb1b8e 100644 --- a/tests/compilation/test_tracer.py +++ b/tests/compilation/test_tracer.py @@ -77,8 +77,8 @@ def test_compilation_tracer_generates_valid_json(tmp_path: Path) -> None: qc_terminal.rz(0.5, 2) # 2. Mock the feature dictionaries that the RL environment normally passes - features_baseline = create_feature_dict(qc_baseline) - features_terminal = create_feature_dict(qc_terminal) + features_baseline = create_feature_dict(qc_baseline, device.num_qubits) + features_terminal = create_feature_dict(qc_terminal, device.num_qubits) # 3. Use the actual `record_step` API instead of manually building CompilationSteps tracer.record_step(