diff --git a/.gitignore b/.gitignore index 17b4dd8c9..455a1bf13 100644 --- a/.gitignore +++ b/.gitignore @@ -53,3 +53,6 @@ venv.bak/ .mypy_cache/ src/**/_version.py + +# RL training checkpoints and logs +events.out.* diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a7b727cd..c837e4992 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning], with the exception that minor rel ### Changed +- ♻️ Restructure existing RL passes into SDK-level action modules ([#680]) ([**@flowerthrower**]) - 🎨 Improve the RL state machine logic ([#677]) ([**@flowerthrower**]) - 🐛 Support BQSKit conversion of IQM's native `r` gate ([#679]) ([**@flowerthrower**]) - 🔧 Replace `mypy` with `ty` ([#572]) ([**@denialhaag**]) @@ -48,6 +49,7 @@ _📚 Refer to the [GitHub Release Notes](https://github.com/munich-quantum-tool +[#680]: https://github.com/munich-quantum-toolkit/predictor/pull/680 [#677]: https://github.com/munich-quantum-toolkit/predictor/pull/677 [#679]: https://github.com/munich-quantum-toolkit/predictor/pull/679 [#572]: https://github.com/munich-quantum-toolkit/predictor/pull/572 diff --git a/src/mqt/predictor/hellinger/utils.py b/src/mqt/predictor/hellinger/utils.py index 3383f0ace..d50454c0e 100644 --- a/src/mqt/predictor/hellinger/utils.py +++ b/src/mqt/predictor/hellinger/utils.py @@ -39,7 +39,7 @@ def calc_device_specific_features( ) -> NDArray[np.float64]: """Creates and returns a device-specific feature vector for a given quantum circuit and device. - Arguments: + Args: qc: The quantum circuit for which the features are calculated. device: The device for which the features are calculated. ignore_gates: A list of gates to be ignored when calculating the features. Defaults to ["barrier", "id", "measure", "if_else"]. diff --git a/src/mqt/predictor/ml/helper.py b/src/mqt/predictor/ml/helper.py index d73a7d85c..22dc15d5d 100644 --- a/src/mqt/predictor/ml/helper.py +++ b/src/mqt/predictor/ml/helper.py @@ -112,7 +112,7 @@ def dict_to_featurevector(gate_dict: dict[str, int]) -> dict[str, int]: def create_feature_vector(qc: QuantumCircuit) -> list[int | float]: """Creates and returns a feature dictionary for a given quantum circuit. - Arguments: + Args: qc: The quantum circuit to be compiled. Returns: diff --git a/src/mqt/predictor/ml/predictor.py b/src/mqt/predictor/ml/predictor.py index 8863f337e..664ce5e0f 100644 --- a/src/mqt/predictor/ml/predictor.py +++ b/src/mqt/predictor/ml/predictor.py @@ -66,7 +66,7 @@ def setup_device_predictor( ) -> bool: """Sets up the device predictor for the given figure of merit. - Arguments: + Args: devices: The devices to be used for training. figure_of_merit: The figure of merit to be used for training. Defaults to "expected_fidelity". path_uncompiled_circuits: The path to the directory containing the circuits to be compiled. Defaults to None. @@ -127,7 +127,7 @@ def __init__( ) -> None: """Initializes the Predictor class. - Arguments: + Args: figure_of_merit: The figure of merit to be used for training. devices: The devices to be used for training. logger_level: The level of the logger. Defaults to logging.INFO. @@ -151,7 +151,7 @@ def _compile_all_circuits_devicewise( ) -> None: """Compiles all circuits in the given directory with the given timeout and saves them in the given directory. - Arguments: + Args: device: The device to be used for compilation. timeout: The timeout in seconds for the compilation of a single circuit. path_uncompiled_circuits: The path to the directory containing the circuits to be compiled. Defaults to None. @@ -204,7 +204,7 @@ def compile_training_circuits( ) -> None: """Compiles all circuits in the given directory with the given timeout and saves them in the given directory. - Arguments: + Args: path_uncompiled_circuits: The path to the directory containing the circuits to be compiled. Defaults to None. path_compiled_circuits: The path to the directory where the compiled circuits should be saved. Defaults to None. timeout: The timeout in seconds for the compilation of a single circuit. Defaults to 600. @@ -237,7 +237,7 @@ def generate_training_data( ) -> None: """Creates and saves training data from all generated training samples. - Arguments: + Args: path_uncompiled_circuits: The path to the directory containing the uncompiled circuits. Defaults to None. path_compiled_circuits: The path to the directory containing the compiled circuits. Defaults to None. path_training_data: The path to the directory where the generated training data should be saved. Defaults to None. @@ -295,7 +295,7 @@ def _generate_training_sample( ) -> tuple[tuple[list[Any], Any], str, list[float]]: """Handles to create a training sample from a given file. - Arguments: + Args: file: The name of the file to be used for training. path_uncompiled_circuit: The path to the directory containing the uncompiled circuits. Defaults to None. path_compiled_circuits: The path to the directory containing the compiled circuits. Defaults to None. @@ -365,7 +365,7 @@ def train_random_forest_model( ) -> RandomForestRegressor | RandomForestClassifier: """Trains a random forest model for the given figure of merit. - Arguments: + Args: training_data: The training data to be used for training the model. If None, the training data is loaded from the pre-prepared training data files. Returns: @@ -453,7 +453,7 @@ def predict_device_for_figure_of_merit( ) -> Target: """Returns the probabilities for all supported quantum devices to be the most suitable one for the given quantum circuit. - Arguments: + Args: qc: The QuantumCircuit or Path to the respective qasm file. figure_of_merit: The figure of merit to be used for compilation. diff --git a/src/mqt/predictor/qcompile.py b/src/mqt/predictor/qcompile.py index d65be982e..91cd88a54 100644 --- a/src/mqt/predictor/qcompile.py +++ b/src/mqt/predictor/qcompile.py @@ -27,7 +27,7 @@ def qcompile( ) -> tuple[QuantumCircuit, list[str], str]: """Compiles a given quantum circuit to a device with the highest predicted figure of merit. - Arguments: + Args: qc: The quantum circuit to be compiled. figure_of_merit: The figure of merit to be used for compilation. Defaults to "expected_fidelity". diff --git a/src/mqt/predictor/reward.py b/src/mqt/predictor/reward.py index e8c1b4742..2009eba36 100644 --- a/src/mqt/predictor/reward.py +++ b/src/mqt/predictor/reward.py @@ -46,7 +46,7 @@ def crit_depth(qc: QuantumCircuit, precision: int = 10) -> float: def expected_fidelity(qc: QuantumCircuit, device: Target, precision: int = 10) -> float: """Calculates the expected fidelity of a given quantum circuit on a given device. - Arguments: + Args: qc: The quantum circuit to be compiled. device: The device to be used for compilation. precision: The precision of the returned value. Defaults to 10. @@ -83,7 +83,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: It is calculated by multiplying the expected fidelity with a min(T1,T2)-dependent decay factor during qubit idle times. To this end, the circuit is scheduled using ASAP scheduling. - Arguments: + Args: qc: The quantum circuit to be compiled. device: The device to be used for compilation. precision: The precision of the returned value. Defaults to 10. @@ -210,7 +210,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: def esp_data_available(device: Target) -> bool: """Check if calibration data to calculate ESP is available for the device. - Arguments: + Args: device: The device to be checked for calibration data. Returns: @@ -310,7 +310,7 @@ def estimated_hellinger_distance( ) -> float: """Calculates the estimated Hellinger distance of a given quantum circuit on a given device. - Arguments: + Args: qc: The quantum circuit to be compiled. device: The device to be used for compilation. model: The pre-trained model to use for prediction (optional). If not provided, the model will try to be loaded from files. diff --git a/src/mqt/predictor/rl/actions.py b/src/mqt/predictor/rl/actions.py deleted file mode 100644 index 598dfe0e6..000000000 --- a/src/mqt/predictor/rl/actions.py +++ /dev/null @@ -1,487 +0,0 @@ -# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM -# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH -# All rights reserved. -# -# SPDX-License-Identifier: MIT -# -# Licensed under the MIT License - -"""This module provides the actions that can be used in the reinforcement learning environment.""" - -from __future__ import annotations - -import os -from collections import defaultdict -from dataclasses import dataclass -from enum import Enum -from typing import TYPE_CHECKING, cast - -from bqskit import MachineModel -from bqskit import compile as bqskit_compile -from pytket.architecture import Architecture -from pytket.passes import ( - CliffordSimp, - FullPeepholeOptimise, - PeepholeOptimise2Q, - RemoveRedundancies, - RoutingPass, -) -from qiskit.circuit import StandardEquivalenceLibrary -from qiskit.circuit.library import ( - CXGate, - CYGate, - CZGate, - ECRGate, - HGate, - SdgGate, - SGate, - SwapGate, - SXdgGate, - SXGate, - TdgGate, - TGate, - XGate, - YGate, - ZGate, -) -from qiskit.passmanager import ConditionalController -from qiskit.transpiler import ( - CouplingMap, -) -from qiskit.transpiler.passes import ( - ApplyLayout, - BasisTranslator, - Collect2qBlocks, - CommutativeCancellation, - CommutativeInverseCancellation, - ConsolidateBlocks, - DenseLayout, - Depth, - EnlargeWithAncilla, - FixedPoint, - FullAncillaAllocation, - GatesInBasis, - InverseCancellation, - MinimumPoint, - Optimize1qGatesDecomposition, - OptimizeCliffords, - RemoveDiagonalGatesBeforeMeasure, - SabreLayout, - Size, - UnitarySynthesis, - VF2Layout, - VF2PostLayout, -) -from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason -from qiskit.transpiler.preset_passmanagers import common - -from mqt.predictor.rl.parsing import ( - PreProcessTKETRoutingAfterQiskitLayout, - get_bqskit_native_gates, -) - -if TYPE_CHECKING: - from collections.abc import Callable - from typing import Any - - from bqskit import Circuit - from pytket._tket.passes import BasePass as TketBasePass - from qiskit.passmanager.base_tasks import Task - - TaskList = list[Task | TketBasePass | PreProcessTKETRoutingAfterQiskitLayout] - from qiskit.passmanager import PropertySet - - -class CompilationOrigin(str, Enum): - """Enumeration of the origin of the compilation action.""" - - QISKIT = "qiskit" - TKET = "tket" - BQSKIT = "bqskit" - GENERAL = "general" - - -class PassType(str, Enum): - """Enumeration of the type of compilation pass.""" - - OPT = "optimization" - SYNTHESIS = "synthesis" - MAPPING = "mapping" - LAYOUT = "layout" - ROUTING = "routing" - FINAL_OPT = "final_optimization" - TERMINATE = "terminate" - - -@dataclass -class Action: - """Base class for all actions in the reinforcement learning environment.""" - - name: str - origin: CompilationOrigin - pass_type: PassType - transpile_pass: ( - TaskList - | Callable[..., TaskList] - | Callable[ - ..., - Callable[..., tuple[Any, ...] | Circuit], - ] - ) - - -@dataclass -class DeviceIndependentAction(Action): - """Action that represents a static compilation pass that can be applied directly.""" - - -@dataclass -class DeviceDependentAction(Action): - """Action that represents a device-specific compilation pass that can be applied to a specific device.""" - - transpile_pass: ( - Callable[..., TaskList] - | Callable[ - ..., - Callable[..., tuple[Any, ...] | Circuit], - ] - ) - do_while: Callable[[PropertySet], bool] | None = None - - -# Registry of actions -_ACTIONS: dict[str, Action] = {} - - -def register_action(action: Action) -> Action: - """Registers a new action in the global actions registry. - - Raises: - ValueError: If an action with the same name is already registered. - """ - if action.name in _ACTIONS: - msg = f"Action with name {action.name} already registered." - raise ValueError(msg) - _ACTIONS[action.name] = action - return action - - -def remove_action(name: str) -> None: - """Removes an action from the global actions registry by name. - - Raises: - ValueError: If no action with the given name is registered. - """ - if name not in _ACTIONS: - msg = f"No action with name {name} is registered." - raise KeyError(msg) - del _ACTIONS[name] - - -register_action( - DeviceIndependentAction( - "Optimize1qGatesDecomposition", - CompilationOrigin.QISKIT, - PassType.OPT, - [Optimize1qGatesDecomposition()], - ) -) - -register_action( - DeviceIndependentAction( - "CommutativeCancellation", - CompilationOrigin.QISKIT, - PassType.OPT, - [CommutativeCancellation()], - ) -) - -register_action( - DeviceIndependentAction( - "CommutativeInverseCancellation", - CompilationOrigin.QISKIT, - PassType.OPT, - [CommutativeInverseCancellation()], - ) -) - -register_action( - DeviceIndependentAction( - "RemoveDiagonalGatesBeforeMeasure", - CompilationOrigin.QISKIT, - PassType.OPT, - [RemoveDiagonalGatesBeforeMeasure()], - ) -) - -register_action( - DeviceIndependentAction( - "InverseCancellation", - CompilationOrigin.QISKIT, - PassType.OPT, - [ - InverseCancellation([ - CXGate(), - ECRGate(), - CZGate(), - CYGate(), - XGate(), - YGate(), - ZGate(), - HGate(), - SwapGate(), - (TGate(), TdgGate()), - (SGate(), SdgGate()), - (SXGate(), SXdgGate()), - ]) - ], - ) -) - -register_action( - DeviceIndependentAction( - "OptimizeCliffords", - CompilationOrigin.QISKIT, - PassType.OPT, - [OptimizeCliffords()], - ) -) - -register_action( - DeviceIndependentAction( - "Opt2qBlocks", - CompilationOrigin.QISKIT, - PassType.OPT, - [Collect2qBlocks(), ConsolidateBlocks(), UnitarySynthesis()], - ) -) - -register_action( - DeviceIndependentAction( - "PeepholeOptimise2Q", - CompilationOrigin.TKET, - PassType.OPT, - [PeepholeOptimise2Q()], - ) -) - -register_action( - DeviceIndependentAction( - "CliffordSimp", - CompilationOrigin.TKET, - PassType.OPT, - [CliffordSimp()], - ) -) - -register_action( - DeviceIndependentAction( - "FullPeepholeOptimiseCX", - CompilationOrigin.TKET, - PassType.OPT, - [FullPeepholeOptimise()], - ) -) - -register_action( - DeviceIndependentAction( - "RemoveRedundancies", - CompilationOrigin.TKET, - PassType.OPT, - [RemoveRedundancies()], - ) -) - -register_action( - DeviceDependentAction( - "QiskitO3", - CompilationOrigin.QISKIT, - PassType.OPT, - transpile_pass=lambda native_gate, coupling_map: cast( - "TaskList", - [ - Collect2qBlocks(), - ConsolidateBlocks(basis_gates=native_gate), - UnitarySynthesis(basis_gates=native_gate, coupling_map=coupling_map), - Optimize1qGatesDecomposition(basis=native_gate), - CommutativeCancellation(basis_gates=native_gate), - GatesInBasis(native_gate), - ConditionalController( - common.generate_translation_passmanager( - target=None, basis_gates=native_gate, coupling_map=coupling_map - ).to_flow_controller(), - condition=lambda property_set: not property_set["all_gates_in_basis"], - ), - Depth(recurse=True), - FixedPoint("depth"), - Size(recurse=True), - FixedPoint("size"), - MinimumPoint(["depth", "size"], "optimization_loop"), - ], - ), - do_while=lambda property_set: not property_set["optimization_loop_minimum_point"], - ) -) - -register_action( - DeviceDependentAction( - "BQSKitO2", - CompilationOrigin.BQSKIT, - PassType.OPT, - transpile_pass=lambda circuit: bqskit_compile( - circuit, - optimization_level=1 if os.getenv("GITHUB_ACTIONS") == "true" else 2, - synthesis_epsilon=1e-1 if os.getenv("GITHUB_ACTIONS") == "true" else 1e-8, - max_synthesis_size=3, - seed=10, - num_workers=1 if os.getenv("GITHUB_ACTIONS") == "true" else -1, - ), - ) -) - -register_action( - DeviceDependentAction( - "VF2PostLayout", - CompilationOrigin.QISKIT, - PassType.FINAL_OPT, - transpile_pass=lambda device: VF2PostLayout(target=device), - ) -) - -register_action( - DeviceDependentAction( - "DenseLayout", - CompilationOrigin.QISKIT, - PassType.LAYOUT, - transpile_pass=lambda device: cast( - "TaskList", - [ - DenseLayout(coupling_map=CouplingMap(device.build_coupling_map())), - FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), - EnlargeWithAncilla(), - ApplyLayout(), - ], - ), - ) -) - -register_action( - DeviceDependentAction( - "VF2Layout", - CompilationOrigin.QISKIT, - PassType.LAYOUT, - transpile_pass=lambda device: cast( - "TaskList", - [ - VF2Layout(target=device), - ConditionalController( - [ - FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), - EnlargeWithAncilla(), - ApplyLayout(), - ], - condition=lambda property_set: ( - property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND - ), - ), - ], - ), - ) -) - -register_action( - DeviceDependentAction( - "RoutingPass", - CompilationOrigin.TKET, - PassType.ROUTING, - transpile_pass=lambda device: cast( - "TaskList", - [ - PreProcessTKETRoutingAfterQiskitLayout(), - RoutingPass(Architecture(list(device.build_coupling_map()))), - ], - ), - ) -) - -register_action( - DeviceDependentAction( - "SabreMapping", - CompilationOrigin.QISKIT, - PassType.MAPPING, - transpile_pass=lambda device: cast( - "TaskList", [SabreLayout(coupling_map=CouplingMap(device.build_coupling_map()), skip_routing=False)] - ), - ) -) - -register_action( - DeviceDependentAction( - "BQSKitMapping", - CompilationOrigin.BQSKIT, - PassType.MAPPING, - transpile_pass=lambda device: ( - lambda bqskit_circuit: bqskit_compile( - bqskit_circuit, - model=MachineModel( - num_qudits=device.num_qubits, - gate_set=get_bqskit_native_gates(device), - coupling_graph=[(elem[0], elem[1]) for elem in device.build_coupling_map()], - ), - with_mapping=True, - optimization_level=1 if os.getenv("GITHUB_ACTIONS") == "true" else 2, - synthesis_epsilon=1e-1 if os.getenv("GITHUB_ACTIONS") == "true" else 1e-8, - max_synthesis_size=3, - seed=10, - num_workers=1 if os.getenv("GITHUB_ACTIONS") == "true" else -1, - ) - ), - ) -) - -register_action( - DeviceDependentAction( - "BasisTranslator", - CompilationOrigin.QISKIT, - PassType.SYNTHESIS, - transpile_pass=lambda device: cast( - "TaskList", [BasisTranslator(StandardEquivalenceLibrary, target_basis=device.operation_names)] - ), - ) -) - -register_action( - DeviceDependentAction( - "BQSKitSynthesis", - CompilationOrigin.BQSKIT, - PassType.SYNTHESIS, - transpile_pass=lambda device: ( - lambda bqskit_circuit: bqskit_compile( - bqskit_circuit, - model=MachineModel(bqskit_circuit.num_qudits, gate_set=get_bqskit_native_gates(device)), - optimization_level=1 if os.getenv("GITHUB_ACTIONS") == "true" else 2, - synthesis_epsilon=1e-1 if os.getenv("GITHUB_ACTIONS") == "true" else 1e-8, - max_synthesis_size=3, - seed=10, - num_workers=1 if os.getenv("GITHUB_ACTIONS") == "true" else -1, - ) - ), - ) -) - -register_action( - DeviceIndependentAction( - "terminate", - CompilationOrigin.GENERAL, - PassType.TERMINATE, - transpile_pass=[], - ) -) - - -def get_actions_by_pass_type() -> dict[PassType, list[Action]]: - """Returns a dictionary mapping each PassType to a list of Actions of that type.""" - result: dict[PassType, list[Action]] = defaultdict(list) - for action in _ACTIONS.values(): - result[action.pass_type].append(action) - return result diff --git a/src/mqt/predictor/rl/actions/__init__.py b/src/mqt/predictor/rl/actions/__init__.py new file mode 100644 index 000000000..c059ad12d --- /dev/null +++ b/src/mqt/predictor/rl/actions/__init__.py @@ -0,0 +1,167 @@ +# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM +# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""Actions (i.e. compiler passes) available in the reinforcement learning environment.""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from enum import Enum +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Callable + + from qiskit.passmanager import PropertySet + + +class CompilationOrigin(str, Enum): + """Enumeration of the origin of the compilation action.""" + + QISKIT = "qiskit" + TKET = "tket" + BQSKIT = "bqskit" + + +class PassType(str, Enum): + """Enumeration of the type of compilation pass.""" + + OPT = "optimization" + SYNTHESIS = "synthesis" + MAPPING = "mapping" + LAYOUT = "layout" + ROUTING = "routing" + FINAL_OPT = "final_optimization" + TERMINATE = "terminate" + + +@dataclass +class Action: + """Base class for all actions in the reinforcement learning environment. + + Attributes: + name: Unique action name. + origin: SDK origin of the action; ``None`` for terminate. + pass_type: Category of pass represented by this action. + transpile_pass: Pass object(s) executed for this action. + preserves_layout: Whether action preserves existing layout. + preserves_routing: Whether action preserves existing routing. + preserves_synthesis: Whether action preserves synthesis state. + """ + + name: str + origin: CompilationOrigin | None + pass_type: PassType + transpile_pass: Any + preserves_layout: bool = False + preserves_routing: bool = False + preserves_synthesis: bool = False + + +@dataclass +class DeviceIndependentAction(Action): + """Action that represents a static compilation pass that can be applied directly.""" + + +@dataclass +class DeviceDependentAction(Action): + """Device-specific action that depends on a target device. + + Attributes: + do_while: Optional do-while predicate for pass-manager execution. + """ + + transpile_pass: Any + do_while: Callable[[PropertySet], bool] | None = None + + +_ACTIONS: dict[str, Action] = {} + + +def register_action(action: Action) -> Action: + """Registers a new Action in the global _ACTIONS registry. + + Args: + action: Action to register. + + Returns: + The registered Action. + + Raises: + ValueError: If an action with the same name is already registered. + """ + if action.name in _ACTIONS: + msg = f"Action with name {action.name} already registered." + raise ValueError(msg) + _ACTIONS[action.name] = action + return action + + +def remove_action(name: str) -> None: + """Removes an Action from the global _ACTIONS registry by name. + + Args: + name: Name of the Action to remove. + + Raises: + KeyError: If no action with the given name is registered. + """ + if name not in _ACTIONS: + msg = f"No action with name {name} is registered." + raise KeyError(msg) + del _ACTIONS[name] + + +def get_actions_by_pass_type() -> dict[PassType, list[Action]]: + """Groups registered Actions from the global _ACTIONS registry by PassType. + + Returns: + A dictionary mapping each PassType to the list of registered Actions of that type. + """ + result: dict[PassType, list[Action]] = defaultdict(list) + for action in _ACTIONS.values(): + result[action.pass_type].append(action) + return result + + +from mqt.predictor.rl.actions import bqskit_actions as _bqskit_actions +from mqt.predictor.rl.actions import qiskit_actions as _qiskit_actions +from mqt.predictor.rl.actions import tket_actions as _tket_actions + +for _action in ( + *_qiskit_actions.qiskit_optimization_actions(), + *_tket_actions.tket_optimization_actions(), + _qiskit_actions.qiskit_o3_action(), + _bqskit_actions.bqskit_optimization_action(), + _qiskit_actions.qiskit_final_optimization_action(), + *_qiskit_actions.qiskit_layout_actions(), + _tket_actions.tket_routing_action(), + _qiskit_actions.qiskit_mapping_action(), + _bqskit_actions.bqskit_mapping_action(), + _qiskit_actions.qiskit_synthesis_action(), + _bqskit_actions.bqskit_synthesis_action(), + DeviceIndependentAction( + "terminate", + None, + PassType.TERMINATE, + transpile_pass=[], + ), +): + register_action(_action) + +__all__ = [ + "Action", + "CompilationOrigin", + "DeviceDependentAction", + "DeviceIndependentAction", + "PassType", + "get_actions_by_pass_type", + "register_action", + "remove_action", +] diff --git a/src/mqt/predictor/rl/actions/bqskit_actions.py b/src/mqt/predictor/rl/actions/bqskit_actions.py new file mode 100644 index 000000000..f0f2d5835 --- /dev/null +++ b/src/mqt/predictor/rl/actions/bqskit_actions.py @@ -0,0 +1,306 @@ +# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM +# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""BQSKit actions and execution helpers.""" + +from __future__ import annotations + +import os +import re +from functools import cache +from typing import TYPE_CHECKING, cast + +from bqskit import MachineModel +from bqskit import compile as bqskit_compile +from bqskit.ext import qiskit_to_bqskit +from bqskit.ext.qiskit.translate import OPENQASM2Language +from bqskit.ir import gates +from qiskit import qasm2 +from qiskit.circuit import QuantumRegister +from qiskit.circuit.library import RGate +from qiskit.transpiler import Layout, TranspileLayout + +from mqt.predictor.rl.actions import CompilationOrigin, DeviceDependentAction, PassType + +if TYPE_CHECKING: + from collections.abc import Callable + + from bqskit import Circuit + from bqskit.ir import Gate + from qiskit import QuantumCircuit + from qiskit.circuit import Instruction + from qiskit.transpiler import Target + + from mqt.predictor.rl.actions import Action + + +def _r_gate(theta: float, phi: float) -> Instruction: + """Construct an RGate with the given parameters.""" + return RGate(theta, phi) + + +def _bqskit_compilation_options() -> dict[str, float | int]: + """Returns BQSKit options tuned for local runs and CI.""" + return { + "optimization_level": 1 if os.getenv("GITHUB_ACTIONS") == "true" else 2, + "synthesis_epsilon": 1e-1 if os.getenv("GITHUB_ACTIONS") == "true" else 1e-8, + "max_synthesis_size": 3, + "seed": 10, + "num_workers": 1 if os.getenv("GITHUB_ACTIONS") == "true" else -1, + } + + +def bqskit_to_qiskit(circuit: Circuit) -> QuantumCircuit: + """Convert a BQSKit Circuit to Qiskit's QuantumCircuit. + + This function extends BQSKit's built-in conversion by adding support for + IQM's native 'r' gate. BQSKit represents this as U1qGate, which is converted + to Qiskit's RGate by rewriting the OpenQASM 2 output. + + Args: + circuit: The BQSKit circuit to convert. + + Returns: + The equivalent Qiskit QuantumCircuit with 'r' gates properly mapped. + """ + qasm = OPENQASM2Language().encode(circuit) + qasm = re.sub(r"\bU1q\(", "r(", qasm) + return qasm2.loads( + qasm, + custom_instructions=( + *qasm2.LEGACY_CUSTOM_INSTRUCTIONS, + qasm2.CustomInstruction( + "r", 2, 1, cast("Callable[[tuple[int | float, ...]], Instruction]", _r_gate), builtin=True + ), + ), + ) + + +def bqskit_optimization_action() -> Action: + """Returns the BQSKit optimization action.""" + return DeviceDependentAction( + "BQSKitO2", + CompilationOrigin.BQSKIT, + PassType.OPT, + transpile_pass=lambda circuit: bqskit_compile(circuit, **_bqskit_compilation_options()), + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=False, + ) + + +def bqskit_mapping_action() -> Action: + """Returns the BQSKit mapping action.""" + return DeviceDependentAction( + "BQSKitMapping", + CompilationOrigin.BQSKIT, + PassType.MAPPING, + transpile_pass=lambda device: ( + lambda bqskit_circuit: bqskit_compile( + bqskit_circuit, + model=MachineModel( + num_qudits=device.num_qubits, + gate_set=get_bqskit_native_gates(device), + coupling_graph=[(elem[0], elem[1]) for elem in device.build_coupling_map()], + ), + with_mapping=True, + **_bqskit_compilation_options(), + ) + ), + ) + + +def bqskit_synthesis_action() -> Action: + """Returns the BQSKit synthesis action.""" + return DeviceDependentAction( + "BQSKitSynthesis", + CompilationOrigin.BQSKIT, + PassType.SYNTHESIS, + transpile_pass=lambda device: ( + lambda bqskit_circuit: bqskit_compile( + bqskit_circuit, + model=MachineModel(bqskit_circuit.num_qudits, gate_set=get_bqskit_native_gates(device)), + **_bqskit_compilation_options(), + ) + ), + ) + + +@cache +def get_bqskit_native_gates(device: Target) -> list[Gate]: + """Returns the native gates of the given device. + + Args: + device: Target whose operation names are translated to BQSKit native gates. + + Returns: + The native gates of the given Target as a list of BQSKit Gate objects. + + Raises: + ValueError: If a gate in the device is not supported in BQSKit. + """ + gate_map = { + # --- 1-qubit gates --- + "id": gates.IdentityGate(), + "x": gates.XGate(), + "y": gates.YGate(), + "z": gates.ZGate(), + "h": gates.HGate(), + "s": gates.SGate(), + "sdg": gates.SdgGate(), + "t": gates.TGate(), + "tdg": gates.TdgGate(), + "sx": gates.SXGate(), + "rx": gates.RXGate(), + "ry": gates.RYGate(), + "rz": gates.RZGate(), + "r": gates.U1qGate(), + "u1": gates.U1Gate(), + "u2": gates.U2Gate(), + "u3": gates.U3Gate(), + # --- Controlled 1-qubit gates --- + "cx": gates.CXGate(), + "cy": gates.CYGate(), + "cz": gates.CZGate(), + "ch": gates.CHGate(), + "crx": gates.CRXGate(), + "cry": gates.CRYGate(), + "crz": gates.CRZGate(), + "cp": gates.CPGate(), + "cu": gates.CUGate(), + # --- 2-qubit gates --- + "swap": gates.SwapGate(), + "iswap": gates.ISwapGate(), + "ecr": gates.ECRGate(), + "rzz": gates.RZZGate(), + "rxx": gates.RXXGate(), + "ryy": gates.RYYGate(), + "zz": gates.ZZGate(), + # --- 3-qubit gates --- + "ccx": gates.CCXGate(), + # --- Others / approximations --- + "reset": gates.Reset(), + } + + native_gates = [] + + for instr in device.operation_names: + name = instr + + if name in [ + "barrier", + "measure", + "delay", + "for_loop", + "control", + "while_loop", + "if_test", + "if_else", + "switch_case", + "break", + "continue", + "box", + ]: + continue + + if name not in gate_map: + msg = f"The '{name}' gate of device '{device.description}' is not supported in BQSKIT." + raise ValueError(msg) + + native_gates.append(gate_map[name]) + + return native_gates + + +def final_layout_bqskit_to_qiskit( + bqskit_initial_layout: tuple[int, ...], + bqskit_final_layout: tuple[int, ...], + compiled_qc: QuantumCircuit, + initial_qc: QuantumCircuit, +) -> TranspileLayout: + """Converts a final layout from bqskit to qiskit. + + BQSKit provides an initial layout as a list[int] where each virtual qubit is mapped to a physical qubit + similarly, it provides a final layout as a list[int] representing where each virtual qubit is mapped to at the end + of the circuit. + + Args: + bqskit_initial_layout: Tuple mapping each BQSKit virtual qubit index to its initial physical qubit index. + bqskit_final_layout: Tuple mapping each BQSKit virtual qubit index to its final physical qubit index. + compiled_qc: Compiled QuantumCircuit whose qubits define the output qubit list and final layout values. + initial_qc: Initial QuantumCircuit whose qubits define the input layout values. + + Returns: + A TranspileLayout with a Qiskit Layout for the initial layout, an input-qubit-to-index mapping, and a final + Layout mapping physical qubit indices to compiled QuantumCircuit qubits when BQSKit changed the layout. + """ + ancilla = QuantumRegister(compiled_qc.num_qubits - initial_qc.num_qubits, "ancilla") + qiskit_initial_layout = {} + counter_ancilla_qubit = 0 + for i in range(compiled_qc.num_qubits): + if i in bqskit_initial_layout: + qiskit_initial_layout[i] = initial_qc.qubits[bqskit_initial_layout.index(i)] + else: + qiskit_initial_layout[i] = ancilla[counter_ancilla_qubit] + counter_ancilla_qubit += 1 + + initial_qubit_mapping = {bit: index for index, bit in enumerate(compiled_qc.qubits)} + + if bqskit_initial_layout == bqskit_final_layout: + qiskit_final_layout = None + else: + qiskit_final_layout = {} + for i in range(compiled_qc.num_qubits): + if i in bqskit_final_layout: + qiskit_final_layout[i] = compiled_qc.qubits[bqskit_initial_layout[bqskit_final_layout.index(i)]] + else: + qiskit_final_layout[i] = compiled_qc.qubits[i] + + return TranspileLayout( + initial_layout=Layout(input_dict=qiskit_initial_layout), + input_qubit_mapping=initial_qubit_mapping, + final_layout=Layout(input_dict=qiskit_final_layout) if qiskit_final_layout else None, + _output_qubit_list=compiled_qc.qubits, + _input_qubit_count=initial_qc.num_qubits, + ) + + +def run_bqskit_action( + action: Action, + circuit: QuantumCircuit, + device: Target, + layout: TranspileLayout | None, +) -> tuple[QuantumCircuit, TranspileLayout | None]: + """Apply a BQSKit action and return the updated circuit and layout metadata.""" + bqskit_qc = qiskit_to_bqskit(circuit) + if action.pass_type == PassType.OPT: + transpile = cast("Callable[[Circuit], Circuit]", action.transpile_pass) + bqskit_compiled_qc = transpile(bqskit_qc) + elif action.pass_type == PassType.SYNTHESIS: + factory = cast("Callable[[Target], Callable[[Circuit], Circuit]]", action.transpile_pass) + bqskit_compiled_qc = factory(device)(bqskit_qc) + elif action.pass_type == PassType.MAPPING: + factory = cast( + "Callable[[Target], Callable[[Circuit], tuple[Circuit, tuple[int, ...], tuple[int, ...]]]]", + action.transpile_pass, + ) + bqskit_compiled_qc, initial, final = factory(device)(bqskit_qc) + compiled_qiskit_qc = bqskit_to_qiskit(bqskit_compiled_qc) + return compiled_qiskit_qc, final_layout_bqskit_to_qiskit(initial, final, compiled_qiskit_qc, circuit) + else: + msg = f"Unhandled BQSKit pass type: {action.pass_type}" + raise ValueError(msg) + + return bqskit_to_qiskit(bqskit_compiled_qc), layout + + +def is_bqskit_action_available(*, has_layout: bool, has_parameterized_gates: bool) -> bool: + """Return whether a BQSKit action is available for the current circuit state.""" + # Using BQSKit after a layout is set can result in an error # TODO: Why? + # BQSKit does not support parameterized gates + return not (has_layout or has_parameterized_gates) diff --git a/src/mqt/predictor/rl/actions/qiskit_actions.py b/src/mqt/predictor/rl/actions/qiskit_actions.py new file mode 100644 index 000000000..1e5c20e31 --- /dev/null +++ b/src/mqt/predictor/rl/actions/qiskit_actions.py @@ -0,0 +1,387 @@ +# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM +# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""Qiskit actions and execution helpers.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, cast + +from qiskit.circuit import StandardEquivalenceLibrary +from qiskit.circuit.library import ( + CXGate, + CYGate, + CZGate, + ECRGate, + HGate, + SdgGate, + SGate, + SwapGate, + SXdgGate, + SXGate, + TdgGate, + TGate, + XGate, + YGate, + ZGate, +) +from qiskit.converters import circuit_to_dag, dag_to_circuit +from qiskit.passmanager import ConditionalController +from qiskit.passmanager.flow_controllers import DoWhileController +from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout +from qiskit.transpiler.passes import ( + ApplyLayout, + BasisTranslator, + Collect2qBlocks, + CommutativeCancellation, + CommutativeInverseCancellation, + ConsolidateBlocks, + DenseLayout, + Depth, + EnlargeWithAncilla, + FixedPoint, + FullAncillaAllocation, + GatesInBasis, + InverseCancellation, + MinimumPoint, + Optimize1qGatesDecomposition, + OptimizeCliffords, + RemoveDiagonalGatesBeforeMeasure, + SabreLayout, + Size, + UnitarySynthesis, + VF2Layout, + VF2PostLayout, +) +from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason +from qiskit.transpiler.preset_passmanagers import common + +from mqt.predictor.rl.actions import ( + CompilationOrigin, + DeviceDependentAction, + DeviceIndependentAction, + PassType, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from qiskit import QuantumCircuit + from qiskit.passmanager import PropertySet + from qiskit.passmanager.base_tasks import Task + from qiskit.transpiler import Layout, Target + + from mqt.predictor.rl.actions import ( + Action, + ) + +logger = logging.getLogger("mqt-predictor") + + +def qiskit_optimization_actions() -> list[Action]: + """Returns the Qiskit optimization actions.""" + return [ + DeviceIndependentAction( + "Optimize1qGatesDecomposition", + CompilationOrigin.QISKIT, + PassType.OPT, + [Optimize1qGatesDecomposition()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=False, + ), + DeviceIndependentAction( + "CommutativeCancellation", + CompilationOrigin.QISKIT, + PassType.OPT, + [CommutativeCancellation()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + DeviceIndependentAction( + "CommutativeInverseCancellation", + CompilationOrigin.QISKIT, + PassType.OPT, + [CommutativeInverseCancellation()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + DeviceIndependentAction( + "RemoveDiagonalGatesBeforeMeasure", + CompilationOrigin.QISKIT, + PassType.OPT, + [RemoveDiagonalGatesBeforeMeasure()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + DeviceIndependentAction( + "InverseCancellation", + CompilationOrigin.QISKIT, + PassType.OPT, + [ + InverseCancellation([ + CXGate(), + ECRGate(), + CZGate(), + CYGate(), + XGate(), + YGate(), + ZGate(), + HGate(), + SwapGate(), + (TGate(), TdgGate()), + (SGate(), SdgGate()), + (SXGate(), SXdgGate()), + ]) + ], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + DeviceIndependentAction( + "OptimizeCliffords", + CompilationOrigin.QISKIT, + PassType.OPT, + [OptimizeCliffords()], + preserves_layout=True, + preserves_routing=False, + preserves_synthesis=False, + ), + DeviceIndependentAction( + "Opt2qBlocks", + CompilationOrigin.QISKIT, + PassType.OPT, + [Collect2qBlocks(), ConsolidateBlocks(), UnitarySynthesis()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=False, + ), + ] + + +def qiskit_o3_action() -> Action: + """Returns the Qiskit level-3 optimization action.""" + return DeviceDependentAction( + "QiskitO3", + CompilationOrigin.QISKIT, + PassType.OPT, + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + transpile_pass=lambda native_gate, coupling_map: cast( + "list[Task]", + [ + Collect2qBlocks(), + ConsolidateBlocks(basis_gates=native_gate), + UnitarySynthesis(basis_gates=native_gate, coupling_map=coupling_map), + Optimize1qGatesDecomposition(basis=native_gate), + CommutativeCancellation(basis_gates=native_gate), + GatesInBasis(native_gate), + ConditionalController( + common.generate_translation_passmanager( + target=None, basis_gates=native_gate, coupling_map=coupling_map + ).to_flow_controller(), + condition=lambda property_set: not property_set["all_gates_in_basis"], + ), + Depth(recurse=True), + FixedPoint("depth"), + Size(recurse=True), + FixedPoint("size"), + MinimumPoint(["depth", "size"], "optimization_loop"), + ], + ), + do_while=lambda property_set: not property_set["optimization_loop_minimum_point"], + ) + + +def qiskit_final_optimization_action() -> Action: + """Returns the Qiskit final layout optimization action.""" + return DeviceDependentAction( + "VF2PostLayout", + CompilationOrigin.QISKIT, + PassType.FINAL_OPT, + transpile_pass=lambda device: [VF2PostLayout(target=device)], + ) + + +def qiskit_layout_actions() -> list[Action]: + """Returns the Qiskit layout actions.""" + return [ + DeviceDependentAction( + "DenseLayout", + CompilationOrigin.QISKIT, + PassType.LAYOUT, + transpile_pass=lambda device: cast( + "list[Task]", + [ + DenseLayout(coupling_map=CouplingMap(device.build_coupling_map())), + FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ], + ), + ), + DeviceDependentAction( + "VF2Layout", + CompilationOrigin.QISKIT, + PassType.LAYOUT, + transpile_pass=lambda device: cast( + "list[Task]", + [ + VF2Layout(target=device), + ConditionalController( + [ + FullAncillaAllocation(coupling_map=CouplingMap(device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ], + condition=lambda property_set: ( + property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND + ), + ), + ], + ), + ), + ] + + +def qiskit_mapping_action() -> Action: + """Returns the Qiskit mapping action.""" + return DeviceDependentAction( + "SabreMapping", + CompilationOrigin.QISKIT, + PassType.MAPPING, + transpile_pass=lambda device: cast( + "list[Task]", [SabreLayout(coupling_map=CouplingMap(device.build_coupling_map()), skip_routing=False)] + ), + ) + + +def qiskit_synthesis_action() -> Action: + """Returns the Qiskit synthesis action.""" + return DeviceDependentAction( + "BasisTranslator", + CompilationOrigin.QISKIT, + PassType.SYNTHESIS, + transpile_pass=lambda device: cast( + "list[Task]", [BasisTranslator(StandardEquivalenceLibrary, target_basis=device.operation_names)] + ), + ) + + +def postprocess_vf2postlayout( + qc: QuantumCircuit, post_layout: Layout, layout_before: TranspileLayout +) -> tuple[QuantumCircuit, ApplyLayout]: + """Postprocess a quantum circuit after VF2 layout assignment. + + Args: + qc: The quantum circuit to transform. + post_layout: The layout computed after routing. + layout_before: The layout before post-routing adjustment. + + Returns: + A tuple of the transformed circuit and the ApplyLayout used. + """ + apply_layout = ApplyLayout() + apply_layout.property_set["layout"] = layout_before.initial_layout + apply_layout.property_set["original_qubit_indices"] = layout_before.input_qubit_mapping + apply_layout.property_set["final_layout"] = layout_before.final_layout + apply_layout.property_set["post_layout"] = post_layout + + altered_qc = apply_layout.run(circuit_to_dag(qc)) + return dag_to_circuit(altered_qc), apply_layout + + +def _qiskit_passes(action: Action, device: Target, layout: TranspileLayout | None) -> list[Task]: + """Build the concrete Qiskit pass list for an action.""" + if action.name == "QiskitO3" and isinstance(action, DeviceDependentAction): + factory = cast("Callable[[list[str], CouplingMap | None], list[Task]]", action.transpile_pass) + return factory( + device.operation_names, + CouplingMap(device.build_coupling_map()) if layout else None, + ) + if callable(action.transpile_pass): + factory = cast("Callable[[Target], list[Task]]", action.transpile_pass) + return factory(device) + return cast("list[Task]", action.transpile_pass) + + +def _postprocess_layout_action( + action: Action, + property_set: PropertySet, + altered_qc: QuantumCircuit, + layout: TranspileLayout | None, + input_qubit_count: int | None = None, +) -> tuple[QuantumCircuit, TranspileLayout | None]: + """Update Qiskit's layout metadata after passes that can create or alter layouts.""" + if action.name == "VF2PostLayout": + assert property_set["VF2PostLayout_stop_reason"] is not None + post_layout = property_set["post_layout"] + if post_layout: + assert layout is not None + altered_qc, apply_layout = postprocess_vf2postlayout(altered_qc, post_layout, layout) + property_set = apply_layout.property_set + elif action.name == "VF2Layout": + if property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: + logger.warning( + "VF2Layout pass did not find a solution. Reason: %s", + property_set["VF2Layout_stop_reason"], + ) + else: + assert property_set["layout"] + else: + assert property_set["layout"] + + if property_set["layout"]: + return altered_qc, TranspileLayout( + initial_layout=property_set["layout"], + input_qubit_mapping=property_set["original_qubit_indices"], + final_layout=property_set["final_layout"], + _input_qubit_count=input_qubit_count, + _output_qubit_list=altered_qc.qubits, + ) + return altered_qc, layout + + +def run_qiskit_action( + action: Action, + circuit: QuantumCircuit, + device: Target, + layout: TranspileLayout | None, + input_qubit_count: int | None = None, +) -> tuple[QuantumCircuit, TranspileLayout | None]: + """Apply a Qiskit action and return the updated circuit and layout metadata.""" + passes = _qiskit_passes(action, device, layout) + if action.name == "QiskitO3" and isinstance(action, DeviceDependentAction): + assert action.do_while is not None + pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) + else: + pm = PassManager(passes) + + altered_qc = pm.run(circuit) + + if action.pass_type in {PassType.LAYOUT, PassType.MAPPING, PassType.FINAL_OPT}: + altered_qc, layout = _postprocess_layout_action(action, pm.property_set, altered_qc, layout, input_qubit_count) + elif action.pass_type == PassType.ROUTING and layout and pm.property_set["final_layout"] is not None: + layout.final_layout = pm.property_set["final_layout"] + + 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") + + return altered_qc, layout + + +def is_qiskit_action_available(action: Action, device: Target) -> bool: + """Return whether a Qiskit action is available for the current device.""" + # Only allow VF2PostLayout if "ibm" is in the device name # TODO: Why? + return action.name != "VF2PostLayout" or "ibm" in device.description diff --git a/src/mqt/predictor/rl/actions/tket_actions.py b/src/mqt/predictor/rl/actions/tket_actions.py new file mode 100644 index 000000000..42cef28f0 --- /dev/null +++ b/src/mqt/predictor/rl/actions/tket_actions.py @@ -0,0 +1,162 @@ +# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM +# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""TKET actions and execution helpers.""" + +from __future__ import annotations + +import operator +from typing import TYPE_CHECKING, cast + +from pytket import Qubit +from pytket._tket.passes import BasePass as TketBasePass # noqa: PLC2701 +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 mqt.predictor.rl.actions import CompilationOrigin, DeviceDependentAction, DeviceIndependentAction, PassType + +if TYPE_CHECKING: + from collections.abc import Callable + + from pytket import Circuit + from qiskit import QuantumCircuit + from qiskit.passmanager.base_tasks import Task + from qiskit.transpiler import Target, TranspileLayout + + from mqt.predictor.rl.actions import Action + + +class PreProcessTKETRoutingAfterQiskitLayout: + """Pre-process TKET routing for circuits that already carry a Qiskit layout. + + Qiskit layout passes rewrite the circuit into physical-qubit order. Before + TKET routing can operate on that circuit, it needs an equivalent trivial + placement so the current wire order is treated as the starting placement. + """ + + def apply(self, circuit: Circuit) -> None: + """Applies the pre-processing step to route a circuit with tket after a Qiskit Layout pass has been applied.""" + mapping = {Qubit(i): Node(i) for i in range(circuit.n_qubits)} + place_with_map(circuit=circuit, qmap=mapping) + + +def tket_optimization_actions() -> list[Action]: + """Returns the TKET optimization actions.""" + return [ + DeviceIndependentAction( + "PeepholeOptimise2Q", + CompilationOrigin.TKET, + PassType.OPT, + [PeepholeOptimise2Q()], + preserves_layout=False, + preserves_routing=False, + preserves_synthesis=False, + ), + DeviceIndependentAction( + "CliffordSimp", + CompilationOrigin.TKET, + PassType.OPT, + [CliffordSimp()], + preserves_layout=False, + preserves_routing=False, + preserves_synthesis=False, + ), + DeviceIndependentAction( + "FullPeepholeOptimiseCX", + CompilationOrigin.TKET, + PassType.OPT, + [FullPeepholeOptimise()], + preserves_layout=False, + preserves_routing=False, + preserves_synthesis=False, + ), + DeviceIndependentAction( + "RemoveRedundancies", + CompilationOrigin.TKET, + PassType.OPT, + [RemoveRedundancies()], + preserves_layout=True, + preserves_routing=True, + preserves_synthesis=True, + ), + ] + + +def tket_routing_action() -> Action: + """Returns the TKET routing action.""" + return DeviceDependentAction( + "RoutingPass", + CompilationOrigin.TKET, + PassType.ROUTING, + transpile_pass=lambda device: cast( + "list[Task]", + [ + PreProcessTKETRoutingAfterQiskitLayout(), + RoutingPass(Architecture(list(device.build_coupling_map()))), + ], + ), + ) + + +def final_layout_pytket_to_qiskit(pytket_circuit: Circuit, qiskit_circuit: QuantumCircuit) -> Layout: + """Converts a final layout from pytket to qiskit.""" + pytket_layout = pytket_circuit.qubit_readout + size_circuit = pytket_circuit.n_qubits + qiskit_layout = {} + qiskit_qreg = qiskit_circuit.qregs[0] + + pytket_layout = dict(sorted(pytket_layout.items(), key=operator.itemgetter(1))) + + for node, qubit_index in pytket_layout.items(): + qiskit_layout[node.index[0]] = qiskit_qreg[qubit_index] + + for i in range(size_circuit): + if i not in set(pytket_layout.values()): + qiskit_layout[i] = qiskit_qreg[i] + + return Layout(input_dict=qiskit_layout) + + +def run_tket_action( + action: Action, + circuit: QuantumCircuit, + device: Target, + layout: TranspileLayout | None, +) -> tuple[QuantumCircuit, TranspileLayout | None]: + """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) + passes = factory(device) + else: + passes = cast("list[Task]", action.transpile_pass) + for pass_ in passes: + assert isinstance(pass_, TketBasePass | PreProcessTKETRoutingAfterQiskitLayout) + pass_.apply(tket_qc) + + qbs = tket_qc.qubits + tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) + altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + + if action.pass_type == PassType.ROUTING: + assert layout is not None + layout.final_layout = final_layout_pytket_to_qiskit(tket_qc, altered_qc) + + return altered_qc, layout + + +def is_tket_action_available(*, action: Action, has_layout: bool) -> bool: + """Return whether a TKET action is available for the current layout state.""" + # 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. + return not has_layout or action.pass_type == PassType.ROUTING diff --git a/src/mqt/predictor/rl/helper.py b/src/mqt/predictor/rl/helper.py index f2d8c01e8..49532414f 100644 --- a/src/mqt/predictor/rl/helper.py +++ b/src/mqt/predictor/rl/helper.py @@ -32,7 +32,7 @@ def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generator) -> tuple[QuantumCircuit, str]: """Returns a random quantum circuit from the training circuits folder. - Arguments: + Args: max_qubits: The maximum number of qubits the returned quantum circuit may have. If no limit is set, it defaults to None. path_training_circuits: The path to the training circuits folder. rng: A random number generator to select a random quantum circuit. @@ -73,7 +73,7 @@ def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generat def create_feature_dict(qc: QuantumCircuit) -> dict[str, int | NDArray[np.float32]]: """Creates a feature dictionary for a given quantum circuit. - Arguments: + Args: qc: The quantum circuit for which the feature dictionary is created. Returns: diff --git a/src/mqt/predictor/rl/parsing.py b/src/mqt/predictor/rl/parsing.py deleted file mode 100644 index 5e7daaee5..000000000 --- a/src/mqt/predictor/rl/parsing.py +++ /dev/null @@ -1,302 +0,0 @@ -# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM -# Copyright (c) 2025 - 2026 Munich Quantum Software Company GmbH -# All rights reserved. -# -# SPDX-License-Identifier: MIT -# -# Licensed under the MIT License - -"""Helper methods necessary for parsing between circuit formats.""" - -from __future__ import annotations - -import operator -import re -from functools import cache -from typing import TYPE_CHECKING, cast - -from bqskit.ir import gates -from bqskit.ir.lang.qasm2 import OPENQASM2Language -from pytket import Qubit -from pytket.circuit import Node -from pytket.placement import place_with_map -from qiskit import qasm2 -from qiskit.circuit import QuantumCircuit, QuantumRegister -from qiskit.circuit.library import RGate -from qiskit.converters import circuit_to_dag, dag_to_circuit -from qiskit.transpiler import Layout, TranspileLayout -from qiskit.transpiler.passes import ApplyLayout - -if TYPE_CHECKING: - from collections.abc import Callable - - from bqskit.ir import Circuit as BqskitCircuit - from bqskit.ir import Gate - from pytket import Circuit as TketCircuit - from qiskit import QuantumCircuit - from qiskit.circuit import Instruction - from qiskit.transpiler import Target - - -class PreProcessTKETRoutingAfterQiskitLayout: - """Pre-processing step to route a circuit with TKET after a Qiskit Layout pass has been applied. - - The reason why we can apply the trivial layout here is that the circuit already got assigned a layout by qiskit. - Implicitly, Qiskit is reordering its qubits in a sequential manner, i.e., the qubit with the lowest *physical* qubit - first. - - Assuming, the layouted circuit is given by - - ┌───┐ ░ ┌─┐ - q_2 -> 0 ┤ H ├──■────────░───────┤M├ - └───┘┌─┴─┐ ░ ┌─┐└╥┘ - q_1 -> 1 ─────┤ X ├──■───░────┤M├─╫─ - └───┘┌─┴─┐ ░ ┌─┐└╥┘ ║ - q_0 -> 2 ──────────┤ X ├─░─┤M├─╫──╫─ - └───┘ ░ └╥┘ ║ ║ - ancilla_0 -> 3 ───────────────────╫──╫──╫─ - ║ ║ ║ - ancilla_1 -> 4 ───────────────────╫──╫──╫─ - ║ ║ ║ - meas: 3/═══════════════════╩══╩══╩═ - 0 1 2 - - Applying the trivial layout, we get the same qubit order as in the original circuit and can be respectively - routed. This results in: - ┌───┐ ░ ┌─┐ - q_0: ┤ H ├──■────────░───────┤M├ - └───┘┌─┴─┐ ░ ┌─┐└╥┘ - q_1: ─────┤ X ├──■───░────┤M├─╫─ - └───┘┌─┴─┐ ░ ┌─┐└╥┘ ║ - q_2: ──────────┤ X ├─░─┤M├─╫──╫─ - └───┘ ░ └╥┘ ║ ║ - q_3: ───────────────────╫──╫──╫─ - ║ ║ ║ - q_4: ───────────────────╫──╫──╫─ - ║ ║ ║ - meas: 3/═══════════════════╩══╩══╩═ - 0 1 2 - - - If we would not apply the trivial layout, no layout would be considered resulting, e.g., in the following circuit: - ┌───┐ ░ ┌─┐ - q_0: ─────┤ X ├─────■───░────┤M├─── - ┌───┐└─┬─┘ ┌─┴─┐ ░ ┌─┐└╥┘ - q_1: ┤ H ├──■───X─┤ X ├─░─┤M├─╫──── - └───┘ │ └───┘ ░ └╥┘ ║ ┌─┐ - q_2: ───────────X───────░──╫──╫─┤M├ - ░ ║ ║ └╥┘ - q_3: ──────────────────────╫──╫──╫─ - ║ ║ ║ - q_4: ──────────────────────╫──╫──╫─ - ║ ║ ║ - meas: 3/══════════════════════╩══╩══╩═ - 0 1 2 - - """ - - def apply(self, circuit: TketCircuit) -> None: - """Applies the pre-processing step to route a circuit with tket after a Qiskit Layout pass has been applied.""" - mapping = {Qubit(i): Node(i) for i in range(circuit.n_qubits)} - place_with_map(circuit=circuit, qmap=mapping) - - -@cache -def get_bqskit_native_gates(device: Target) -> list[Gate]: - """Returns the native gates of the given device. - - Arguments: - device: The device for which the native gates are returned. - - Returns: - The native gates of the given device as BQSKit gates. - - Raises: - ValueError: If a gate in the device is not supported in BQSKit. - """ - gate_map = { - # --- 1-qubit gates --- - "id": gates.IdentityGate(), - "x": gates.XGate(), - "y": gates.YGate(), - "z": gates.ZGate(), - "h": gates.HGate(), - "s": gates.SGate(), - "sdg": gates.SdgGate(), - "t": gates.TGate(), - "tdg": gates.TdgGate(), - "sx": gates.SXGate(), - "rx": gates.RXGate(), - "ry": gates.RYGate(), - "rz": gates.RZGate(), - "r": gates.U1qGate(), - "u1": gates.U1Gate(), - "u2": gates.U2Gate(), - "u3": gates.U3Gate(), - # --- Controlled 1-qubit gates --- - "cx": gates.CXGate(), - "cy": gates.CYGate(), - "cz": gates.CZGate(), - "ch": gates.CHGate(), - "crx": gates.CRXGate(), - "cry": gates.CRYGate(), - "crz": gates.CRZGate(), - "cp": gates.CPGate(), - "cu": gates.CUGate(), - # --- 2-qubit gates --- - "swap": gates.SwapGate(), - "iswap": gates.ISwapGate(), - "ecr": gates.ECRGate(), - "rzz": gates.RZZGate(), - "rxx": gates.RXXGate(), - "ryy": gates.RYYGate(), - "zz": gates.ZZGate(), - # --- 3-qubit gates --- - "ccx": gates.CCXGate(), - # --- Others / approximations --- - "reset": gates.Reset(), - } - - native_gates = [] - - for instr in device.operation_names: - name = instr - - if name in [ - "barrier", - "measure", - "delay", - "for_loop", - "control", - "while_loop", - "if_test", - "if_else", - "switch_case", - "break", - "continue", - "box", - ]: - continue - - if name not in gate_map: - msg = f"The '{name}' gate of device '{device.description}' is not supported in BQSKIT." - raise ValueError(msg) - - native_gates.append(gate_map[name]) - - return native_gates - - -def bqskit_to_qiskit(circuit: BqskitCircuit) -> QuantumCircuit: - """Convert a BQSKit Circuit to Qiskit's QuantumCircuit. - - This function extends BQSKit's built-in conversion by adding support for - IQM's native 'r' gate. BQSKit represents this as U1qGate, which is converted - to Qiskit's RGate by rewriting the OpenQASM 2 output. - - Args: - circuit: The BQSKit circuit to convert. - - Returns: - The equivalent Qiskit QuantumCircuit with 'r' gates properly mapped. - """ - qasm = OPENQASM2Language().encode(circuit) - qasm = re.sub(r"\bU1q\(", "r(", qasm) - - def r_gate(theta: float, phi: float) -> RGate: - return RGate(theta, phi) - - r_gate_constructor = cast("Callable[[tuple[int | float, ...]], Instruction]", r_gate) - - return qasm2.loads( - qasm, - custom_instructions=( - *qasm2.LEGACY_CUSTOM_INSTRUCTIONS, - qasm2.CustomInstruction("r", 2, 1, r_gate_constructor, builtin=True), - ), - ) - - -def final_layout_pytket_to_qiskit(pytket_circuit: TketCircuit, qiskit_circuit: QuantumCircuit) -> Layout: - """Converts a final layout from pytket to qiskit.""" - pytket_layout = pytket_circuit.qubit_readout - size_circuit = pytket_circuit.n_qubits - qiskit_layout = {} - qiskit_qreg = qiskit_circuit.qregs[0] - - pytket_layout = dict(sorted(pytket_layout.items(), key=operator.itemgetter(1))) - - for node, qubit_index in pytket_layout.items(): - qiskit_layout[node.index[0]] = qiskit_qreg[qubit_index] - - for i in range(size_circuit): - if i not in set(pytket_layout.values()): - qiskit_layout[i] = qiskit_qreg[i] - - return Layout(input_dict=qiskit_layout) - - -def final_layout_bqskit_to_qiskit( - bqskit_initial_layout: tuple[int, ...], - bqskit_final_layout: tuple[int, ...], - compiled_qc: QuantumCircuit, - initial_qc: QuantumCircuit, -) -> TranspileLayout: - """Converts a final layout from bqskit to qiskit. - - BQSKit provides an initial layout as a list[int] where each virtual qubit is mapped to a physical qubit - similarly, it provides a final layout as a list[int] representing where each virtual qubit is mapped to at the end - of the circuit. - """ - ancilla = QuantumRegister(compiled_qc.num_qubits - initial_qc.num_qubits, "ancilla") - qiskit_initial_layout = {} - counter_ancilla_qubit = 0 - for i in range(compiled_qc.num_qubits): - if i in bqskit_initial_layout: - qiskit_initial_layout[i] = initial_qc.qubits[bqskit_initial_layout.index(i)] - else: - qiskit_initial_layout[i] = ancilla[counter_ancilla_qubit] - counter_ancilla_qubit += 1 - - initial_qubit_mapping = {bit: index for index, bit in enumerate(compiled_qc.qubits)} - - if bqskit_initial_layout == bqskit_final_layout: - qiskit_final_layout = None - else: - qiskit_final_layout = {} - for i in range(compiled_qc.num_qubits): - if i in bqskit_final_layout: - qiskit_final_layout[i] = compiled_qc.qubits[bqskit_initial_layout[bqskit_final_layout.index(i)]] - else: - qiskit_final_layout[i] = compiled_qc.qubits[i] - - return TranspileLayout( - initial_layout=Layout(input_dict=qiskit_initial_layout), - input_qubit_mapping=initial_qubit_mapping, - final_layout=Layout(input_dict=qiskit_final_layout) if qiskit_final_layout else None, - _output_qubit_list=compiled_qc.qubits, - _input_qubit_count=initial_qc.num_qubits, - ) - - -def postprocess_vf2postlayout( - qc: QuantumCircuit, post_layout: Layout, layout_before: TranspileLayout -) -> tuple[QuantumCircuit, ApplyLayout]: - """Postprocess a quantum circuit after VF2 layout assignment. - - Args: - qc: The quantum circuit to transform. - post_layout: The layout computed after routing. - layout_before: The layout before post-routing adjustment. - - Returns: - A tuple of the transformed circuit and the ApplyLayout used. - """ - apply_layout = ApplyLayout() - apply_layout.property_set["layout"] = layout_before.initial_layout - apply_layout.property_set["original_qubit_indices"] = layout_before.input_qubit_mapping - apply_layout.property_set["final_layout"] = layout_before.final_layout - apply_layout.property_set["post_layout"] = post_layout - - altered_qc = apply_layout.run(circuit_to_dag(qc)) - return dag_to_circuit(altered_qc), apply_layout diff --git a/src/mqt/predictor/rl/predictor.py b/src/mqt/predictor/rl/predictor.py index c1578125a..3c57eb647 100644 --- a/src/mqt/predictor/rl/predictor.py +++ b/src/mqt/predictor/rl/predictor.py @@ -54,7 +54,7 @@ def compile_as_predicted( ) -> 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: + Args: qc: The quantum circuit to be compiled or the path to a qasm file containing the quantum circuit. Returns: @@ -93,7 +93,7 @@ def train_model( ) -> None: """Trains all models for the given reward functions and device. - Arguments: + Args: timesteps: The number of timesteps to train the model. Defaults to 1000. verbose: The verbosity level. Defaults to 2. test: Whether to train the model for testing purposes. Defaults to False. @@ -136,7 +136,7 @@ def train_model( def load_model(model_name: str) -> MaskablePPO: """Loads a trained model from the trained model folder. - Arguments: + Args: model_name: The name of the model to be loaded. Returns: @@ -162,7 +162,7 @@ def rl_compile( ) -> tuple[QuantumCircuit, list[str]]: """Compiles a given quantum circuit to a device optimizing for the given figure of merit. - Arguments: + Args: qc: The quantum circuit to be compiled. If a string is given, it is assumed to be a path to a qasm file. device: The device to compile to. figure_of_merit: The figure of merit to be used for compilation. Defaults to "expected_fidelity". diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 95d632463..b3be49793 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -11,37 +11,23 @@ from __future__ import annotations import logging +import warnings from typing import TYPE_CHECKING, Any -from pytket._tket.passes import BasePass as TketBasePass # noqa: PLC2701 - if TYPE_CHECKING: - from collections.abc import Callable from pathlib import Path - from bqskit import Circuit from gymnasium.spaces import Space - from qiskit.passmanager.base_tasks import Task from qiskit.transpiler import Layout, Target from mqt.predictor.reward import figure_of_merit - from mqt.predictor.rl.actions import Action - - -import warnings -from typing import cast import numpy as np -from bqskit.ext import qiskit_to_bqskit from gymnasium import Env from gymnasium.spaces import Box, Dict, Discrete from joblib import load -from pytket.circuit import Qubit -from pytket.extensions.qiskit import qiskit_to_tk, tk_to_qiskit from qiskit import QuantumCircuit -from qiskit.passmanager.flow_controllers import DoWhileController -from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout -from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason +from qiskit.transpiler import CouplingMap, TranspileLayout from mqt.predictor.hellinger import get_hellinger_model_path from mqt.predictor.reward import ( @@ -51,15 +37,15 @@ estimated_success_probability, expected_fidelity, ) -from mqt.predictor.rl.actions import CompilationOrigin, DeviceDependentAction, PassType, get_actions_by_pass_type -from mqt.predictor.rl.helper import create_feature_dict, get_path_training_circuits, get_state_sample -from mqt.predictor.rl.parsing import ( - PreProcessTKETRoutingAfterQiskitLayout, - bqskit_to_qiskit, - final_layout_bqskit_to_qiskit, - final_layout_pytket_to_qiskit, - postprocess_vf2postlayout, +from mqt.predictor.rl.actions import ( + CompilationOrigin, + PassType, + get_actions_by_pass_type, ) +from mqt.predictor.rl.actions.bqskit_actions import is_bqskit_action_available, run_bqskit_action +from mqt.predictor.rl.actions.qiskit_actions import is_qiskit_action_available, run_qiskit_action +from mqt.predictor.rl.actions.tket_actions import is_tket_action_available, run_tket_action +from mqt.predictor.rl.helper import create_feature_dict, get_path_training_circuits, get_state_sample logger = logging.getLogger("mqt-predictor") @@ -75,7 +61,7 @@ def __init__( ) -> None: """Initializes the PredictorEnv object. - Arguments: + Args: device: The target device to be used for compilation. reward_function: The figure of merit to be used for the reward function. Defaults to "expected_fidelity". path_training_circuits: The path to the training circuits folder. Defaults to None, which uses the default path. @@ -149,8 +135,8 @@ def __init__( self.num_qubits_uncompiled_circuit = 0 # Canonical layout state for the current circuit. It is mirrored to - # QuantumCircuit.layout for callers, but kept here because TKET and - # BQSKit conversions do not preserve Qiskit's layout metadata. + # QuantumCircuit.layout for callers, but kept here because some action + # implementations do not preserve layout metadata across conversions. self.layout: TranspileLayout | None = None self.has_parameterized_gates = False @@ -171,7 +157,7 @@ def __init__( def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any, Any]]: """Executes the given action and returns the new state, the reward, whether the episode is done, whether the episode is truncated and additional information. - Arguments: + Args: action: The action to be executed, represented by its index in the action set. Returns: @@ -180,25 +166,33 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any Raises: RuntimeError: If no valid actions are left. """ - self.used_actions.append(str(self.action_set[action].name)) - altered_qc = self.apply_action(action) + try: + self.used_actions.append(str(self.action_set[action].name)) + altered_qc = self.apply_action(action) + except Exception as exc: # noqa: BLE001 + # Different passes may fail for various reasons (e.g., found no routing solution). + self.error_occurred = True + return ( + create_feature_dict(self.state), # features + 0, # reward + False, # terminated + True, # truncated + {"Truncated because of error": f"{type(exc).__name__}: {exc}"}, # info + ) + if not altered_qc: + # TODO: Remove as dead code? return ( - create_feature_dict(self.state), - 0, - True, - False, - {}, + create_feature_dict(self.state), # features + 0, # reward + True, # terminated + False, # truncated + {}, # info ) self.state: QuantumCircuit = altered_qc self.num_steps += 1 - # in case a Qiskit.QuantumCircuit has `unitary` gates in it, decompose them (otherwise qiskit will throw an error when applying BasisTranslator) - # TODO: will be improved by addressing issue https://github.com/munich-quantum-toolkit/predictor/issues/668 - if self.state.count_ops().get("unitary"): - self.state = self.state.decompose(gates_to_decompose="unitary") - self.state._layout = self.layout # noqa: SLF001 self.valid_actions = self.determine_valid_actions_for_state() @@ -207,7 +201,6 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any raise RuntimeError(msg) if action == self.action_terminate_index: - assert action in self.valid_actions, "Terminate action is not valid but was chosen." reward_val = self.calculate_reward() done = True else: @@ -242,7 +235,7 @@ def reset( ) -> tuple[dict[str, Any], dict[str, Any]]: """Resets the environment to the given state or a random state. - Arguments: + Args: qc: The quantum circuit to be compiled or the path to a qasm file containing the quantum circuit. Defaults to None. seed: The seed to be used for the random number generator. Defaults to None. options: Additional options. Defaults to None. @@ -273,47 +266,59 @@ def reset( return create_feature_dict(self.state), {} def action_masks(self) -> list[bool]: - """Returns a list of valid actions for the current state.""" - action_mask = [action in self.valid_actions for action in self.action_set] + """Build the boolean action mask exposed to the RL policy. + ``self.valid_actions`` contains the structurally valid action indices + for the current circuit state. This method expands that sparse list to + one boolean per registered action and applies SDK-specific availability + filters that depend on circuit features or the selected device. + Terminate has no SDK origin and is accepted solely from the structural + candidate list. + + Returns: + A dense boolean mask ordered like ``self.action_set``. + """ has_layout = self.layout is not None + valid_action_indices = set(self.valid_actions) + action_mask: list[bool] = [] + + for action_index in range(len(self.action_set)): + action = self.action_set[action_index] + + if action_index not in valid_action_indices: + action_mask.append(False) + continue + if action.pass_type == PassType.TERMINATE: + action_mask.append(True) + continue + 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)) + elif action.origin == CompilationOrigin.BQSKIT: + action_mask.append( + is_bqskit_action_available( + has_layout=has_layout, + has_parameterized_gates=self.has_parameterized_gates, + ) + ) + else: + msg = f"Origin {action.origin} not supported." + raise ValueError(msg) - if has_layout: - # 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 are - # designed to work after a Qiskit layout via PreProcessTKETRoutingAfterQiskitLayout. - action_mask = [ - action_mask[i] - and (self.action_set[i].origin != CompilationOrigin.TKET or i in self.actions_routing_indices) - for i in range(len(action_mask)) - ] - - if self.has_parameterized_gates or has_layout: - # remove all actions that are from "origin"=="bqskit" because they are not supported for parameterized gates - # or after layout since using BQSKit after a layout is set can result in an error - action_mask = [ - action_mask[i] and self.action_set[i].origin != CompilationOrigin.BQSKIT - for i in range(len(action_mask)) - ] - - # only allow VF2PostLayout if "ibm" is in the device name - if "ibm" not in self.device.description: - action_mask = [ - action_mask[i] and self.action_set[i].name != "VF2PostLayout" for i in range(len(action_mask)) - ] return action_mask - def apply_action(self, action_index: int) -> QuantumCircuit | None: + def apply_action(self, action_index: int) -> QuantumCircuit: """Applies the given action to the current state and returns the altered state. - Arguments: + Args: action_index: The index of the action to be applied, which must be in the action set. Returns: - The altered quantum circuit after applying the action, or None if the action is to terminate the compilation. + The altered quantum circuit after applying the action. Raises: - ValueError: If the action index is not in the action set or if the action origin is not supported. + ValueError: If the action index is not in the action set or if the action cannot be applied. """ if action_index not in self.action_set: msg = f"Action {action_index} not supported." @@ -321,140 +326,36 @@ def apply_action(self, action_index: int) -> QuantumCircuit | None: action = self.action_set[action_index] - if action.name == "terminate": + if action.pass_type == PassType.TERMINATE: return self.state + if action.origin == CompilationOrigin.QISKIT: - return self._apply_qiskit_action(action, action_index) - if action.origin == CompilationOrigin.TKET: - return self._apply_tket_action(action, action_index) - if action.origin == CompilationOrigin.BQSKIT: - return self._apply_bqskit_action(action, action_index) - msg = f"Origin {action.origin} not supported." - raise ValueError(msg) - - def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCircuit: - if action.name == "QiskitO3" and isinstance(action, DeviceDependentAction): - factory = cast("Callable[[list[str], CouplingMap | None], list[Task]]", action.transpile_pass) - passes = factory( - self.device.operation_names, - CouplingMap(self.device.build_coupling_map()) if self.layout else None, + altered_qc, self.layout = run_qiskit_action( + action=action, + circuit=self.state, + device=self.device, + layout=self.layout, + input_qubit_count=self.num_qubits_uncompiled_circuit, ) - assert action.do_while is not None - pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) - else: - if callable(action.transpile_pass): - factory = cast("Callable[[Target], list[Task]]", action.transpile_pass) - passes = factory(self.device) - else: - passes = cast("list[Task]", action.transpile_pass) - pm = PassManager(passes) - - altered_qc = pm.run(self.state) - - if action_index in ( - self.actions_layout_indices + self.actions_mapping_indices + self.actions_final_optimization_indices - ): - altered_qc = self._handle_qiskit_layout_postprocessing(action, pm, altered_qc) - - elif ( - action_index in self.actions_routing_indices and self.layout and pm.property_set["final_layout"] is not None - ): - self.layout.final_layout = pm.property_set["final_layout"] - - # BasisTranslator errors on unitary gates; decompose them immediately so - # the circuit is always in a consistent state after a Qiskit action. - if altered_qc.count_ops().get("unitary"): - altered_qc = altered_qc.decompose(gates_to_decompose="unitary") - - return altered_qc - - def _handle_qiskit_layout_postprocessing( - self, action: Action, pm: PassManager, altered_qc: QuantumCircuit - ) -> QuantumCircuit: - if action.name == "VF2PostLayout": - assert pm.property_set["VF2PostLayout_stop_reason"] is not None - post_layout = pm.property_set["post_layout"] - if post_layout: - assert self.layout is not None - altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) - elif action.name == "VF2Layout": - if pm.property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: - logger.warning( - "VF2Layout pass did not find a solution. Reason: %s", - pm.property_set["VF2Layout_stop_reason"], - ) - else: - assert pm.property_set["layout"] - else: - assert pm.property_set["layout"] - - if pm.property_set["layout"]: - # Layout/mapping passes create the base logical-to-physical mapping; - # later routing actions only update final_layout. - self.layout = TranspileLayout( - initial_layout=pm.property_set["layout"], - input_qubit_mapping=pm.property_set["original_qubit_indices"], - final_layout=pm.property_set["final_layout"], - _output_qubit_list=altered_qc.qubits, - _input_qubit_count=self.num_qubits_uncompiled_circuit, + elif action.origin == CompilationOrigin.TKET: + altered_qc, self.layout = run_tket_action( + action=action, + circuit=self.state, + device=self.device, + layout=self.layout, ) - return altered_qc - - def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircuit: - tket_qc = qiskit_to_tk(self.state, preserve_param_uuid=True) - if callable(action.transpile_pass): - factory = cast("Callable[[Target], list[Task]]", action.transpile_pass) - passes = factory(self.device) - else: - passes = cast("list[Task]", action.transpile_pass) - for pass_ in passes: - assert isinstance(pass_, TketBasePass | PreProcessTKETRoutingAfterQiskitLayout) - pass_.apply(tket_qc) - - qbs = tket_qc.qubits - tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) - - if action_index in self.actions_routing_indices: - assert self.layout is not None - self.layout.final_layout = final_layout_pytket_to_qiskit(tket_qc, altered_qc) - - return altered_qc - - def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCircuit: - """Applies the given BQSKit action to the current state and returns the altered state. - - Arguments: - action: The BQSKit action to be applied. - action_index: The index of the action in the action set. - - Returns: - The altered quantum circuit after applying the action. - - Raises: - ValueError: If the action index is not in the action set or if the action origin is not supported. - """ - bqskit_qc = qiskit_to_bqskit(self.state) - if action_index in self.actions_opt_indices: - transpile = cast("Callable[[Circuit], Circuit]", action.transpile_pass) - bqskit_compiled_qc = transpile(bqskit_qc) - elif action_index in self.actions_synthesis_indices: - factory = cast("Callable[[Target], Callable[[Circuit], Circuit]]", action.transpile_pass) - bqskit_compiled_qc = factory(self.device)(bqskit_qc) - elif action_index in self.actions_mapping_indices: - factory = cast( - "Callable[[Target], Callable[[Circuit], tuple[Circuit, tuple[int, ...], tuple[int, ...]]]]", - action.transpile_pass, + elif action.origin == CompilationOrigin.BQSKIT: + altered_qc, self.layout = run_bqskit_action( + action=action, + circuit=self.state, + device=self.device, + layout=self.layout, ) - bqskit_compiled_qc, initial, final = factory(self.device)(bqskit_qc) - compiled_qiskit_qc = bqskit_to_qiskit(bqskit_compiled_qc) - self.layout = final_layout_bqskit_to_qiskit(initial, final, compiled_qiskit_qc, self.state) - return compiled_qiskit_qc else: - msg = f"Unhandled BQSKit action index: {action_index}" + msg = f"Origin {action.origin} not supported." raise ValueError(msg) - return bqskit_to_qiskit(bqskit_compiled_qc) + return altered_qc def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | Layout) -> bool: """True if every logical qubit in the circuit has a physical assignment.""" @@ -510,7 +411,17 @@ def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) return True def determine_valid_actions_for_state(self) -> list[int]: - """Determine valid actions based on circuit state: synthesized, mapped, routed.""" + """Select structurally valid action indices for the current circuit state. + + The circuit is classified by compilation progress: synthesized to the + target gate set, laid out to physical qubits, and routed against the + directed coupling map. This method only determines which pass types can + advance that state; SDK/backend-specific availability filters are applied later + in ``action_masks``. + + Returns: + Action indices whose pass type can be attempted from the current state. + """ synthesized = self.is_circuit_synthesized(self.state) laid_out = self.is_circuit_laid_out(self.state, self.layout) if self.layout else False # Routing is only allowed after layout @@ -530,20 +441,17 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - # Possible state because optimization can destroy the native gate set - # State is not explicitly *depicted* in original paper + # Not *depicted* in paper; necessary because optimization can destroy the native gate set if not synthesized and laid_out and not routed: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - # Possible state because there are layout-only passes - # State is not explicitly *depicted* in original paper + # Not *depicted* in paper; necessary because of layout-only passes if synthesized and laid_out and not routed: actions.extend(self.actions_routing_indices) - # Possible state because routing may add SWAP gates which are not necessarily native gates - # State is not explicitly *depicted* in original paper + # Not *depicted* in paper; necessary because routing can insert non-native SWAPs if not synthesized and laid_out and routed: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) diff --git a/src/mqt/predictor/utils.py b/src/mqt/predictor/utils.py index 82631bd07..c30d2cfb0 100644 --- a/src/mqt/predictor/utils.py +++ b/src/mqt/predictor/utils.py @@ -39,7 +39,7 @@ def timeout_watcher( ) -> tuple[QuantumCircuit, list[str]] | bool: """Method that stops a function call after a given timeout limit. - Arguments: + Args: func: The function to be called. args: The arguments to be passed to the function. timeout: The timeout limit in seconds. diff --git a/tests/compilation/test_helper_rl.py b/tests/compilation/test_helper_rl.py index c324106a5..b99ad0fca 100644 --- a/tests/compilation/test_helper_rl.py +++ b/tests/compilation/test_helper_rl.py @@ -14,16 +14,21 @@ from typing import TYPE_CHECKING, cast import numpy as np -from bqskit.ir import Circuit, gates +from bqskit.ir import gates +from bqskit.ir.circuit import Circuit from mqt.bench import BenchmarkLevel, get_benchmark from mqt.bench.targets import get_device from qiskit import transpile from qiskit.transpiler import PassManager from qiskit.transpiler.passes.layout.vf2_post_layout import VF2PostLayoutStopReason -from mqt.predictor.rl.actions import PassType, get_actions_by_pass_type +from mqt.predictor.rl.actions import ( + PassType, + get_actions_by_pass_type, +) +from mqt.predictor.rl.actions.bqskit_actions import bqskit_to_qiskit, get_bqskit_native_gates +from mqt.predictor.rl.actions.qiskit_actions import postprocess_vf2postlayout from mqt.predictor.rl.helper import create_feature_dict, get_path_trained_model, get_path_training_circuits -from mqt.predictor.rl.parsing import bqskit_to_qiskit, get_bqskit_native_gates, postprocess_vf2postlayout if TYPE_CHECKING: from collections.abc import Callable diff --git a/tests/compilation/test_integration_further_SDKs.py b/tests/compilation/test_integration_further_SDKs.py index b9b5c077f..713d24f03 100644 --- a/tests/compilation/test_integration_further_SDKs.py +++ b/tests/compilation/test_integration_further_SDKs.py @@ -10,247 +10,246 @@ from __future__ import annotations -import re -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING import pytest -from bqskit.ext import qiskit_to_bqskit -from bqskit.ir.circuit import Circuit -from mqt.bench.targets import get_available_device_names, get_device -from pytket.circuit import Qubit -from pytket.extensions.qiskit import qiskit_to_tk, tk_to_qiskit +from mqt.bench.targets import get_device from qiskit import QuantumCircuit -from qiskit.transpiler import PassManager -from qiskit.transpiler.layout import TranspileLayout -from qiskit.transpiler.passes import CheckMap, GatesInBasis - -from mqt.predictor.rl.actions import CompilationOrigin, PassType, get_actions_by_pass_type -from mqt.predictor.rl.parsing import ( - bqskit_to_qiskit, - final_layout_bqskit_to_qiskit, - final_layout_pytket_to_qiskit, +from qiskit.circuit import StandardEquivalenceLibrary +from qiskit.transpiler import PassManager, TranspileLayout +from qiskit.transpiler.passes import ( + ApplyLayout, + BasisTranslator, + EnlargeWithAncilla, + FullAncillaAllocation, + SabreSwap, + TrivialLayout, ) -if TYPE_CHECKING: - from collections.abc import Callable +from mqt.predictor.rl.actions import PassType +from mqt.predictor.rl.predictorenv import PredictorEnv - from pytket._tket.passes import BasePass as TketBasePass - from qiskit.passmanager.base_tasks import Task +if TYPE_CHECKING: from qiskit.transpiler import Target - from mqt.predictor.rl.actions import Action - from mqt.predictor.rl.parsing import ( - PreProcessTKETRoutingAfterQiskitLayout, - ) - - -@pytest.fixture -def available_actions_dict() -> dict[PassType, list[Action]]: - """Return a dictionary of available actions.""" - return get_actions_by_pass_type() - - -def test_bqskit_o2_action(available_actions_dict: dict[PassType, list[Action]]) -> None: - """Test the BQSKitO2 action.""" - action_bqskit_o2: Action | None = None - for action in available_actions_dict[PassType.OPT]: - if action.name == "BQSKitO2": - action_bqskit_o2 = action - assert action_bqskit_o2 is not None - qc = QuantumCircuit(2) - qc.h(0) - qc.cx(0, 1) - - bqskit_qc = qiskit_to_bqskit(qc) - factory = cast("Callable[[Circuit], Circuit]", action_bqskit_o2.transpile_pass) - bqskit_qc_optimized = factory(bqskit_qc) - assert isinstance(bqskit_qc_optimized, Circuit) - optimized_qc = bqskit_to_qiskit(bqskit_qc_optimized) +def _setup_env(env: PredictorEnv, circuit: QuantumCircuit, layout: TranspileLayout | None, n_qubits: int) -> None: + """Reset env to the given circuit/layout state without starting a full RL episode.""" + env.reset(qc=circuit.copy()) + env.layout = layout + env.num_qubits_uncompiled_circuit = n_qubits - assert optimized_qc != qc +def _is_available(env: PredictorEnv, idx: int) -> bool: + """Return whether action idx is structurally and SDK-valid for the current env state.""" + env.valid_actions = env.determine_valid_actions_for_state() + return env.action_masks()[idx] -@pytest.mark.parametrize("device", [get_device(name) for name in get_available_device_names()]) -def test_bqskit_synthesis_action(device: Target, available_actions_dict: dict[PassType, list[Action]]) -> None: - """Test the BQSKitSynthesis action for all devices.""" - action_bqskit_synthesis_action: Action | None = None - for action in available_actions_dict[PassType.SYNTHESIS]: - if action.name == "BQSKitSynthesis": - action_bqskit_synthesis_action = action - assert action_bqskit_synthesis_action is not None - qc = QuantumCircuit(2) - qc.h(0) - qc.cx(0, 1) - - check_nat_gates = GatesInBasis(basis_gates=device.operation_names) - check_nat_gates(qc) - assert not check_nat_gates.property_set["all_gates_in_basis"] - - factory = cast("Callable[[Target], Callable[[Circuit], Circuit]]", action_bqskit_synthesis_action.transpile_pass) - lambda_ = factory(device) - bqskit_qc = qiskit_to_bqskit(qc) - if "rigetti" in device.description or "ionq" in device.description: - with pytest.raises(ValueError, match=re.escape("not supported in BQSKIT")): - bqskit_qc_compiled = lambda_(bqskit_qc) - return - bqskit_qc_compiled = lambda_(bqskit_qc) - assert isinstance(bqskit_qc_compiled, Circuit) - native_gates_qc = bqskit_to_qiskit(bqskit_qc_compiled) - - check_nat_gates = GatesInBasis(basis_gates=device.operation_names) - check_nat_gates(native_gates_qc) - only_nat_gates = check_nat_gates.property_set["all_gates_in_basis"] - assert only_nat_gates - - -def test_bqskit_mapping_action_swaps_necessary(available_actions_dict: dict[PassType, list[Action]]) -> None: - """Test the BQSKitMapping action for quantum circuit that requires SWAP gates.""" - bqskit_mapping_action = None - for action in available_actions_dict[PassType.MAPPING]: - if action.name == "BQSKitMapping": - bqskit_mapping_action = action - - assert bqskit_mapping_action is not None - - qc = QuantumCircuit(8) - qc.h(0) - qc.cx(0, 1) - qc.cx(0, 2) - qc.cx(0, 3) - qc.cx(0, 4) - qc.cx(0, 5) - qc.cx(0, 6) - qc.cx(0, 7) - - device = get_device("ibm_falcon_27") - bqskit_qc = qiskit_to_bqskit(qc) - factory = cast( - "Callable[[Target], Callable[[Circuit], tuple[Circuit, tuple[int, ...], tuple[int, ...]]]]", - bqskit_mapping_action.transpile_pass, +def _lay_out(circuit: QuantumCircuit, target: Target) -> tuple[QuantumCircuit, TranspileLayout]: + """Apply a trivial Qiskit layout to the circuit.""" + coupling_map = target.build_coupling_map() + layout_pm = PassManager([ + TrivialLayout(coupling_map), + FullAncillaAllocation(coupling_map), + EnlargeWithAncilla(), + ApplyLayout(), + ]) + laid_out = layout_pm.run(circuit.copy()) + layout = TranspileLayout( + initial_layout=layout_pm.property_set["layout"], + input_qubit_mapping=dict(layout_pm.property_set["original_qubit_indices"]), + final_layout=layout_pm.property_set.get("final_layout"), + _output_qubit_list=laid_out.qubits, + _input_qubit_count=circuit.num_qubits, ) - bqskit_qc_mapped, input_mapping, output_mapping = factory(device)(bqskit_qc) - mapped_qc = bqskit_to_qiskit(bqskit_qc_mapped) - layout = final_layout_bqskit_to_qiskit(input_mapping, output_mapping, mapped_qc, qc) - - assert input_mapping != output_mapping - assert layout.final_layout is not None - check_mapped_circuit(initial_qc=qc, mapped_qc=mapped_qc, device=device, layout=layout) - - -def check_mapped_circuit( - initial_qc: QuantumCircuit, mapped_qc: QuantumCircuit, device: Target, layout: TranspileLayout -) -> None: - """Check if the mapped quantum circuit is correctly mapped to the device.""" - # check if the altered circuit is correctly mapped to the device - check_mapping = CheckMap(coupling_map=device.build_coupling_map()) - check_mapping(mapped_qc) - mapped = check_mapping.property_set["is_swap_mapped"] - assert mapped - assert mapped_qc != initial_qc - assert layout is not None - assert len(layout.initial_layout) == device.num_qubits - if layout.final_layout is not None: - assert len(layout.final_layout) == device.num_qubits - - # each qubit of the initial layout is part of the initial quantum circuit and the register name is correctly set - for assigned_physical_qubit in layout.initial_layout._p2v.values(): # noqa: SLF001 - qreg = assigned_physical_qubit._register # noqa: SLF001 - assert qreg.name in {"q", "ancilla"} - - # assigned_physical_qubit is part of the original quantum circuit - if qreg.name == "q": - assert qreg.size == initial_qc.num_qubits - # each qubit is also part of the initial uncompiled quantum circuit - assert initial_qc.find_bit(assigned_physical_qubit).registers[0][0].name == "q" - # assigned_physical_qubit is an ancilla qubit - else: - assert qreg.size == device.num_qubits - initial_qc.num_qubits - # each qubit of the final layout is part of the mapped quantum circuit and the register name is correctly set - if layout.final_layout is not None: - for assigned_physical_qubit in layout.final_layout._p2v.values(): # noqa: SLF001 - assert mapped_qc.find_bit(assigned_physical_qubit).registers[0][0].name == "q" - # each virtual qubit of the original quantum circuit is part of the initial layout - for virtual_qubit in initial_qc.qubits: - assert virtual_qubit in layout.initial_layout._p2v.values() # noqa: SLF001 - - -def test_bqskit_mapping_action_no_swaps_necessary(available_actions_dict: dict[PassType, list[Action]]) -> None: - """Test the BQSKitMapping action for a simple quantum circuit that does not require SWAP gates.""" - bqskit_mapping_action: Action | None = None - for action in available_actions_dict[PassType.MAPPING]: - if action.name == "BQSKitMapping": - bqskit_mapping_action = action - assert bqskit_mapping_action is not None - - qc_no_swap_needed = QuantumCircuit(2) - qc_no_swap_needed.h(0) - qc_no_swap_needed.cx(0, 1) - - device = get_device("quantinuum_h2_56") - - bqskit_qc = qiskit_to_bqskit(qc_no_swap_needed) - factory = cast( - "Callable[[Target], Callable[[Circuit], tuple[Circuit, tuple[int, ...], tuple[int, ...]]]]", - bqskit_mapping_action.transpile_pass, + return laid_out, layout + + +def _route(circuit: QuantumCircuit, layout: TranspileLayout, target: Target) -> tuple[QuantumCircuit, TranspileLayout]: + """Route the laid-out circuit with SabreSwap.""" + coupling_map = target.build_coupling_map() + routing_pm = PassManager([SabreSwap(coupling_map=coupling_map)]) + routed = routing_pm.run(circuit.copy()) + routed_layout = TranspileLayout( + initial_layout=layout.initial_layout, + input_qubit_mapping=dict(layout.input_qubit_mapping), + final_layout=routing_pm.property_set.get("final_layout"), + _output_qubit_list=routed.qubits, + _input_qubit_count=len(layout.input_qubit_mapping), ) - bqskit_qc_mapped, input_mapping, output_mapping = factory(device)(bqskit_qc) - mapped_qc = bqskit_to_qiskit(bqskit_qc_mapped) - layout = final_layout_bqskit_to_qiskit(input_mapping, output_mapping, mapped_qc, qc_no_swap_needed) - assert layout is not None - assert input_mapping == output_mapping - assert layout.final_layout is None + return routed, routed_layout + + +def _synthesize( + circuit: QuantumCircuit, layout: TranspileLayout, target: Target +) -> tuple[QuantumCircuit, TranspileLayout]: + """Translate the circuit to the target basis without changing its layout.""" + synthesis_pm = PassManager([BasisTranslator(StandardEquivalenceLibrary, target_basis=target.operation_names)]) + synthesized = synthesis_pm.run(circuit.copy()) + synthesized_layout = TranspileLayout( + initial_layout=layout.initial_layout, + input_qubit_mapping=dict(layout.input_qubit_mapping), + final_layout=layout.final_layout, + _output_qubit_list=synthesized.qubits, + _input_qubit_count=len(layout.input_qubit_mapping), + ) + return synthesized, synthesized_layout + - check_mapped_circuit(qc_no_swap_needed, mapped_qc, device, layout) +@pytest.fixture +def target() -> Target: + """Fixture to provide the target device for testing.""" + return get_device("ibm_falcon_27") -def test_tket_routing(available_actions_dict: dict[PassType, list[Action]]) -> None: - """Test the TKETRouting action.""" - qc = QuantumCircuit(5) +@pytest.fixture +def simple_circuit() -> QuantumCircuit: + """Return a small circuit used to probe action invariants. + + CX(0, 2) is intentional: qubits 0 and 2 are not adjacent on ibm_falcon_27 + (qubit 0 only connects to 1), so SabreSwap inserts at least one SWAP. + This ensures the routed fixture carries a real SWAP so routing-preservation + checks are non-trivial. + """ + qc = QuantumCircuit(3) qc.h(0) - qc.cx(0, 1) qc.cx(0, 2) - qc.cx(0, 3) - qc.cx(0, 4) - - device = get_device("quantinuum_h2_56") - - layout_action = available_actions_dict[PassType.LAYOUT][0] - factory = cast("Callable[[Target], list[Task]]", layout_action.transpile_pass) - passes_ = factory(device) - pm = PassManager(passes_) - layouted_qc = pm.run(qc) - initial_layout = pm.property_set["layout"] - input_qubit_mapping = pm.property_set["original_qubit_indices"] - - routing_action: Action | None = None - for action in available_actions_dict[PassType.ROUTING]: - if action.origin == CompilationOrigin.TKET: - routing_action = action - assert routing_action is not None - - tket_qc = qiskit_to_tk(layouted_qc, preserve_param_uuid=True) - factory = cast( - "Callable[[Target], list[TketBasePass | PreProcessTKETRoutingAfterQiskitLayout]]", routing_action.transpile_pass - ) - passes = factory(device) - for pass_ in passes: - pass_.apply(tket_qc) - - qbs = tket_qc.qubits - tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) + qc.cx(1, 2) + return qc - mapped_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) - final_layout = final_layout_pytket_to_qiskit(tket_qc, mapped_qc) +@pytest.fixture +def env(target: Target) -> PredictorEnv: + """Create a PredictorEnv for state-based invariant checking.""" + return PredictorEnv(device=target, reward_function="expected_fidelity") - layout = TranspileLayout( - initial_layout=initial_layout, - input_qubit_mapping=input_qubit_mapping, - final_layout=final_layout, - _output_qubit_list=mapped_qc.qubits, - _input_qubit_count=qc.num_qubits, - ) - check_mapped_circuit(qc, mapped_qc, device, layout) +def test_synthesis_actions_produce_native_gates( + simple_circuit: QuantumCircuit, + env: PredictorEnv, +) -> None: + """Invariant: every synthesis action produces only native gates for all circuit states.""" + n_qubits = simple_circuit.num_qubits + qc_laid_out, laid_layout = _lay_out(simple_circuit, env.device) + qc_routed, routed_layout = _route(qc_laid_out, laid_layout, env.device) + + test_cases = [ + ("uncompiled", simple_circuit, None), + ("laid-out", qc_laid_out, laid_layout), + ("routed", qc_routed, routed_layout), + ] + + for idx, action in env.action_set.items(): + if action.pass_type != PassType.SYNTHESIS: + continue + for kind, circuit, layout in test_cases: + _setup_env(env, circuit, layout, n_qubits) + if not _is_available(env, idx): + continue + compiled = env.apply_action(idx) + assert env.is_circuit_synthesized(compiled), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT: " + f"synthesis produced non-native gates for {kind} circuit. " + f"Device native gates: {env.device.operation_names}. " + f"Circuit gates: {set(compiled.count_ops().keys())}" + ) + + +def test_layout_actions_establish_layout( + simple_circuit: QuantumCircuit, + env: PredictorEnv, +) -> None: + """Invariant: every layout action establishes a valid qubit assignment.""" + n_qubits = simple_circuit.num_qubits + + for idx, action in env.action_set.items(): + if action.pass_type != PassType.LAYOUT: + continue + _setup_env(env, simple_circuit, None, n_qubits) + if not _is_available(env, idx): + continue + compiled = env.apply_action(idx) + assert env.layout is not None, ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT: failed to establish layout" + ) + assert env.is_circuit_laid_out(compiled, env.layout), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT: " + f"did not establish valid layout. Layout: {env.layout}" + ) + + +def test_routing_actions_route_circuit( + simple_circuit: QuantumCircuit, + env: PredictorEnv, +) -> None: + """Invariant: every routing action produces a circuit where all 2-qubit gates respect the coupling map.""" + qc_laid_out, layout = _lay_out(simple_circuit, env.device) + n_qubits = qc_laid_out.num_qubits + coupling_map = env.device.build_coupling_map() + + for idx, action in env.action_set.items(): + if action.pass_type != PassType.ROUTING: + continue + _setup_env(env, qc_laid_out, layout, n_qubits) + if not _is_available(env, idx): + continue + routed = env.apply_action(idx) + assert env.is_circuit_routed(routed, coupling_map), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT: circuit not properly routed after action" + ) + + +def test_optimization_actions_preserve_invariants( + simple_circuit: QuantumCircuit, + env: PredictorEnv, +) -> None: + """Invariant: OPT actions honour their declared preserves_layout/routing/synthesis contracts.""" + qc_laid_out, laid_layout = _lay_out(simple_circuit, env.device) + qc_routed, layout = _route(qc_laid_out, laid_layout, env.device) + qc_synthesized, layout_synth = _synthesize(qc_routed, layout, env.device) + n_qubits = qc_routed.num_qubits + coupling_map = env.device.build_coupling_map() + + for idx, action in env.action_set.items(): + if action.pass_type != PassType.OPT: + continue + + if action.preserves_layout: + pre_v2p = dict(layout.initial_layout.get_virtual_bits()) + _setup_env(env, qc_routed, layout, n_qubits) + if _is_available(env, idx): + compiled = env.apply_action(idx) + assert env.is_circuit_laid_out(compiled, layout), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT preserves_layout: " + f"circuit no longer has a valid layout after action" + ) + assert env.layout is not None, ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT preserves_layout: " + f"layout metadata was removed" + ) + post_v2p = dict(env.layout.initial_layout.get_virtual_bits()) + assert post_v2p == pre_v2p, ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT preserves_layout: " + f"initial qubit assignment changed. Before: {pre_v2p}, After: {post_v2p}" + ) + + if action.preserves_routing: + _setup_env(env, qc_routed, layout, n_qubits) + if _is_available(env, idx): + compiled = env.apply_action(idx) + assert env.is_circuit_routed(compiled, coupling_map), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT preserves_routing: " + f"produced gates on non-adjacent qubits" + ) + + if action.preserves_synthesis: + _setup_env(env, qc_synthesized, layout_synth, n_qubits) + if _is_available(env, idx): + compiled = env.apply_action(idx) + assert env.is_circuit_synthesized(compiled), ( + f"{action.name} on {env.device.description} VIOLATED INVARIANT preserves_synthesis: " + f"introduced non-native gates. " + f"Device native gates: {env.device.operation_names}. " + f"Circuit gates: {set(compiled.count_ops().keys())}" + ) diff --git a/tests/compilation/test_predictor_rl.py b/tests/compilation/test_predictor_rl.py index 5cebe65d8..763ea54f9 100644 --- a/tests/compilation/test_predictor_rl.py +++ b/tests/compilation/test_predictor_rl.py @@ -22,6 +22,7 @@ from qiskit.transpiler import InstructionProperties, Layout, Target, TranspileLayout from qiskit.transpiler.passes import GatesInBasis +import mqt.predictor.rl.actions.qiskit_actions as qiskit_actions from mqt.predictor.rl import Predictor, rl_compile from mqt.predictor.rl import predictorenv as predictorenv_module from mqt.predictor.rl.actions import ( @@ -171,15 +172,16 @@ def __init__(self, _passes: object) -> None: def run(self, circuit: QuantumCircuit) -> QuantumCircuit: return circuit - monkeypatch.setattr(predictorenv_module, "PassManager", FakePassManager) + monkeypatch.setattr(qiskit_actions, "PassManager", FakePassManager) action = DeviceIndependentAction( name="SyntheticQiskitRouting", pass_type=PassType.ROUTING, transpile_pass=[], origin=CompilationOrigin.QISKIT, ) - - altered_qc = env._apply_qiskit_action(action, env.actions_routing_indices[0]) # noqa: SLF001 + routing_action_index = next(iter(env.actions_routing_indices)) + env.action_set[routing_action_index] = action + altered_qc = env.apply_action(action_index=routing_action_index) assert altered_qc is env.state assert env.layout.final_layout is final_layout