diff --git a/CHANGELOG.md b/CHANGELOG.md index fc26c5b61..15fa023a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,8 @@ releases may include breaking changes. ### Changed +- ✨ Enable configurable intermediate rewards for RL training by default + ([#799]) ([**@flowerthrower**]) - 🐛 Restore nondeterministic RL circuit sampling without a seed and make explicitly seeded sampling and randomized Qiskit actions reproducible ([#797]) ([**@flowerthrower**]) @@ -116,6 +118,7 @@ for previous changelogs._ [#796]: https://github.com/munich-quantum-toolkit/predictor/pull/796 [#795]: https://github.com/munich-quantum-toolkit/predictor/pull/795 [#794]: https://github.com/munich-quantum-toolkit/predictor/pull/794 +[#799]: https://github.com/munich-quantum-toolkit/predictor/pull/799 [#758]: https://github.com/munich-quantum-toolkit/predictor/pull/758 [#755]: https://github.com/munich-quantum-toolkit/predictor/pull/755 [#731]: https://github.com/munich-quantum-toolkit/predictor/pull/731 diff --git a/UPGRADING.md b/UPGRADING.md index bda476727..ca3689bd4 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -6,6 +6,15 @@ of changes including minor and patch releases, please refer to the ## [Unreleased] +### Intermediate RL rewards + +`PredictorEnv` now enables intermediate rewards by default. For +`expected_fidelity` and `estimated_success_probability`, comparable non-terminal +steps are rewarded from changes in the exact or approximate figure of merit. +Optimization actions with no measurable change receive `-0.001`. Set +`intermediate_reward=False` to retain the previous terminal-only reward +behavior; use `reward_scale` and `no_effect_penalty` to tune reward shaping. + ### Expanded TKET action set The RL action space now includes TKET's `KAKDecomposition` optimization action diff --git a/src/mqt/predictor/reward.py b/src/mqt/predictor/reward.py index 225b05c2b..e1248294a 100644 --- a/src/mqt/predictor/reward.py +++ b/src/mqt/predictor/reward.py @@ -15,13 +15,15 @@ import numpy as np from joblib import load -from qiskit import __version__ as qiskit_version -from qiskit import transpile +from qiskit.transpiler import InstructionDurations, PassManager +from qiskit.transpiler.passes import ASAPScheduleAnalysis from mqt.predictor.hellinger import calc_device_specific_features, get_hellinger_model_path from mqt.predictor.utils import calc_supermarq_features if TYPE_CHECKING: + from collections.abc import Iterable + from qiskit import QuantumCircuit from qiskit.transpiler import Target from sklearn.ensemble import RandomForestRegressor @@ -91,124 +93,75 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: Returns: The expected success probability of the given quantum circuit on the given device. """ - exec_time_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0) - - op_times, active_qubits = [], set() + operation_times: list[tuple[str, Iterable[int] | None, float, str]] = [] for instr in qc.data: - instruction = instr.operation - qargs = instr.qubits - gate_type = instruction.name - - if gate_type == "barrier" or gate_type == "id": + gate_type = str(instr.operation.name) + if gate_type in {"barrier", "id"}: continue - assert len(qargs) in (1, 2) - first_qubit_idx = qc.find_bit(qargs[0]).index - active_qubits.add(first_qubit_idx) - - if len(qargs) == 1: # single-qubit gate - duration = device[gate_type][first_qubit_idx,].duration - op_times.append(( - gate_type, - [ - first_qubit_idx, - ], - duration, - "s", - )) - exec_time_per_qubit[first_qubit_idx] += duration - else: # multi-qubit gate - second_qubit_idx = qc.find_bit(qargs[1]).index - active_qubits.add(second_qubit_idx) - duration = device[gate_type][first_qubit_idx, second_qubit_idx].duration - op_times.append((gate_type, [first_qubit_idx, second_qubit_idx], duration, "s")) - exec_time_per_qubit[first_qubit_idx] += duration - exec_time_per_qubit[second_qubit_idx] += duration - - if qiskit_version < "2.0.0": - from qiskit.transpiler import ( # ruff:ignore[import-outside-top-level] - InstructionDurations, - Layout, - PassManager, - passes, - ) - from qiskit.transpiler.passes import ApplyLayout, SetLayout # ruff:ignore[import-outside-top-level] - - if qc.qregs[0].name != "q": - # create a layout that maps the (tket) 'node' registers to the (qiskit) 'q' registers - layouts = [ - SetLayout(Layout({node_qubit: i for i, node_qubit in enumerate(node_reg)})) for node_reg in qc.qregs - ] - # create a pass manager with the SetLayout and ApplyLayout passes - pm = PassManager(list(layouts)) - pm.append(ApplyLayout()) - - # replace the 'node' register with the 'q' register in the circuit - qc = pm.run(qc) - assert qc.qregs[0].name == "q" - - sched_pass = passes.ASAPScheduleAnalysis(InstructionDurations(op_times)) - delay_pass = passes.PadDelay() - pm = PassManager([sched_pass, delay_pass]) - scheduled_circ = pm.run(qc) - - else: - scheduled_circ = transpile( - qc, - target=device, - scheduling_method="asap", - optimization_level=0, - initial_layout=None, - routing_method=None, - layout_method=None, - ) - overall_estimated_duration = scheduled_circ.estimate_duration(target=device) + qubit_indices = [int(qc.find_bit(qubit).index) for qubit in instr.qubits] + properties = device[gate_type].get(tuple(qubit_indices)) + if properties is None or properties.duration is None: + msg = f"Duration for gate {gate_type} on qubits {tuple(qubit_indices)} not found in device properties." + raise ValueError(msg) + operation_times.append((gate_type, qubit_indices, float(properties.duration), "s")) + + durations = InstructionDurations(operation_times, dt=device.dt) + pass_manager = PassManager([ASAPScheduleAnalysis(durations=durations)]) + pass_manager.run(qc) + + time_unit = pass_manager.property_set["time_unit"] + execution_time_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0) + last_end_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0) + last_operation_per_qubit = dict.fromkeys(range(device.num_qubits), "") + circuit_duration = 0.0 + + for node, start_time in pass_manager.property_set["node_start_time"].items(): + qubit_indices = [qc.find_bit(qubit).index for qubit in node.qargs] + duration = float(durations.get(node.name, qubit_indices, unit=time_unit)) + end_time = float(start_time) + duration + circuit_duration = max(circuit_duration, end_time) + for qubit in qubit_indices: + execution_time_per_qubit[qubit] += duration + if end_time >= last_end_per_qubit[qubit]: + last_end_per_qubit[qubit] = end_time + last_operation_per_qubit[qubit] = node.name res = 1.0 - for instr in scheduled_circ.data: + active_qubits = set() + for instr in qc.data: instruction = instr.operation qargs = instr.qubits gate_type = instruction.name - if gate_type == "barrier" or gate_type == "id": + if gate_type in {"barrier", "id"}: continue assert len(qargs) in (1, 2) - first_qubit_idx = scheduled_circ.find_bit(qargs[0]).index + qubit_indices = [qc.find_bit(qubit).index for qubit in qargs] + active_qubits.update(qubit_indices) + first_qubit_idx = qubit_indices[0] if len(qargs) == 1: - if gate_type == "measure": - res *= 1 - device[gate_type][first_qubit_idx,].error - continue - if gate_type == "delay": - if qiskit_version < "2.0.0": - continue - # only consider active qubits - if first_qubit_idx not in active_qubits: - continue - - dt = device.dt # instruction durations are stored in unit dt - res *= np.exp( - -instruction.duration - * dt - / min(device.qubit_properties[first_qubit_idx].t1, device.qubit_properties[first_qubit_idx].t2) - ) - continue res *= 1 - device[gate_type][first_qubit_idx,].error else: - second_qubit_idx = scheduled_circ.find_bit(qargs[1]).index + second_qubit_idx = qubit_indices[1] try: res *= 1 - device[gate_type][first_qubit_idx, second_qubit_idx].error except KeyError: msg = f"Error rate for gate {gate_type} on qubits {first_qubit_idx} and {second_qubit_idx} not found in device properties." raise KeyError(msg) from None - if qiskit_version >= "2.0.0": - for i in range(device.num_qubits): - qubit_execution_time = exec_time_per_qubit[i] - if qubit_execution_time == 0: - continue - idle_time = overall_estimated_duration - qubit_execution_time - res *= np.exp(-idle_time / min(device.qubit_properties[i].t1, device.qubit_properties[i].t2)) + assert device.qubit_properties is not None + for qubit in active_qubits: + properties = device.qubit_properties[qubit] + assert properties is not None + assert properties.t1 is not None + assert properties.t2 is not None + live_end = ( + last_end_per_qubit[qubit] if last_operation_per_qubit[qubit] in {"measure", "reset"} else circuit_duration + ) + idle_time = max(live_end - execution_time_per_qubit[qubit], 0.0) + res *= np.exp(-idle_time / min(properties.t1, properties.t2)) return float(np.round(res, precision).item()) diff --git a/src/mqt/predictor/rl/approx_reward.py b/src/mqt/predictor/rl/approx_reward.py new file mode 100644 index 000000000..9885e1ca4 --- /dev/null +++ b/src/mqt/predictor/rl/approx_reward.py @@ -0,0 +1,185 @@ +# 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 + +"""Approximate reward calculations for intermediate RL compilation states.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import numpy as np +from qiskit.circuit.equivalence_library import SessionEquivalenceLibrary +from qiskit.transpiler import PassManager +from qiskit.transpiler.passes import BasisTranslator + +if TYPE_CHECKING: + from qiskit import QuantumCircuit + from qiskit.transpiler import InstructionProperties, Target + + +_EXCLUDED_OPERATIONS = { + "barrier", + "delay", + "id", + "if_else", + "while_loop", + "for_loop", + "switch_case", + "box", + "break", + "continue", +} + + +def _operation_arity(device: Target, name: str) -> int | None: + """Return the arity of a target operation if it behaves like a gate.""" + try: + operation = device.operation_from_name(name) + except KeyError: + return None + + try: + return int(operation.num_qubits) + except (AttributeError, TypeError, ValueError): + return None + + +def _basis_gates(device: Target) -> list[str]: + """Return gate-like target operations used for approximate rewards.""" + basis_gates = [ + name + for name in device.operation_names + if name not in _EXCLUDED_OPERATIONS and _operation_arity(device, name) is not None + ] + if "reset" in basis_gates and not device["reset"]: + basis_gates.remove("reset") + return sorted(basis_gates) + + +def _basis_gate_counts(qc: QuantumCircuit, basis_gates: list[str]) -> dict[str, int]: + """Translate a circuit to the target basis and count its operations.""" + translated = PassManager([BasisTranslator(SessionEquivalenceLibrary, basis_gates)]).run(qc) + counts = dict.fromkeys(basis_gates, 0) + for instruction in translated.data: + name = instruction.operation.name + if name in counts: + counts[name] += 1 + return counts + + +def approximate_expected_fidelity( + qc: QuantumCircuit, + *, + device: Target, + error_rates: dict[str, float], +) -> float: + """Approximate expected fidelity using average per-gate error rates.""" + counts = _basis_gate_counts(qc, _basis_gates(device)) + fidelity = 1.0 + for gate, count in counts.items(): + fidelity *= (1.0 - error_rates.get(gate, 0.0)) ** count + return float(np.clip(fidelity, 0.0, 1.0)) + + +def approximate_estimated_success_probability( + qc: QuantumCircuit, + *, + device: Target, + error_rates: dict[str, float], + gate_durations: dict[str, float], + coherence_time: float | None, + parallelism: float, + liveness: float, +) -> float: + """Approximate ESP from gate errors, duration, and circuit-level features.""" + basis_gates = _basis_gates(device) + counts = _basis_gate_counts(qc, basis_gates) + + gate_fidelity = 1.0 + for gate, count in counts.items(): + gate_fidelity *= (1.0 - error_rates.get(gate, 0.0)) ** count + + effective_parallelism = 1.0 + (max(qc.num_qubits, 1) - 1.0) * parallelism + total_gate_time = sum(counts[gate] * gate_durations.get(gate, 0.0) for gate in basis_gates) / effective_parallelism + idle_fraction = max(0.0, 1.0 - liveness) + idle_factor = ( + 1.0 + if coherence_time is None or coherence_time <= 0.0 + else float(np.exp(-(total_gate_time * idle_fraction) / coherence_time)) + ) + return float(np.clip(gate_fidelity * idle_factor, 0.0, 1.0)) + + +def average_target_calibration( + device: Target, +) -> tuple[dict[str, float], dict[str, float], float | None]: + """Return per-gate error/duration averages and a representative coherence time.""" + try: + num_qubits = device.num_qubits + coupling_map = device.build_coupling_map() + qubit_properties = device.qubit_properties + except AttributeError as exc: + msg = "Device target does not expose the required API for approximate reward computation." + raise RuntimeError(msg) from exc + + basis_gates = _basis_gates(device) + edges = coupling_map.get_edges() if coupling_map is not None else [] + + def get_properties(name: str, qubits: tuple[int, ...]) -> InstructionProperties | None: + return device[name].get(qubits) + + error_samples: dict[str, list[float]] = {name: [] for name in basis_gates} + duration_samples: dict[str, list[float]] = {name: [] for name in basis_gates} + + for name in basis_gates: + arity = _operation_arity(device, name) + qubit_tuples: list[tuple[int, ...]] + if arity == 1: + qubit_tuples = [(qubit,) for qubit in range(num_qubits)] + elif arity == 2: + qubit_tuples = [tuple(edge) for edge in edges] + else: + continue + + for qubits in qubit_tuples: + properties = get_properties(name, qubits) + if properties is None and len(qubits) == 2: + properties = get_properties(name, (qubits[1], qubits[0])) + if properties is None: + continue + if properties.error is not None: + error_samples[name].append(float(properties.error)) + if properties.duration is not None: + duration_samples[name].append(float(properties.duration)) + + all_errors = [error for samples in error_samples.values() for error in samples] + all_durations = [duration for samples in duration_samples.values() for duration in samples] + if not all_errors and not all_durations: + msg = "No valid calibration data found in target; cannot compute approximate reward." + raise RuntimeError(msg) + + fallback_error = float(np.mean(all_errors)) if all_errors else 0.0 + fallback_duration = float(np.mean(all_durations)) if all_durations else 0.0 + error_rates = { + name: float(np.mean(samples)) if samples else fallback_error for name, samples in error_samples.items() + } + gate_durations = { + name: float(np.mean(samples)) if samples else fallback_duration for name, samples in duration_samples.items() + } + + coherence_samples: list[float] = [] + if qubit_properties: + for properties in qubit_properties: + if properties is None: + continue + values = [value for value in (properties.t1, properties.t2) if value is not None] + if values: + coherence_samples.append(float(min(values))) + + coherence_time = float(np.median(coherence_samples)) if coherence_samples else None + return error_rates, gate_durations, coherence_time diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 1f5a0e891..d0b494a97 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -17,6 +17,7 @@ import threading import time import warnings +from math import isclose from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, get_args @@ -52,6 +53,11 @@ 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.approx_reward import ( + approximate_estimated_success_probability, + approximate_expected_fidelity, + average_target_calibration, +) from mqt.predictor.rl.helper import ( OBSERVATION_OPERATIONS, create_feature_dict, @@ -59,6 +65,7 @@ get_state_sample, ) from mqt.predictor.rl.tracer import CompilationTracer, FigureOfMeritMetric, FigureOfMeritMetrics +from mqt.predictor.utils import calc_supermarq_features logger = logging.getLogger("mqt-predictor") @@ -114,6 +121,9 @@ def __init__( max_steps: int | None = None, tracer_output_path: str | Path | None = None, mdp: MDPPolicy = "v3", + intermediate_reward: bool = True, + reward_scale: float = 1.0, + no_effect_penalty: float = -0.001, pass_timeout: float | None = None, ) -> None: """Initializes the PredictorEnv object. @@ -127,6 +137,11 @@ def __init__( mdp: The MDP transition policy. ``v2`` is the original MQT Predictor strategy. ``v3`` is the default and is permissive before layout while preserving the established compilation structure afterwards. + intermediate_reward: Whether to reward changes to the figure of merit + before termination. + reward_scale: Multiplier applied to intermediate reward changes. + no_effect_penalty: Reward for optimization actions that do not improve + the figure of merit. pass_timeout: Maximum duration in seconds for one compilation pass. Defaults to None, which disables pass timeouts. @@ -147,6 +162,9 @@ def __init__( self.path_training_circuits = path_training_circuits or get_path_training_circuits() self.max_steps = max_steps self.mdp = mdp + self.intermediate_reward = intermediate_reward + self.reward_scale = reward_scale + self.no_effect_penalty = no_effect_penalty self.pass_timeout = pass_timeout self.action_set = {} @@ -246,6 +264,7 @@ def __init__( self._current_laid_out = False self._current_routed = False self._current_foms: dict[str, float] = {} + self._approximate_calibration: tuple[dict[str, float], dict[str, float], float | None] | None = None self.observation_space = Dict(spaces) self.filename = "" @@ -353,10 +372,20 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any action_obj = self.action_set[action] action_name = str(action_obj.name) action_type = action_obj.pass_type.value + previous_compilation_state = ( + self._current_synthesized, + self._current_laid_out, + self._current_routed, + ) start_time = time.perf_counter() try: self.used_actions.append(action_name) + previous_reward = ( + self._get_stepwise_reward() + if self.intermediate_reward and action != self.action_terminate_index + else None + ) with _enforce_pass_timeout(self.pass_timeout): altered_qc = self.apply_action(action) action_duration = time.perf_counter() - start_time @@ -372,13 +401,13 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any action_name=action_name, action_type=action_type, action_duration=action_duration, - reward_val=0.0, + reward_val=self.no_effect_penalty, feature_vector=obs, done=True, ) return ( obs, # features - 0, # reward + self.no_effect_penalty, # reward False, # terminated True, # truncated {"Truncated because of error": f"{type(exc).__name__}: {exc}"}, # info @@ -400,6 +429,13 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any if action == self.action_terminate_index: reward_val = self.calculate_reward() done = True + elif previous_reward is not None: + reward_val = self._calculate_intermediate_reward( + action, + previous_reward, + previous_compilation_state, + ) + done = False else: reward_val = 0 done = False @@ -456,6 +492,82 @@ def calculate_reward(self) -> float: """Calculates and returns the reward for the current state.""" return self.get_fom(self.reward_function) + def _get_stepwise_reward(self) -> tuple[float, str]: + """Return the current reward and whether it is exact or approximate.""" + if self.reward_function == "critical_depth": + return self.calculate_reward(), "exact" + if self.reward_function not in {"expected_fidelity", "estimated_success_probability"}: + return 0.0, "unavailable" + + if self._current_synthesized and self._current_laid_out and self._current_routed: + return self.calculate_reward(), "exact" + return self._approximate_reward(), "approximate" + + def _approximate_reward(self) -> float: + """Estimate the current figure of merit before exact evaluation is possible.""" + cache_key = f"approximate_{self.reward_function}" + if cache_key in self._current_foms: + return self._current_foms[cache_key] + + if self._approximate_calibration is None: + self._approximate_calibration = average_target_calibration(self.device) + error_rates, gate_durations, coherence_time = self._approximate_calibration + + if self.reward_function == "expected_fidelity": + reward = approximate_expected_fidelity( + self.state, + device=self.device, + error_rates=error_rates, + ) + else: + features = calc_supermarq_features(self.state) + reward = approximate_estimated_success_probability( + self.state, + device=self.device, + error_rates=error_rates, + gate_durations=gate_durations, + coherence_time=coherence_time, + parallelism=float(features.parallelism), + liveness=float(features.liveness), + ) + + self._current_foms[cache_key] = reward + return reward + + def _calculate_intermediate_reward( + self, + action: int, + previous_reward: tuple[float, str], + previous_compilation_state: tuple[bool, bool, bool], + ) -> float: + """Calculate the shaped reward for a non-terminal action.""" + previous_value, previous_kind = previous_reward + current_value, current_kind = self._get_stepwise_reward() + pass_type = self.action_set[action].pass_type + structural_progress = pass_type in { + PassType.SYNTHESIS, + PassType.LAYOUT, + PassType.ROUTING, + PassType.MAPPING, + } and any( + not previous and current + for previous, current in zip( + previous_compilation_state, + (self._current_synthesized, self._current_laid_out, self._current_routed), + strict=True, + ) + ) + if "unavailable" in {previous_kind, current_kind} or previous_kind != current_kind or structural_progress: + return 0.0 + + delta = current_value - previous_value + if not isclose(delta, 0.0, abs_tol=1e-12): + return self.reward_scale * delta + + if pass_type in {PassType.OPT, PassType.FINAL_OPT}: + return self.no_effect_penalty + return 0.0 + def render(self) -> None: """Renders the current state.""" print(self.state.draw()) diff --git a/tests/compilation/test_predictor_rl.py b/tests/compilation/test_predictor_rl.py index 1866f92ca..78598fa43 100644 --- a/tests/compilation/test_predictor_rl.py +++ b/tests/compilation/test_predictor_rl.py @@ -101,14 +101,16 @@ def test_predictor_env_rejects_nonpositive_pass_timeout(pass_timeout: float) -> @pytest.mark.skipif(not hasattr(signal, "SIGALRM"), reason="SIGALRM is unavailable") def test_predictor_env_truncates_timed_out_pass(monkeypatch: pytest.MonkeyPatch) -> None: """Test that a pass timeout truncates the current episode.""" - env = predictorenv_module.PredictorEnv(device=get_device("ibm_falcon_27"), pass_timeout=0.01) + env = predictorenv_module.PredictorEnv( + device=get_device("ibm_falcon_27"), pass_timeout=0.01, intermediate_reward=False + ) qc = QuantumCircuit(1) env.reset(qc) monkeypatch.setattr(env, "apply_action", lambda _action: time.sleep(1)) _, reward_val, terminated, truncated, info = env.step(env.actions_opt_indices[0]) - assert reward_val == 0 + assert reward_val == env.no_effect_penalty assert not terminated assert truncated assert info == { @@ -289,7 +291,7 @@ def test_warning_for_unidirectional_device() -> None: def test_predictor_env_truncates_at_max_steps() -> None: """Test that the environment truncates episodes that hit the step limit.""" device = get_device("ibm_falcon_27") - env = predictorenv_module.PredictorEnv(device=device, max_steps=1) + env = predictorenv_module.PredictorEnv(device=device, max_steps=1, intermediate_reward=False) qc = QuantumCircuit(1) qc.h(0) env.reset(qc) diff --git a/tests/compilation/test_reward.py b/tests/compilation/test_reward.py index e0843dbc9..c1ac95036 100644 --- a/tests/compilation/test_reward.py +++ b/tests/compilation/test_reward.py @@ -19,7 +19,6 @@ from qiskit.circuit.library import CXGate, Measure, XGate from qiskit.transpiler import InstructionProperties, Target -from mqt.predictor import reward as reward_module from mqt.predictor.reward import crit_depth, esp_data_available, estimated_success_probability, expected_fidelity try: @@ -133,34 +132,20 @@ def test_esp_data_available_invalid_target(kwargs: dict[str, float | bool]) -> N @pytest.mark.parametrize("reward_function", ["expected_fidelity", "estimated_success_probability"]) -def test_reward_missing_two_qubit_error(reward_function: str, monkeypatch: pytest.MonkeyPatch) -> None: - """Test that reward functions report missing two-qubit error rates descriptively.""" +def test_reward_missing_two_qubit_calibration(reward_function: str) -> None: + """Test that reward functions report missing two-qubit calibration data descriptively.""" target = make_target() del target["cx"][0, 1] qc = QuantumCircuit(2) + qc.cx(0, 1) if reward_function == "estimated_success_probability": - qc.x(0) - - scheduled_qc = QuantumCircuit(2) - scheduled_qc.cx(0, 1) - - def fake_transpile(*_args: object, **_kwargs: object) -> QuantumCircuit: - return scheduled_qc - - def estimate_duration(*, target: Target) -> float: - assert target.num_qubits == 2 - return 0.0 - - monkeypatch.setattr(reward_module, "qiskit_version", "2.0.0") - monkeypatch.setattr(reward_module, "transpile", fake_transpile) - monkeypatch.setattr( - scheduled_qc, "estimate_duration", estimate_duration, raising=False - ) # not available in qiskit<2.0 + error_type = ValueError + expected_message = "Duration for gate cx on qubits (0, 1) not found in device properties." else: - qc.cx(0, 1) + error_type = KeyError + expected_message = "Error rate for gate cx on qubits 0 and 1 not found in device properties." reward = estimated_success_probability if reward_function == "estimated_success_probability" else expected_fidelity - expected_message = "Error rate for gate cx on qubits 0 and 1 not found in device properties." - with pytest.raises(KeyError, match=re.escape(expected_message)): + with pytest.raises(error_type, match=re.escape(expected_message)): reward(qc, target)