From d30e7eb8fc45f9cb7e1446990ab709a2023ac37a Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Tue, 2 Jun 2026 08:09:34 -0700 Subject: [PATCH 01/15] feat: add incremental add/remove helpers for qubits, channels, and ports Add standalone helper functions to modify an existing QUAM machine without re-running build_quam: - modify_quam.py: add_qubit, remove_qubit, add_channel, remove_channel - modify_ports.py: add_port, remove_port - Comprehensive tests for all helpers including equivalence checks --- .../builder/qop_connectivity/__init__.py | 1 + .../builder/qop_connectivity/modify_ports.py | 210 +++++++ .../builder/superconducting/__init__.py | 1 + .../builder/superconducting/modify_quam.py | 248 ++++++++ tests/test_modify_quam.py | 591 ++++++++++++++++++ 5 files changed, 1051 insertions(+) create mode 100644 quam_builder/builder/qop_connectivity/modify_ports.py create mode 100644 quam_builder/builder/superconducting/modify_quam.py create mode 100644 tests/test_modify_quam.py diff --git a/quam_builder/builder/qop_connectivity/__init__.py b/quam_builder/builder/qop_connectivity/__init__.py index 885f110c..a907f15d 100644 --- a/quam_builder/builder/qop_connectivity/__init__.py +++ b/quam_builder/builder/qop_connectivity/__init__.py @@ -1 +1,2 @@ from .build_quam_wiring import build_quam_wiring +from .modify_ports import add_port, remove_port diff --git a/quam_builder/builder/qop_connectivity/modify_ports.py b/quam_builder/builder/qop_connectivity/modify_ports.py new file mode 100644 index 00000000..44ed1155 --- /dev/null +++ b/quam_builder/builder/qop_connectivity/modify_ports.py @@ -0,0 +1,210 @@ +"""Add and remove ports from any QUAM machine with a ports container. + +Wraps ``FEMPortsContainer`` and ``OPXPlusPortsContainer`` methods into a +uniform interface that works across all modalities (superconducting, +quantum dots, NV centers). + +Example:: + + from quam_builder.builder.qop_connectivity.modify_ports import add_port, remove_port + + port = add_port(machine, "mw_output", "con1", fem_id=1, port_id=3, band=1) + remove_port(machine, "mw_output", "con1", fem_id=1, port_id=3) +""" + +from typing import Optional, Protocol, Union + +from quam.components.ports import FEMPortsContainer, OPXPlusPortsContainer +from quam.components.ports.base_ports import BasePort + +__all__ = ["add_port", "remove_port"] + +_PortsContainer = Union[FEMPortsContainer, OPXPlusPortsContainer] + +_FEM_PORT_TYPES = ("mw_output", "mw_input", "analog_output", "analog_input", "digital_output") +_OPXPLUS_PORT_TYPES = ("analog_output", "analog_input", "digital_output", "digital_input") + + +class _HasPorts(Protocol): + ports: Union[FEMPortsContainer, OPXPlusPortsContainer] + + +def _get_ports_container(machine: _HasPorts) -> _PortsContainer: + ports = getattr(machine, "ports", None) + if ports is None: + raise TypeError(f"{type(machine).__name__} does not have a 'ports' attribute") + if not isinstance(ports, (FEMPortsContainer, OPXPlusPortsContainer)): + raise TypeError( + f"Expected FEMPortsContainer or OPXPlusPortsContainer, " f"got {type(ports).__name__}" + ) + return ports + + +def _add_fem_port( + container: FEMPortsContainer, + port_type: str, + controller_id: Union[str, int], + fem_id: int, + port_id: int, + **kwargs, +) -> BasePort: + if port_type == "mw_output": + return container.get_mw_output(controller_id, fem_id, port_id, create=True, **kwargs) + elif port_type == "mw_input": + return container.get_mw_input(controller_id, fem_id, port_id, create=True, **kwargs) + elif port_type == "analog_output": + return container.get_analog_output(controller_id, fem_id, port_id, create=True, **kwargs) + elif port_type == "analog_input": + return container.get_analog_input(controller_id, fem_id, port_id, create=True, **kwargs) + elif port_type == "digital_output": + return container.get_digital_output(controller_id, fem_id, port_id, create=True, **kwargs) + else: + raise ValueError( + f"Unsupported FEM port type '{port_type}'. Expected one of: {_FEM_PORT_TYPES}" + ) + + +def _add_opxplus_port( + container: OPXPlusPortsContainer, + port_type: str, + controller_id: Union[str, int], + port_id: int, + **kwargs, +) -> BasePort: + if port_type == "analog_output": + return container.get_analog_output(controller_id, port_id, create=True, **kwargs) + elif port_type == "analog_input": + return container.get_analog_input(controller_id, port_id, create=True, **kwargs) + elif port_type == "digital_output": + return container.get_digital_output(controller_id, port_id, create=True, **kwargs) + elif port_type == "digital_input": + return container.get_digital_input(controller_id, port_id, create=True, **kwargs) + else: + raise ValueError( + f"Unsupported OPX+ port type '{port_type}'. Expected one of: {_OPXPLUS_PORT_TYPES}" + ) + + +def _remove_fem_port( + container: FEMPortsContainer, + port_type: str, + controller_id: Union[str, int], + fem_id: int, + port_id: int, +) -> BasePort: + if port_type == "mw_output": + return container.mw_outputs[controller_id][fem_id].pop(port_id) + elif port_type == "mw_input": + return container.mw_inputs[controller_id][fem_id].pop(port_id) + elif port_type == "analog_output": + return container.analog_outputs[controller_id][fem_id].pop(port_id) + elif port_type == "analog_input": + return container.analog_inputs[controller_id][fem_id].pop(port_id) + elif port_type == "digital_output": + return container.digital_outputs[controller_id][fem_id].pop(port_id) + else: + raise ValueError( + f"Unsupported FEM port type '{port_type}'. Expected one of: {_FEM_PORT_TYPES}" + ) + + +def _remove_opxplus_port( + container: OPXPlusPortsContainer, + port_type: str, + controller_id: Union[str, int], + port_id: int, +) -> BasePort: + if port_type == "analog_output": + return container.analog_outputs[controller_id].pop(port_id) + elif port_type == "analog_input": + return container.analog_inputs[controller_id].pop(port_id) + elif port_type == "digital_output": + return container.digital_outputs[controller_id].pop(port_id) + elif port_type == "digital_input": + return container.digital_inputs[controller_id].pop(port_id) + else: + raise ValueError( + f"Unsupported OPX+ port type '{port_type}'. Expected one of: {_OPXPLUS_PORT_TYPES}" + ) + + +def add_port( + machine: _HasPorts, + port_type: str, + controller_id: Union[str, int], + fem_id: Optional[int] = None, + port_id: Optional[int] = None, + **kwargs, +) -> BasePort: + """Create a port and register it in the machine's ports container. + + For FEM-based machines, ``fem_id`` and ``port_id`` are both required. + For OPX+-based machines, only ``port_id`` is required. + + Args: + machine: A QUAM machine with a ``ports`` attribute (``FEMPortsContainer`` + or ``OPXPlusPortsContainer``). + port_type: Port kind e.g. ``"mw_output"``, ``"analog_input"``. + controller_id: Controller identifier (e.g. ``"con1"``). + fem_id: FEM slot number. Required for FEM containers. + port_id: Port number on the FEM or controller. + **kwargs: Forwarded to the port constructor (e.g. ``band=1``). + + Returns: + The newly created (or already existing) port. + + Raises: + TypeError: If ``machine`` has no supported ports container. + ValueError: If required identifiers are missing or ``port_type`` + is not recognised by the container. + """ + container = _get_ports_container(machine) + + if isinstance(container, FEMPortsContainer): + if fem_id is None or port_id is None: + raise ValueError("FEM ports require both fem_id and port_id") + return _add_fem_port(container, port_type, controller_id, fem_id, port_id, **kwargs) + else: + if port_id is None: + raise ValueError("OPX+ ports require port_id") + return _add_opxplus_port(container, port_type, controller_id, port_id, **kwargs) + + +def remove_port( + machine: _HasPorts, + port_type: str, + controller_id: Union[str, int], + fem_id: Optional[int] = None, + port_id: Optional[int] = None, +) -> BasePort: + """Remove a port from the machine's ports container. + + Args: + machine: A QUAM machine with a ``ports`` attribute. + port_type: Port kind e.g. ``"mw_output"``, ``"analog_input"``. + controller_id: Controller identifier. + fem_id: FEM slot number. Required for FEM containers. + port_id: Port number. + + Returns: + The removed port. + + Raises: + TypeError: If ``machine`` has no supported ports container. + KeyError: If the specified port does not exist. + ValueError: If required identifiers are missing or ``port_type`` + is not recognised by the container. + """ + container = _get_ports_container(machine) + + if isinstance(container, FEMPortsContainer): + if fem_id is None or port_id is None: + raise ValueError("FEM ports require both fem_id and port_id") + removed = _remove_fem_port(container, port_type, controller_id, fem_id, port_id) + else: + if port_id is None: + raise ValueError("OPX+ ports require port_id") + removed = _remove_opxplus_port(container, port_type, controller_id, port_id) + + removed.parent = None + return removed diff --git a/quam_builder/builder/superconducting/__init__.py b/quam_builder/builder/superconducting/__init__.py index 3dfcbbd1..eaf6c363 100644 --- a/quam_builder/builder/superconducting/__init__.py +++ b/quam_builder/builder/superconducting/__init__.py @@ -1 +1,2 @@ from .build_quam import build_quam +from .modify_quam import add_qubit, remove_qubit, add_channel, remove_channel diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py new file mode 100644 index 00000000..5fcb2733 --- /dev/null +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -0,0 +1,248 @@ +"""Add and remove qubits and channels from an existing superconducting QUAM. + +Reuses the same atomic component adders that ``build_quam()`` uses, so +incrementally-added objects are identical to batch-built ones. + +Line type keys match ``WiringLineType`` enum values: ``"xy"`` (drive), +``"rr"`` (resonator), ``"z"`` (flux). + +Example:: + + from quam_builder.builder.superconducting.modify_quam import ( + add_qubit, remove_qubit, add_channel, remove_channel, + ) + + add_qubit( + machine, + qubit_id="q5", + wiring={ + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/5"}, + "rr": {"opx_output": "#/ports/mw_outputs/con1/2/5", + "opx_input": "#/ports/mw_inputs/con1/2/1"}, + }, + ) + + add_channel(machine, "q5", "z", {"opx_output": "#/ports/analog_outputs/con1/3/1"}) + + remove_qubit(machine, "q5") +""" + +from typing import Dict, Optional + +from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType + +from quam_builder.architecture.superconducting.qpu import AnyQuam +from quam_builder.architecture.superconducting.qubit import AnyTransmon +from quam_builder.builder.superconducting.add_default_pulses import ( + add_default_transmon_pulses, +) +from quam_builder.builder.superconducting.add_transmon_drive_component import ( + add_transmon_drive_component, +) +from quam_builder.builder.superconducting.add_transmon_flux_component import ( + add_transmon_flux_component, +) +from quam_builder.builder.superconducting.add_transmon_resonator_component import ( + add_transmon_resonator_component, +) + +__all__ = ["add_qubit", "remove_qubit", "add_channel", "remove_channel"] + +_LINE_TYPE_TO_ADDER = { + WiringLineType.DRIVE.value: add_transmon_drive_component, + WiringLineType.RESONATOR.value: add_transmon_resonator_component, + WiringLineType.FLUX.value: add_transmon_flux_component, +} + +_LINE_TYPE_TO_FIELD = { + WiringLineType.DRIVE.value: "xy", + WiringLineType.RESONATOR.value: "resonator", + WiringLineType.FLUX.value: "z", +} + + +def _create_ports(machine: AnyQuam, ports: Dict[str, str]): + """Ensure every port reference in *ports* exists in ``machine.ports``.""" + for ref in ports.values(): + if isinstance(ref, str) and "ports" in ref and machine.ports is not None: + machine.ports.reference_to_port(ref, create=True) + + +def add_qubit( + machine: AnyQuam, + qubit_id: str, + wiring: Optional[Dict[str, Dict[str, str]]] = None, + add_default_pulses: bool = True, +) -> AnyTransmon: + """Add a single qubit to an existing machine. + + Args: + machine: The QUAM machine instance. + qubit_id: Name for the qubit (e.g. ``"q5"``). + wiring: Dict mapping line types (``"xy"``, ``"rr"``, ``"z"``) to port + dicts, e.g.:: + + { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/5"}, + "rr": {"opx_output": "#/ports/...","opx_input": "#/ports/..."}, + "z": {"opx_output": "#/ports/analog_outputs/con1/3/1"}, + } + + If ``None``, the wiring must already exist in + ``machine.wiring["qubits"][qubit_id]``. + add_default_pulses: Seed default pulse operations on the new channels. + + Returns: + The newly created qubit, fully wired with channels and (optionally) pulses. + + Raises: + KeyError: If a qubit with ``qubit_id`` already exists. + """ + if qubit_id in machine.qubits: + raise KeyError(f"Qubit '{qubit_id}' already exists") + + if wiring is not None: + machine.wiring.setdefault("qubits", {}) + machine.wiring["qubits"][qubit_id] = wiring + + qubit_wiring = machine.wiring.get("qubits", {}).get(qubit_id, {}) + + try: + transmon = machine.qubit_type(id=qubit_id) + machine.qubits[qubit_id] = transmon + except AttributeError as e: + raise TypeError( + f"{type(machine).__name__} does not define qubit_type. " + "Use FixedFrequencyQuam or FluxTunableQuam." + ) from e + + for line_type, ports in qubit_wiring.items(): + _create_ports(machine, ports) + wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" + adder = _LINE_TYPE_TO_ADDER.get(line_type) + if adder is None: + raise ValueError(f"Unknown line type: {line_type}") + adder(transmon, wiring_path, ports) + + if add_default_pulses: + add_default_transmon_pulses(transmon) + + if qubit_id not in machine.active_qubit_names: + machine.active_qubit_names.append(transmon.name) + + return transmon + + +def remove_qubit(machine: AnyQuam, qubit_id: str) -> AnyTransmon: + """Remove a qubit and its channels from the machine. + + The qubit is removed from ``machine.qubits`` and ``active_qubit_names``, + and its wiring entry is cleaned up. The returned qubit has its parent + cleared so it can be garbage-collected or re-attached elsewhere. + + Args: + machine: The QUAM machine instance. + qubit_id: The id of the qubit to remove. + + Returns: + The detached qubit object. + + Raises: + KeyError: If no qubit with ``qubit_id`` exists. + """ + if qubit_id not in machine.qubits: + raise KeyError(f"Qubit '{qubit_id}' not found") + + transmon = machine.qubits.pop(qubit_id) + transmon.parent = None + + if transmon.name in machine.active_qubit_names: + machine.active_qubit_names.remove(transmon.name) + + if "qubits" in machine.wiring and qubit_id in machine.wiring["qubits"]: + del machine.wiring["qubits"][qubit_id] + + return transmon + + +def add_channel( + machine: AnyQuam, + qubit_id: str, + line_type: str, + ports: Dict[str, str], + add_default_pulses: bool = True, +) -> None: + """Add a single channel to an existing qubit. + + Args: + machine: The QUAM machine instance. + qubit_id: The id of the target qubit. + line_type: One of ``"xy"`` (drive), ``"rr"`` (resonator), or ``"z"`` (flux). + ports: Dict with port references, e.g. + ``{"opx_output": "#/ports/mw_outputs/con1/1/5"}``. + add_default_pulses: Seed default pulse operations on the new channel. + + Raises: + KeyError: If the qubit doesn't exist. + ValueError: If the channel slot is already occupied. + """ + if qubit_id not in machine.qubits: + raise KeyError(f"Qubit '{qubit_id}' not found") + + transmon = machine.qubits[qubit_id] + field_name = _LINE_TYPE_TO_FIELD.get(line_type) + if field_name is None: + raise ValueError(f"Unknown line type: {line_type}") + + if getattr(transmon, field_name, None) is not None: + raise ValueError( + f"Channel '{line_type}' (field '{field_name}') already exists on qubit '{qubit_id}'" + ) + + _create_ports(machine, ports) + + machine.wiring.setdefault("qubits", {}) + machine.wiring["qubits"].setdefault(qubit_id, {}) + machine.wiring["qubits"][qubit_id][line_type] = ports + + wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" + adder = _LINE_TYPE_TO_ADDER[line_type] + adder(transmon, wiring_path, ports) + + if add_default_pulses: + add_default_transmon_pulses(transmon) + + +def remove_channel(machine: AnyQuam, qubit_id: str, line_type: str) -> None: + """Remove a single channel from an existing qubit. + + Args: + machine: The QUAM machine instance. + qubit_id: The id of the target qubit. + line_type: One of ``"xy"`` (drive), ``"rr"`` (resonator), or ``"z"`` (flux). + + Raises: + KeyError: If the qubit doesn't exist. + ValueError: If the channel slot is already empty. + """ + if qubit_id not in machine.qubits: + raise KeyError(f"Qubit '{qubit_id}' not found") + + transmon = machine.qubits[qubit_id] + field_name = _LINE_TYPE_TO_FIELD.get(line_type) + if field_name is None: + raise ValueError(f"Unknown line type: {line_type}") + + channel = getattr(transmon, field_name, None) + if channel is None: + raise ValueError( + f"Channel '{line_type}' (field '{field_name}') does not exist on qubit '{qubit_id}'" + ) + + channel.parent = None + setattr(transmon, field_name, None) + + if "qubits" in machine.wiring: + qubit_wiring = machine.wiring["qubits"].get(qubit_id, {}) + if line_type in qubit_wiring: + del qubit_wiring[line_type] diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py new file mode 100644 index 00000000..36817d49 --- /dev/null +++ b/tests/test_modify_quam.py @@ -0,0 +1,591 @@ +"""Tests for the machine add/remove helpers. + +Validates that added qubits, channels, and ports produce the +same results as the batch ``build_quam()`` flow: + - Config generation + - Serialization + - iterate_components visibility + - Parent chain integrity + - Active qubit tracking +""" + +import pytest + +from quam.components.ports import ( + FEMPortsContainer, + MWFEMAnalogOutputPort, + MWFEMAnalogInputPort, + LFFEMAnalogOutputPort, +) + +from quam_builder.architecture.superconducting.qpu import ( + FixedFrequencyQuam, + FluxTunableQuam, +) +from quam_builder.architecture.superconducting.qubit import ( + FixedFrequencyTransmon, +) +from quam_builder.architecture.superconducting.components.xy_drive import XYDriveMW +from quam_builder.architecture.superconducting.components.readout_resonator import ( + ReadoutResonatorMW, +) +from quam_builder.architecture.superconducting.components.flux_line import FluxLine + +from quam_builder.builder.superconducting.modify_quam import ( + add_qubit, + remove_qubit, + add_channel, + remove_channel, +) +from quam_builder.builder.qop_connectivity.modify_ports import ( + add_port, + remove_port, +) + + +############################################################################## +############################################################################## +# Fixtures +############################################################################## +############################################################################## + + +@pytest.fixture +def empty_ff_machine() -> FixedFrequencyQuam: + machine = FixedFrequencyQuam( + ports=FEMPortsContainer(), + ) + return machine + + +@pytest.fixture +def mw_wiring() -> dict[str, str]: + """MW-FEM wiring dict for a single qubit with drive + resonator.""" + return { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "rr": { + "opx_output": "#/ports/mw_outputs/con1/2/1", + "opx_input": "#/ports/mw_inputs/con1/2/1", + }, + } + + +@pytest.fixture +def flux_wiring() -> dict[str, str]: + """LF-FEM flux wiring for a single channel.""" + return {"opx_output": "#/ports/analog_outputs/con1/3/1"} + + +############################################################################## +############################################################################## +# Helpers +############################################################################## +############################################################################## + + +def _build_machine() -> FixedFrequencyQuam: + """Build a machine using the standard build_quam sub-functions.""" + from quam_builder.builder.superconducting.build_quam import ( + add_ports, + add_transmons, + add_pulses, + ) + + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + machine.wiring = { + "qubits": { + "q0": { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "rr": { + "opx_output": "#/ports/mw_outputs/con1/2/1", + "opx_input": "#/ports/mw_inputs/con1/2/1", + }, + } + } + } + + add_ports(machine) + add_transmons(machine) + add_pulses(machine) + return machine + + +def _build_machine_with_add_qubit() -> FixedFrequencyQuam: + """Build the same machine via add_qubit.""" + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + wiring = { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "rr": { + "opx_output": "#/ports/mw_outputs/con1/2/1", + "opx_input": "#/ports/mw_inputs/con1/2/1", + }, + } + add_qubit(machine, "q0", wiring) + return machine + + +############################################################################## +############################################################################## +# add_qubit tests +############################################################################## +############################################################################## + + +def test_add_qubit_creates_transmon(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert isinstance(transmon, FixedFrequencyTransmon) + assert transmon.id == "q0" + assert "q0" in empty_ff_machine.qubits + assert empty_ff_machine.qubits["q0"] is transmon + + +def test_add_qubit_sets_parent_chain(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert transmon.parent is empty_ff_machine.qubits + assert transmon.parent.parent is empty_ff_machine + + +def test_add_qubit_creates_channels(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert transmon.xy is not None + assert isinstance(transmon.xy, XYDriveMW) + assert transmon.resonator is not None + assert isinstance(transmon.resonator, ReadoutResonatorMW) + + +def test_add_qubit_creates_ports(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert isinstance(empty_ff_machine.ports.mw_outputs["con1"][1][1], MWFEMAnalogOutputPort) + assert isinstance(empty_ff_machine.ports.mw_outputs["con1"][2][1], MWFEMAnalogOutputPort) + assert isinstance(empty_ff_machine.ports.mw_inputs["con1"][2][1], MWFEMAnalogInputPort) + + +def test_add_qubit_adds_default_pulses(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert "saturation" in transmon.xy.operations + assert "readout" in transmon.resonator.operations + + +def test_add_qubit_skips_pulses_when_disabled(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring, add_default_pulses=False) + + assert len(transmon.xy.operations) == 0 + + +def test_add_qubit_updates_active_names(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert "q0" in empty_ff_machine.active_qubit_names + + +def test_add_qubit_inserts_wiring(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + assert "q0" in empty_ff_machine.wiring["qubits"] + assert "xy" in empty_ff_machine.wiring["qubits"]["q0"] + + +def test_add_qubit_duplicate_raises(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + with pytest.raises(KeyError, match="already exists"): + add_qubit(empty_ff_machine, "q0", mw_wiring) + + +def test_add_qubit_generates_config(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + config = empty_ff_machine.generate_config() + element_names = set(config.get("elements", {}).keys()) + assert "q0.xy" in element_names or any("q0" in n for n in element_names) + + +############################################################################## +############################################################################## +# remove_qubit tests +############################################################################## +############################################################################## + + +def test_remove_qubit_returns_transmon(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + removed = remove_qubit(empty_ff_machine, "q0") + assert isinstance(removed, FixedFrequencyTransmon) + assert removed.id == "q0" + + +def test_remove_qubit_clears_parent(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + removed = remove_qubit(empty_ff_machine, "q0") + assert removed.parent is None + + +def test_remove_qubit_removes_from_dict(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_qubit(empty_ff_machine, "q0") + + assert "q0" not in empty_ff_machine.qubits + + +def test_remove_qubit_updates_active_names(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_qubit(empty_ff_machine, "q0") + + assert "q0" not in empty_ff_machine.active_qubit_names + + +def test_remove_qubit_cleans_wiring(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_qubit(empty_ff_machine, "q0") + + assert "q0" not in empty_ff_machine.wiring.get("qubits", {}) + + +def test_remove_qubit_not_found_raises(empty_ff_machine): + with pytest.raises(KeyError, match="not found"): + remove_qubit(empty_ff_machine, "qfake") + + +def test_remove_qubit_no_longer_in_config(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_qubit(empty_ff_machine, "q0") + + config = empty_ff_machine.generate_config() + element_names = set(config.get("elements", {}).keys()) + assert not any("q0" in n for n in element_names) + +############################################################################## +############################################################################## +# add_channel tests +############################################################################## +############################################################################## + + +def test_add_flux_channel_to_existing_qubit() -> None: + machine = FluxTunableQuam(ports=FEMPortsContainer()) + flux_ports = {"opx_output": "#/ports/analog_outputs/con1/3/1"} + drive_wiring = {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}} + + transmon = add_qubit(machine, "q0", drive_wiring, add_default_pulses=False) + assert transmon.z is None + + add_channel(machine, "q0", "z", flux_ports) + + assert transmon.z is not None + assert isinstance(transmon.z, FluxLine) + + +def test_add_channel_duplicate_raises(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + with pytest.raises(ValueError, match="already exists"): + add_channel( + empty_ff_machine, + "q0", + "xy", + {"opx_output": "#/ports/mw_outputs/con1/1/2"}, + ) + + +def test_add_channel_qubit_not_found_raises(empty_ff_machine): + with pytest.raises(KeyError, match="not found"): + add_channel( + empty_ff_machine, + "q99", + "xy", + {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + ) + + +def test_add_channel_unknown_type_raises(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + + with pytest.raises(ValueError, match="Unknown line type"): + add_channel(empty_ff_machine, "q0", "fake_type", {}) + + +def test_add_channel_updates_wiring() -> None: + machine = FluxTunableQuam(ports=FEMPortsContainer()) + drive_wiring = {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}} + add_qubit(machine, "q0", drive_wiring, add_default_pulses=False) + + flux_ports = {"opx_output": "#/ports/analog_outputs/con1/3/1"} + add_channel(machine, "q0", "z", flux_ports) + + assert "z" in machine.wiring["qubits"]["q0"] + + +def test_add_channel_visible_in_qubit_channels() -> None: + machine = FluxTunableQuam(ports=FEMPortsContainer()) + drive_wiring = {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}} + add_qubit(machine, "q0", drive_wiring, add_default_pulses=False) + add_channel( + machine, + "q0", + "z", + {"opx_output": "#/ports/analog_outputs/con1/3/1"}, + add_default_pulses=False, + ) + + transmon = machine.qubits["q0"] + assert "z" in transmon.channels + + +############################################################################## +############################################################################## +# remove_channel tests +############################################################################## +############################################################################## + + +def test_remove_channel_clears_field(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_channel(empty_ff_machine, "q0", "xy") + + assert empty_ff_machine.qubits["q0"].xy is None + + +def test_remove_channel_clears_parent(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + xy_before = transmon.xy + + remove_channel(empty_ff_machine, "q0", "xy") + assert xy_before.parent is None + + +def test_remove_channel_empty_slot_raises(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + machine = FluxTunableQuam(ports=FEMPortsContainer()) + drive_wiring = {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}} + add_qubit(machine, "q0", drive_wiring) + + with pytest.raises(ValueError, match="does not exist"): + remove_channel(machine, "q0", "z") + + +def test_remove_channel_updates_wiring(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + remove_channel(empty_ff_machine, "q0", "xy") + + assert "xy" not in empty_ff_machine.wiring["qubits"]["q0"] + + +def test_remove_channel_not_in_iterate_components(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: + transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) + xy = transmon.xy + remove_channel(empty_ff_machine, "q0", "xy") + + components = list(empty_ff_machine.iterate_components()) + assert xy not in components + + +############################################################################## +############################################################################## +# add_port / remove_port tests +############################################################################## +############################################################################## + +def test_add_mw_output_port(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + port = add_port( + machine, + "mw_output", + "con1", + fem_id=1, + port_id=3, + band=1, + upconverter_frequency=5e9, + ) + + assert isinstance(port, MWFEMAnalogOutputPort) + assert port.controller_id == "con1" + assert port.fem_id == 1 + assert port.port_id == 3 + assert machine.ports.mw_outputs["con1"][1][3] is port + + +def test_add_analog_output_port(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + port = add_port(machine, "analog_output", "con1", fem_id=5, port_id=1, offset=0.0) + + assert isinstance(port, LFFEMAnalogOutputPort) + assert machine.ports.analog_outputs["con1"][5][1] is port + + +def test_add_port_invalid_type_raises(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + + with pytest.raises(ValueError, match="Unsupported"): + add_port(machine, "teleporter", "con1", fem_id=1, port_id=1) + + +def test_add_port_visible_in_iterate_components(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + port = add_port( + machine, + "mw_output", + "con1", + fem_id=1, + port_id=1, + band=1, + upconverter_frequency=5e9, + ) + + components = list(machine.iterate_components()) + assert port in components + + +def test_remove_port(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + port = add_port( + machine, + "mw_output", + "con1", + fem_id=1, + port_id=1, + band=1, + upconverter_frequency=5e9, + ) + + removed = remove_port(machine, "mw_output", "con1", fem_id=1, port_id=1) + assert removed is port + assert removed.parent is None + assert 1 not in machine.ports.mw_outputs["con1"][1] + + +def test_remove_port_not_found_raises(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + + with pytest.raises(KeyError): + remove_port(machine, "mw_output", "con1", fem_id=1, port_id=99) + + +def test_remove_port_not_in_iterate_components(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + port = add_port( + machine, + "mw_output", + "con1", + fem_id=1, + port_id=1, + band=1, + upconverter_frequency=5e9, + ) + remove_port(machine, "mw_output", "con1", fem_id=1, port_id=1) + + components = list(machine.iterate_components()) + assert port not in components + + +############################################################################## +############################################################################## +# Add vs original tests +############################################################################## +############################################################################## + + +def test_same_channels_created(): + machine = _build_machine() + machine_add = _build_machine_with_add_qubit() + + assert type(machine.qubits["q0"].xy) is type(machine_add.qubits["q0"].xy) + assert type(machine.qubits["q0"].resonator) is type(machine_add.qubits["q0"].resonator) + + +def test_same_pulse_operations(): + machine = _build_machine() + machine_add = _build_machine_with_add_qubit() + + assert set(machine.qubits["q0"].xy.operations.keys()) == set( + machine_add.qubits["q0"].xy.operations.keys() + ) + assert set(machine.qubits["q0"].resonator.operations.keys()) == set( + machine_add.qubits["q0"].resonator.operations.keys() + ) + + +def test_same_active_qubit_names(): + machine = _build_machine() + machine_add = _build_machine_with_add_qubit() + + assert machine.active_qubit_names == machine_add.active_qubit_names + + +def test_same_port_types_created(): + machine = _build_machine() + machine_add = _build_machine_with_add_qubit() + + assert type(machine.ports.mw_outputs["con1"][1][1]) is type( + machine_add.ports.mw_outputs["con1"][1][1] + ) + + +def test_same_config_element_names(): + machine = _build_machine() + machine_add = _build_machine_with_add_qubit() + + machine_config = machine.generate_config() + machine_add_config = machine_add.generate_config() + + assert set(machine_config.get("elements", {}).keys()) == set( + machine_add_config.get("elements", {}).keys() + ) + + +def test_serialization_round_trip(tmp_path): + """Machine built with add_qubit survives save/load.""" + machine = _build_machine_with_add_qubit() + + machine.save(tmp_path / "state") + loaded = FixedFrequencyQuam.load(tmp_path / "state") + + assert "q0" in loaded.qubits + assert loaded.qubits["q0"].xy is not None + assert loaded.qubits["q0"].resonator is not None + assert "saturation" in loaded.qubits["q0"].xy.operations + assert "readout" in loaded.qubits["q0"].resonator.operations + + +def test_add_then_remove_leaves_clean_machine(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + wiring = { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "rr": { + "opx_output": "#/ports/mw_outputs/con1/2/1", + "opx_input": "#/ports/mw_inputs/con1/2/1", + }, + } + add_qubit(machine, "q0", wiring) + remove_qubit(machine, "q0") + + assert len(machine.qubits) == 0 + assert len(machine.active_qubit_names) == 0 + assert "q0" not in machine.wiring.get("qubits", {}) + + +def test_add_multiple_qubits(): + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + + for i in range(3): + wiring = { + "xy": {"opx_output": f"#/ports/mw_outputs/con1/1/{i + 1}"}, + "rr": { + "opx_output": f"#/ports/mw_outputs/con1/2/{i + 1}", + "opx_input": f"#/ports/mw_inputs/con1/2/{i + 1}", + }, + } + add_qubit(machine, f"q{i}", wiring) + + assert len(machine.qubits) == 3 + assert len(machine.active_qubit_names) == 3 + + config = machine.generate_config() + elements = set(config.get("elements", {}).keys()) + for i in range(3): + assert any(f"q{i}" in name for name in elements) From 4aeecf5b5058541f0dddddd539e1ac2f5ad7b39e Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Tue, 2 Jun 2026 09:09:30 -0700 Subject: [PATCH 02/15] docs: Changelog --- CHANGELOG.md | 5 +++++ quam_builder/builder/qop_connectivity/modify_ports.py | 6 ++++++ 2 files changed, 11 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 86bef6e9..0373afe6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Fixed - `add_default_transmon_pair_macros` now passes `flux_pulse_qubit="const"` (was `flux_pulse_control="const"`) to `CZGate`, matching the field rename introduced in v0.4.0 and aligning with the `"const"` pulse added to every `FluxTunableTransmon` Z line by `add_default_transmon_pulses`. +- Added incremental add/remove helpers for qubits, channels, and ports. + +### Changed + +- Formatting all files with black. ## [0.4.0] - 2026-05-26 diff --git a/quam_builder/builder/qop_connectivity/modify_ports.py b/quam_builder/builder/qop_connectivity/modify_ports.py index 44ed1155..2e77f39d 100644 --- a/quam_builder/builder/qop_connectivity/modify_ports.py +++ b/quam_builder/builder/qop_connectivity/modify_ports.py @@ -141,6 +141,9 @@ def add_port( For FEM-based machines, ``fem_id`` and ``port_id`` are both required. For OPX+-based machines, only ``port_id`` is required. + FEM port types: ("mw_output", "mw_input", "analog_output", "analog_input", "digital_output") + OPX+ port types: ("analog_output", "analog_input", "digital_output", "digital_input") + Args: machine: A QUAM machine with a ``ports`` attribute (``FEMPortsContainer`` or ``OPXPlusPortsContainer``). @@ -179,6 +182,9 @@ def remove_port( ) -> BasePort: """Remove a port from the machine's ports container. + FEM port types: ("mw_output", "mw_input", "analog_output", "analog_input", "digital_output") + OPX+ port types: ("analog_output", "analog_input", "digital_output", "digital_input") + Args: machine: A QUAM machine with a ``ports`` attribute. port_type: Port kind e.g. ``"mw_output"``, ``"analog_input"``. From f9e394274ccb121ad507691ab6f60748c9e271b7 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Tue, 2 Jun 2026 13:06:56 -0700 Subject: [PATCH 03/15] fix: updated to work with quam<0.5.0 --- .../builder/qop_connectivity/modify_ports.py | 28 +++++++++++++++++-- .../builder/superconducting/modify_quam.py | 18 +++++++++++- tests/test_modify_quam.py | 13 +++++++++ 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/quam_builder/builder/qop_connectivity/modify_ports.py b/quam_builder/builder/qop_connectivity/modify_ports.py index 2e77f39d..9cad04c5 100644 --- a/quam_builder/builder/qop_connectivity/modify_ports.py +++ b/quam_builder/builder/qop_connectivity/modify_ports.py @@ -40,6 +40,18 @@ def _get_ports_container(machine: _HasPorts) -> _PortsContainer: return ports +def _sanitize_fem_mw_kwargs(port_type: str, kwargs: dict) -> dict: + """Drop kwargs that ``FEMPortsContainer`` already passes explicitly.""" + sanitized = dict(kwargs) + if port_type == "mw_output": + sanitized.pop("band", None) + sanitized.pop("upconverter_frequency", None) + elif port_type == "mw_input": + sanitized.pop("band", None) + sanitized.pop("downconverter_frequency", None) + return sanitized + + def _add_fem_port( container: FEMPortsContainer, port_type: str, @@ -49,9 +61,21 @@ def _add_fem_port( **kwargs, ) -> BasePort: if port_type == "mw_output": - return container.get_mw_output(controller_id, fem_id, port_id, create=True, **kwargs) + return container.get_mw_output( + controller_id, + fem_id, + port_id, + create=True, + **_sanitize_fem_mw_kwargs(port_type, kwargs), + ) elif port_type == "mw_input": - return container.get_mw_input(controller_id, fem_id, port_id, create=True, **kwargs) + return container.get_mw_input( + controller_id, + fem_id, + port_id, + create=True, + **_sanitize_fem_mw_kwargs(port_type, kwargs), + ) elif port_type == "analog_output": return container.get_analog_output(controller_id, fem_id, port_id, create=True, **kwargs) elif port_type == "analog_input": diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 5fcb2733..37a76fec 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -30,6 +30,7 @@ from typing import Dict, Optional from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType +from quam.core.quam_classes import QuamDict from quam_builder.architecture.superconducting.qpu import AnyQuam from quam_builder.architecture.superconducting.qubit import AnyTransmon @@ -61,9 +62,24 @@ } +def _port_reference_values(ports: Dict[str, str]): + """Yield port reference strings without resolving through ``machine.wiring``. + + Wiring port dicts stored on the machine are ``QuamDict`` instances. Reading + ``ports[key]`` resolves references, but the port objects may not exist yet. + Use ``get_raw_value`` to read the stored reference string instead. + """ + for key in ports: + if isinstance(ports, QuamDict): + ref = ports.get_raw_value(key) + else: + ref = ports[key] + yield ref + + def _create_ports(machine: AnyQuam, ports: Dict[str, str]): """Ensure every port reference in *ports* exists in ``machine.ports``.""" - for ref in ports.values(): + for ref in _port_reference_values(ports): if isinstance(ref, str) and "ports" in ref and machine.ports is not None: machine.ports.reference_to_port(ref, create=True) diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index 36817d49..9562de5f 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -11,6 +11,10 @@ import pytest +from quam.config.models.quam import QuamConfig +from quam.config.resolvers import get_quam_config +from quam.config.vars import CONFIG_PATH_ENV_NAME + from quam.components.ports import ( FEMPortsContainer, MWFEMAnalogOutputPort, @@ -50,6 +54,15 @@ ############################################################################## +@pytest.fixture(autouse=True) +def compatible_quam_config(tmp_path, monkeypatch): + """Use a qualibrate config that matches the installed quam package version.""" + config_file = tmp_path / "config.toml" + config_file.write_text(f"[quam]\nversion = {QuamConfig.version}\n") + monkeypatch.setenv(CONFIG_PATH_ENV_NAME, str(config_file)) + get_quam_config.cache_clear() + + @pytest.fixture def empty_ff_machine() -> FixedFrequencyQuam: machine = FixedFrequencyQuam( From 4cf3275dd021d0298601c859e288359faf91af9a Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 4 Jun 2026 15:30:10 -0700 Subject: [PATCH 04/15] docs: replacing changelog message --- CHANGELOG.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0373afe6..6b586794 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) - `add_default_transmon_pair_macros` now passes `flux_pulse_qubit="const"` (was `flux_pulse_control="const"`) to `CZGate`, matching the field rename introduced in v0.4.0 and aligning with the `"const"` pulse added to every `FluxTunableTransmon` Z line by `add_default_transmon_pulses`. - Added incremental add/remove helpers for qubits, channels, and ports. -### Changed - -- Formatting all files with black. - ## [0.4.0] - 2026-05-26 ### Added From 9cabed938c569def3d317fbbfb79ec5de578958a Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 4 Jun 2026 15:38:00 -0700 Subject: [PATCH 05/15] refactor: formatting --- tests/test_modify_quam.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index 9562de5f..6adf5024 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -46,7 +46,6 @@ remove_port, ) - ############################################################################## ############################################################################## # Fixtures @@ -184,7 +183,9 @@ def test_add_qubit_adds_default_pulses(empty_ff_machine: FixedFrequencyQuam, mw_ assert "readout" in transmon.resonator.operations -def test_add_qubit_skips_pulses_when_disabled(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: +def test_add_qubit_skips_pulses_when_disabled( + empty_ff_machine: FixedFrequencyQuam, mw_wiring +) -> None: transmon = add_qubit(empty_ff_machine, "q0", mw_wiring, add_default_pulses=False) assert len(transmon.xy.operations) == 0 @@ -274,6 +275,7 @@ def test_remove_qubit_no_longer_in_config(empty_ff_machine: FixedFrequencyQuam, element_names = set(config.get("elements", {}).keys()) assert not any("q0" in n for n in element_names) + ############################################################################## ############################################################################## # add_channel tests @@ -389,7 +391,9 @@ def test_remove_channel_updates_wiring(empty_ff_machine: FixedFrequencyQuam, mw_ assert "xy" not in empty_ff_machine.wiring["qubits"]["q0"] -def test_remove_channel_not_in_iterate_components(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: +def test_remove_channel_not_in_iterate_components( + empty_ff_machine: FixedFrequencyQuam, mw_wiring +) -> None: transmon = add_qubit(empty_ff_machine, "q0", mw_wiring) xy = transmon.xy remove_channel(empty_ff_machine, "q0", "xy") @@ -404,6 +408,7 @@ def test_remove_channel_not_in_iterate_components(empty_ff_machine: FixedFrequen ############################################################################## ############################################################################## + def test_add_mw_output_port(): machine = FixedFrequencyQuam(ports=FEMPortsContainer()) port = add_port( From 2b0aa13e3d476cd12c43bd589682e5a7e9622bf0 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 4 Jun 2026 15:54:52 -0700 Subject: [PATCH 06/15] docs: QL-1225 link PR to Jira From 98d8c297649ca3064f9fe090491686251f2d55d8 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Mon, 8 Jun 2026 09:29:17 -0700 Subject: [PATCH 07/15] QL-1225 within QuAM instance, be able to add/remove physical channels, qubits, etc From c6f5202ea2ec9e57ca9491fe9e0e6aacf5b23920 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Tue, 9 Jun 2026 15:00:11 -0700 Subject: [PATCH 08/15] guard against removing pairs on remove_qubit --- .../builder/superconducting/modify_quam.py | 44 ++++++++++++++++--- tests/test_modify_quam.py | 31 +++++++++++++ 2 files changed, 69 insertions(+), 6 deletions(-) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 37a76fec..9633de88 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -27,8 +27,6 @@ remove_qubit(machine, "q5") """ -from typing import Dict, Optional - from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType from quam.core.quam_classes import QuamDict @@ -62,7 +60,7 @@ } -def _port_reference_values(ports: Dict[str, str]): +def _port_reference_values(ports: dict[str, str]): """Yield port reference strings without resolving through ``machine.wiring``. Wiring port dicts stored on the machine are ``QuamDict`` instances. Reading @@ -77,17 +75,42 @@ def _port_reference_values(ports: Dict[str, str]): yield ref -def _create_ports(machine: AnyQuam, ports: Dict[str, str]): +def _create_ports(machine: AnyQuam, ports: dict[str, str]): """Ensure every port reference in *ports* exists in ``machine.ports``.""" for ref in _port_reference_values(ports): if isinstance(ref, str) and "ports" in ref and machine.ports is not None: machine.ports.reference_to_port(ref, create=True) +def _get_referencing_qubit_pair_ids(machine: AnyQuam, qubit_id: str) -> list[str]: + """Pair ids that reference ``qubit_id`` in ``machine.qubit_pairs`` or wiring.""" + found: set[str] = set() + + for pair_id, pair in machine.qubit_pairs.items(): + for ref in (pair.qubit_control, pair.qubit_target): + if ref is None: + continue + if isinstance(ref, str): + name = ref.rstrip("/").split("/")[-1] + else: + name = ref.name + if name == qubit_id: + found.add(pair_id) + break + + for pair_id in machine.wiring.get("qubit_pairs", {}): + qc, qt = pair_id.split("-", 1) + qt = qt if str(qt).startswith("q") else f"q{qt}" + if qubit_id in (qc, qt): + found.add(pair_id) + + return sorted(found) + + def add_qubit( machine: AnyQuam, qubit_id: str, - wiring: Optional[Dict[str, Dict[str, str]]] = None, + wiring: dict[str, dict[str, str]] | None = None, add_default_pulses: bool = True, ) -> AnyTransmon: """Add a single qubit to an existing machine. @@ -165,10 +188,19 @@ def remove_qubit(machine: AnyQuam, qubit_id: str) -> AnyTransmon: Raises: KeyError: If no qubit with ``qubit_id`` exists. + ValueError: If the qubit participates in one or more qubit pairs. """ if qubit_id not in machine.qubits: raise KeyError(f"Qubit '{qubit_id}' not found") + pairs = _get_referencing_qubit_pair_ids(machine, qubit_id) + if pairs: + pair_list = ", ".join(pairs) + raise ValueError( + f"Cannot remove qubit '{qubit_id}': referenced by qubit pair(s): {pair_list}. " + "Remove the qubit pair(s) first." + ) + transmon = machine.qubits.pop(qubit_id) transmon.parent = None @@ -185,7 +217,7 @@ def add_channel( machine: AnyQuam, qubit_id: str, line_type: str, - ports: Dict[str, str], + ports: dict[str, str], add_default_pulses: bool = True, ) -> None: """Add a single channel to an existing qubit. diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index 6adf5024..fac8093d 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -34,6 +34,7 @@ ReadoutResonatorMW, ) from quam_builder.architecture.superconducting.components.flux_line import FluxLine +from quam_builder.architecture.superconducting.qubit_pair import FixedFrequencyTransmonPair from quam_builder.builder.superconducting.modify_quam import ( add_qubit, @@ -276,6 +277,36 @@ def test_remove_qubit_no_longer_in_config(empty_ff_machine: FixedFrequencyQuam, assert not any("q0" in n for n in element_names) +def test_remove_qubit_rejects_when_in_pair( + empty_ff_machine: FixedFrequencyQuam, mw_wiring +) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + add_qubit(empty_ff_machine, "q1", mw_wiring) + empty_ff_machine.qubit_pairs["q0-q1"] = FixedFrequencyTransmonPair( + id="q0-q1", + qubit_control="#/qubits/q0", + qubit_target="#/qubits/q1", + ) + + with pytest.raises(ValueError, match="qubit pair"): + remove_qubit(empty_ff_machine, "q0") + + assert "q0" in empty_ff_machine.qubits + + +def test_remove_qubit_rejects_when_in_pair_wiring_only( + empty_ff_machine: FixedFrequencyQuam, mw_wiring +) -> None: + add_qubit(empty_ff_machine, "q0", mw_wiring) + empty_ff_machine.wiring.setdefault("qubit_pairs", {}) + empty_ff_machine.wiring["qubit_pairs"]["q0-q1"] = {} + + with pytest.raises(ValueError, match="qubit pair"): + remove_qubit(empty_ff_machine, "q0") + + assert "q0" in empty_ff_machine.qubits + + ############################################################################## ############################################################################## # add_channel tests From abf3058372d814557347fe36686bf0bdd9906e1f Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Tue, 9 Jun 2026 15:32:10 -0700 Subject: [PATCH 09/15] add channel does not reseed default pulses on the whole transmon, only the new channel --- .../superconducting/add_default_pulses.py | 29 +++++++++++++++++++ .../builder/superconducting/modify_quam.py | 11 +++---- tests/test_modify_quam.py | 20 +++++++++++++ 3 files changed, 55 insertions(+), 5 deletions(-) diff --git a/quam_builder/builder/superconducting/add_default_pulses.py b/quam_builder/builder/superconducting/add_default_pulses.py index f912aee4..10ab69ce 100644 --- a/quam_builder/builder/superconducting/add_default_pulses.py +++ b/quam_builder/builder/superconducting/add_default_pulses.py @@ -4,6 +4,7 @@ DragCosinePulse, ) from qualang_tools.units import unit +from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType from quam_builder.architecture.superconducting.qubit import ( FixedFrequencyTransmon, FluxTunableTransmon, @@ -266,6 +267,34 @@ def add_Square_pulses( transmon.set_gate_shape("Square") +def add_default_transmon_channel_pulses( + transmon: Union[FixedFrequencyTransmon, FluxTunableTransmon], + line_type: str, +) -> None: + """Seed default pulse operations for a single channel. + + Only adds operations that are not already present on the channel. + """ + if line_type == WiringLineType.DRIVE.value: + if transmon.xy is not None and "saturation" not in transmon.xy.operations: + transmon.xy.operations["saturation"] = SquarePulse( + amplitude=0.25, length=20 * u.us, axis_angle=0 + ) + elif line_type == WiringLineType.FLUX.value: + if transmon.z is not None and "const" not in transmon.z.operations: + transmon.z.operations["const"] = SquarePulse(amplitude=0.1, length=100) + elif line_type == WiringLineType.RESONATOR.value: + if transmon.resonator is not None: + if "readout" not in transmon.resonator.operations: + transmon.resonator.operations["readout"] = SquareReadoutPulse( + length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" + ) + if "readout_GEF" not in transmon.resonator.operations: + transmon.resonator.operations["readout_GEF"] = SquareReadoutPulse( + length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" + ) + + def add_default_transmon_pulses(transmon: Union[FixedFrequencyTransmon, FluxTunableTransmon]): """Adds default pulses to a transmon qubit: * transmon.xy.operations["saturation"] = SquarePulse(amplitude=0.25, length=20 * u.us, axis_angle=0) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 9633de88..5270850d 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -33,7 +33,7 @@ from quam_builder.architecture.superconducting.qpu import AnyQuam from quam_builder.architecture.superconducting.qubit import AnyTransmon from quam_builder.builder.superconducting.add_default_pulses import ( - add_default_transmon_pulses, + add_default_transmon_channel_pulses, ) from quam_builder.builder.superconducting.add_transmon_drive_component import ( add_transmon_drive_component, @@ -129,7 +129,7 @@ def add_qubit( If ``None``, the wiring must already exist in ``machine.wiring["qubits"][qubit_id]``. - add_default_pulses: Seed default pulse operations on the new channels. + add_default_pulses: Seed default pulse operations on the new channels only. Returns: The newly created qubit, fully wired with channels and (optionally) pulses. @@ -164,7 +164,8 @@ def add_qubit( adder(transmon, wiring_path, ports) if add_default_pulses: - add_default_transmon_pulses(transmon) + for line_type in qubit_wiring: + add_default_transmon_channel_pulses(transmon, line_type) if qubit_id not in machine.active_qubit_names: machine.active_qubit_names.append(transmon.name) @@ -228,7 +229,7 @@ def add_channel( line_type: One of ``"xy"`` (drive), ``"rr"`` (resonator), or ``"z"`` (flux). ports: Dict with port references, e.g. ``{"opx_output": "#/ports/mw_outputs/con1/1/5"}``. - add_default_pulses: Seed default pulse operations on the new channel. + add_default_pulses: Seed default pulse operations on the new channel only. Raises: KeyError: If the qubit doesn't exist. @@ -258,7 +259,7 @@ def add_channel( adder(transmon, wiring_path, ports) if add_default_pulses: - add_default_transmon_pulses(transmon) + add_default_transmon_channel_pulses(transmon, line_type) def remove_channel(machine: AnyQuam, qubit_id: str, line_type: str) -> None: diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index fac8093d..37ff71b3 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -15,6 +15,7 @@ from quam.config.resolvers import get_quam_config from quam.config.vars import CONFIG_PATH_ENV_NAME +from quam.components.pulses import SquarePulse from quam.components.ports import ( FEMPortsContainer, MWFEMAnalogOutputPort, @@ -328,6 +329,25 @@ def test_add_flux_channel_to_existing_qubit() -> None: assert isinstance(transmon.z, FluxLine) +def test_add_channel_preserves_existing_pulses() -> None: + """Adding a channel must not overwrite calibrated pulses on other channels.""" + machine = FluxTunableQuam(ports=FEMPortsContainer()) + drive_wiring = {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}} + transmon = add_qubit(machine, "q0", drive_wiring, add_default_pulses=True) + transmon.xy.operations["saturation"] = SquarePulse(amplitude=0.99, length=100) + + add_channel( + machine, + "q0", + "z", + {"opx_output": "#/ports/analog_outputs/con1/3/1"}, + add_default_pulses=True, + ) + + assert transmon.xy.operations["saturation"].amplitude == 0.99 + assert "const" in transmon.z.operations + + def test_add_channel_duplicate_raises(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: add_qubit(empty_ff_machine, "q0", mw_wiring) From 56238acc4dfd7c812cf256eabd1b0cb71686b005 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Wed, 10 Jun 2026 17:28:27 -0700 Subject: [PATCH 10/15] add validation before mutation --- .../builder/superconducting/modify_quam.py | 80 +++++++++++++++---- tests/test_modify_quam.py | 46 +++++++++++ 2 files changed, 109 insertions(+), 17 deletions(-) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 5270850d..24b7df57 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -6,6 +6,9 @@ Line type keys match ``WiringLineType`` enum values: ``"xy"`` (drive), ``"rr"`` (resonator), ``"z"`` (flux). +``add_qubit`` and ``add_channel`` validate wiring and port mappings before +mutating the machine, so invalid input raises without changing state. + Example:: from quam_builder.builder.superconducting.modify_quam import ( @@ -32,6 +35,12 @@ from quam_builder.architecture.superconducting.qpu import AnyQuam from quam_builder.architecture.superconducting.qubit import AnyTransmon +from quam_builder.builder.qop_connectivity.channel_ports import ( + iq_in_out_channel_ports, + iq_out_channel_ports, + mw_in_out_channel_ports, + mw_out_channel_ports, +) from quam_builder.builder.superconducting.add_default_pulses import ( add_default_transmon_channel_pulses, ) @@ -82,6 +91,42 @@ def _create_ports(machine: AnyQuam, ports: dict[str, str]): machine.ports.reference_to_port(ref, create=True) +def _validate_port_mapping(line_type: str, ports: dict[str, str]) -> None: + keys = set(ports.keys()) + if line_type == WiringLineType.FLUX.value: + if "opx_output" in keys: + return + elif line_type == WiringLineType.DRIVE.value: + if all(key in keys for key in iq_out_channel_ports) or all( + key in keys for key in mw_out_channel_ports + ): + return + elif line_type == WiringLineType.RESONATOR.value: + if all(key in keys for key in iq_in_out_channel_ports) or all( + key in keys for key in mw_in_out_channel_ports + ): + return + raise ValueError(f"Unimplemented mapping of port keys to channel for ports: {ports}") + + +def _validate_qubit_wiring(qubit_wiring: dict[str, dict[str, str]]) -> None: + if not qubit_wiring: + raise ValueError("Qubit wiring cannot be empty") + for line_type, ports in qubit_wiring.items(): + if line_type not in _LINE_TYPE_TO_ADDER: + raise ValueError(f"Unknown line type: {line_type}. Valid line types are: {_LINE_TYPE_TO_ADDER.keys()}") + _validate_port_mapping(line_type, ports) + + +def _wire_qubit_channels( + machine: AnyQuam, transmon: AnyTransmon, qubit_id: str, qubit_wiring: dict[str, dict[str, str]] +) -> None: + for line_type, ports in qubit_wiring.items(): + _create_ports(machine, ports) + wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" + _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) + + def _get_referencing_qubit_pair_ids(machine: AnyQuam, qubit_id: str) -> list[str]: """Pair ids that reference ``qubit_id`` in ``machine.qubit_pairs`` or wiring.""" found: set[str] = set() @@ -136,32 +181,33 @@ def add_qubit( Raises: KeyError: If a qubit with ``qubit_id`` already exists. + ValueError: If wiring is invalid. """ if qubit_id in machine.qubits: raise KeyError(f"Qubit '{qubit_id}' already exists") - if wiring is not None: - machine.wiring.setdefault("qubits", {}) - machine.wiring["qubits"][qubit_id] = wiring - - qubit_wiring = machine.wiring.get("qubits", {}).get(qubit_id, {}) + qubit_wiring = ( + wiring + if wiring is not None + else machine.wiring.get("qubits", {}).get(qubit_id, {}) + ) + _validate_qubit_wiring(qubit_wiring) try: transmon = machine.qubit_type(id=qubit_id) - machine.qubits[qubit_id] = transmon except AttributeError as e: raise TypeError( f"{type(machine).__name__} does not define qubit_type. " "Use FixedFrequencyQuam or FluxTunableQuam." ) from e - for line_type, ports in qubit_wiring.items(): - _create_ports(machine, ports) - wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" - adder = _LINE_TYPE_TO_ADDER.get(line_type) - if adder is None: - raise ValueError(f"Unknown line type: {line_type}") - adder(transmon, wiring_path, ports) + if wiring is not None: + machine.wiring.setdefault("qubits", {}) + machine.wiring["qubits"][qubit_id] = wiring + + machine.qubits[qubit_id] = transmon + + _wire_qubit_channels(machine, transmon, qubit_id, qubit_wiring) if add_default_pulses: for line_type in qubit_wiring: @@ -233,7 +279,7 @@ def add_channel( Raises: KeyError: If the qubit doesn't exist. - ValueError: If the channel slot is already occupied. + ValueError: If the channel slot is already occupied or ports are invalid. """ if qubit_id not in machine.qubits: raise KeyError(f"Qubit '{qubit_id}' not found") @@ -248,15 +294,15 @@ def add_channel( f"Channel '{line_type}' (field '{field_name}') already exists on qubit '{qubit_id}'" ) - _create_ports(machine, ports) + _validate_port_mapping(line_type, ports) machine.wiring.setdefault("qubits", {}) machine.wiring["qubits"].setdefault(qubit_id, {}) machine.wiring["qubits"][qubit_id][line_type] = ports wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" - adder = _LINE_TYPE_TO_ADDER[line_type] - adder(transmon, wiring_path, ports) + _create_ports(machine, ports) + _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) if add_default_pulses: add_default_transmon_channel_pulses(transmon, line_type) diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index 37ff71b3..d1a84325 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -308,6 +308,52 @@ def test_remove_qubit_rejects_when_in_pair_wiring_only( assert "q0" in empty_ff_machine.qubits +def test_add_qubit_rejects_invalid_wiring_before_mutating( + empty_ff_machine: FixedFrequencyQuam, +) -> None: + wiring = { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "rr": {"not_a_valid_port_key": "#/ports/mw_outputs/con1/2/1"}, + } + + with pytest.raises(ValueError, match="Unimplemented mapping"): + add_qubit(empty_ff_machine, "q0", wiring) + + assert "q0" not in empty_ff_machine.qubits + assert "q0" not in empty_ff_machine.wiring.get("qubits", {}) + + +def test_add_qubit_rejects_unknown_line_type_before_mutating( + empty_ff_machine: FixedFrequencyQuam, +) -> None: + wiring = { + "xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}, + "coupler": {"opx_output": "#/ports/analog_outputs/con1/3/1"}, + } + + with pytest.raises(ValueError, match="Unknown line type"): + add_qubit(empty_ff_machine, "q0", wiring) + + assert "q0" not in empty_ff_machine.qubits + assert "q0" not in empty_ff_machine.wiring.get("qubits", {}) + + +def test_add_channel_rejects_invalid_ports_before_mutating() -> None: + machine = FluxTunableQuam(ports=FEMPortsContainer()) + add_qubit( + machine, + "q0", + {"xy": {"opx_output": "#/ports/mw_outputs/con1/1/1"}}, + add_default_pulses=False, + ) + + with pytest.raises(ValueError, match="Unimplemented mapping"): + add_channel(machine, "q0", "z", {"not_a_valid_port_key": "#/ports/analog_outputs/con1/3/1"}) + + assert machine.qubits["q0"].z is None + assert "z" not in machine.wiring["qubits"]["q0"] + + ############################################################################## ############################################################################## # add_channel tests From f8e3428e86b206c5f5b1906ca5bbfd948cc1a1d2 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 11 Jun 2026 13:22:08 -0700 Subject: [PATCH 11/15] add support for octave/external mixer wirings --- .../builder/superconducting/modify_quam.py | 86 +++++++++++++++++-- tests/test_modify_quam.py | 72 ++++++++++++++++ 2 files changed, 151 insertions(+), 7 deletions(-) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 24b7df57..90c1505c 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -8,6 +8,8 @@ ``add_qubit`` and ``add_channel`` validate wiring and port mappings before mutating the machine, so invalid input raises without changing state. +Port dicts may reference ``#/ports/...``, ``#/octaves/...``, and +``#/mixers/...``; the latter two are created on the machine when missing. Example:: @@ -30,9 +32,13 @@ remove_qubit(machine, "q5") """ +from pathlib import Path + from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType +from quam.components import FrequencyConverter, LocalOscillator, Octave from quam.core.quam_classes import QuamDict +from quam_builder.architecture.superconducting.components.mixer import StandaloneMixer from quam_builder.architecture.superconducting.qpu import AnyQuam from quam_builder.architecture.superconducting.qubit import AnyTransmon from quam_builder.builder.qop_connectivity.channel_ports import ( @@ -68,6 +74,11 @@ WiringLineType.FLUX.value: "z", } +_MIXER_CHANNEL_FIELD = { + WiringLineType.DRIVE.value: "xy", + WiringLineType.RESONATOR.value: "resonator", +} + def _port_reference_values(ports: dict[str, str]): """Yield port reference strings without resolving through ``machine.wiring``. @@ -84,11 +95,56 @@ def _port_reference_values(ports: dict[str, str]): yield ref -def _create_ports(machine: AnyQuam, ports: dict[str, str]): - """Ensure every port reference in *ports* exists in ``machine.ports``.""" +def _calibration_db_path( + machine: AnyQuam, calibration_db_path: Path | str | None +) -> Path: + if calibration_db_path is None: + calibration_db_path = machine.get_serialiser()._get_state_path().parent + if isinstance(calibration_db_path, str): + calibration_db_path = Path(calibration_db_path) + return calibration_db_path + + +def _create_line_refs( + machine: AnyQuam, + ports: dict[str, str], + qubit_id: str, + line_type: str, + calibration_db_path: Path | str | None = None, +) -> None: + """Create ports, octaves, and mixers referenced by a wiring port dict if missing.""" + for ref in _port_reference_values(ports): - if isinstance(ref, str) and "ports" in ref and machine.ports is not None: + if not isinstance(ref, str): + continue + if "ports" in ref and machine.ports is not None: machine.ports.reference_to_port(ref, create=True) + elif "octaves" in ref: + db_path = _calibration_db_path(machine, calibration_db_path) + octave_name = ref.split("/")[2] + if octave_name not in machine.octaves: + octave = Octave( + name=octave_name, + calibration_db_path=str(db_path), + ) + machine.octaves[octave_name] = octave + octave.initialize_frequency_converters() + elif "mixers" in ref: + channel_field = _MIXER_CHANNEL_FIELD.get(line_type) + if channel_field is None: + raise ValueError( + f"Cannot create mixer for line type '{line_type}' on qubit '{qubit_id}'" + ) + mixer_name = ref.split("/")[2] + if mixer_name not in machine.mixers: + machine.mixers[mixer_name] = FrequencyConverter( + local_oscillator=LocalOscillator(), + mixer=StandaloneMixer( + intermediate_frequency=( + f"#/qubits/{qubit_id}/{channel_field}/intermediate_frequency" + ), + ), + ) def _validate_port_mapping(line_type: str, ports: dict[str, str]) -> None: @@ -119,10 +175,16 @@ def _validate_qubit_wiring(qubit_wiring: dict[str, dict[str, str]]) -> None: def _wire_qubit_channels( - machine: AnyQuam, transmon: AnyTransmon, qubit_id: str, qubit_wiring: dict[str, dict[str, str]] + machine: AnyQuam, + transmon: AnyTransmon, + qubit_id: str, + qubit_wiring: dict[str, dict[str, str]], + calibration_db_path: Path | str | None = None, ) -> None: for line_type, ports in qubit_wiring.items(): - _create_ports(machine, ports) + _create_line_refs( + machine, ports, qubit_id, line_type, calibration_db_path + ) wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) @@ -157,6 +219,7 @@ def add_qubit( qubit_id: str, wiring: dict[str, dict[str, str]] | None = None, add_default_pulses: bool = True, + calibration_db_path: Path | str | None = None, ) -> AnyTransmon: """Add a single qubit to an existing machine. @@ -175,6 +238,8 @@ def add_qubit( If ``None``, the wiring must already exist in ``machine.wiring["qubits"][qubit_id]``. add_default_pulses: Seed default pulse operations on the new channels only. + calibration_db_path: Path to the Octave calibration database. Defaults to + the machine state directory. Returns: The newly created qubit, fully wired with channels and (optionally) pulses. @@ -207,7 +272,9 @@ def add_qubit( machine.qubits[qubit_id] = transmon - _wire_qubit_channels(machine, transmon, qubit_id, qubit_wiring) + _wire_qubit_channels( + machine, transmon, qubit_id, qubit_wiring, calibration_db_path + ) if add_default_pulses: for line_type in qubit_wiring: @@ -266,6 +333,7 @@ def add_channel( line_type: str, ports: dict[str, str], add_default_pulses: bool = True, + calibration_db_path: Path | str | None = None, ) -> None: """Add a single channel to an existing qubit. @@ -276,6 +344,8 @@ def add_channel( ports: Dict with port references, e.g. ``{"opx_output": "#/ports/mw_outputs/con1/1/5"}``. add_default_pulses: Seed default pulse operations on the new channel only. + calibration_db_path: Path to the Octave calibration database. Defaults to + the machine state directory. Raises: KeyError: If the qubit doesn't exist. @@ -301,7 +371,9 @@ def add_channel( machine.wiring["qubits"][qubit_id][line_type] = ports wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" - _create_ports(machine, ports) + _create_line_refs( + machine, ports, qubit_id, line_type, calibration_db_path + ) _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) if add_default_pulses: diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index d1a84325..508f30d6 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -354,6 +354,78 @@ def test_add_channel_rejects_invalid_ports_before_mutating() -> None: assert "z" not in machine.wiring["qubits"]["q0"] +def test_add_qubit_octave_wiring_creates_octave(tmp_path) -> None: + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + machine.save(tmp_path / "state") + wiring = { + "xy": { + "opx_output_I": "#/ports/analog_outputs/con1/1/1", + "opx_output_Q": "#/ports/analog_outputs/con1/1/2", + "frequency_converter_up": "#/octaves/oct1/RF_outputs/1", + }, + } + + add_qubit( + machine, + "q0", + wiring, + add_default_pulses=False, + calibration_db_path=tmp_path, + ) + + assert "oct1" in machine.octaves + assert machine.qubits["q0"].xy is not None + + +def test_add_qubit_mixer_wiring_creates_mixer() -> None: + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + wiring = { + "xy": { + "opx_output_I": "#/ports/analog_outputs/con1/1/1", + "opx_output_Q": "#/ports/analog_outputs/con1/1/2", + "frequency_converter_up": "#/mixers/mixercon1.q0.xy", + }, + } + + add_qubit(machine, "q0", wiring, add_default_pulses=False) + + assert "mixercon1.q0.xy" in machine.mixers + assert machine.qubits["q0"].xy is not None + + +def test_add_qubit_octave_wiring_reuses_existing_octave(tmp_path) -> None: + machine = FixedFrequencyQuam(ports=FEMPortsContainer()) + machine.save(tmp_path / "state") + octave_wiring = { + "xy": { + "opx_output_I": "#/ports/analog_outputs/con1/1/1", + "opx_output_Q": "#/ports/analog_outputs/con1/1/2", + "frequency_converter_up": "#/octaves/oct1/RF_outputs/1", + }, + } + + add_qubit( + machine, "q0", octave_wiring, add_default_pulses=False, calibration_db_path=tmp_path + ) + existing_octave = machine.octaves["oct1"] + + add_qubit( + machine, + "q1", + { + "xy": { + "opx_output_I": "#/ports/analog_outputs/con1/2/1", + "opx_output_Q": "#/ports/analog_outputs/con1/2/2", + "frequency_converter_up": "#/octaves/oct1/RF_outputs/2", + }, + }, + add_default_pulses=False, + calibration_db_path=tmp_path, + ) + + assert machine.octaves["oct1"] is existing_octave + + ############################################################################## ############################################################################## # add_channel tests From cb6ac3ef0ae79c635501578a97ae67a2c417b769 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 11 Jun 2026 13:24:54 -0700 Subject: [PATCH 12/15] fix changelog --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b586794..d9bdf930 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,11 +9,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Added - Added `custom_gates` section to `architecture/superconducting/README.md` documenting single-qubit macros (`MeasureMacro`, `ResetMacro`, `VirtualZMacro`, `DelayMacro`, `IdMacro`) and the `CZGate` two-qubit macro, including pulse-naming conventions and usage examples. Gate macros are currently specific to the superconducting architecture; `CZGate` requires `FluxTunableTransmonPair`. +- Added incremental add/remove helpers for qubits, channels, and ports. ### Fixed - `add_default_transmon_pair_macros` now passes `flux_pulse_qubit="const"` (was `flux_pulse_control="const"`) to `CZGate`, matching the field rename introduced in v0.4.0 and aligning with the `"const"` pulse added to every `FluxTunableTransmon` Z line by `add_default_transmon_pulses`. -- Added incremental add/remove helpers for qubits, channels, and ports. ## [0.4.0] - 2026-05-26 From b6ba26d026069d9866894d7ca3fc9837869b8121 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 11 Jun 2026 13:26:40 -0700 Subject: [PATCH 13/15] formatting --- .../builder/superconducting/modify_quam.py | 24 +++++++------------ tests/test_modify_quam.py | 8 ++----- 2 files changed, 10 insertions(+), 22 deletions(-) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 90c1505c..5c38f295 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -95,9 +95,7 @@ def _port_reference_values(ports: dict[str, str]): yield ref -def _calibration_db_path( - machine: AnyQuam, calibration_db_path: Path | str | None -) -> Path: +def _calibration_db_path(machine: AnyQuam, calibration_db_path: Path | str | None) -> Path: if calibration_db_path is None: calibration_db_path = machine.get_serialiser()._get_state_path().parent if isinstance(calibration_db_path, str): @@ -170,7 +168,9 @@ def _validate_qubit_wiring(qubit_wiring: dict[str, dict[str, str]]) -> None: raise ValueError("Qubit wiring cannot be empty") for line_type, ports in qubit_wiring.items(): if line_type not in _LINE_TYPE_TO_ADDER: - raise ValueError(f"Unknown line type: {line_type}. Valid line types are: {_LINE_TYPE_TO_ADDER.keys()}") + raise ValueError( + f"Unknown line type: {line_type}. Valid line types are: {_LINE_TYPE_TO_ADDER.keys()}" + ) _validate_port_mapping(line_type, ports) @@ -182,9 +182,7 @@ def _wire_qubit_channels( calibration_db_path: Path | str | None = None, ) -> None: for line_type, ports in qubit_wiring.items(): - _create_line_refs( - machine, ports, qubit_id, line_type, calibration_db_path - ) + _create_line_refs(machine, ports, qubit_id, line_type, calibration_db_path) wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) @@ -252,9 +250,7 @@ def add_qubit( raise KeyError(f"Qubit '{qubit_id}' already exists") qubit_wiring = ( - wiring - if wiring is not None - else machine.wiring.get("qubits", {}).get(qubit_id, {}) + wiring if wiring is not None else machine.wiring.get("qubits", {}).get(qubit_id, {}) ) _validate_qubit_wiring(qubit_wiring) @@ -272,9 +268,7 @@ def add_qubit( machine.qubits[qubit_id] = transmon - _wire_qubit_channels( - machine, transmon, qubit_id, qubit_wiring, calibration_db_path - ) + _wire_qubit_channels(machine, transmon, qubit_id, qubit_wiring, calibration_db_path) if add_default_pulses: for line_type in qubit_wiring: @@ -371,9 +365,7 @@ def add_channel( machine.wiring["qubits"][qubit_id][line_type] = ports wiring_path = f"#/wiring/qubits/{qubit_id}/{line_type}" - _create_line_refs( - machine, ports, qubit_id, line_type, calibration_db_path - ) + _create_line_refs(machine, ports, qubit_id, line_type, calibration_db_path) _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) if add_default_pulses: diff --git a/tests/test_modify_quam.py b/tests/test_modify_quam.py index 508f30d6..2d6f5a28 100644 --- a/tests/test_modify_quam.py +++ b/tests/test_modify_quam.py @@ -278,9 +278,7 @@ def test_remove_qubit_no_longer_in_config(empty_ff_machine: FixedFrequencyQuam, assert not any("q0" in n for n in element_names) -def test_remove_qubit_rejects_when_in_pair( - empty_ff_machine: FixedFrequencyQuam, mw_wiring -) -> None: +def test_remove_qubit_rejects_when_in_pair(empty_ff_machine: FixedFrequencyQuam, mw_wiring) -> None: add_qubit(empty_ff_machine, "q0", mw_wiring) add_qubit(empty_ff_machine, "q1", mw_wiring) empty_ff_machine.qubit_pairs["q0-q1"] = FixedFrequencyTransmonPair( @@ -404,9 +402,7 @@ def test_add_qubit_octave_wiring_reuses_existing_octave(tmp_path) -> None: }, } - add_qubit( - machine, "q0", octave_wiring, add_default_pulses=False, calibration_db_path=tmp_path - ) + add_qubit(machine, "q0", octave_wiring, add_default_pulses=False, calibration_db_path=tmp_path) existing_octave = machine.octaves["oct1"] add_qubit( From bc42feb525ec178e52ea3dcbcb3b5bb9814d92b2 Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Thu, 11 Jun 2026 14:12:41 -0700 Subject: [PATCH 14/15] clearer helper naming --- CHANGELOG.md | 2 +- .../builder/qop_connectivity/__init__.py | 11 +- .../builder/qop_connectivity/modify_ports.py | 140 +++++++++++++++++- 3 files changed, 143 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d9bdf930..4ad38a9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Added - Added `custom_gates` section to `architecture/superconducting/README.md` documenting single-qubit macros (`MeasureMacro`, `ResetMacro`, `VirtualZMacro`, `DelayMacro`, `IdMacro`) and the `CZGate` two-qubit macro, including pulse-naming conventions and usage examples. Gate macros are currently specific to the superconducting architecture; `CZGate` requires `FluxTunableTransmonPair`. -- Added incremental add/remove helpers for qubits, channels, and ports. +- Added incremental add/remove helpers for qubits, channels, and ports, including typed port helpers ``add_mw_port`` (MW-FEM) and ``add_lf_port`` (LF-FEM / OPX+ baseband) with a required ``type="input"`` or ``type="output"`` argument. ### Fixed diff --git a/quam_builder/builder/qop_connectivity/__init__.py b/quam_builder/builder/qop_connectivity/__init__.py index a907f15d..90c237c3 100644 --- a/quam_builder/builder/qop_connectivity/__init__.py +++ b/quam_builder/builder/qop_connectivity/__init__.py @@ -1,2 +1,11 @@ from .build_quam_wiring import build_quam_wiring -from .modify_ports import add_port, remove_port +from .modify_ports import ( + add_analog_port, + add_digital_port, + add_mw_port, + add_port, + remove_analog_port, + remove_digital_port, + remove_mw_port, + remove_port, +) diff --git a/quam_builder/builder/qop_connectivity/modify_ports.py b/quam_builder/builder/qop_connectivity/modify_ports.py index 9cad04c5..a7a1e80a 100644 --- a/quam_builder/builder/qop_connectivity/modify_ports.py +++ b/quam_builder/builder/qop_connectivity/modify_ports.py @@ -1,25 +1,38 @@ """Add and remove ports from any QUAM machine with a ports container. -Wraps ``FEMPortsContainer`` and ``OPXPlusPortsContainer`` methods into a -uniform interface that works across all modalities (superconducting, -quantum dots, NV centers). +Wraps ``FEMPortsContainer`` and ``OPXPlusPortsContainer`` methods into typed +helpers that work across all modalities (superconducting, quantum dots, +NV centers). Example:: - from quam_builder.builder.qop_connectivity.modify_ports import add_port, remove_port + from quam_builder.builder.qop_connectivity.modify_ports import ( + add_mw_port, + remove_mw_port, + ) - port = add_port(machine, "mw_output", "con1", fem_id=1, port_id=3, band=1) - remove_port(machine, "mw_output", "con1", fem_id=1, port_id=3) + port = add_mw_port(machine, con=1, slot=2, port=1, type="output", band=1) + remove_mw_port(machine, con=1, slot=2, port=1, type="output") """ -from typing import Optional, Protocol, Union +from typing import Literal, Optional, Protocol, Union from quam.components.ports import FEMPortsContainer, OPXPlusPortsContainer from quam.components.ports.base_ports import BasePort -__all__ = ["add_port", "remove_port"] +__all__ = [ + "add_analog_port", + "add_digital_port", + "add_mw_port", + "add_port", + "remove_analog_port", + "remove_digital_port", + "remove_mw_port", + "remove_port", +] _PortsContainer = Union[FEMPortsContainer, OPXPlusPortsContainer] +_PortDirection = Literal["input", "output"] _FEM_PORT_TYPES = ("mw_output", "mw_input", "analog_output", "analog_input", "digital_output") _OPXPLUS_PORT_TYPES = ("analog_output", "analog_input", "digital_output", "digital_input") @@ -29,6 +42,18 @@ class _HasPorts(Protocol): ports: Union[FEMPortsContainer, OPXPlusPortsContainer] +def _normalize_con(con: Union[str, int]) -> Union[str, int]: + if isinstance(con, int): + return f"con{con}" + return con + + +def _resolve_port_type(category: str, type: _PortDirection) -> str: + if type not in ("input", "output"): + raise ValueError("type must be 'input' or 'output'") + return f"{category}_{type}" + + def _get_ports_container(machine: _HasPorts) -> _PortsContainer: ports = getattr(machine, "ports", None) if ports is None: @@ -238,3 +263,102 @@ def remove_port( removed.parent = None return removed + + +def add_mw_port( + machine: _HasPorts, + con: Union[str, int], + slot: int, + port: int, + *, + type: _PortDirection, + **kwargs, +) -> BasePort: + """Create an MW-FEM port.""" + port_type = _resolve_port_type("mw", type) + return add_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port, **kwargs) + + +def remove_mw_port( + machine: _HasPorts, + con: Union[str, int], + slot: int, + port: int, + *, + type: _PortDirection, +) -> BasePort: + """Remove an MW-FEM port.""" + port_type = _resolve_port_type("mw", type) + return remove_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port) + + +def add_analog_port( + machine: _HasPorts, + con: Union[str, int], + port: int, + *, + type: _PortDirection, + slot: Optional[int] = None, + **kwargs, +) -> BasePort: + """Create an analog port. + + For FEM machines, ``slot`` is required. For OPX+ machines, omit ``slot``. + """ + port_type = _resolve_port_type("analog", type) + return add_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port, **kwargs) + + +def remove_analog_port( + machine: _HasPorts, + con: Union[str, int], + port: int, + *, + type: _PortDirection, + slot: Optional[int] = None, +) -> BasePort: + """Remove an analog port. + + For FEM machines, ``slot`` is required. For OPX+ machines, omit ``slot``. + """ + port_type = _resolve_port_type("analog", type) + return remove_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port) + + +def add_digital_port( + machine: _HasPorts, + con: Union[str, int], + port: int, + *, + type: _PortDirection, + slot: Optional[int] = None, + **kwargs, +) -> BasePort: + """Create a digital port. + + For FEM machines, only ``type="output"`` is supported and ``slot`` is + required. For OPX+ machines, omit ``slot``. + """ + if type == "input" and isinstance(_get_ports_container(machine), FEMPortsContainer): + raise ValueError("FEM containers only support digital output ports") + port_type = _resolve_port_type("digital", type) + return add_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port, **kwargs) + + +def remove_digital_port( + machine: _HasPorts, + con: Union[str, int], + port: int, + *, + type: _PortDirection, + slot: Optional[int] = None, +) -> BasePort: + """Remove a digital port. + + For FEM machines, only ``type="output"`` is supported and ``slot`` is + required. For OPX+ machines, omit ``slot``. + """ + if type == "input" and isinstance(_get_ports_container(machine), FEMPortsContainer): + raise ValueError("FEM containers only support digital output ports") + port_type = _resolve_port_type("digital", type) + return remove_port(machine, port_type, _normalize_con(con), fem_id=slot, port_id=port) From 603139ebde8df7664c8a005652d7afb9d2d4487e Mon Sep 17 00:00:00 2001 From: Jacob Fast Date: Mon, 15 Jun 2026 13:11:10 -0700 Subject: [PATCH 15/15] seeding default pulses moved to modify_quam --- .../superconducting/add_default_pulses.py | 29 -------------- .../builder/superconducting/modify_quam.py | 39 ++++++++++++++++--- 2 files changed, 34 insertions(+), 34 deletions(-) diff --git a/quam_builder/builder/superconducting/add_default_pulses.py b/quam_builder/builder/superconducting/add_default_pulses.py index 10ab69ce..f912aee4 100644 --- a/quam_builder/builder/superconducting/add_default_pulses.py +++ b/quam_builder/builder/superconducting/add_default_pulses.py @@ -4,7 +4,6 @@ DragCosinePulse, ) from qualang_tools.units import unit -from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType from quam_builder.architecture.superconducting.qubit import ( FixedFrequencyTransmon, FluxTunableTransmon, @@ -267,34 +266,6 @@ def add_Square_pulses( transmon.set_gate_shape("Square") -def add_default_transmon_channel_pulses( - transmon: Union[FixedFrequencyTransmon, FluxTunableTransmon], - line_type: str, -) -> None: - """Seed default pulse operations for a single channel. - - Only adds operations that are not already present on the channel. - """ - if line_type == WiringLineType.DRIVE.value: - if transmon.xy is not None and "saturation" not in transmon.xy.operations: - transmon.xy.operations["saturation"] = SquarePulse( - amplitude=0.25, length=20 * u.us, axis_angle=0 - ) - elif line_type == WiringLineType.FLUX.value: - if transmon.z is not None and "const" not in transmon.z.operations: - transmon.z.operations["const"] = SquarePulse(amplitude=0.1, length=100) - elif line_type == WiringLineType.RESONATOR.value: - if transmon.resonator is not None: - if "readout" not in transmon.resonator.operations: - transmon.resonator.operations["readout"] = SquareReadoutPulse( - length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" - ) - if "readout_GEF" not in transmon.resonator.operations: - transmon.resonator.operations["readout_GEF"] = SquareReadoutPulse( - length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" - ) - - def add_default_transmon_pulses(transmon: Union[FixedFrequencyTransmon, FluxTunableTransmon]): """Adds default pulses to a transmon qubit: * transmon.xy.operations["saturation"] = SquarePulse(amplitude=0.25, length=20 * u.us, axis_angle=0) diff --git a/quam_builder/builder/superconducting/modify_quam.py b/quam_builder/builder/superconducting/modify_quam.py index 5c38f295..bc50b42f 100644 --- a/quam_builder/builder/superconducting/modify_quam.py +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -34,8 +34,10 @@ from pathlib import Path +from qualang_tools.units import unit from qualang_tools.wirer.connectivity.wiring_spec import WiringLineType from quam.components import FrequencyConverter, LocalOscillator, Octave +from quam.components.pulses import SquarePulse, SquareReadoutPulse from quam.core.quam_classes import QuamDict from quam_builder.architecture.superconducting.components.mixer import StandaloneMixer @@ -47,9 +49,6 @@ mw_in_out_channel_ports, mw_out_channel_ports, ) -from quam_builder.builder.superconducting.add_default_pulses import ( - add_default_transmon_channel_pulses, -) from quam_builder.builder.superconducting.add_transmon_drive_component import ( add_transmon_drive_component, ) @@ -62,6 +61,8 @@ __all__ = ["add_qubit", "remove_qubit", "add_channel", "remove_channel"] +_u = unit(coerce_to_integer=True) + _LINE_TYPE_TO_ADDER = { WiringLineType.DRIVE.value: add_transmon_drive_component, WiringLineType.RESONATOR.value: add_transmon_resonator_component, @@ -174,6 +175,34 @@ def _validate_qubit_wiring(qubit_wiring: dict[str, dict[str, str]]) -> None: _validate_port_mapping(line_type, ports) +def _seed_default_pulses_for_line(transmon: AnyTransmon, line_type: str) -> None: + """Map wiring line type to a transmon channel and seed missing default pulses.""" + field_name = _LINE_TYPE_TO_FIELD.get(line_type) + if field_name is None: + return + channel = getattr(transmon, field_name, None) + if channel is None: + return + + if field_name == "xy": + if "saturation" not in channel.operations: + channel.operations["saturation"] = SquarePulse( + amplitude=0.25, length=20 * _u.us, axis_angle=0 + ) + elif field_name == "z": + if "const" not in channel.operations: + channel.operations["const"] = SquarePulse(amplitude=0.1, length=100) + elif field_name == "resonator": + if "readout" not in channel.operations: + channel.operations["readout"] = SquareReadoutPulse( + length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" + ) + if "readout_GEF" not in channel.operations: + channel.operations["readout_GEF"] = SquareReadoutPulse( + length=2000, amplitude=0.01, threshold=0.0, digital_marker="ON" + ) + + def _wire_qubit_channels( machine: AnyQuam, transmon: AnyTransmon, @@ -272,7 +301,7 @@ def add_qubit( if add_default_pulses: for line_type in qubit_wiring: - add_default_transmon_channel_pulses(transmon, line_type) + _seed_default_pulses_for_line(transmon, line_type) if qubit_id not in machine.active_qubit_names: machine.active_qubit_names.append(transmon.name) @@ -369,7 +398,7 @@ def add_channel( _LINE_TYPE_TO_ADDER[line_type](transmon, wiring_path, ports) if add_default_pulses: - add_default_transmon_channel_pulses(transmon, line_type) + _seed_default_pulses_for_line(transmon, line_type) def remove_channel(machine: AnyQuam, qubit_id: str, line_type: str) -> None: