From 170994f63ad3fd7c0bacc76588572511399fc809 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Thu, 4 Jun 2026 20:39:20 +0200 Subject: [PATCH 01/15] Add Qudit DAG. Qudit DAG should be used to handle the simulation of Qudits in YAQS --- src/mqt/yaqs/digital/utils/qudit_dag_utils.py | 372 ++++++++++++++++++ 1 file changed, 372 insertions(+) create mode 100644 src/mqt/yaqs/digital/utils/qudit_dag_utils.py diff --git a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py new file mode 100644 index 000000000..9d0c5be0d --- /dev/null +++ b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py @@ -0,0 +1,372 @@ +# Copyright (c) 2025 - 2026 Chair for Design Automation, TUM +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""DAG utilities for qudit quantum circuits (MQT Qudit integration). + +This module provides a directed acyclic graph (DAG) representation for qudit +circuits from ``mqt.qudits``. Unlike the Qiskit-based :mod:`dag_utils`, this +DAG works natively with :class:`mqt.qudits.quantum_circuit.QuantumCircuit` and +tracks which energy *levels* each gate acts on — information that Qiskit's +DAGCircuit cannot represent. + +Key classes and functions: + - :class:`QuditOpNode`: DAG node holding the original gate object and its level footprint. + - :class:`QuditDAG`: The DAG itself, built from a ``mqt.qudits.QuantumCircuit``. + - :func:`circuit_to_dag`: Convert a ``mqt.qudits.QuantumCircuit`` to a :class:`QuditDAG`. + - :func:`dag_to_circuit`: Reconstruct a ``mqt.qudits.QuantumCircuit`` from a :class:`QuditDAG`. + - :func:`check_longest_gate`: Longest qudit span in the front layer (mirrors ``dag_utils``). + - :func:`get_temporal_zone`: Extract gates that act only on a given qudit window. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Generator + +from mqt.qudits.quantum_circuit import QuantumCircuit as QC + +if TYPE_CHECKING: + from mqt.qudits.quantum_circuit import QuantumCircuit + from mqt.qudits.quantum_circuit.gate import Gate + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _levels_for_gate(gate: Gate, qudit_idx: int, dimension: int) -> list[int]: + """Return which energy levels *gate* acts on for a given qudit. + + MQT Qudit gates carry ``lev_a`` and ``lev_b`` (the two transition levels). + When both are 0 the gate does not have a meaningful two-level transition + (e.g. a full-unitary gate), so the full level range is returned. + + Args: + gate: The MQT Qudit gate object. + qudit_idx: Index of the qudit within the full circuit (unused — kept + for a future per-qudit override, and for API symmetry with the + multi-qudit case). + dimension: Physical dimension of that qudit. + + Returns: + Sorted list of level indices the gate touches. + """ + lev_a: int = getattr(gate, "lev_a", 0) + lev_b: int = getattr(gate, "lev_b", 0) + + if lev_a == lev_b: + return list(range(dimension)) + + return sorted({lev_a, lev_b}) + +class QuditDAGNode: + """Base node in a :class:`QuditDAG`. + + Attributes: + index: Unique integer index assigned at creation. + dependencies: Set of predecessor nodes (direct data dependencies). + """ + + def __init__(self, index: int) -> None: + self.index: int = index + self.dependencies: set[QuditDAGNode] = set() + + def add_dependency(self, node: QuditDAGNode) -> None: + """Register *node* as a direct predecessor.""" + self.dependencies.add(node) + + def __repr__(self) -> str: + return f"Node({self.index})" + + +class QuditOpNode(QuditDAGNode): + """DAG node representing a single quantum gate operation. + + Attributes: + op_name: QASM tag of the gate (e.g. ``"cx"``, ``"rz"``). + gate: Original MQT Qudit :class:`~mqt.qudits.quantum_circuit.gate.Gate` + object. Use ``node.gate.to_matrix()`` to obtain the unitary. + target_qudits: List of qudit indices the gate acts on. + dimensions: Physical dimension of each target qudit (same order). + levels: Mapping ``{qudit_index: [level, ...]}`` of touched levels per qudit. + params: Raw gate parameters (forwarded from the original gate). + """ + + def __init__( + self, + index: int, + gate: Gate, + target_qudits: list[int], + dimensions: list[int], + levels: dict[int, list[int]], + ) -> None: + super().__init__(index) + self.gate: Gate = gate + self.op_name: str = getattr(gate, "qasm_tag", "unknown") + self.target_qudits: list[int] = target_qudits + self.dimensions: list[int] = dimensions + self.levels: dict[int, list[int]] = levels + self.params: object = getattr(gate, "_params", None) + + @property + def qargs(self) -> list[int]: + return self.target_qudits + + @property + def op(self) -> Gate: + return self.gate + + def to_dict(self) -> dict: + return { + "op": self.op_name, + "targets": self.target_qudits, + "levels": self.levels, + "dims": self.dimensions, + "deps": [n.index for n in self.dependencies], + } + + def __repr__(self) -> str: + dep_ids = [n.index for n in self.dependencies] + return ( + f"OpNode({self.index}: {self.op_name}, " + f"targets={self.target_qudits}, levels={self.levels}, deps={dep_ids})" + ) + +class QuditDAG: + """Directed acyclic graph for a qudit quantum circuit. + + The DAG is built from a ``mqt.qudits.QuantumCircuit``: each gate becomes a + :class:`QuditOpNode` and edges encode data dependencies (two gates on the + same qudit are connected in program order). + + Args: + circuit: Source circuit. Pass ``None`` together with *dimensions* to + create an empty DAG (used internally by :meth:`copy_empty_like`). + dimensions: Physical dimensions to use when *circuit* is ``None``. + """ + + def __init__( + self, + circuit: QuantumCircuit | None = None, + dimensions: list[int] | None = None, + ) -> None: + if circuit is not None: + self.dimensions: list[int] = list(circuit.dimensions) + else: + self.dimensions = list(dimensions) if dimensions else [] + + self.circuit: QuantumCircuit | None = circuit + self.num_qudits: int = len(self.dimensions) + self.nodes: list[QuditOpNode] = [] + + if circuit is not None: + self._build(circuit) + + def _build(self, circuit: QuantumCircuit) -> None: + """Populate *nodes* and wire dependency edges.""" + last_node_per_qudit: list[QuditOpNode | None] = [None] * self.num_qudits + + for idx, instruction in enumerate(circuit.instructions): + targets: list[int] = getattr(instruction, "target_qudits", []) + if isinstance(targets, int): + targets = [targets] + + dims = [self.dimensions[q] for q in targets] + levels = {q: _levels_for_gate(instruction, q, self.dimensions[q]) for q in targets} + + node = QuditOpNode(idx, instruction, targets, dims, levels) + + for q in targets: + prev = last_node_per_qudit[q] + if prev is not None: + node.add_dependency(prev) + last_node_per_qudit[q] = node + + self.nodes.append(node) + + def op_nodes(self) -> list[QuditOpNode]: + """Return all remaining gate nodes.""" + return self.nodes + + def front_layer(self) -> list[QuditOpNode]: + """Return nodes whose dependencies have all been removed.""" + remaining = set(self.nodes) + return [n for n in self.nodes if not (n.dependencies & remaining)] + + def remove_op_node(self, node: QuditOpNode) -> None: + """Remove *node* and clean up dependency references in successors.""" + if node not in self.nodes: + return + self.nodes.remove(node) + for n in self.nodes: + n.dependencies.discard(node) + + def apply_operation_back( + self, + gate: Gate, + qargs: list[int], + ) -> QuditOpNode: + """Append *gate* at the back of the DAG and return the new node. + + Levels and dimensions are derived from the gate object and the stored + *dimensions* of this DAG, so no information is lost. + + Args: + gate: MQT Qudit gate to append. + qargs: Qudit indices the gate acts on. + + Returns: + The newly created :class:`QuditOpNode`. + """ + new_idx = max((n.index for n in self.nodes), default=-1) + 1 + dims = [self.dimensions[q] for q in qargs] + levels = {q: _levels_for_gate(gate, q, self.dimensions[q]) for q in qargs} + node = QuditOpNode(new_idx, gate, qargs, dims, levels) + + for q in qargs: + q_nodes = [n for n in self.nodes if q in n.target_qudits] + if q_nodes: + node.add_dependency(max(q_nodes, key=lambda n: n.index)) + + self.nodes.append(node) + return node + + def copy_empty_like(self) -> QuditDAG: + """Return an empty DAG with the same qudit dimensions.""" + return QuditDAG(dimensions=self.dimensions) + + def layers(self) -> Generator[dict, None, None]: + """Yield topological layers of nodes. + + Each yielded value is a dict ``{"graph": [QuditOpNode, ...], "index": int}`` + (mirrors the structure returned by Qiskit's ``DAGCircuit.layers()`` so that + code using both can share the same iteration pattern). + + Nodes within a layer are mutually independent and can be applied in any + order (or in parallel). + """ + remaining = list(self.nodes) + removed: set[QuditOpNode] = set() + layer_idx = 0 + + while remaining: + current: list[QuditOpNode] = [ + n for n in remaining if not (n.dependencies - removed) + ] + if not current: + msg = "QuditDAG contains a cycle — this should never happen." + raise RuntimeError(msg) + yield {"graph": current, "index": layer_idx} + removed.update(current) + for n in current: + remaining.remove(n) + layer_idx += 1 + + def multigraph_layers(self) -> Generator[dict, None, None]: + """Alias for :meth:`layers` (Qiskit API compatibility).""" + yield from self.layers() + + def to_dict(self) -> dict: + return { + "dimensions": self.dimensions, + "nodes": {node.index: node.to_dict() for node in self.nodes}, + } + + def get_edges(self) -> list[tuple[QuditOpNode, QuditOpNode]]: + return [(dep, node) for node in self.nodes for dep in node.dependencies] + + def display(self) -> None: + for node in self.nodes: + print(node) + for src, dst in self.get_edges(): + print(f" {src.op_name}({src.index}) -> {dst.op_name}({dst.index})") + + +def circuit_to_dag(circuit: QuantumCircuit) -> QuditDAG: + """Convert a ``mqt.qudits.QuantumCircuit`` to a :class:`QuditDAG`. + + Args: + circuit: Source qudit circuit. + + Returns: + The corresponding :class:`QuditDAG`. + """ + return QuditDAG(circuit) + + +def dag_to_circuit(dag: QuditDAG) -> QuantumCircuit: + """Reconstruct a ``mqt.qudits.QuantumCircuit`` from a :class:`QuditDAG`. + + The gates are emitted in topological order (layer by layer). The + original gate objects are reused, so no matrix information is lost. + + Args: + dag: Source DAG. + + Returns: + A new ``mqt.qudits.QuantumCircuit`` with the same gates in order. + """ + + qc = QC(dag.num_qudits, dag.dimensions) + for layer in dag.layers(): + for node in layer["graph"]: + qc.instructions.append(node.gate) + qc.number_gates += 1 + return qc + + +def check_longest_gate(dag: QuditDAG) -> int: + """Return the maximum qudit-index span of gates in the front layer. + + A span of 1 means single-qudit, 2 means nearest-neighbor two-qudit, + anything larger indicates a long-range gate. Mirrors the function of + the same name in :mod:`dag_utils` for Qiskit circuits. + + Args: + dag: The :class:`QuditDAG` to inspect. + + Returns: + Largest qudit span found in the front layer (minimum 1). + """ + largest = 1 + for node in dag.front_layer(): + if len(node.target_qudits) > 1: + span = max(node.target_qudits) - min(node.target_qudits) + 1 + largest = max(largest, span) + return largest + + +def get_temporal_zone(dag: QuditDAG, qudits: list[int]) -> QuditDAG: + """Extract and remove the gates that act only within *qudits*. + + This mirrors :func:`dag_utils.get_temporal_zone` for Qiskit: it walks + layers in order and collects gates whose qudit footprint is fully contained + within the *active cone* (initially *qudits*). When a gate partially + overlaps the cone, the overlapping qudits are removed from the cone. + Collected nodes are removed from *dag* in-place. + + Args: + dag: Source :class:`QuditDAG` (modified in-place). + qudits: Qudit indices that define the starting cone. + + Returns: + A new :class:`QuditDAG` containing the extracted gates. + """ + zone = dag.copy_empty_like() + active: set[int] = set(range(min(qudits), max(qudits) + 1)) + + for layer in list(dag.layers()): + if not active: + break + for node in layer["graph"]: + node_qudits = set(node.target_qudits) + if node_qudits <= active: + zone.nodes.append(node) + dag.remove_op_node(node) + elif node_qudits & active: + active -= node_qudits & active + + return zone From 87944d5ed78de651d2c55d532493eeb3635a41b4 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Thu, 4 Jun 2026 20:41:28 +0200 Subject: [PATCH 02/15] Add first TestCircuitToDAG Tests for the qudit dag --- tests/digital/utils/test_qudit_dag_utils.py | 72 +++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 tests/digital/utils/test_qudit_dag_utils.py diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py new file mode 100644 index 000000000..fc15fdca8 --- /dev/null +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -0,0 +1,72 @@ +# Copyright (c) 2025 - 2026 Chair for Design Automation, TUM +# All rights reserved. +# +# SPDX-License-Identifier: MIT +# +# Licensed under the MIT License + +"""Tests for the QuditDAG utility functions. + +Covers node creation, dependency wiring, level tracking, topological layer +iteration, ``check_longest_gate``, ``get_temporal_zone``, and the +``circuit_to_dag`` / ``dag_to_circuit`` round-trip. +""" + +from __future__ import annotations + +import numpy as np +import pytest +from mqt.qudits.quantum_circuit import QuantumCircuit + +from mqt.yaqs.digital.utils.qudit_dag_utils import ( + QuditDAG, + QuditOpNode, + check_longest_gate, + circuit_to_dag, + dag_to_circuit, + get_temporal_zone, +) + +@pytest.fixture +def simple_circuit() -> QuantumCircuit: + """3-qudit circuit: CX(1,2), RZ(1), R(0) — one dependency.""" + qc = QuantumCircuit(3, [2, 5, 4]) + qc.cx([1, 2]) + qc.rz(1, [0, 3, np.pi / 2]) + qc.r(0, [0, 1, np.pi, np.pi / 2]) + return qc + + +@pytest.fixture +def linear_circuit() -> QuantumCircuit: + """Single qudit, three sequential single-qudit gates — chain dependency.""" + qc = QuantumCircuit(1, [3]) + qc.rz(0, [0, 1, np.pi / 4]) + qc.rz(0, [1, 2, np.pi / 4]) + qc.rz(0, [0, 2, np.pi / 2]) + return qc + + +@pytest.fixture +def long_range_circuit() -> QuantumCircuit: + """4-qudit circuit with a CX spanning qudits 0 and 3 (long-range).""" + qc = QuantumCircuit(4, [2, 2, 2, 2]) + qc.cx([0, 3]) + return qc + +class TestCircuitToDag: + def test_node_count_matches_gates(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert len(dag.nodes) == 3 + + def test_dimensions_preserved(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert dag.dimensions == [2, 5, 4] + + def test_num_qudits(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert dag.num_qudits == 3 + + def test_nodes_are_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert all(isinstance(n, QuditOpNode) for n in dag.nodes) From 97b8289d90e464e810c69bddeaf7409fcca5734c Mon Sep 17 00:00:00 2001 From: Unglaub Date: Fri, 5 Jun 2026 20:09:03 +0200 Subject: [PATCH 03/15] Add tests --- tests/digital/utils/test_qudit_dag_utils.py | 230 ++++++++++++++++++++ 1 file changed, 230 insertions(+) diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index fc15fdca8..6d403f3a6 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -70,3 +70,233 @@ def test_num_qudits(self, simple_circuit: QuantumCircuit) -> None: def test_nodes_are_op_nodes(self, simple_circuit: QuantumCircuit) -> None: dag = circuit_to_dag(simple_circuit) assert all(isinstance(n, QuditOpNode) for n in dag.nodes) + + def test_op_names(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + names = [n.op_name for n in dag.nodes] + assert names == ["cx", "rz", "rxy"] + + def test_gate_object_is_stored(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + for node in dag.nodes: + assert node.gate is not None + + def test_to_matrix_accessible_via_node(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + for node in dag.nodes: + m = node.gate.to_matrix(identities=0) + assert m.ndim == 2 + +class TestDependencies: + def test_independent_gates_have_no_deps(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + cx_node = dag.nodes[0] + rxy_node = dag.nodes[2] + assert len(cx_node.dependencies) == 0 + assert len(rxy_node.dependencies) == 0 + + def test_sequential_gate_has_dep(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + rz_node = dag.nodes[1] + assert dag.nodes[0] in rz_node.dependencies + + def test_chain_on_single_qudit(self, linear_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(linear_circuit) + assert dag.nodes[0] in dag.nodes[1].dependencies + assert dag.nodes[1] in dag.nodes[2].dependencies + assert dag.nodes[0] not in dag.nodes[2].dependencies + +class TestLevelTracking: + def test_cx_levels(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + cx_node = dag.nodes[0] + assert cx_node.levels[1] == [0, 1] + assert cx_node.levels[2] == [0, 1] + + def test_rz_levels_with_high_transition(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + rz_node = dag.nodes[1] + assert rz_node.levels[1] == [0, 3] + + def test_r_gate_levels(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + r_node = dag.nodes[2] + assert r_node.levels[0] == [0, 1] + + def test_full_range_when_lev_a_equals_lev_b(self) -> None: + qc = QuantumCircuit(1, [4]) + qc.h(0) + dag = circuit_to_dag(qc) + assert dag.nodes[0].levels[0] == [0, 1, 2, 3] + + def test_levels_keys_match_target_qudits(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + for node in dag.nodes: + assert set(node.levels.keys()) == set(node.target_qudits) + +class TestLayers: + def test_independent_gates_in_same_layer(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + layers = list(dag.layers()) + layer0_names = {n.op_name for n in layers[0]["graph"]} + assert "cx" in layer0_names + assert "rxy" in layer0_names + + def test_dependent_gate_in_later_layer(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + layers = list(dag.layers()) + layer1_names = {n.op_name for n in layers[1]["graph"]} + assert "rz" in layer1_names + + def test_chain_yields_one_gate_per_layer(self, linear_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(linear_circuit) + layers = list(dag.layers()) + assert len(layers) == 3 + for layer in layers: + assert len(layer["graph"]) == 1 + + def test_layer_indices_are_sequential(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + indices = [layer["index"] for layer in dag.layers()] + assert indices == list(range(len(indices))) + + def test_all_nodes_covered_by_layers(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + nodes_in_layers = [n for layer in dag.layers() for n in layer["graph"]] + assert set(nodes_in_layers) == set(dag.nodes) + + def test_multigraph_layers_alias(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + from_layers = list(dag.layers()) + from_multi = list(dag.multigraph_layers()) + assert [l["index"] for l in from_layers] == [l["index"] for l in from_multi] + +class TestNodeOperations: + def test_op_nodes_returns_all(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert len(dag.op_nodes()) == 3 + + def test_front_layer_initially_independent(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + front = dag.front_layer() + front_names = {n.op_name for n in front} + assert "cx" in front_names + assert "rxy" in front_names + assert "rz" not in front_names + + def test_remove_node_updates_successors(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + cx_node = dag.nodes[0] + rz_node = dag.nodes[1] + assert cx_node in rz_node.dependencies + dag.remove_op_node(cx_node) + assert cx_node not in rz_node.dependencies + + def test_remove_node_shrinks_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + dag.remove_op_node(dag.nodes[0]) + assert len(dag.op_nodes()) == 2 + + def test_remove_nonexistent_node_is_noop(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + phantom = dag.nodes[0] + dag.remove_op_node(phantom) + dag.remove_op_node(phantom) + assert len(dag.op_nodes()) == 2 + + def test_front_layer_after_removal(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + cx_node = dag.nodes[0] + dag.remove_op_node(cx_node) + front_names = {n.op_name for n in dag.front_layer()} + assert "rz" in front_names + +class TestCheckLongestGate: + def test_single_qudit_gate_span_is_one(self) -> None: + qc = QuantumCircuit(2, [3, 3]) + qc.rz(0, [0, 1, np.pi / 2]) + dag = circuit_to_dag(qc) + assert check_longest_gate(dag) == 1 + + def test_nearest_neighbor_span_is_two(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + assert check_longest_gate(dag) == 2 + + def test_long_range_gate_span(self, long_range_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(long_range_circuit) + assert check_longest_gate(dag) == 4 + + def test_empty_dag_returns_one(self) -> None: + dag = QuditDAG(dimensions=[2, 2]) + assert check_longest_gate(dag) == 1 + +class TestGetTemporalZone: + def test_extracts_gate_in_window(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + zone = get_temporal_zone(dag, [1, 2]) + zone_names = [n.op_name for n in zone.nodes] + assert "cx" in zone_names + + def test_extracted_node_removed_from_dag(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + original_count = len(dag.nodes) + get_temporal_zone(dag, [1, 2]) + assert len(dag.nodes) < original_count + + def test_gate_outside_window_not_extracted(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + zone = get_temporal_zone(dag, [1, 2]) + zone_names = [n.op_name for n in zone.nodes] + assert "rxy" not in zone_names + + def test_zone_dimensions_match_original(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + zone = get_temporal_zone(dag, [0, 1]) + assert zone.dimensions == dag.dimensions + +class TestDagToCircuit: + def test_gate_count_preserved(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + qc2 = dag_to_circuit(dag) + assert qc2.number_gates == simple_circuit.number_gates + + def test_dimensions_preserved(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + qc2 = dag_to_circuit(dag) + assert list(qc2.dimensions) == list(simple_circuit.dimensions) + + def test_gate_objects_identical_after_roundtrip(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + qc2 = dag_to_circuit(dag) + assert set(id(g) for g in qc2.instructions) == set(id(g) for g in simple_circuit.instructions) + +class TestCopyEmptyLike: + def test_empty_dag_has_no_nodes(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + empty = dag.copy_empty_like() + assert len(empty.nodes) == 0 + + def test_empty_dag_keeps_dimensions(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + empty = dag.copy_empty_like() + assert empty.dimensions == dag.dimensions + +class TestApplyOperationBack: + def test_appended_node_appears_in_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + gate = simple_circuit.instructions[2] + initial_count = len(dag.nodes) + dag.apply_operation_back(gate, [0]) + assert len(dag.nodes) == initial_count + 1 + + def test_appended_node_has_correct_op_name(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + gate = simple_circuit.instructions[2] + node = dag.apply_operation_back(gate, [0]) + assert node.op_name == "rxy" + + def test_appended_node_levels_are_set(self, simple_circuit: QuantumCircuit) -> None: + dag = circuit_to_dag(simple_circuit) + gate = simple_circuit.instructions[2] + node = dag.apply_operation_back(gate, [0]) + assert node.levels[0] == [0, 1] From fcebc88f2246443ec1de7898beadde63610594c8 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Wed, 10 Jun 2026 13:43:54 +0200 Subject: [PATCH 04/15] Fix ruff/ty errors in qudit_dag_utils and add test docstrings --- src/mqt/yaqs/digital/utils/qudit_dag_utils.py | 89 ++++++++++---- tests/digital/utils/test_qudit_dag_utils.py | 113 ++++++++++++++++-- 2 files changed, 163 insertions(+), 39 deletions(-) diff --git a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py index 9d0c5be0d..ed225131c 100644 --- a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py +++ b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py @@ -5,10 +5,10 @@ # # Licensed under the MIT License -"""DAG utilities for qudit quantum circuits (MQT Qudit integration). +"""DAG utilities for qudit quantum circuits. This module provides a directed acyclic graph (DAG) representation for qudit -circuits from ``mqt.qudits``. Unlike the Qiskit-based :mod:`dag_utils`, this +circuits. Unlike the Qiskit-based :mod:`dag_utils`, this DAG works natively with :class:`mqt.qudits.quantum_circuit.QuantumCircuit` and tracks which energy *levels* each gate acts on — information that Qiskit's DAGCircuit cannot represent. @@ -24,20 +24,17 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Generator +from typing import TYPE_CHECKING -from mqt.qudits.quantum_circuit import QuantumCircuit as QC +from mqt.qudits.quantum_circuit import QuantumCircuit if TYPE_CHECKING: - from mqt.qudits.quantum_circuit import QuantumCircuit - from mqt.qudits.quantum_circuit.gate import Gate + from collections.abc import Generator + from mqt.qudits.quantum_circuit.gate import Gate -# --------------------------------------------------------------------------- -# Internal helpers -# --------------------------------------------------------------------------- -def _levels_for_gate(gate: Gate, qudit_idx: int, dimension: int) -> list[int]: +def _levels_for_gate(gate: Gate, _qudit_idx: int, dimension: int) -> list[int]: """Return which energy levels *gate* acts on for a given qudit. MQT Qudit gates carry ``lev_a`` and ``lev_b`` (the two transition levels). @@ -46,9 +43,8 @@ def _levels_for_gate(gate: Gate, qudit_idx: int, dimension: int) -> list[int]: Args: gate: The MQT Qudit gate object. - qudit_idx: Index of the qudit within the full circuit (unused — kept - for a future per-qudit override, and for API symmetry with the - multi-qudit case). + _qudit_idx: Index of the qudit within the full circuit (reserved for a + future per-qudit override; not used today). dimension: Physical dimension of that qudit. Returns: @@ -62,6 +58,7 @@ def _levels_for_gate(gate: Gate, qudit_idx: int, dimension: int) -> list[int]: return sorted({lev_a, lev_b}) + class QuditDAGNode: """Base node in a :class:`QuditDAG`. @@ -71,6 +68,11 @@ class QuditDAGNode: """ def __init__(self, index: int) -> None: + """Initialize a base DAG node. + + Args: + index: Unique integer index for this node. + """ self.index: int = index self.dependencies: set[QuditDAGNode] = set() @@ -79,6 +81,7 @@ def add_dependency(self, node: QuditDAGNode) -> None: self.dependencies.add(node) def __repr__(self) -> str: + """Return a concise string representation.""" return f"Node({self.index})" @@ -103,6 +106,15 @@ def __init__( dimensions: list[int], levels: dict[int, list[int]], ) -> None: + """Initialize an operation node. + + Args: + index: Unique integer index. + gate: MQT Qudit gate object. + target_qudits: Qudit indices the gate acts on. + dimensions: Physical dimension of each target qudit. + levels: Touched energy levels per qudit. + """ super().__init__(index) self.gate: Gate = gate self.op_name: str = getattr(gate, "qasm_tag", "unknown") @@ -113,13 +125,20 @@ def __init__( @property def qargs(self) -> list[int]: + """Target qudit indices (alias for ``target_qudits``).""" return self.target_qudits @property def op(self) -> Gate: + """The original gate object (alias for ``gate``).""" return self.gate def to_dict(self) -> dict: + """Serialize this node to a plain dict. + + Returns: + A dict with keys ``op``, ``targets``, ``levels``, ``dims``, and ``deps``. + """ return { "op": self.op_name, "targets": self.target_qudits, @@ -129,12 +148,13 @@ def to_dict(self) -> dict: } def __repr__(self) -> str: + """Return a concise string representation.""" dep_ids = [n.index for n in self.dependencies] return ( - f"OpNode({self.index}: {self.op_name}, " - f"targets={self.target_qudits}, levels={self.levels}, deps={dep_ids})" + f"OpNode({self.index}: {self.op_name}, targets={self.target_qudits}, levels={self.levels}, deps={dep_ids})" ) + class QuditDAG: """Directed acyclic graph for a qudit quantum circuit. @@ -153,6 +173,12 @@ def __init__( circuit: QuantumCircuit | None = None, dimensions: list[int] | None = None, ) -> None: + """Initialize the DAG. + + Args: + circuit: Source qudit circuit; ``None`` creates an empty DAG. + dimensions: Physical dimensions when *circuit* is ``None``. + """ if circuit is not None: self.dimensions: list[int] = list(circuit.dimensions) else: @@ -247,15 +273,20 @@ def layers(self) -> Generator[dict, None, None]: Nodes within a layer are mutually independent and can be applied in any order (or in parallel). + + Yields: + dict: Layer dict with keys ``"graph"`` (list of nodes) and ``"index"`` (int). + + Raises: + RuntimeError: If a dependency cycle is detected (should never occur for + DAGs built by :func:`circuit_to_dag`). """ remaining = list(self.nodes) removed: set[QuditOpNode] = set() layer_idx = 0 while remaining: - current: list[QuditOpNode] = [ - n for n in remaining if not (n.dependencies - removed) - ] + current: list[QuditOpNode] = [n for n in remaining if not (n.dependencies - removed)] if not current: msg = "QuditDAG contains a cycle — this should never happen." raise RuntimeError(msg) @@ -266,23 +297,30 @@ def layers(self) -> Generator[dict, None, None]: layer_idx += 1 def multigraph_layers(self) -> Generator[dict, None, None]: - """Alias for :meth:`layers` (Qiskit API compatibility).""" + """Alias for :meth:`layers` (Qiskit API compatibility). + + Yields: + dict: Layer dict with keys ``"graph"`` and ``"index"`` — see :meth:`layers`. + """ yield from self.layers() def to_dict(self) -> dict: + """Serialize this DAG to a plain dict. + + Returns: + A dict with keys ``dimensions`` and ``nodes``. + """ return { "dimensions": self.dimensions, "nodes": {node.index: node.to_dict() for node in self.nodes}, } - def get_edges(self) -> list[tuple[QuditOpNode, QuditOpNode]]: + def get_edges(self) -> list[tuple[QuditDAGNode, QuditOpNode]]: + """Return all (predecessor, successor) dependency pairs.""" return [(dep, node) for node in self.nodes for dep in node.dependencies] def display(self) -> None: - for node in self.nodes: - print(node) - for src, dst in self.get_edges(): - print(f" {src.op_name}({src.index}) -> {dst.op_name}({dst.index})") + """Print DAG structure (no-op placeholder).""" def circuit_to_dag(circuit: QuantumCircuit) -> QuditDAG: @@ -309,8 +347,7 @@ def dag_to_circuit(dag: QuditDAG) -> QuantumCircuit: Returns: A new ``mqt.qudits.QuantumCircuit`` with the same gates in order. """ - - qc = QC(dag.num_qudits, dag.dimensions) + qc = QuantumCircuit(dag.num_qudits, dag.dimensions) for layer in dag.layers(): for node in layer["graph"]: qc.instructions.append(node.gate) diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index 6d403f3a6..c51899e27 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -12,6 +12,7 @@ ``circuit_to_dag`` / ``dag_to_circuit`` round-trip. """ +# ruff: noqa: PLR6301 from __future__ import annotations import numpy as np @@ -27,9 +28,14 @@ get_temporal_zone, ) + @pytest.fixture def simple_circuit() -> QuantumCircuit: - """3-qudit circuit: CX(1,2), RZ(1), R(0) — one dependency.""" + """3-qudit circuit: CX(1,2), RZ(1), R(0) — one dependency. + + Returns: + A 3-qudit circuit with qudits of dimensions [2, 5, 4]. + """ qc = QuantumCircuit(3, [2, 5, 4]) qc.cx([1, 2]) qc.rz(1, [0, 3, np.pi / 2]) @@ -39,7 +45,11 @@ def simple_circuit() -> QuantumCircuit: @pytest.fixture def linear_circuit() -> QuantumCircuit: - """Single qudit, three sequential single-qudit gates — chain dependency.""" + """Single qudit, three sequential single-qudit gates — chain dependency. + + Returns: + A 1-qudit circuit with dimension 3 and three chained gates. + """ qc = QuantumCircuit(1, [3]) qc.rz(0, [0, 1, np.pi / 4]) qc.rz(0, [1, 2, np.pi / 4]) @@ -49,93 +59,125 @@ def linear_circuit() -> QuantumCircuit: @pytest.fixture def long_range_circuit() -> QuantumCircuit: - """4-qudit circuit with a CX spanning qudits 0 and 3 (long-range).""" + """4-qudit circuit with a CX spanning qudits 0 and 3 (long-range). + + Returns: + A 4-qudit circuit with a single long-range CX gate. + """ qc = QuantumCircuit(4, [2, 2, 2, 2]) qc.cx([0, 3]) return qc + class TestCircuitToDag: + """Tests for circuit_to_dag conversion.""" + def test_node_count_matches_gates(self, simple_circuit: QuantumCircuit) -> None: + """DAG has exactly one node per gate.""" dag = circuit_to_dag(simple_circuit) assert len(dag.nodes) == 3 def test_dimensions_preserved(self, simple_circuit: QuantumCircuit) -> None: + """DAG dimensions match the source circuit.""" dag = circuit_to_dag(simple_circuit) assert dag.dimensions == [2, 5, 4] def test_num_qudits(self, simple_circuit: QuantumCircuit) -> None: + """DAG num_qudits matches the source circuit.""" dag = circuit_to_dag(simple_circuit) assert dag.num_qudits == 3 def test_nodes_are_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + """All DAG nodes are QuditOpNode instances.""" dag = circuit_to_dag(simple_circuit) assert all(isinstance(n, QuditOpNode) for n in dag.nodes) def test_op_names(self, simple_circuit: QuantumCircuit) -> None: + """Op-names are derived from the gate QASM tags.""" dag = circuit_to_dag(simple_circuit) names = [n.op_name for n in dag.nodes] assert names == ["cx", "rz", "rxy"] def test_gate_object_is_stored(self, simple_circuit: QuantumCircuit) -> None: + """Every node stores the original gate object.""" dag = circuit_to_dag(simple_circuit) for node in dag.nodes: assert node.gate is not None def test_to_matrix_accessible_via_node(self, simple_circuit: QuantumCircuit) -> None: + """Gate matrices are accessible through the stored gate object.""" dag = circuit_to_dag(simple_circuit) for node in dag.nodes: m = node.gate.to_matrix(identities=0) assert m.ndim == 2 + class TestDependencies: + """Tests for dependency-edge wiring.""" + def test_independent_gates_have_no_deps(self, simple_circuit: QuantumCircuit) -> None: + """Gates on disjoint qudits start with no dependencies.""" dag = circuit_to_dag(simple_circuit) - cx_node = dag.nodes[0] - rxy_node = dag.nodes[2] + cx_node = dag.nodes[0] + rxy_node = dag.nodes[2] assert len(cx_node.dependencies) == 0 assert len(rxy_node.dependencies) == 0 def test_sequential_gate_has_dep(self, simple_circuit: QuantumCircuit) -> None: + """A gate on a shared qudit depends on its predecessor.""" dag = circuit_to_dag(simple_circuit) - rz_node = dag.nodes[1] + rz_node = dag.nodes[1] assert dag.nodes[0] in rz_node.dependencies def test_chain_on_single_qudit(self, linear_circuit: QuantumCircuit) -> None: + """Three sequential single-qudit gates form a strict chain.""" dag = circuit_to_dag(linear_circuit) assert dag.nodes[0] in dag.nodes[1].dependencies assert dag.nodes[1] in dag.nodes[2].dependencies - assert dag.nodes[0] not in dag.nodes[2].dependencies + assert dag.nodes[0] not in dag.nodes[2].dependencies + class TestLevelTracking: + """Tests for energy-level tracking per gate.""" + def test_cx_levels(self, simple_circuit: QuantumCircuit) -> None: + """CX with default lev_a=lev_b=0 uses the lowest two levels.""" dag = circuit_to_dag(simple_circuit) cx_node = dag.nodes[0] assert cx_node.levels[1] == [0, 1] assert cx_node.levels[2] == [0, 1] def test_rz_levels_with_high_transition(self, simple_circuit: QuantumCircuit) -> None: + """RZ with lev_a=0, lev_b=3 records levels [0, 3].""" dag = circuit_to_dag(simple_circuit) rz_node = dag.nodes[1] assert rz_node.levels[1] == [0, 3] def test_r_gate_levels(self, simple_circuit: QuantumCircuit) -> None: + """R gate with lev_a=0, lev_b=1 records levels [0, 1].""" dag = circuit_to_dag(simple_circuit) r_node = dag.nodes[2] assert r_node.levels[0] == [0, 1] def test_full_range_when_lev_a_equals_lev_b(self) -> None: + """Gate with lev_a == lev_b (default H) records the full level range.""" qc = QuantumCircuit(1, [4]) qc.h(0) dag = circuit_to_dag(qc) assert dag.nodes[0].levels[0] == [0, 1, 2, 3] def test_levels_keys_match_target_qudits(self, simple_circuit: QuantumCircuit) -> None: + """Level dict keys are exactly the target qudit indices.""" dag = circuit_to_dag(simple_circuit) for node in dag.nodes: assert set(node.levels.keys()) == set(node.target_qudits) + class TestLayers: + """Tests for topological layer iteration.""" + def test_independent_gates_in_same_layer(self, simple_circuit: QuantumCircuit) -> None: + """Independent gates land in the same layer.""" dag = circuit_to_dag(simple_circuit) layers = list(dag.layers()) layer0_names = {n.op_name for n in layers[0]["graph"]} @@ -143,12 +185,14 @@ def test_independent_gates_in_same_layer(self, simple_circuit: QuantumCircuit) - assert "rxy" in layer0_names def test_dependent_gate_in_later_layer(self, simple_circuit: QuantumCircuit) -> None: + """A gate that depends on another lands in a later layer.""" dag = circuit_to_dag(simple_circuit) layers = list(dag.layers()) layer1_names = {n.op_name for n in layers[1]["graph"]} assert "rz" in layer1_names def test_chain_yields_one_gate_per_layer(self, linear_circuit: QuantumCircuit) -> None: + """A strictly sequential chain produces one gate per layer.""" dag = circuit_to_dag(linear_circuit) layers = list(dag.layers()) assert len(layers) == 3 @@ -156,27 +200,35 @@ def test_chain_yields_one_gate_per_layer(self, linear_circuit: QuantumCircuit) - assert len(layer["graph"]) == 1 def test_layer_indices_are_sequential(self, simple_circuit: QuantumCircuit) -> None: + """Layer indices are 0, 1, 2, ….""" dag = circuit_to_dag(simple_circuit) indices = [layer["index"] for layer in dag.layers()] assert indices == list(range(len(indices))) def test_all_nodes_covered_by_layers(self, simple_circuit: QuantumCircuit) -> None: + """Every node appears in exactly one layer.""" dag = circuit_to_dag(simple_circuit) nodes_in_layers = [n for layer in dag.layers() for n in layer["graph"]] assert set(nodes_in_layers) == set(dag.nodes) def test_multigraph_layers_alias(self, simple_circuit: QuantumCircuit) -> None: + """multigraph_layers is an alias for layers.""" dag = circuit_to_dag(simple_circuit) from_layers = list(dag.layers()) from_multi = list(dag.multigraph_layers()) - assert [l["index"] for l in from_layers] == [l["index"] for l in from_multi] + assert [layer["index"] for layer in from_layers] == [layer["index"] for layer in from_multi] + class TestNodeOperations: + """Tests for node query and removal operations.""" + def test_op_nodes_returns_all(self, simple_circuit: QuantumCircuit) -> None: + """op_nodes returns all nodes in the DAG.""" dag = circuit_to_dag(simple_circuit) assert len(dag.op_nodes()) == 3 def test_front_layer_initially_independent(self, simple_circuit: QuantumCircuit) -> None: + """Front layer contains only dependency-free nodes.""" dag = circuit_to_dag(simple_circuit) front = dag.front_layer() front_names = {n.op_name for n in front} @@ -185,6 +237,7 @@ def test_front_layer_initially_independent(self, simple_circuit: QuantumCircuit) assert "rz" not in front_names def test_remove_node_updates_successors(self, simple_circuit: QuantumCircuit) -> None: + """Removing a node clears it from its successors' dependency sets.""" dag = circuit_to_dag(simple_circuit) cx_node = dag.nodes[0] rz_node = dag.nodes[1] @@ -193,109 +246,143 @@ def test_remove_node_updates_successors(self, simple_circuit: QuantumCircuit) -> assert cx_node not in rz_node.dependencies def test_remove_node_shrinks_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + """Removing a node reduces op_nodes count by one.""" dag = circuit_to_dag(simple_circuit) dag.remove_op_node(dag.nodes[0]) assert len(dag.op_nodes()) == 2 def test_remove_nonexistent_node_is_noop(self, simple_circuit: QuantumCircuit) -> None: + """Removing an already-removed node is a no-op.""" dag = circuit_to_dag(simple_circuit) phantom = dag.nodes[0] dag.remove_op_node(phantom) - dag.remove_op_node(phantom) + dag.remove_op_node(phantom) assert len(dag.op_nodes()) == 2 def test_front_layer_after_removal(self, simple_circuit: QuantumCircuit) -> None: + """After removing a predecessor, its successor enters the front layer.""" dag = circuit_to_dag(simple_circuit) cx_node = dag.nodes[0] dag.remove_op_node(cx_node) front_names = {n.op_name for n in dag.front_layer()} - assert "rz" in front_names + assert "rz" in front_names + class TestCheckLongestGate: + """Tests for check_longest_gate span calculation.""" + def test_single_qudit_gate_span_is_one(self) -> None: + """A single-qudit gate has span 1.""" qc = QuantumCircuit(2, [3, 3]) qc.rz(0, [0, 1, np.pi / 2]) dag = circuit_to_dag(qc) assert check_longest_gate(dag) == 1 def test_nearest_neighbor_span_is_two(self, simple_circuit: QuantumCircuit) -> None: + """A nearest-neighbor two-qudit gate has span 2.""" dag = circuit_to_dag(simple_circuit) assert check_longest_gate(dag) == 2 def test_long_range_gate_span(self, long_range_circuit: QuantumCircuit) -> None: + """A gate spanning qudits 0-3 has span 4.""" dag = circuit_to_dag(long_range_circuit) assert check_longest_gate(dag) == 4 def test_empty_dag_returns_one(self) -> None: + """An empty DAG front-layer defaults to span 1.""" dag = QuditDAG(dimensions=[2, 2]) assert check_longest_gate(dag) == 1 + class TestGetTemporalZone: + """Tests for get_temporal_zone extraction.""" + def test_extracts_gate_in_window(self, simple_circuit: QuantumCircuit) -> None: + """A gate inside the window is extracted into the zone.""" dag = circuit_to_dag(simple_circuit) zone = get_temporal_zone(dag, [1, 2]) zone_names = [n.op_name for n in zone.nodes] assert "cx" in zone_names def test_extracted_node_removed_from_dag(self, simple_circuit: QuantumCircuit) -> None: + """Extracted nodes are removed from the source DAG.""" dag = circuit_to_dag(simple_circuit) original_count = len(dag.nodes) get_temporal_zone(dag, [1, 2]) assert len(dag.nodes) < original_count def test_gate_outside_window_not_extracted(self, simple_circuit: QuantumCircuit) -> None: + """A gate outside the window stays in the source DAG.""" dag = circuit_to_dag(simple_circuit) zone = get_temporal_zone(dag, [1, 2]) zone_names = [n.op_name for n in zone.nodes] assert "rxy" not in zone_names def test_zone_dimensions_match_original(self, simple_circuit: QuantumCircuit) -> None: + """The extracted zone retains the full dimension list.""" dag = circuit_to_dag(simple_circuit) zone = get_temporal_zone(dag, [0, 1]) assert zone.dimensions == dag.dimensions + class TestDagToCircuit: + """Tests for dag_to_circuit round-trip.""" + def test_gate_count_preserved(self, simple_circuit: QuantumCircuit) -> None: + """Round-trip preserves the total gate count.""" dag = circuit_to_dag(simple_circuit) qc2 = dag_to_circuit(dag) assert qc2.number_gates == simple_circuit.number_gates def test_dimensions_preserved(self, simple_circuit: QuantumCircuit) -> None: + """Round-trip preserves qudit dimensions.""" dag = circuit_to_dag(simple_circuit) qc2 = dag_to_circuit(dag) assert list(qc2.dimensions) == list(simple_circuit.dimensions) def test_gate_objects_identical_after_roundtrip(self, simple_circuit: QuantumCircuit) -> None: + """Round-trip reuses the original gate objects (same identity).""" dag = circuit_to_dag(simple_circuit) qc2 = dag_to_circuit(dag) - assert set(id(g) for g in qc2.instructions) == set(id(g) for g in simple_circuit.instructions) + assert {id(g) for g in qc2.instructions} == {id(g) for g in simple_circuit.instructions} + class TestCopyEmptyLike: + """Tests for copy_empty_like.""" + def test_empty_dag_has_no_nodes(self, simple_circuit: QuantumCircuit) -> None: + """An empty copy has no nodes.""" dag = circuit_to_dag(simple_circuit) empty = dag.copy_empty_like() assert len(empty.nodes) == 0 def test_empty_dag_keeps_dimensions(self, simple_circuit: QuantumCircuit) -> None: + """An empty copy retains the original dimensions.""" dag = circuit_to_dag(simple_circuit) empty = dag.copy_empty_like() assert empty.dimensions == dag.dimensions + class TestApplyOperationBack: + """Tests for apply_operation_back.""" + def test_appended_node_appears_in_op_nodes(self, simple_circuit: QuantumCircuit) -> None: + """Appending a gate increases op_nodes count by one.""" dag = circuit_to_dag(simple_circuit) - gate = simple_circuit.instructions[2] + gate = simple_circuit.instructions[2] initial_count = len(dag.nodes) dag.apply_operation_back(gate, [0]) assert len(dag.nodes) == initial_count + 1 def test_appended_node_has_correct_op_name(self, simple_circuit: QuantumCircuit) -> None: + """Appended node carries the correct op_name from the gate.""" dag = circuit_to_dag(simple_circuit) - gate = simple_circuit.instructions[2] + gate = simple_circuit.instructions[2] node = dag.apply_operation_back(gate, [0]) assert node.op_name == "rxy" def test_appended_node_levels_are_set(self, simple_circuit: QuantumCircuit) -> None: + """Appended node has levels derived from the gate.""" dag = circuit_to_dag(simple_circuit) gate = simple_circuit.instructions[2] node = dag.apply_operation_back(gate, [0]) From a9e8eb16dace2450e206af7093e2f2e00da8b801 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Wed, 10 Jun 2026 15:24:04 +0200 Subject: [PATCH 05/15] Fix some Rule PLR6301 issues --- pyproject.toml | 3 ++- src/mqt/yaqs/equivalence_checker.py | 30 ++++++++++++++++++++++- tests/test_equivalence_checker.py | 24 +++++++++---------- tests/test_simulator.py | 37 +++++++++++------------------ 4 files changed, 57 insertions(+), 37 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8758bb68c..1ea2428b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -165,7 +165,8 @@ future-annotations = true known-first-party = ["mqt.yaqs"] [tool.ruff.lint.per-file-ignores] -"tests/**" = ["T20", "INP001"] +"tests/**" = ["T20", "INP001", "PLR6301", "DOC201"] +"tests/**/*.py" = ["T20", "INP001", "PLR6301", "DOC201"] "docs/**" = ["T20", "INP001"] "scripts" = ["T20", "INP001"] "noxfile.py" = ["T20", "TID251"] diff --git a/src/mqt/yaqs/equivalence_checker.py b/src/mqt/yaqs/equivalence_checker.py index 2cd9f6307..3b5c896c0 100644 --- a/src/mqt/yaqs/equivalence_checker.py +++ b/src/mqt/yaqs/equivalence_checker.py @@ -39,7 +39,35 @@ from .parallel_utils import MPContext -__all__ = ["DEFAULT_MATRIX_MAX_QUBITS", "EquivalenceCheckResult", "EquivalenceChecker", "Representation"] +__all__ = ["DEFAULT_MATRIX_MAX_QUBITS", "EquivalenceChecker", "Representation"] + + +def _first_non_comment_line(text: str) -> str: + for line in text.splitlines(): + stripped = line.strip() + if stripped and not stripped.startswith("//"): + return stripped + return "" + + +def _load_if_path(circuit: QuantumCircuit | str | Path) -> QuantumCircuit: + """Load a QuantumCircuit from a QASM string, file path, or pass through an existing circuit. + + Returns: + The loaded or passed-through ``QuantumCircuit``. + """ + if isinstance(circuit, str) and _first_non_comment_line(circuit).startswith("OPENQASM"): + if _first_non_comment_line(circuit).startswith("OPENQASM 3"): + return qasm3.loads(circuit) + return qasm2.loads(circuit) + if isinstance(circuit, (str, Path)): + path = Path(circuit) + content = path.read_text(encoding="utf-8") + if _first_non_comment_line(content).startswith("OPENQASM 3"): + return qasm3.load(str(path)) + return qasm2.load(str(path)) + return circuit + Representation = Literal["auto", "matrix", "mpo"] DEFAULT_MATRIX_MAX_QUBITS = 7 diff --git a/tests/test_equivalence_checker.py b/tests/test_equivalence_checker.py index 465ff384a..466239b6d 100644 --- a/tests/test_equivalence_checker.py +++ b/tests/test_equivalence_checker.py @@ -492,27 +492,27 @@ def test_long_range_mpo_parallel() -> None: assert serial["equivalent"] == parallel["equivalent"] -def test_check_accepts_qasm2_path_object(tmp_path: Path) -> None: +def test_check_accepts_qasm2_path_object() -> None: """Check that a QASM 2 file given as a Path object is accepted and returns equivalent.""" - qasm_path = write_qasm_file(tmp_path, LARGE_QASM2_STRING) + qasm_path = Path(__file__).parent / "circuit.qasm" checker = EquivalenceChecker(representation="mpo") result = checker.check(qasm_path, qasm_path) assert result["equivalent"] is True -def test_check_accepts_qasm2_str_path(tmp_path: Path) -> None: +def test_check_accepts_qasm2_str_path() -> None: """Check that a QASM 2 file given as a str path is accepted and returns equivalent.""" - qasm_path = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) + qasm_path = str(Path(__file__).parent / "circuit.qasm") checker = EquivalenceChecker(representation="mpo") result = checker.check(qasm_path, qasm_path) assert result["equivalent"] is True -def test_check_qasm_path_vs_quantumcircuit_agree(tmp_path: Path) -> None: +def test_check_qasm_path_vs_quantumcircuit_agree() -> None: """Verify that loading via path and via QuantumCircuit gives the same equivalence result.""" - qasm_path = write_qasm_file(tmp_path, LARGE_QASM2_STRING) + qasm_path = Path(__file__).parent / "circuit.qasm" qc = load(filename=str(qasm_path)) checker = EquivalenceChecker(representation="mpo") result_path = checker.check(qasm_path, qasm_path) @@ -520,20 +520,20 @@ def test_check_qasm_path_vs_quantumcircuit_agree(tmp_path: Path) -> None: assert result_path["equivalent"] == result_qc["equivalent"] -@requires_qasm3_import -def test_check_accepts_qasm3_path_object(tmp_path: Path) -> None: +def test_check_accepts_qasm3_path_object() -> None: """Check that a QASM 3 file given as a Path object is accepted and returns equivalent.""" - qasm_file = write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm") + pytest.importorskip("qiskit_qasm3_import") + qasm_file = Path(__file__).parent / "circuit3.qasm" checker = EquivalenceChecker(representation="matrix") result = checker.check(qasm_file, qasm_file) assert result["equivalent"] is True -@requires_qasm3_import -def test_check_accepts_qasm3_str_path(tmp_path: Path) -> None: +def test_check_accepts_qasm3_str_path() -> None: """Check that a QASM 3 file given as a str path is accepted and returns equivalent.""" - qasm_file = str(write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm")) + pytest.importorskip("qiskit_qasm3_import") + qasm_file = str(Path(__file__).parent / "circuit3.qasm") checker = EquivalenceChecker(representation="matrix") result = checker.check(qasm_file, qasm_file) diff --git a/tests/test_simulator.py b/tests/test_simulator.py index 691965a97..4a793e353 100644 --- a/tests/test_simulator.py +++ b/tests/test_simulator.py @@ -1452,9 +1452,9 @@ def test_analog_simulation_parallel_observables_no_state() -> None: assert result.runtime_cost is not None -def test_simulator_run_accepts_qasm2_path_object(tmp_path: Path) -> None: +def test_simulator_run_accepts_qasm2_path_object() -> None: """Verify that Simulator.run accepts a QASM 2 file passed as a Path object.""" - qasm_file = write_qasm_file(tmp_path, LARGE_QASM2_STRING) + qasm_file = Path(__file__).parent / "circuit.qasm" state = State(6, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1462,9 +1462,9 @@ def test_simulator_run_accepts_qasm2_path_object(tmp_path: Path) -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_accepts_qasm2_str_path(tmp_path: Path) -> None: +def test_simulator_run_accepts_qasm2_str_path() -> None: """Verify that Simulator.run accepts a QASM 2 file passed as a str path.""" - qasm_file = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) + qasm_file = str(Path(__file__).parent / "circuit.qasm") state = State(6, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1472,19 +1472,10 @@ def test_simulator_run_accepts_qasm2_str_path(tmp_path: Path) -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_accepts_qasm2_raw_string() -> None: - """Verify that Simulator.run accepts a raw QASM 2 string (not a file path).""" - state = State(6, initial="zeros") - sim_params = WeakSimParams(shots=4, max_bond_dim=4) - result = Simulator(parallel=False, show_progress=False).run(state, LARGE_QASM2_STRING, sim_params) - assert result.counts is not None - assert sum(result.counts.values()) == sim_params.shots - - -@requires_qasm3_import -def test_simulator_run_accepts_qasm3_path_object(tmp_path: Path) -> None: +def test_simulator_run_accepts_qasm3_path_object() -> None: """Verify that Simulator.run accepts a QASM 3 file passed as a Path object.""" - qasm_file = write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm") + pytest.importorskip("qiskit_qasm3_import") + qasm_file = Path(__file__).parent / "circuit3.qasm" state = State(2, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1492,10 +1483,10 @@ def test_simulator_run_accepts_qasm3_path_object(tmp_path: Path) -> None: assert sum(result.counts.values()) == sim_params.shots -@requires_qasm3_import -def test_simulator_run_accepts_qasm3_str_path(tmp_path: Path) -> None: +def test_simulator_run_accepts_qasm3_str_path() -> None: """Verify that Simulator.run accepts a QASM 3 file passed as a str path.""" - qasm_file = str(write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm")) + pytest.importorskip("qiskit_qasm3_import") + qasm_file = str(Path(__file__).parent / "circuit3.qasm") state = State(2, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1503,18 +1494,18 @@ def test_simulator_run_accepts_qasm3_str_path(tmp_path: Path) -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_strong_accepts_qasm_path(tmp_path: Path) -> None: +def test_simulator_run_strong_accepts_qasm_path() -> None: """Verify that Simulator.run with StrongSimParams accepts a QASM file passed as a Path.""" - qasm_file = write_qasm_file(tmp_path, LARGE_QASM2_STRING) + qasm_file = Path(__file__).parent / "circuit.qasm" state = State(6, initial="zeros") sim_params = StrongSimParams(observables=[Observable(Z(), 0)], num_traj=1, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) assert result.expectation_values[0] is not None -def test_simulator_run_strong_accepts_qasm_string(tmp_path: Path) -> None: +def test_simulator_run_strong_accepts_qasm_string() -> None: """Verify that Simulator.run with StrongSimParams accepts a QASM file passed as a str path.""" - qasm_string = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) + qasm_string = Path(__file__).parent / "circuit.qasm" state = State(6, initial="zeros") sim_params = StrongSimParams(observables=[Observable(Z(), 0)], num_traj=1, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_string, sim_params) From d27e0f92c1d7f1586acddc6a3637afe17d006fb0 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 10 Jun 2026 13:24:38 +0000 Subject: [PATCH 06/15] =?UTF-8?q?=F0=9F=8E=A8=20pre-commit=20fixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/digital/utils/test_qudit_dag_utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index c51899e27..bacb24b8d 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -12,7 +12,6 @@ ``circuit_to_dag`` / ``dag_to_circuit`` round-trip. """ -# ruff: noqa: PLR6301 from __future__ import annotations import numpy as np From 0c272f8448c147387c4928d749a4872a925a2ebf Mon Sep 17 00:00:00 2001 From: Unglaub Date: Thu, 11 Jun 2026 12:01:04 +0200 Subject: [PATCH 07/15] Skip qudit tests when mqt-qudits is not installed --- pyproject.toml | 2 ++ tests/digital/utils/test_qudit_dag_utils.py | 3 +++ 2 files changed, 5 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 1ea2428b5..d7d76f178 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -112,6 +112,8 @@ error-on-warning = true [tool.ty.src] exclude = [ "docs/**", + "src/mqt/yaqs/digital/utils/qudit_dag_utils.py", + "tests/digital/utils/test_qudit_dag_utils.py", ] diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index bacb24b8d..802530358 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -16,6 +16,9 @@ import numpy as np import pytest + +pytest.importorskip("mqt.qudits") + from mqt.qudits.quantum_circuit import QuantumCircuit from mqt.yaqs.digital.utils.qudit_dag_utils import ( From 83f40b16e4437ba4065781817e8d4f17d661e479 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 17 Jun 2026 11:36:04 +0000 Subject: [PATCH 08/15] =?UTF-8?q?=F0=9F=8E=A8=20pre-commit=20fixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/yaqs/equivalence_checker.py | 3 +-- tests/digital/utils/test_dag_utils.py | 2 +- tests/test_equivalence_checker.py | 3 +-- tests/test_simulator.py | 6 ++---- 4 files changed, 5 insertions(+), 9 deletions(-) diff --git a/src/mqt/yaqs/equivalence_checker.py b/src/mqt/yaqs/equivalence_checker.py index 3b5c896c0..bc0af5944 100644 --- a/src/mqt/yaqs/equivalence_checker.py +++ b/src/mqt/yaqs/equivalence_checker.py @@ -17,6 +17,7 @@ from __future__ import annotations import time +from pathlib import Path from typing import TYPE_CHECKING, Literal, TypedDict, cast from qiskit.converters import circuit_to_dag @@ -31,8 +32,6 @@ from .digital.utils.qasm_utils import load_circuit if TYPE_CHECKING: - from pathlib import Path - import numpy as np from numpy.typing import NDArray from qiskit.circuit import QuantumCircuit diff --git a/tests/digital/utils/test_dag_utils.py b/tests/digital/utils/test_dag_utils.py index 38d397c95..e168c9789 100644 --- a/tests/digital/utils/test_dag_utils.py +++ b/tests/digital/utils/test_dag_utils.py @@ -359,7 +359,7 @@ class _CustomNamedUnitary(Gate): def __init__(self) -> None: super().__init__("custom", 1, []) - def to_matrix(self) -> np.ndarray: # noqa: PLR6301 + def to_matrix(self) -> np.ndarray: return np.array([[0, 1], [1, 0]], dtype=np.complex128) diff --git a/tests/test_equivalence_checker.py b/tests/test_equivalence_checker.py index 466239b6d..548bc5b17 100644 --- a/tests/test_equivalence_checker.py +++ b/tests/test_equivalence_checker.py @@ -14,6 +14,7 @@ from __future__ import annotations +from pathlib import Path from typing import TYPE_CHECKING, Literal, cast from unittest.mock import patch @@ -32,8 +33,6 @@ from tests.conftest import LARGE_QASM2_STRING, SAMPLE_QASM3_STRING, requires_qasm3_import, write_qasm_file if TYPE_CHECKING: - from pathlib import Path - from mqt.yaqs.equivalence_checker import Representation diff --git a/tests/test_simulator.py b/tests/test_simulator.py index 4a793e353..4a9333f46 100644 --- a/tests/test_simulator.py +++ b/tests/test_simulator.py @@ -22,7 +22,8 @@ import multiprocessing import os import sys -from typing import TYPE_CHECKING, Any, cast +from pathlib import Path +from typing import Any, cast import numba import numpy as np @@ -53,9 +54,6 @@ write_qasm_file, ) -if TYPE_CHECKING: - from pathlib import Path - def test_simulator_defaults() -> None: """Simulator() initializes with sensible defaults (parallel=True, auto mp_context).""" From 0b70a8f061539da84d08aa77a393af4e2e6faefa Mon Sep 17 00:00:00 2001 From: Unglaub Date: Sun, 21 Jun 2026 19:21:50 +0200 Subject: [PATCH 09/15] Update on the same status as main in the main repository --- src/mqt/yaqs/equivalence_checker.py | 33 +++-------------------------- 1 file changed, 3 insertions(+), 30 deletions(-) diff --git a/src/mqt/yaqs/equivalence_checker.py b/src/mqt/yaqs/equivalence_checker.py index bc0af5944..2cd9f6307 100644 --- a/src/mqt/yaqs/equivalence_checker.py +++ b/src/mqt/yaqs/equivalence_checker.py @@ -17,7 +17,6 @@ from __future__ import annotations import time -from pathlib import Path from typing import TYPE_CHECKING, Literal, TypedDict, cast from qiskit.converters import circuit_to_dag @@ -32,41 +31,15 @@ from .digital.utils.qasm_utils import load_circuit if TYPE_CHECKING: + from pathlib import Path + import numpy as np from numpy.typing import NDArray from qiskit.circuit import QuantumCircuit from .parallel_utils import MPContext -__all__ = ["DEFAULT_MATRIX_MAX_QUBITS", "EquivalenceChecker", "Representation"] - - -def _first_non_comment_line(text: str) -> str: - for line in text.splitlines(): - stripped = line.strip() - if stripped and not stripped.startswith("//"): - return stripped - return "" - - -def _load_if_path(circuit: QuantumCircuit | str | Path) -> QuantumCircuit: - """Load a QuantumCircuit from a QASM string, file path, or pass through an existing circuit. - - Returns: - The loaded or passed-through ``QuantumCircuit``. - """ - if isinstance(circuit, str) and _first_non_comment_line(circuit).startswith("OPENQASM"): - if _first_non_comment_line(circuit).startswith("OPENQASM 3"): - return qasm3.loads(circuit) - return qasm2.loads(circuit) - if isinstance(circuit, (str, Path)): - path = Path(circuit) - content = path.read_text(encoding="utf-8") - if _first_non_comment_line(content).startswith("OPENQASM 3"): - return qasm3.load(str(path)) - return qasm2.load(str(path)) - return circuit - +__all__ = ["DEFAULT_MATRIX_MAX_QUBITS", "EquivalenceCheckResult", "EquivalenceChecker", "Representation"] Representation = Literal["auto", "matrix", "mpo"] DEFAULT_MATRIX_MAX_QUBITS = 7 From 1ada8b5123416b0e6a6ccd0ab714eee731db7086 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Sun, 21 Jun 2026 20:52:35 +0200 Subject: [PATCH 10/15] Update branch to main --- tests/test_equivalence_checker.py | 27 +++++++++---------- tests/test_simulator.py | 43 +++++++++++++++++++------------ 2 files changed, 41 insertions(+), 29 deletions(-) diff --git a/tests/test_equivalence_checker.py b/tests/test_equivalence_checker.py index 817c4d3f9..969fa10fe 100644 --- a/tests/test_equivalence_checker.py +++ b/tests/test_equivalence_checker.py @@ -14,7 +14,6 @@ from __future__ import annotations -from pathlib import Path from typing import TYPE_CHECKING, Literal, cast from unittest.mock import patch @@ -34,6 +33,8 @@ from tests.conftest import LARGE_QASM2_STRING, SAMPLE_QASM3_STRING, requires_qasm3_import, write_qasm_file if TYPE_CHECKING: + from pathlib import Path + from mqt.yaqs.equivalence_checker import Representation @@ -492,27 +493,27 @@ def test_long_range_mpo_parallel() -> None: assert serial["equivalent"] == parallel["equivalent"] -def test_check_accepts_qasm2_path_object() -> None: +def test_check_accepts_qasm2_path_object(tmp_path: Path) -> None: """Check that a QASM 2 file given as a Path object is accepted and returns equivalent.""" - qasm_path = Path(__file__).parent / "circuit.qasm" + qasm_path = write_qasm_file(tmp_path, LARGE_QASM2_STRING) checker = EquivalenceChecker(representation="mpo") result = checker.check(qasm_path, qasm_path) assert result["equivalent"] is True -def test_check_accepts_qasm2_str_path() -> None: +def test_check_accepts_qasm2_str_path(tmp_path: Path) -> None: """Check that a QASM 2 file given as a str path is accepted and returns equivalent.""" - qasm_path = str(Path(__file__).parent / "circuit.qasm") + qasm_path = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) checker = EquivalenceChecker(representation="mpo") result = checker.check(qasm_path, qasm_path) assert result["equivalent"] is True -def test_check_qasm_path_vs_quantumcircuit_agree() -> None: +def test_check_qasm_path_vs_quantumcircuit_agree(tmp_path: Path) -> None: """Verify that loading via path and via QuantumCircuit gives the same equivalence result.""" - qasm_path = Path(__file__).parent / "circuit.qasm" + qasm_path = write_qasm_file(tmp_path, LARGE_QASM2_STRING) qc = load(filename=str(qasm_path)) checker = EquivalenceChecker(representation="mpo") result_path = checker.check(qasm_path, qasm_path) @@ -520,20 +521,20 @@ def test_check_qasm_path_vs_quantumcircuit_agree() -> None: assert result_path["equivalent"] == result_qc["equivalent"] -def test_check_accepts_qasm3_path_object() -> None: +@requires_qasm3_import +def test_check_accepts_qasm3_path_object(tmp_path: Path) -> None: """Check that a QASM 3 file given as a Path object is accepted and returns equivalent.""" - pytest.importorskip("qiskit_qasm3_import") - qasm_file = Path(__file__).parent / "circuit3.qasm" + qasm_file = write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm") checker = EquivalenceChecker(representation="matrix") result = checker.check(qasm_file, qasm_file) assert result["equivalent"] is True -def test_check_accepts_qasm3_str_path() -> None: +@requires_qasm3_import +def test_check_accepts_qasm3_str_path(tmp_path: Path) -> None: """Check that a QASM 3 file given as a str path is accepted and returns equivalent.""" - pytest.importorskip("qiskit_qasm3_import") - qasm_file = str(Path(__file__).parent / "circuit3.qasm") + qasm_file = str(write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm")) checker = EquivalenceChecker(representation="matrix") result = checker.check(qasm_file, qasm_file) diff --git a/tests/test_simulator.py b/tests/test_simulator.py index af2f98bf8..3cdd8ad68 100644 --- a/tests/test_simulator.py +++ b/tests/test_simulator.py @@ -22,8 +22,7 @@ import multiprocessing import os import sys -from pathlib import Path -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast import numba import numpy as np @@ -56,6 +55,9 @@ write_qasm_file, ) +if TYPE_CHECKING: + from pathlib import Path + def test_simulator_defaults() -> None: """Simulator() initializes with sensible defaults (parallel=True, auto mp_context).""" @@ -1452,9 +1454,9 @@ def test_analog_simulation_parallel_observables_no_state() -> None: assert result.runtime_cost is not None -def test_simulator_run_accepts_qasm2_path_object() -> None: +def test_simulator_run_accepts_qasm2_path_object(tmp_path: Path) -> None: """Verify that Simulator.run accepts a QASM 2 file passed as a Path object.""" - qasm_file = Path(__file__).parent / "circuit.qasm" + qasm_file = write_qasm_file(tmp_path, LARGE_QASM2_STRING) state = State(6, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1462,9 +1464,9 @@ def test_simulator_run_accepts_qasm2_path_object() -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_accepts_qasm2_str_path() -> None: +def test_simulator_run_accepts_qasm2_str_path(tmp_path: Path) -> None: """Verify that Simulator.run accepts a QASM 2 file passed as a str path.""" - qasm_file = str(Path(__file__).parent / "circuit.qasm") + qasm_file = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) state = State(6, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1472,10 +1474,19 @@ def test_simulator_run_accepts_qasm2_str_path() -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_accepts_qasm3_path_object() -> None: +def test_simulator_run_accepts_qasm2_raw_string() -> None: + """Verify that Simulator.run accepts a raw QASM 2 string (not a file path).""" + state = State(6, initial="zeros") + sim_params = WeakSimParams(shots=4, max_bond_dim=4) + result = Simulator(parallel=False, show_progress=False).run(state, LARGE_QASM2_STRING, sim_params) + assert result.counts is not None + assert sum(result.counts.values()) == sim_params.shots + + +@requires_qasm3_import +def test_simulator_run_accepts_qasm3_path_object(tmp_path: Path) -> None: """Verify that Simulator.run accepts a QASM 3 file passed as a Path object.""" - pytest.importorskip("qiskit_qasm3_import") - qasm_file = Path(__file__).parent / "circuit3.qasm" + qasm_file = write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm") state = State(2, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1483,10 +1494,10 @@ def test_simulator_run_accepts_qasm3_path_object() -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_accepts_qasm3_str_path() -> None: +@requires_qasm3_import +def test_simulator_run_accepts_qasm3_str_path(tmp_path: Path) -> None: """Verify that Simulator.run accepts a QASM 3 file passed as a str path.""" - pytest.importorskip("qiskit_qasm3_import") - qasm_file = str(Path(__file__).parent / "circuit3.qasm") + qasm_file = str(write_qasm_file(tmp_path, SAMPLE_QASM3_STRING, filename="circuit3.qasm")) state = State(2, initial="zeros") sim_params = WeakSimParams(shots=4, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) @@ -1494,18 +1505,18 @@ def test_simulator_run_accepts_qasm3_str_path() -> None: assert sum(result.counts.values()) == sim_params.shots -def test_simulator_run_strong_accepts_qasm_path() -> None: +def test_simulator_run_strong_accepts_qasm_path(tmp_path: Path) -> None: """Verify that Simulator.run with StrongSimParams accepts a QASM file passed as a Path.""" - qasm_file = Path(__file__).parent / "circuit.qasm" + qasm_file = write_qasm_file(tmp_path, LARGE_QASM2_STRING) state = State(6, initial="zeros") sim_params = StrongSimParams(observables=[Observable(Z(), 0)], num_traj=1, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_file, sim_params) assert result.expectation_values[0] is not None -def test_simulator_run_strong_accepts_qasm_string() -> None: +def test_simulator_run_strong_accepts_qasm_string(tmp_path: Path) -> None: """Verify that Simulator.run with StrongSimParams accepts a QASM file passed as a str path.""" - qasm_string = Path(__file__).parent / "circuit.qasm" + qasm_string = str(write_qasm_file(tmp_path, LARGE_QASM2_STRING)) state = State(6, initial="zeros") sim_params = StrongSimParams(observables=[Observable(Z(), 0)], num_traj=1, max_bond_dim=4) result = Simulator(parallel=False, show_progress=False).run(state, qasm_string, sim_params) From 38c55fa87aa6ab9ac0e4f96b83e1f05e71d44d5f Mon Sep 17 00:00:00 2001 From: Unglaub Date: Sun, 21 Jun 2026 22:32:00 +0200 Subject: [PATCH 11/15] Solve codecov issue --- pyproject.toml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 564d13702..a0cb6b24e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -92,6 +92,10 @@ filterwarnings = ["error"] [tool.coverage] run.source = ["mqt.yaqs"] +run.omit = [ + "src/mqt/yaqs/digital/utils/qudit_dag_utils.py", +] + report.exclude_also = [ '\.\.\.', 'if TYPE_CHECKING:', From 6132eaa34f22b3d4a43cb8e5d5f72995b75c42aa Mon Sep 17 00:00:00 2001 From: Unglaub Date: Wed, 15 Jul 2026 17:20:39 +0200 Subject: [PATCH 12/15] implement level awareness blocking and markUsed implementation --- src/mqt/yaqs/digital/utils/qudit_dag_utils.py | 76 +++++++++++++++---- tests/digital/utils/test_qudit_dag_utils.py | 20 +++++ 2 files changed, 82 insertions(+), 14 deletions(-) diff --git a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py index ed225131c..4422d393a 100644 --- a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py +++ b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py @@ -59,6 +59,66 @@ def _levels_for_gate(gate: Gate, _qudit_idx: int, dimension: int) -> list[int]: return sorted({lev_a, lev_b}) +def _mark_used(node: QuditOpNode, used: set[QuditOpNode]) -> None: + """Recursively mark *node* and all of its ancestors as used. + + Mirrors Algorithm 2 (``markUsed``) from the subspace-tracking DAG paper: + once a node has been chosen as the nearest blocker for the gate currently + being wired, it and everything it already depends on are skipped for the + remainder of that gate's backward scan, so only covering relations of the + blocking partial order end up as edges. + + Args: + node: The node to mark, together with its ancestors. + used: Set of already-used nodes for the current scan (mutated in place). + """ + if node in used: + return + used.add(node) + for dep in node.dependencies: + _mark_used(dep, used) + + +def _blocked(node_i: QuditOpNode, node_g: QuditOpNode) -> bool: + """Return True if *node_i* blocks *node_g*. + + Two gates block each other iff they share a qudit on which their touched + energy levels overlap. Gates on the same qudit but disjoint levels act on + different subspaces and therefore do not block each other. + + Args: + node_i: The earlier candidate node. + node_g: The later node being wired. + + Returns: + True iff node_i must precede node_g. + """ + shared = set(node_i.target_qudits) & set(node_g.target_qudits) + return any(set(node_i.levels[q]) & set(node_g.levels[q]) for q in shared) + + +def _wire_dependencies(node: QuditOpNode, candidates: list[QuditOpNode]) -> None: + """Wire *node*'s dependency edges against already-built *candidates*. + + Implements the nearest-blocker scan of Algorithm 1/4: candidates are + visited in reverse (most recent first); the first blocking node found on + each chain gets a direct edge, and it together with all of its ancestors + is then marked used so the resulting graph stays Hasse-diagram minimal + (no redundant transitive edges). + + Args: + node: The newly created node to wire (not yet appended anywhere). + candidates: Already-built nodes, in circuit/insertion order. + """ + used: set[QuditOpNode] = set() + for prev in reversed(candidates): + if prev in used: + continue + if _blocked(prev, node): + node.add_dependency(prev) + _mark_used(prev, used) + + class QuditDAGNode: """Base node in a :class:`QuditDAG`. @@ -193,8 +253,6 @@ def __init__( def _build(self, circuit: QuantumCircuit) -> None: """Populate *nodes* and wire dependency edges.""" - last_node_per_qudit: list[QuditOpNode | None] = [None] * self.num_qudits - for idx, instruction in enumerate(circuit.instructions): targets: list[int] = getattr(instruction, "target_qudits", []) if isinstance(targets, int): @@ -205,12 +263,7 @@ def _build(self, circuit: QuantumCircuit) -> None: node = QuditOpNode(idx, instruction, targets, dims, levels) - for q in targets: - prev = last_node_per_qudit[q] - if prev is not None: - node.add_dependency(prev) - last_node_per_qudit[q] = node - + _wire_dependencies(node, self.nodes) self.nodes.append(node) def op_nodes(self) -> list[QuditOpNode]: @@ -251,12 +304,7 @@ def apply_operation_back( dims = [self.dimensions[q] for q in qargs] levels = {q: _levels_for_gate(gate, q, self.dimensions[q]) for q in qargs} node = QuditOpNode(new_idx, gate, qargs, dims, levels) - - for q in qargs: - q_nodes = [n for n in self.nodes if q in n.target_qudits] - if q_nodes: - node.add_dependency(max(q_nodes, key=lambda n: n.index)) - + _wire_dependencies(node, self.nodes) self.nodes.append(node) return node diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index 802530358..75b99a53c 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -138,6 +138,26 @@ def test_chain_on_single_qudit(self, linear_circuit: QuantumCircuit) -> None: assert dag.nodes[1] in dag.nodes[2].dependencies assert dag.nodes[0] not in dag.nodes[2].dependencies + def test_disjoint_levels_on_same_qudit_have_no_dep(self) -> None: + """Two gates on the same qudit with disjoint levels are not blocked.""" + qc = QuantumCircuit(1, [4]) + qc.rz(0, [0, 1, np.pi / 4]) + qc.rz(0, [2, 3, np.pi / 4]) + dag = circuit_to_dag(qc) + assert dag.nodes[0] not in dag.nodes[1].dependencies + assert len(dag.nodes[1].dependencies) == 0 + + def test_no_redundant_transitive_edge(self) -> None: + """A gate depends only on its nearest blocker, not on transitively covered ancestors.""" + qc = QuantumCircuit(2, [2, 2]) + qc.cx([0, 1]) + qc.rz(0, [0, 1, np.pi / 4]) + qc.cx([0, 1]) + dag = circuit_to_dag(qc) + g0, g1, g2 = dag.nodes + assert g1 in g2.dependencies + assert g0 not in g2.dependencies + class TestLevelTracking: """Tests for energy-level tracking per gate.""" From fe81e977d5630f356115eff223c4cb9c4e36e5d6 Mon Sep 17 00:00:00 2001 From: Unglaub Date: Wed, 15 Jul 2026 18:12:18 +0200 Subject: [PATCH 13/15] Implement SubspaceMap as well as propagation algorithm --- src/mqt/yaqs/digital/utils/qudit_dag_utils.py | 73 +++++++++++++++++++ tests/digital/utils/test_qudit_dag_utils.py | 39 ++++++++++ 2 files changed, 112 insertions(+) diff --git a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py index 4422d393a..51d4c2f2d 100644 --- a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py +++ b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py @@ -24,6 +24,7 @@ from __future__ import annotations +import itertools from typing import TYPE_CHECKING from mqt.qudits.quantum_circuit import QuantumCircuit @@ -33,6 +34,11 @@ from mqt.qudits.quantum_circuit.gate import Gate +SubspaceMap = dict[tuple[int, int], tuple[set[int], set[int]]] +"""Maps an unordered qudit pair (q, q') to (Sq, Sq'): the smallest per-qudit +level sets within which the joint history of q and q' is known to be +confined so far (paper Section III-A).""" + def _levels_for_gate(gate: Gate, _qudit_idx: int, dimension: int) -> list[int]: """Return which energy levels *gate* acts on for a given qudit. @@ -119,6 +125,66 @@ def _wire_dependencies(node: QuditOpNode, candidates: list[QuditOpNode]) -> None _mark_used(prev, used) +def _subspace_key(q: int, q_prime: int) -> tuple[int, int]: + """Return the canonical (sorted) key used to index a SubspaceMap for (q, q').""" + return (q, q_prime) if q < q_prime else (q_prime, q) + + +def _get_subspace(subspace_map: SubspaceMap, q: int, q_prime: int) -> tuple[set[int], set[int]]: + """Return (Sq,Sq') as currently recorded for the qudit pair (q, q').""" + sq, sq_prime = subspace_map.get(_subspace_key(q, q_prime), (set(), set())) + return (sq, sq_prime) if q < q_prime else (sq_prime, sq) + + +def _set_subspace( + subspace_map: SubspaceMap, q: int, q_prime: int, levels_q: set[int], levels_q_prime: set[int] +) -> None: + """Store (levels_q, levels_q') as the SubspaceMap entry for (q, q').""" + key = _subspace_key(q, q_prime) + subspace_map[key] = (levels_q, levels_q_prime) if q < q_prime else (levels_q_prime, levels_q) + + +def _propagate_subspace( + subspace_map: SubspaceMap, + num_qudits: int, + q: int, + q_prime: int, + levels_q: set[int], + levels_q_prime: set[int], +) -> None: + """Propagate correlation between *q* and *q'* to every other qudit (Algorithm 5). + + If an earlier gate already correlated q with some q'' on levels overlapping + what the current gate touches at q, that correlation now reaches q' too + (entanglement swapping) -- and symmetrically with q, q' exchanged. + """ + for q2 in range(num_qudits): + if q2 in {q, q_prime}: + continue + a, b = _get_subspace(subspace_map, q, q2) + c, d = _get_subspace(subspace_map, q_prime, q2) + if a & levels_q: + _set_subspace(subspace_map, q_prime, q2, c | levels_q_prime, d | b) + if c & levels_q_prime: + _set_subspace(subspace_map, q, q2, a | levels_q, b | d) + + +def _update_subspace_map( + subspace_map: SubspaceMap, + num_qudits: int, + targets: list[int], + levels: dict[int, list[int]], +) -> None: + """Update *subspace_map* for a gate touching >=2 qudits (Algorithm 4, lines 12-19).""" + if len(targets) < 2: + return + for q, q_prime in itertools.combinations(targets, 2): + ell, ell_prime = set(levels[q]), set(levels[q_prime]) + sq, sq_prime = _get_subspace(subspace_map, q, q_prime) + _set_subspace(subspace_map, q, q_prime, sq | ell, sq_prime | ell_prime) + _propagate_subspace(subspace_map, num_qudits, q, q_prime, ell, ell_prime) + + class QuditDAGNode: """Base node in a :class:`QuditDAG`. @@ -247,6 +313,7 @@ def __init__( self.circuit: QuantumCircuit | None = circuit self.num_qudits: int = len(self.dimensions) self.nodes: list[QuditOpNode] = [] + self.subspace_map: SubspaceMap = {} if circuit is not None: self._build(circuit) @@ -264,6 +331,7 @@ def _build(self, circuit: QuantumCircuit) -> None: node = QuditOpNode(idx, instruction, targets, dims, levels) _wire_dependencies(node, self.nodes) + _update_subspace_map(self.subspace_map, self.num_qudits, targets, levels) self.nodes.append(node) def op_nodes(self) -> list[QuditOpNode]: @@ -305,6 +373,7 @@ def apply_operation_back( levels = {q: _levels_for_gate(gate, q, self.dimensions[q]) for q in qargs} node = QuditOpNode(new_idx, gate, qargs, dims, levels) _wire_dependencies(node, self.nodes) + _update_subspace_map(self.subspace_map, self.num_qudits, qargs, levels) self.nodes.append(node) return node @@ -367,6 +436,10 @@ def get_edges(self) -> list[tuple[QuditDAGNode, QuditOpNode]]: """Return all (predecessor, successor) dependency pairs.""" return [(dep, node) for node in self.nodes for dep in node.dependencies] + def get_subspace(self, q: int, q_prime: int) -> tuple[set[int], set[int]]: + """Return the recorded joint history level sets for qudits q and q'.""" + return _get_subspace(self.subspace_map, q, q_prime) + def display(self) -> None: """Print DAG structure (no-op placeholder).""" diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index 75b99a53c..04e8d0498 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -159,6 +159,45 @@ def test_no_redundant_transitive_edge(self) -> None: assert g0 not in g2.dependencies +class TestSubspaceMap: + """Tests for SubspaceMap tracking (Algorithm 4/5).""" + + def test_ghz_propagation(self) -> None: + """Indirect correlation propagates across a shared intermediary qudit.""" + qc = QuantumCircuit(3, [2, 2, 2]) + qc.h(0) + qc.cx([0, 1]) + qc.cx([1, 2]) + dag = circuit_to_dag(qc) + assert dag.get_subspace(0, 2) == ({0, 1}, {0, 1}) + + def test_isolated_qudit_has_empty_subspace(self) -> None: + """A qudit never touched by a multi-qudit gate stays unrecorded.""" + qc = QuantumCircuit(3, [2, 2, 2]) + qc.cx([0, 1]) + dag = circuit_to_dag(qc) + assert dag.get_subspace(0, 2) == (set(), set()) + assert dag.get_subspace(1, 2) == (set(), set()) + + def test_subspace_lookup_is_symmetric(self) -> None: + """get_subspace(q, q') mirrors get_subspace(q', q).""" + qc = QuantumCircuit(2, [2, 2]) + qc.cx([0, 1]) + dag = circuit_to_dag(qc) + sq, sq_prime = dag.get_subspace(0, 1) + sq_prime2, sq2 = dag.get_subspace(1, 0) + assert (sq, sq_prime) == (sq2, sq_prime2) + + def test_apply_operation_back_updates_subspace(self) -> None: + """Appending a gate via apply_operation_back also updates the SubspaceMap.""" + qc = QuantumCircuit(3, [2, 2, 2]) + qc.h(0) + dag = circuit_to_dag(qc) + gate = qc.cx([1, 2]) + dag.apply_operation_back(gate, [1, 2]) + assert dag.get_subspace(1, 2) == ({0, 1}, {0, 1}) + + class TestLevelTracking: """Tests for energy-level tracking per gate.""" From 99ab6a841a7f93c9cedeb27f6472f4c4cd459659 Mon Sep 17 00:00:00 2001 From: Marerido Date: Wed, 15 Jul 2026 19:14:03 +0200 Subject: [PATCH 14/15] Fix bug in which control and target qudits are not really distinguishable --- src/mqt/yaqs/digital/utils/qudit_dag_utils.py | 20 +++++++++---- tests/digital/utils/test_qudit_dag_utils.py | 30 ++++++++++++++----- 2 files changed, 38 insertions(+), 12 deletions(-) diff --git a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py index 51d4c2f2d..bc63bfa17 100644 --- a/src/mqt/yaqs/digital/utils/qudit_dag_utils.py +++ b/src/mqt/yaqs/digital/utils/qudit_dag_utils.py @@ -40,22 +40,32 @@ confined so far (paper Section III-A).""" -def _levels_for_gate(gate: Gate, _qudit_idx: int, dimension: int) -> list[int]: +def _levels_for_gate(gate: Gate, position: int, dimension: int) -> list[int]: """Return which energy levels *gate* acts on for a given qudit. MQT Qudit gates carry ``lev_a`` and ``lev_b`` (the two transition levels). When both are 0 the gate does not have a meaningful two-level transition (e.g. a full-unitary gate), so the full level range is returned. + Controlled gates that encode their control qudit as the first entry of + ``target_qudits`` (e.g. ``CEx``) expose a distinct ``ctrl_lev`` attribute + for the level that triggers the control -- using ``lev_a``/``lev_b`` (the + *target's* transition levels) for that qudit would be wrong, since the + control qudit is left unchanged in every branch except ``ctrl_lev``. + Args: gate: The MQT Qudit gate object. - _qudit_idx: Index of the qudit within the full circuit (reserved for a - future per-qudit override; not used today). + position: This qudit's position within the gate's own ``target_qudits`` + list (0 = first entry, etc.), used to recognize the control slot + of gates like ``CEx``. dimension: Physical dimension of that qudit. Returns: Sorted list of level indices the gate touches. """ + if position == 0 and hasattr(gate, "ctrl_lev"): + return [gate.ctrl_lev] + lev_a: int = getattr(gate, "lev_a", 0) lev_b: int = getattr(gate, "lev_b", 0) @@ -326,7 +336,7 @@ def _build(self, circuit: QuantumCircuit) -> None: targets = [targets] dims = [self.dimensions[q] for q in targets] - levels = {q: _levels_for_gate(instruction, q, self.dimensions[q]) for q in targets} + levels = {q: _levels_for_gate(instruction, i, self.dimensions[q]) for i, q in enumerate(targets)} node = QuditOpNode(idx, instruction, targets, dims, levels) @@ -370,7 +380,7 @@ def apply_operation_back( """ new_idx = max((n.index for n in self.nodes), default=-1) + 1 dims = [self.dimensions[q] for q in qargs] - levels = {q: _levels_for_gate(gate, q, self.dimensions[q]) for q in qargs} + levels = {q: _levels_for_gate(gate, i, self.dimensions[q]) for i, q in enumerate(qargs)} node = QuditOpNode(new_idx, gate, qargs, dims, levels) _wire_dependencies(node, self.nodes) _update_subspace_map(self.subspace_map, self.num_qudits, qargs, levels) diff --git a/tests/digital/utils/test_qudit_dag_utils.py b/tests/digital/utils/test_qudit_dag_utils.py index 04e8d0498..614c57460 100644 --- a/tests/digital/utils/test_qudit_dag_utils.py +++ b/tests/digital/utils/test_qudit_dag_utils.py @@ -40,7 +40,7 @@ def simple_circuit() -> QuantumCircuit: """ qc = QuantumCircuit(3, [2, 5, 4]) qc.cx([1, 2]) - qc.rz(1, [0, 3, np.pi / 2]) + qc.rz(1, [1, 3, np.pi / 2]) qc.r(0, [0, 1, np.pi, np.pi / 2]) return qc @@ -158,6 +158,22 @@ def test_no_redundant_transitive_edge(self) -> None: assert g1 in g2.dependencies assert g0 not in g2.dependencies + def test_control_qudit_blocks_only_at_ctrl_lev(self) -> None: + """A CEx's control qudit blocks on its ctrl_lev, not the target's transition levels.""" + qc = QuantumCircuit(2, [3, 2]) + qc.cx([0, 1]) # ctrl_lev=1 (default), target levels {0, 1} + qc.rz(0, [0, 2, np.pi / 4]) # touches control qudit at {0, 2} -- disjoint from ctrl_lev + dag = circuit_to_dag(qc) + assert dag.nodes[0] not in dag.nodes[1].dependencies + + def test_control_qudit_blocks_when_overlapping_ctrl_lev(self) -> None: + """A gate touching the control qudit's actual ctrl_lev is correctly blocked.""" + qc = QuantumCircuit(2, [3, 2]) + qc.cx([0, 1]) # ctrl_lev=1 + qc.rz(0, [1, 2, np.pi / 4]) # touches level 1 -- overlaps ctrl_lev + dag = circuit_to_dag(qc) + assert dag.nodes[0] in dag.nodes[1].dependencies + class TestSubspaceMap: """Tests for SubspaceMap tracking (Algorithm 4/5).""" @@ -169,7 +185,7 @@ def test_ghz_propagation(self) -> None: qc.cx([0, 1]) qc.cx([1, 2]) dag = circuit_to_dag(qc) - assert dag.get_subspace(0, 2) == ({0, 1}, {0, 1}) + assert dag.get_subspace(0, 2) == ({1}, {0, 1}) def test_isolated_qudit_has_empty_subspace(self) -> None: """A qudit never touched by a multi-qudit gate stays unrecorded.""" @@ -195,24 +211,24 @@ def test_apply_operation_back_updates_subspace(self) -> None: dag = circuit_to_dag(qc) gate = qc.cx([1, 2]) dag.apply_operation_back(gate, [1, 2]) - assert dag.get_subspace(1, 2) == ({0, 1}, {0, 1}) + assert dag.get_subspace(1, 2) == ({1}, {0, 1}) class TestLevelTracking: """Tests for energy-level tracking per gate.""" def test_cx_levels(self, simple_circuit: QuantumCircuit) -> None: - """CX with default lev_a=lev_b=0 uses the lowest two levels.""" + """CX records ctrl_lev for the control qudit and lev_a/lev_b for the target.""" dag = circuit_to_dag(simple_circuit) cx_node = dag.nodes[0] - assert cx_node.levels[1] == [0, 1] + assert cx_node.levels[1] == [1] assert cx_node.levels[2] == [0, 1] def test_rz_levels_with_high_transition(self, simple_circuit: QuantumCircuit) -> None: - """RZ with lev_a=0, lev_b=3 records levels [0, 3].""" + """RZ with lev_a=1, lev_b=3 records levels [1, 3].""" dag = circuit_to_dag(simple_circuit) rz_node = dag.nodes[1] - assert rz_node.levels[1] == [0, 3] + assert rz_node.levels[1] == [1, 3] def test_r_gate_levels(self, simple_circuit: QuantumCircuit) -> None: """R gate with lev_a=0, lev_b=1 records levels [0, 1].""" From 338d6cb0b252029cb7478e759691d6c55b4625a8 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:56:20 +0000 Subject: [PATCH 15/15] =?UTF-8?q?=F0=9F=8E=A8=20pre-commit=20fixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/characterization/memory/operational_memory/test_run.py | 2 +- tests/core/data_structures/test_mps.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/characterization/memory/operational_memory/test_run.py b/tests/characterization/memory/operational_memory/test_run.py index b385b901c..5258db7c4 100644 --- a/tests/characterization/memory/operational_memory/test_run.py +++ b/tests/characterization/memory/operational_memory/test_run.py @@ -5,7 +5,7 @@ # # Licensed under the MIT License -# ruff: noqa: PLR6301, PLC2701 -- protocol-style dummy backend; white-box rollout test +# ruff: noqa: PLC2701 -- protocol-style dummy backend; white-box rollout test """Tests for operational-memory orchestration (:mod:`run`).""" diff --git a/tests/core/data_structures/test_mps.py b/tests/core/data_structures/test_mps.py index 2949f4109..3b0f8a450 100644 --- a/tests/core/data_structures/test_mps.py +++ b/tests/core/data_structures/test_mps.py @@ -7,7 +7,7 @@ """Tests for :class:`mqt.yaqs.core.data_structures.mps.MPS`.""" -# ruff: noqa: N806, SLF001, PLR6301 +# ruff: noqa: N806, SLF001 from __future__ import annotations