diff --git a/CHANGELOG.md b/CHANGELOG.md index 86bef6e9..4ad38a9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +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, 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 885f110c..90c237c3 100644 --- a/quam_builder/builder/qop_connectivity/__init__.py +++ b/quam_builder/builder/qop_connectivity/__init__.py @@ -1 +1,11 @@ from .build_quam_wiring import build_quam_wiring +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 new file mode 100644 index 00000000..a7a1e80a --- /dev/null +++ b/quam_builder/builder/qop_connectivity/modify_ports.py @@ -0,0 +1,364 @@ +"""Add and remove ports from any QUAM machine with a ports container. + +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_mw_port, + remove_mw_port, + ) + + 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 Literal, Optional, Protocol, Union + +from quam.components.ports import FEMPortsContainer, OPXPlusPortsContainer +from quam.components.ports.base_ports import BasePort + +__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") + + +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: + 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 _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, + 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, + **_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, + **_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": + 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. + + 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``). + 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. + + 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"``. + 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 + + +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) 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..bc50b42f --- /dev/null +++ b/quam_builder/builder/superconducting/modify_quam.py @@ -0,0 +1,436 @@ +"""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). + +``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:: + + 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 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 +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_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"] + +_u = unit(coerce_to_integer=True) + +_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", +} + +_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``. + + 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 _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 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: + 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 _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, + 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_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) + + +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: 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. + + 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 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. + + 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") + + 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) + except AttributeError as e: + raise TypeError( + f"{type(machine).__name__} does not define qubit_type. " + "Use FixedFrequencyQuam or FluxTunableQuam." + ) from e + + 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, calibration_db_path) + + if add_default_pulses: + for line_type in qubit_wiring: + _seed_default_pulses_for_line(transmon, line_type) + + 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. + 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 + + 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, + calibration_db_path: Path | str | None = None, +) -> 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 only. + calibration_db_path: Path to the Octave calibration database. Defaults to + the machine state directory. + + Raises: + KeyError: If the qubit doesn't exist. + 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") + + 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}'" + ) + + _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}" + _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: + _seed_default_pulses_for_line(transmon, line_type) + + +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..2d6f5a28 --- /dev/null +++ b/tests/test_modify_quam.py @@ -0,0 +1,774 @@ +"""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.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.pulses import SquarePulse +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.architecture.superconducting.qubit_pair import FixedFrequencyTransmonPair + +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(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( + 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) + + +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 + + +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"] + + +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 +############################################################################## +############################################################################## + + +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_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) + + 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)