From 0fcf167c6e2bfca0558ab8f7bf6d2a477b6cb718 Mon Sep 17 00:00:00 2001 From: David Vesterlund Date: Tue, 27 Oct 2026 17:34:03 +0100 Subject: [PATCH] Fix Qiskit stage plugins and release 0.1.2 The westquant layout/routing/optimization stage plugins crashed on Qiskit 2.x (QuantumCircuit.from_instructions(dag) on a DAGCircuit) and the candidate path relied on the private tqc._layout.dag. Failed candidates were swallowed. - Reimplement each stage as a deterministic best-of over Qiskit built-ins that sets layout/final_layout correctly and reuses Qiskit's embed and routing stage wrappers; record every candidate in property_set and raise when all candidates fail. - Add end-to-end tests: discovery, unitary equivalence, measurement distribution, coupling-map compliance, applied candidate, all-fail error. - Bring the BackendV2 Target support that shipped in the 0.1.1 wheel but was never committed into the source tree, with a test. - CI: visible nightly failures, single gated publish path, Qiskit 2.5.2 and 2.6.1 test matrix. Version 0.1.2 (0.1.0a5 and 0.1.1 are already on PyPI). Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .github/workflows/nightly.yml | 26 ++- .github/workflows/publish.yml | 59 ++++-- .github/workflows/release.yml | 66 ------- .github/workflows/tests.yml | 6 +- CHANGELOG.md | 25 +++ README.md | 29 ++- pyproject.toml | 2 +- src/westquant_qiskit/__init__.py | 2 +- src/westquant_qiskit/adapter.py | 156 ++++++++++++---- src/westquant_qiskit/stage.py | 300 +++++++++++++++---------------- tests/test_adapter.py | 15 ++ tests/test_stage.py | 123 +++++++++++++ 12 files changed, 520 insertions(+), 289 deletions(-) delete mode 100644 .github/workflows/release.yml create mode 100644 CHANGELOG.md create mode 100644 tests/test_stage.py diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index dc48ffd..f6a767f 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -1,5 +1,8 @@ name: Nightly SDK Breakage Detection +# Installs the newest releases of every dependency and fails visibly when the +# install, dependency resolution or tests break. Do not add continue-on-error, +# "|| true" or "|| echo" here: a red nightly run is the signal. on: schedule: - cron: "0 2 * * *" @@ -9,27 +12,22 @@ jobs: nightly: name: Test against latest SDK versions runs-on: ubuntu-latest - continue-on-error: true steps: - uses: actions/checkout@v4 - - name: Checkout WestQuant Core + - name: Checkout WestQuant Core main uses: actions/checkout@v4 with: repository: WestQuantOpen/westquant-core path: _westquant-core - uses: actions/setup-python@v5 with: - python-version: "3.10" - - name: Install latest SDKs + python-version: "3.12" + - name: Install latest dependencies run: | - pip install --upgrade pip - pip install -e ./_westquant-core[test] - pip install --upgrade qiskit pytket pennylane pulser 2>/dev/null || true - pip install -e .[test] 2>/dev/null || pip install -e . + python -m pip install --upgrade pip + pip install -e ./_westquant-core + pip install --upgrade --upgrade-strategy eager -e ".[test]" + pip check + pip freeze - name: Run tests - run: | - pytest tests/ -v || echo "Nightly tests failed — SDK breakage detected" - - name: Report status - if: always() - run: | - echo "Nightly run completed. Check for SDK breakage above." + run: pytest tests/ -v diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index a05c99c..40a9cc5 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -1,34 +1,55 @@ name: Publish to PyPI +# Single publication path. PyPI Trusted Publishing is bound to this file name +# (publish.yml) and the "pypi" environment. Tests must pass and the release tag +# must match the pyproject version before anything is uploaded. on: release: types: [published] workflow_dispatch: - inputs: - confirm: - description: 'Type "publish" to confirm' - required: true - default: 'publish' + +permissions: + contents: read jobs: - publish: - name: Build and publish + test: + uses: ./.github/workflows/tests.yml + + build: + needs: test runs-on: ubuntu-latest - environment: pypi - permissions: - id-token: write steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: - python-version: "3.10" - - name: Build package + python-version: "3.12" + - name: Check release tag matches package version + if: github.event_name == 'release' run: | - pip install build - python -m build - - name: Check package + version=$(python -c "import tomllib; print(tomllib.load(open('pyproject.toml', 'rb'))['project']['version'])") + if [ "${GITHUB_REF_NAME#v}" != "$version" ]; then + echo "::error::Release tag $GITHUB_REF_NAME does not match pyproject version $version" + exit 1 + fi + - name: Build and check distributions run: | - pip install twine - twine check dist/* - - name: Publish to PyPI - uses: pypa/gh-action-pypi-publish@release/v1 + python -m pip install --upgrade build twine + python -m build + twine check --strict dist/* + - uses: actions/upload-artifact@v4 + with: + name: python-package-distributions + path: dist/ + + publish: + needs: build + runs-on: ubuntu-latest + environment: pypi + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: python-package-distributions + path: dist/ + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml deleted file mode 100644 index f55898e..0000000 --- a/.github/workflows/release.yml +++ /dev/null @@ -1,66 +0,0 @@ -name: Release to PyPI - -on: - push: - tags: - - "v*" - -permissions: - contents: read - -jobs: - test: - runs-on: ubuntu-latest - strategy: - matrix: - python-version: ["3.10", "3.11", "3.12"] - steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - - run: | - python -m pip install --upgrade pip - pip install "qiskit~=2.5.2" pytest - pip install . - - run: pytest -q - - name: Verify Qiskit plugin discovery - run: | - python - <<'PY' - from qiskit.transpiler.preset_passmanagers.plugin import list_stage_plugins - for stage in ("layout", "routing", "optimization"): - plugins = list_stage_plugins(stage) - assert "westquant" in plugins, (stage, plugins) - PY - - name: Real sequential smoke search - run: | - westquant-qiskit-search --suite smoke --search sequential --beam-width 2 --output /tmp/wq-seq - test -f /tmp/wq-seq/manifest.json - - build: - needs: test - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 - with: - python-version: "3.12" - - run: python -m pip install --upgrade build - - run: python -m build - - uses: actions/upload-artifact@v4 - with: - name: python-package-distributions - path: dist/ - - publish: - needs: build - runs-on: ubuntu-latest - environment: pypi - permissions: - id-token: write - steps: - - uses: actions/download-artifact@v4 - with: - name: python-package-distributions - path: dist/ - - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 5a63867..4bb3f9f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -1,6 +1,7 @@ name: Tests on: + workflow_call: push: branches: [main] pull_request: @@ -12,6 +13,7 @@ jobs: strategy: matrix: python-version: ["3.10", "3.11", "3.12"] + qiskit-version: ["2.5.2", "2.6.1"] steps: - uses: actions/checkout@v4 - name: Set up Python @@ -20,8 +22,8 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install run: | - pip install westquant-core>=0.2.0a1 - pip install "qiskit~=2.5.2" + pip install "westquant-core>=0.2.0a1" + pip install "qiskit==${{ matrix.qiskit-version }}" pip install -e .[test] - name: Test run: pytest tests/ -v diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..145bc33 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,25 @@ +# Changelog + +## 0.1.2 + +- Fix: the `westquant` layout, routing and optimization stage plugins crashed + on Qiskit 2.x (`QuantumCircuit.from_instructions(dag)` with a `DAGCircuit`) + and the candidate path read a private `tqc._layout.dag` attribute. The stages + are reimplemented as best-of selections over Qiskit built-in strategies that + set `layout` / `final_layout` correctly and use Qiskit's standard embed and + routing stage wrappers. +- Candidate results, including failures, are recorded in + `property_set["westquant_selection"]` and failures are logged; if every + candidate fails a `TranspilerError` is raised instead of silently returning + the input. +- `optimization_level=0` is no longer replaced by 2 inside the plugin. +- Brings the BackendV2 `Target` support in `noise_aware_metrics` that shipped in + the 0.1.1 wheel into the source repository. +- Tests: plugin discovery, unitary equivalence, measurement distribution, + coupling-map compliance, selected-candidate application and all-fail error. + +## 0.1.1 + +Published to PyPI on 2026-09-26 from a working tree that was not committed; +its only source difference from 0.1.0a4 was BackendV2 support in +`adapter.noise_aware_metrics`. diff --git a/README.md b/README.md index 09bf70c..a468708 100644 --- a/README.md +++ b/README.md @@ -75,11 +75,30 @@ the full Pareto front and all losing candidates remain in the dataset. ## Qiskit integration -The package also exposes alpha stage entry points for Qiskit's native -transpiler-plugin mechanism. The plugin stages still delegate to Qiskit -baselines in this Wave 1 build. The deterministic search engine is the first -real WestQuant search implementation and will be wired into stage plugins after -benchmarking and recursion guards are complete. +The package registers `westquant` as a Qiskit transpiler stage plugin for the +`layout`, `routing` and `optimization` stages: + +```python +from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager + +pm = generate_preset_pass_manager( + optimization_level=2, backend=backend, seed_transpiler=7, + layout_method="westquant", routing_method="westquant", optimization_method="westquant", +) +tqc = pm.run(qc) +print(pm.property_set["westquant_selection"]) +``` + +Each stage is a deterministic best-of selection among Qiskit's own built-in +strategies (layout: trivial/dense/sabre; routing: SabreSwap/BasicSwap/ +LookaheadSwap; optimization: levels 0-3), scored by two-qubit cost (SWAP = 3) +and then depth. Every candidate, including failures and their error messages, +is recorded in `property_set["westquant_selection"]`; if every candidate fails a +`TranspilerError` is raised. The routing stage only runs when the circuit is +not already mapped to the device. Tests check unitary equivalence, ideal output +distributions with measurements, coupling-map compliance and that the selected +candidate is the one returned. This is a best-of wrapper over Qiskit, not the +full sequential policy search described below. ## Sequential policy search diff --git a/pyproject.toml b/pyproject.toml index 1122c13..2f02e31 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "westquant-qiskit" -version = "0.1.0a4" +version = "0.1.2" description = "Deterministic representation search, RepGraph tracing, and WQT-ready data generation for Qiskit" readme = "README.md" requires-python = ">=3.10" diff --git a/src/westquant_qiskit/__init__.py b/src/westquant_qiskit/__init__.py index ab83e2a..07d5d4d 100644 --- a/src/westquant_qiskit/__init__.py +++ b/src/westquant_qiskit/__init__.py @@ -5,7 +5,7 @@ from .verification import QiskitEquivalenceVerifier, VerificationReport from .sequential import SequentialSearchEngine, SequentialSearchResult, SequentialSearchSpace -__version__ = "0.1.0a4" +__version__ = "0.1.2" __all__ = [ "QiskitAdapter", diff --git a/src/westquant_qiskit/adapter.py b/src/westquant_qiskit/adapter.py index 6ad1b3f..45f6b58 100644 --- a/src/westquant_qiskit/adapter.py +++ b/src/westquant_qiskit/adapter.py @@ -67,6 +67,87 @@ def numeric_metrics(metrics: dict[str, Any]) -> dict[str, float | int | None]: return {key: metrics.get(key) for key in keys} +def _build_v1_lookup(props: Any) -> dict[str, dict[tuple, dict[str, float | None]]]: + """Build a (gate_name, qubits) -> {error, duration} lookup from a + BackendV1 ``BackendProperties`` object. + + Returns an empty dict if ``props`` does not expose the V1 ``gate_error`` / + ``gate_length`` callable interface. + """ + lookup: dict[str, dict[tuple, dict[str, float | None]]] = {} + if not (callable(getattr(props, "gate_error", None)) + or callable(getattr(props, "gate_length", None))): + return lookup + + # BackendProperties exposes gate names + qubits via .gates list. + gates = getattr(props, "gates", None) + if not gates: + return lookup + + for gate in gates: + name = str(getattr(gate, "name", "")).lower() + qubits = tuple(getattr(gate, "qubits", ())) + if not name or not qubits: + continue + err = None + dur = None + gate_error_fn = getattr(props, "gate_error", None) + gate_length_fn = getattr(props, "gate_length", None) + if callable(gate_error_fn): + try: + err = float(gate_error_fn(name, list(qubits))) + except Exception: + err = None + if callable(gate_length_fn): + try: + dur = float(gate_length_fn(name, list(qubits))) + except Exception: + dur = None + if err is None and dur is None: + continue + lookup.setdefault(name, {})[qubits] = {"error": err, "duration": dur} + return lookup + + +def _build_v2_lookup(target: Any) -> dict[str, dict[tuple, dict[str, float | None]]]: + """Build a (gate_name, qubits) -> {error, duration} lookup from a + BackendV2 ``Target`` object. + + Iterates ``target.instructions`` (list of ``(Instruction, qargs)`` tuples) + and reads ``target.instruction_properties(index).error`` / ``.duration``. + Returns an empty dict if ``target`` does not expose the V2 interface. + """ + lookup: dict[str, dict[tuple, dict[str, float | None]]] = {} + instructions = getattr(target, "instructions", None) + instruction_properties = getattr(target, "instruction_properties", None) + if not instructions or not callable(instruction_properties): + return lookup + + for index, entry in enumerate(instructions): + try: + instr, qargs = entry + except (ValueError, TypeError): + continue + name = str(getattr(instr, "name", "")).lower() + if not name: + continue + qubits = tuple(int(q) for q in qargs) if qargs else () + try: + ip = instruction_properties(index) + except Exception: + continue + if ip is None: + continue + err = getattr(ip, "error", None) + dur = getattr(ip, "duration", None) + err = float(err) if isinstance(err, (int, float)) else None + dur = float(dur) if isinstance(dur, (int, float)) else None + if err is None and dur is None: + continue + lookup.setdefault(name, {})[qubits] = {"error": err, "duration": dur} + return lookup + + def noise_aware_metrics( circuit: Any, backend: Any = None, @@ -74,9 +155,14 @@ def noise_aware_metrics( ) -> dict[str, Any]: """Compute noise/fidelity-aware metrics for a circuit. + Supports both the legacy BackendV1 API (``backend.properties()`` returning + a ``BackendProperties`` object with ``gate_error``/``gate_length``) and + the current BackendV2 API (``backend.target`` returning a ``Target`` + object accessed via ``instruction_properties(index).error``/``.duration``). + Returns a dict with ``estimated_fidelity``, ``total_duration`` and - ``noise_aware_depth``. When ``backend``/``target`` is ``None`` or does not - expose per-gate error/duration data, the corresponding fields are ``None``. + ``noise_aware_depth``. When no per-gate error/duration data is available, + the corresponding fields are ``None``. """ result: dict[str, Any] = { "estimated_fidelity": None, @@ -93,8 +179,12 @@ def noise_aware_metrics( if not ops: return result - # Try to obtain a properties/props source from backend or target. - props = None + # ------------------------------------------------------------------ + # Build a unified lookup: gate_name -> {qubit_tuple -> {error, duration}} + # ------------------------------------------------------------------ + lookup: dict[str, dict[tuple, dict[str, float | None]]] = {} + + # 1. Try BackendV1: backend.properties() -> BackendProperties if backend is not None: props = getattr(backend, "properties", None) if callable(props): @@ -102,32 +192,27 @@ def noise_aware_metrics( props = props() except Exception: props = None - if props is None and target is not None: - props = getattr(target, "props", None) + else: + props = None + if props is not None: + lookup = _build_v1_lookup(props) - if props is None: - return result + # 2. Try BackendV2: backend.target -> Target (or explicit target arg) + if not lookup: + v2_target = None + if target is not None: + v2_target = target + elif backend is not None: + v2_target = getattr(backend, "target", None) + if v2_target is not None: + lookup = _build_v2_lookup(v2_target) - # Build per-gate error rate and duration lookup tables. - def _gate_error(props: Any, name: str, qubits: tuple) -> float | None: - getter = getattr(props, "gate_error", None) - if callable(getter): - try: - return float(getter(name, qubits)) - except Exception: - return None - return None - - def _gate_length(props: Any, name: str, qubits: tuple) -> float | None: - for attr in ("gate_length", "gate_duration"): - getter = getattr(props, attr, None) - if callable(getter): - try: - return float(getter(name, qubits)) - except Exception: - continue - return None + if not lookup: + return result + # ------------------------------------------------------------------ + # Compute fidelity, duration, and noise-aware depth from the lookup. + # ------------------------------------------------------------------ fidelity = 1.0 total_duration = 0.0 weighted_depth = 0.0 @@ -141,15 +226,26 @@ def _gate_length(props: Any, name: str, qubits: tuple) -> float | None: int(getattr(circuit.find_bit(q), "index", 0)) for q in getattr(instruction, "qubits", ()) ) - err = _gate_error(props, name, qubits) - length = _gate_length(props, name, qubits) + + # Look up this gate: try exact qubit match first, then any-qubit fallback. + gate_data = None + name_lookup = lookup.get(name) + if name_lookup: + gate_data = name_lookup.get(qubits) or next(iter(name_lookup.values()), None) + + if gate_data is None: + continue + + err = gate_data.get("error") + length = gate_data.get("duration") + if err is not None: fidelity *= (1.0 - err) have_fidelity = True if length is not None: total_duration += length have_duration = True - # Weight this gate's contribution to depth by its error rate. + # Weight this gate's contribution to depth by its fidelity. weighted_depth += (1.0 - (err if err is not None else 0.0)) * length if have_fidelity: diff --git a/src/westquant_qiskit/stage.py b/src/westquant_qiskit/stage.py index 9e45f3d..386e98b 100644 --- a/src/westquant_qiskit/stage.py +++ b/src/westquant_qiskit/stage.py @@ -1,9 +1,10 @@ """WestQuant Qiskit transpiler stage plugins. -These plugins execute real WestQuant representation search within the -Qiskit transpiler pipeline. When a user selects 'westquant' as a stage -name, the plugin evaluates multiple Qiskit compilation configurations -and selects the best one using WestQuant's Pareto-front evaluation. +These plugins execute a deterministic best-of search within the Qiskit +transpiler pipeline. When a user selects 'westquant' as a stage name, the +plugin evaluates several built-in Qiskit strategies for that stage and keeps +the candidate with the lowest two-qubit cost (SWAP counted as three 2Q gates), +breaking ties by depth. Usage: from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager @@ -12,27 +13,46 @@ backend=backend, layout_method="westquant", routing_method="westquant", + optimization_method="westquant", + seed_transpiler=7, ) tqc = pm.run(qc) - -The WestQuant search evaluates alternative layout/routing/optimization -strategies, measures resource metrics (2Q gates, depth, swaps), and -selects the Pareto-optimal candidate. Failed candidates are retained -for training data. - -Recursion guard: the baseline is always 'default', never 'westquant', -so there is no risk of infinite recursion. + pm.property_set["westquant_selection"] # per-stage candidate records + +Stage semantics: + layout Scores trivial/dense/sabre initial layouts by their fully + transpiled cost, sets ``property_set["layout"]`` to the best + one and then applies Qiskit's standard embedding. + routing Runs SabreSwap, BasicSwap and LookaheadSwap on the physical + DAG and keeps the cheapest routed DAG and its final layout, + wrapped in Qiskit's standard routing stage. + optimization Runs Qiskit's default optimization stage for levels 0-3 and + keeps the cheapest result. + +Every candidate result is recorded. Failed candidates are logged as warnings +and recorded with their error; if every candidate fails, a TranspilerError is +raised instead of silently returning the input. Candidates are always built +from Qiskit's built-in methods, never 'westquant', so there is no recursion. +Results are deterministic for a fixed ``seed_transpiler`` (default seed 0). """ from __future__ import annotations +import copy +import logging + +from qiskit.converters import circuit_to_dag, dag_to_circuit +from qiskit.dagcircuit import DAGCircuit +from qiskit.passmanager import ConditionalController +from qiskit.transpiler import Layout, PassManager, TranspilerError +from qiskit.transpiler.basepasses import TransformationPass +from qiskit.transpiler.passes import BasicSwap, LookaheadSwap, SabreSwap, SetLayout +from qiskit.transpiler.preset_passmanagers import common from qiskit.transpiler.preset_passmanagers.plugin import ( PassManagerStagePlugin, PassManagerStagePluginManager, ) -from qiskit.transpiler import PassManager -from qiskit.transpiler.basepasses import TransformationPass -from qiskit.dagcircuit import DAGCircuit +logger = logging.getLogger(__name__) # Layout methods to try during WestQuant search LAYOUT_CANDIDATES = ["trivial", "dense", "sabre"] @@ -45,175 +65,153 @@ def _count_2q_gates(dag: DAGCircuit) -> int: - """Count two-qubit gates in a DAG.""" - count = 0 - for node in dag.op_nodes(): - if len(node.qargs) == 2: - count += 1 - return count + """Two-qubit cost of a DAG; a SWAP counts as three 2Q gates.""" + return sum(3 if node.name == "swap" else 1 for node in dag.op_nodes() if len(node.qargs) == 2) def _dag_depth(dag: DAGCircuit) -> int: - """Estimate circuit depth from DAG.""" - try: - return dag.depth() - except Exception: - return 0 + return dag.depth() -class WestQuantSearchPass(TransformationPass): - """Custom transpiler pass that runs WestQuant representation search. +def _coupling(config): + target = getattr(config, "target", None) + return getattr(config, "coupling_map", None) or (target.build_coupling_map() if target is not None else None) + - This pass evaluates multiple compilation configurations and selects - the Pareto-optimal result based on 2Q gate count and depth. - """ +class WestQuantSearchPass(TransformationPass): + """Evaluate the candidates of one transpiler stage and apply the best one.""" def __init__(self, stage: str = "optimization", config=None, optimization_level: int | None = None) -> None: super().__init__() + if stage not in ("layout", "routing", "optimization"): + raise TranspilerError(f"Unknown WestQuant stage: {stage!r}") self.stage = stage self.config = config - self.optimization_level = optimization_level or 2 + self.optimization_level = 2 if optimization_level is None else optimization_level + seed = getattr(config, "seed_transpiler", None) + self.seed = 0 if seed is None or seed < 0 else seed + + def _get_candidates(self) -> list: + return {"layout": LAYOUT_CANDIDATES, "routing": ROUTING_CANDIDATES, "optimization": OPT_LEVELS}[self.stage] def run(self, dag: DAGCircuit) -> DAGCircuit: - """Run WestQuant search on the DAG. - - For v0.1, this pass evaluates multiple Qiskit configurations - and selects the best one. The search is deterministic. - """ - # Get the current circuit from the DAG - from qiskit import QuantumCircuit - qc = QuantumCircuit.from_instructions(dag) - - # Get coupling map and basis gates from config - coupling_map = None - basis_gates = None - if self.config: - coupling_map = getattr(self.config, "coupling_map", None) - basis_gates = getattr(self.config, "basis_gates", None) - - # Baseline metrics (current DAG) - baseline_2q = _count_2q_gates(dag) - baseline_depth = _dag_depth(dag) - - best_dag = dag - best_2q = baseline_2q - best_depth = baseline_depth - - # Try alternative configurations based on stage - candidates = self._get_candidates() - - for candidate in candidates: + records, best = [], None + for candidate in self._get_candidates(): try: - candidate_dag = self._try_candidate(qc, candidate, coupling_map, basis_gates) - if candidate_dag is None: - continue - - candidate_2q = _count_2q_gates(candidate_dag) - candidate_depth = _dag_depth(candidate_dag) - - # Pareto comparison: prefer fewer 2Q gates, then less depth - if candidate_2q < best_2q or (candidate_2q == best_2q and candidate_depth < best_depth): - best_dag = candidate_dag - best_2q = candidate_2q - best_depth = candidate_depth - - except Exception: - # Failed candidates are silently skipped in the pass - # (they are retained in the full search engine, not here) + result = self._try_candidate(dag, candidate) + except Exception as exc: # recorded and logged, never silently dropped + logger.warning("WestQuant %s candidate %r failed: %s", self.stage, candidate, exc) + records.append({"candidate": candidate, "ok": False, "error": f"{type(exc).__name__}: {exc}"}) continue - - return best_dag - - def _get_candidates(self) -> list: - """Get candidate configurations for this stage.""" + cost_dag, applied = result + record = {"candidate": candidate, "ok": True, "two_qubit": _count_2q_gates(cost_dag), "depth": _dag_depth(cost_dag)} + records.append(record) + if best is None or (record["two_qubit"], record["depth"]) < (best[0]["two_qubit"], best[0]["depth"]): + best = (record, applied) + if best is None: + raise TranspilerError(f"All WestQuant {self.stage} candidates failed: {records}") + record, applied = best + selection = dict(self.property_set["westquant_selection"] or {}) + selection[self.stage] = {"selected": record["candidate"], "candidates": records} + self.property_set["westquant_selection"] = selection + return self._apply(dag, applied) + + def _try_candidate(self, dag: DAGCircuit, candidate): + """Return (DAG used for scoring, payload needed to apply the candidate).""" if self.stage == "layout": - return LAYOUT_CANDIDATES - elif self.stage == "routing": - return ROUTING_CANDIDATES - elif self.stage == "optimization": - return OPT_LEVELS - return [] - - def _try_candidate(self, qc, candidate, coupling_map, basis_gates) -> DAGCircuit | None: - """Try a candidate configuration and return the resulting DAG.""" - from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager + return self._layout_candidate(dag, candidate) + if self.stage == "routing": + return self._routing_candidate(dag, candidate) + return self._optimization_candidate(dag, candidate) - kwargs = {"optimization_level": self.optimization_level} - if coupling_map: - kwargs["coupling_map"] = coupling_map - if basis_gates: - kwargs["basis_gates"] = basis_gates + def _layout_candidate(self, dag, method): + from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager + cfg = self.config + routing = cfg.routing_method if cfg.routing_method not in (None, "westquant") else "sabre" + pm = generate_preset_pass_manager( + optimization_level=self.optimization_level, + target=cfg.target, + coupling_map=None if cfg.target is not None else cfg.coupling_map, + basis_gates=None if cfg.target is not None else cfg.basis_gates, + layout_method=method, + routing_method=routing, + seed_transpiler=self.seed, + ) + tqc = pm.run(dag_to_circuit(dag)) + physical = tqc.layout.initial_index_layout(filter_ancillas=True) + return circuit_to_dag(tqc), Layout({dag.qubits[i]: p for i, p in enumerate(physical)}) + + def _routing_candidate(self, dag, method): + coupling = _coupling(self.config) + if coupling is None: + raise TranspilerError("Routing requires a coupling map or target.") + routing_pass = { + "sabre": lambda: SabreSwap(coupling, heuristic="decay", seed=self.seed), + "basic": lambda: BasicSwap(coupling), + "lookahead": lambda: LookaheadSwap(coupling), + }[method]() + routed = routing_pass.run(copy.deepcopy(dag)) + return routed, (routed, routing_pass.property_set["final_layout"]) + + def _optimization_candidate(self, dag, level): + pm = PassManagerStagePluginManager().get_passmanager_stage("optimization", "default", self.config, optimization_level=level) + out = circuit_to_dag(pm.run(dag_to_circuit(dag))) if pm is not None else copy.deepcopy(dag) + return out, out + + def _apply(self, dag, applied): if self.stage == "layout": - kwargs["layout_method"] = candidate - elif self.stage == "routing": - kwargs["routing_method"] = candidate - elif self.stage == "optimization": - kwargs["optimization_level"] = candidate - - try: - pm = generate_preset_pass_manager(**kwargs) - tqc = pm.run(qc) - return tqc._layout.dag if hasattr(tqc, "_layout") and hasattr(tqc._layout, "dag") else None - except Exception: - return None + self.property_set["layout"] = applied + return dag + if self.stage == "routing": + routed, final_layout = applied + if final_layout is not None: + current = self.property_set["final_layout"] + self.property_set["final_layout"] = final_layout if current is None else current.compose(final_layout, dag.qubits) + return routed + return applied class WestQuantLayoutPlugin(PassManagerStagePlugin): - """WestQuant layout plugin — evaluates multiple layout methods. - - Instead of always using Qiskit's default (sabre), this plugin - evaluates trivial, dense, and sabre layouts and selects the - Pareto-optimal one based on 2Q gate count and depth. - """ - - stage = "layout" + """WestQuant layout plugin: best of trivial, dense and sabre layouts.""" def pass_manager(self, pass_manager_config, optimization_level=None): - # Build a PassManager that includes the WestQuant search pass - pm = PassManager() - pm.append(WestQuantSearchPass( - stage="layout", - config=pass_manager_config, - optimization_level=optimization_level, + pm = PassManager([SetLayout(pass_manager_config.initial_layout)]) + coupling = pass_manager_config.target or pass_manager_config.coupling_map + if coupling is None: + return pm + pm.append(ConditionalController( + WestQuantSearchPass(stage="layout", config=pass_manager_config, optimization_level=optimization_level), + condition=lambda property_set: not property_set["layout"], )) + pm.append(common.generate_embed_passmanager(coupling).to_flow_controller()) return pm class WestQuantRoutingPlugin(PassManagerStagePlugin): - """WestQuant routing plugin — evaluates multiple routing methods. - - Instead of always using Qiskit's default (sabre), this plugin - evaluates sabre, basic, and lookahead routing and selects the - Pareto-optimal one. - """ - - stage = "routing" + """WestQuant routing plugin: best of SabreSwap, BasicSwap and LookaheadSwap.""" def pass_manager(self, pass_manager_config, optimization_level=None): - pm = PassManager() - pm.append(WestQuantSearchPass( - stage="routing", - config=pass_manager_config, - optimization_level=optimization_level, - )) - return pm + level = 2 if optimization_level is None else optimization_level + vf2_call_limit, vf2_max_trials = common.get_vf2_limits( + level, pass_manager_config.layout_method, pass_manager_config.initial_layout + ) + return common.generate_routing_passmanager( + WestQuantSearchPass(stage="routing", config=pass_manager_config, optimization_level=level), + pass_manager_config.target, + coupling_map=pass_manager_config.coupling_map, + vf2_call_limit=None if level == 0 else vf2_call_limit, + vf2_max_trials=None if level == 0 else vf2_max_trials, + seed_transpiler=-1, + check_trivial=level == 1, + use_barrier_before_measurement=True, + ) class WestQuantOptimizationPlugin(PassManagerStagePlugin): - """WestQuant optimization plugin — evaluates multiple optimization levels. - - Instead of always using the specified optimization level, this plugin - evaluates levels 0-3 and selects the Pareto-optimal one. - """ - - stage = "optimization" + """WestQuant optimization plugin: best of Qiskit optimization levels 0-3.""" def pass_manager(self, pass_manager_config, optimization_level=None): - pm = PassManager() - pm.append(WestQuantSearchPass( - stage="optimization", - config=pass_manager_config, - optimization_level=optimization_level, - )) - return pm + return PassManager([ + WestQuantSearchPass(stage="optimization", config=pass_manager_config, optimization_level=optimization_level) + ]) diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 6d37b8c..e8b52d2 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -19,3 +19,18 @@ def test_metrics(): m = circuit_metrics(FakeCircuit()) assert m["two_qubit_gates"] == 2 assert m["depth"] == 3 + + +def test_noise_aware_metrics_backend_v2_target(): + from qiskit import QuantumCircuit, transpile + from qiskit.providers.fake_provider import GenericBackendV2 + from westquant_qiskit.adapter import noise_aware_metrics + + backend = GenericBackendV2(num_qubits=3, seed=1) + qc = QuantumCircuit(2) + qc.h(0) + qc.cx(0, 1) + tqc = transpile(qc, backend=backend, seed_transpiler=1) + metrics = noise_aware_metrics(tqc, backend=backend) + assert 0 < metrics["estimated_fidelity"] < 1 + assert metrics["total_duration"] > 0 diff --git a/tests/test_stage.py b/tests/test_stage.py new file mode 100644 index 0000000..fd4619d --- /dev/null +++ b/tests/test_stage.py @@ -0,0 +1,123 @@ +import numpy as np +import pytest +from qiskit import QuantumCircuit +from qiskit.converters import circuit_to_dag +from qiskit.quantum_info import Operator, Statevector +from qiskit.transpiler import CouplingMap, PassManagerConfig, TranspilerError +from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager +from qiskit.transpiler.preset_passmanagers.plugin import list_stage_plugins + +from westquant_qiskit.stage import WestQuantSearchPass, _count_2q_gates, _dag_depth + +COUPLING = CouplingMap.from_line(5) +BASIS = ["cx", "rz", "sx", "x"] +STAGES = {"layout": "layout_method", "routing": "routing_method", "optimization": "optimization_method"} + + +def circuit(measure=False): + qc = QuantumCircuit(4) + qc.h(0) + qc.cx(0, 3) + qc.ry(0.4, 1) + qc.cx(1, 3) + qc.cx(0, 2) + qc.cx(1, 2) # interaction 4-cycle: cannot be embedded in a line, routing is required + qc.rz(0.3, 2) + qc.cx(2, 0) + qc.cx(3, 1) + if measure: + qc.measure_all() + return qc + + +def transpile(qc, level=2, **methods): + pm = generate_preset_pass_manager(optimization_level=level, coupling_map=COUPLING, basis_gates=BASIS, seed_transpiler=7, **methods) + return pm, pm.run(qc) + + +def assert_on_coupling(tqc): + edges = set(COUPLING.get_edges()) + for item in tqc.data: + if len(item.qubits) == 2: + pair = tuple(tqc.find_bit(q).index for q in item.qubits) + assert pair in edges or pair[::-1] in edges + + +def assert_same_distribution(qc, tqc): + ideal = Statevector(qc.remove_final_measurements(inplace=False)).probabilities_dict() + physical = tqc.layout.final_index_layout() + routed = Statevector(tqc.remove_final_measurements(inplace=False)).probabilities_dict(qargs=physical) + for key in set(ideal) | set(routed): + assert abs(ideal.get(key, 0) - routed.get(key, 0)) < 1e-8 + + +def assert_selected_is_best(selection): + ok = [c for c in selection["candidates"] if c["ok"]] + best = min(ok, key=lambda c: (c["two_qubit"], c["depth"])) + assert selection["selected"] == best["candidate"] + + +@pytest.mark.parametrize("stage", list(STAGES)) +def test_plugins_are_discoverable(stage): + assert "westquant" in list_stage_plugins(stage) + + +@pytest.mark.parametrize("stage", list(STAGES)) +def test_stage_runs_and_preserves_unitary(stage): + qc = circuit() + pm, tqc = transpile(qc, **{STAGES[stage]: "westquant"}) + selection = pm.property_set["westquant_selection"][stage] + assert [c["candidate"] for c in selection["candidates"]] == WestQuantSearchPass(stage=stage)._get_candidates() + assert all(c["ok"] or c["error"] for c in selection["candidates"]), selection + assert_selected_is_best(selection) + assert_on_coupling(tqc) + ancillas = tqc.num_qubits - qc.num_qubits + assert Operator.from_circuit(tqc).equiv(Operator(qc).expand(Operator(np.eye(2**ancillas)))) + + +def test_all_stages_with_measurements_preserve_distribution(): + qc = circuit(measure=True) + pm, tqc = transpile(qc, layout_method="westquant", routing_method="westquant", optimization_method="westquant") + assert set(pm.property_set["westquant_selection"]) == set(STAGES) + assert tqc.count_ops()["measure"] == 4 + assert_on_coupling(tqc) + assert_same_distribution(qc, tqc) + + +def test_routing_pass_returns_selected_candidate_dag(): + physical = QuantumCircuit(5) # already on physical qubits, but not routed + physical.cx(0, 4) + physical.cx(1, 3) + physical.cx(0, 2) + physical.cx(4, 1) + dag = circuit_to_dag(physical) + search = WestQuantSearchPass(stage="routing", config=PassManagerConfig(coupling_map=COUPLING, seed_transpiler=7)) + out = search(physical) + assert_on_coupling(out) + record = search.property_set["westquant_selection"]["routing"] + chosen = next(c for c in record["candidates"] if c["candidate"] == record["selected"]) + out_dag = circuit_to_dag(out) + assert (_count_2q_gates(out_dag), _dag_depth(out_dag)) == (chosen["two_qubit"], chosen["depth"]) + assert search.property_set["final_layout"] is not None + assert _count_2q_gates(out_dag) > _count_2q_gates(dag) # SWAPs were inserted: the pass acted + + +@pytest.mark.parametrize("level", [1, 2, 3]) +def test_optimization_never_worse_than_default(level): + qc = circuit() + _, default = transpile(qc, level=level) + _, westquant = transpile(qc, level=level, optimization_method="westquant") + def cost(c): + dag = circuit_to_dag(c) + return _count_2q_gates(dag), _dag_depth(dag) + assert cost(westquant) <= cost(default) + + +def test_all_candidates_failing_raises(): + search = WestQuantSearchPass(stage="routing", config=PassManagerConfig(coupling_map=None)) + with pytest.raises(TranspilerError, match="All WestQuant routing candidates failed"): + search(circuit()) + + +def test_optimization_level_zero_is_respected(): + assert WestQuantSearchPass(stage="layout", optimization_level=0).optimization_level == 0