Source code for quantify_scheduler.device_under_test.quantum_device

# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
"""Module containing the QuantumDevice object."""
from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from typing import Any

from qcodes.instrument.base import Instrument
from qcodes.instrument.parameter import InstrumentRefParameter, ManualParameter
from qcodes.utils import validators

from quantify_core.data.handling import get_datadir
from quantify_scheduler.backends.graph_compilation import (
    DeviceCompilationConfig,
    SerialCompilationConfig,
    SimpleNodeConfig,
)
from quantify_scheduler.backends.qblox.helpers import (
    _preprocess_legacy_hardware_config,
)
from quantify_scheduler.backends.qblox_backend import QbloxHardwareCompilationConfig
from quantify_scheduler.backends.types.common import HardwareCompilationConfig
from quantify_scheduler.device_under_test.device_element import DeviceElement
from quantify_scheduler.device_under_test.edge import Edge
from quantify_scheduler.helpers.importers import import_python_object_from_string
from quantify_scheduler.json_utils import SchedulerJSONDecoder, SchedulerJSONEncoder


[docs] class QuantumDevice(Instrument): """ The QuantumDevice directly represents the device under test (DUT). This contains a description of the connectivity to the control hardware as well as parameters specifying quantities like cross talk, attenuation and calibrated cable-delays. The QuantumDevice also contains references to individual DeviceElements, representations of elements on a device (e.g, a transmon qubit) containing the (calibrated) control-pulse parameters. This object can be used to generate configuration files for the compilation step from the gate-level to the pulse level description. These configuration files should be compatible with the :meth:`~quantify_scheduler.backends.graph_compilation.QuantifyCompiler.compile` function. """ def __init__(self, name: str) -> None: super().__init__(name=name) self.elements = ManualParameter( "elements", initial_value=list(), vals=validators.Lists(validators.Strings()), docstring="A list containing the names of all elements that" " are located on this QuantumDevice.", instrument=self, ) self.edges = ManualParameter( "edges", initial_value=list(), vals=validators.Lists(validators.Strings()), docstring="A list containing the names of all the edges which connect the" " DeviceElements within this QuantumDevice", instrument=self, ) self.instr_measurement_control = InstrumentRefParameter( "instr_measurement_control", docstring="A reference to the measurement control instrument.", vals=validators.MultiType(validators.Strings(), validators.Enum(None)), instrument=self, ) self.instr_instrument_coordinator = InstrumentRefParameter( "instr_instrument_coordinator", docstring="A reference to the instrument_coordinator instrument.", vals=validators.MultiType(validators.Strings(), validators.Enum(None)), instrument=self, ) self.cfg_sched_repetitions = ManualParameter( "cfg_sched_repetitions", initial_value=1024, docstring=( "The number of times execution of the schedule gets repeated when " "performing experiments, i.e. used to set the repetitions attribute of " "the Schedule objects generated." ), vals=validators.Ints(min_value=1), instrument=self, ) self.keep_original_schedule = ManualParameter( "keep_original_schedule", initial_value=True, docstring=( "If `True`, the compiler will not modify the schedule argument. " "If `False`, the compilation modifies the schedule, thereby " "making the original schedule unusable for further usage; this " "improves compilation time. Warning: if `False`, the returned schedule " "references objects from the original schedule, please refrain from modifying " "the original schedule after compilation in this case!" ), vals=validators.Bool(), instrument=self, ) self.hardware_config = ManualParameter( "hardware_config", docstring=( "The input dictionary used to generate a valid HardwareCompilationConfig " "using quantum_device.generate_hardware_compilation_config(). This configures " "the compilation from the quantum-device layer to the control-hardware layer." ), initial_value=None, instrument=self, ) self.scheduling_strategy = ManualParameter( "scheduling_strategy", docstring=("Scheduling strategy used to calculate absolute timing."), vals=validators.Enum("asap", "alap"), initial_value="asap", ) self._instrument_references = {} def __getstate__(self) -> dict[str, Any]: """ Serializes :class:`~QuantumDevice` into a dict containing serialized :class:`~DeviceElement` and :class:`~Edge` objects plus ``cfg_sched_repetitions``. """ data = {"name": self.name} data["elements"] = { element_name: json.dumps( self.get_element(element_name), cls=SchedulerJSONEncoder ) for element_name in self.elements() } data["edges"] = { edge_name: json.dumps(self.get_edge(edge_name), cls=SchedulerJSONEncoder) for edge_name in self.edges() } data["cfg_sched_repetitions"] = str(self.cfg_sched_repetitions()) state = { "deserialization_type": self.__class__.__name__, "data": data, } return state def __setstate__(self, state: dict[str, Any]) -> None: """ Deserializes a dict of serialized :class:`~DeviceElement` and :class:`~Edge` objects into a `QuantumDevice`. """ self.__init__(state["data"]["name"]) for element_name, serialized_element in state["data"]["elements"].items(): self._instrument_references[element_name] = json.loads( serialized_element, cls=SchedulerJSONDecoder ) self.add_element(self._instrument_references[element_name]) for edge_name, serialized_edge in state["data"]["edges"].items(): self._instrument_references[edge_name] = json.loads( serialized_edge, cls=SchedulerJSONDecoder ) self.add_edge(self._instrument_references[edge_name]) self.cfg_sched_repetitions(int(state["data"]["cfg_sched_repetitions"]))
[docs] def to_json(self) -> str: """ Convert the :class:`~QuantumDevice` data structure to a JSON string. Returns ------- : The json string containing the serialized `QuantumDevice`. """ device_instruments = [] if hasattr(self, "elements"): device_instruments += self.elements() if hasattr(self, "edges"): device_instruments += self.edges() if not device_instruments: raise RuntimeError( f"Cannot serialize '{self.name}'. All attached instruments have been " f"closed and their information cannot be retrieved any longer." ) closed_instruments = [] for device_name in device_instruments: try: Instrument.find_instrument(device_name) except KeyError: closed_instruments.append(device_name) if closed_instruments: raise RuntimeError( f"Cannot serialize '{self.name}'. Instruments '{closed_instruments}' have " f"been closed and their information cannot be retrieved any longer. " f"If you do not wish to include these in the " f"serialization, please remove using `QuantumDevice.remove_element` or " f"`QuantumDevice.remove_edge`." ) return json.dumps(self, cls=SchedulerJSONEncoder)
[docs] def to_json_file(self, path: str | None = None) -> str: """ Convert the `QuantumDevice` data structure to a JSON string and store it in a file. Parameters ---------- path The path to the directory where the file is created. Returns ------- : The name of the file containing the serialized `QuantumDevice`. """ if path is None: path = get_datadir() timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S_%Z") filename = os.path.join(path, f"{self.name}_{timestamp}.json") with open(filename, "w") as file: file.write(self.to_json()) return filename
@classmethod
[docs] def from_json(cls, data: str) -> QuantumDevice: """ Convert the JSON data to a `QuantumDevice`. Parameters ---------- data The JSON data in str format. Returns ------- : The deserialized :class:`~QuantumDevice` object. """ return json.loads(data, cls=SchedulerJSONDecoder)
@classmethod
[docs] def from_json_file(cls, filename: str) -> QuantumDevice: """ Read JSON data from a file and convert it to a `QuantumDevice`. Parameters ---------- filename The name of the file containing the serialized `QuantumDevice`. Returns ------- : The deserialized :class:`~QuantumDevice` object. """ with open(filename) as file: deserialized_device = cls.from_json(file.read()) return deserialized_device
[docs] def generate_compilation_config(self) -> SerialCompilationConfig: """Generate a config for use with a :class:`~.graph_compilation.QuantifyCompiler`.""" return SerialCompilationConfig( name="QuantumDevice-generated SerialCompilationConfig", keep_original_schedule=self.keep_original_schedule(), device_compilation_config=self.generate_device_config(), hardware_compilation_config=self.generate_hardware_compilation_config(), )
[docs] def generate_hardware_config(self) -> dict[str, Any]: """ Generate a valid hardware configuration describing the quantum device. Returns ------- The hardware configuration file used for compiling from the quantum-device layer to a hardware backend. .. warning: The config currently has to be specified by the user using the :code:`hardware_config` parameter. """ return self.hardware_config()
[docs] def generate_device_config(self) -> DeviceCompilationConfig: """ Generate a device config. This config is used to compile from the quantum-circuit to the quantum-device layer. """ clocks = {} elements_cfg = {} edges_cfg = {} # iterate over the elements on the device for element_name in self.elements(): element = self.get_element(element_name) element_cfg = element.generate_device_config() clocks.update(element_cfg.clocks) elements_cfg.update(element_cfg.elements) # iterate over the edges on the device for edge_name in self.edges(): edge = self.get_edge(edge_name) edge_cfg = edge.generate_edge_config() edges_cfg.update(edge_cfg) device_config = DeviceCompilationConfig( elements=elements_cfg, clocks=clocks, edges=edges_cfg, scheduling_strategy=self.scheduling_strategy(), ) return device_config
[docs] def generate_hardware_compilation_config(self) -> HardwareCompilationConfig | None: """ Generate a hardware compilation config. The compilation config is used to compile from the quantum-device to the control-hardware layer. """ hardware_config = self.hardware_config() if hardware_config is None: return None elif isinstance(hardware_config, HardwareCompilationConfig): # Hardware config is already a valid HardwareCompilationConfig DataStructure return hardware_config elif not any( [ key in hardware_config for key in [ "config_type", "hardware_description", "hardware_options", "connectivity", ] ] ): # Legacy support for the old hardware config dict: if ( hardware_config["backend"] == "quantify_scheduler.backends.qblox_backend.hardware_compile" ): hardware_config = _preprocess_legacy_hardware_config(hardware_config) compilation_passes = [ SimpleNodeConfig( name="compile_long_square_pulses_to_awg_offsets", compilation_func="quantify_scheduler.backends.qblox_backend" + ".compile_long_square_pulses_to_awg_offsets", ), SimpleNodeConfig( name="qblox_hardware_compile", compilation_func=hardware_config["backend"], ), ] hardware_compilation_config = QbloxHardwareCompilationConfig( hardware_description={}, hardware_options={}, connectivity=hardware_config, compilation_passes=compilation_passes, ) elif ( hardware_config["backend"] == "quantify_scheduler.backends.zhinst_backend.compile_backend" ): compilation_passes = [ SimpleNodeConfig( name="zhinst_compile_backend", compilation_func=hardware_config["backend"], ), ] hardware_compilation_config = HardwareCompilationConfig( hardware_description={}, hardware_options={}, connectivity=hardware_config, compilation_passes=compilation_passes, ) else: compilation_passes = [ SimpleNodeConfig( name="custom_hardware_backend", compilation_func=hardware_config["backend"], ), ] hardware_compilation_config = HardwareCompilationConfig( hardware_description={}, hardware_options={}, connectivity=hardware_config, compilation_passes=compilation_passes, ) else: # Parse a (backend-specific) HardwareCompilationConfig if "backend" in hardware_config: raise ValueError( f"`{HardwareCompilationConfig.__name__}` no longer takes a" f" 'backend' field; instead, specify the 'config_type', which should" " contain a string reference to the backend-specific datastructure" " that should be parsed." ) hardware_compilation_config_model = import_python_object_from_string( hardware_config["config_type"] ) hardware_compilation_config = ( hardware_compilation_config_model.model_validate(hardware_config) ) return hardware_compilation_config
[docs] def get_element(self, name: str) -> DeviceElement: """ Return a :class:`~quantify_scheduler.device_under_test.device_element.DeviceElement` by name. Parameters ---------- name The element name. Returns ------- : The element. Raises ------ KeyError If key ``name`` is not present in `self.elements`. """ if name in self.elements(): return self.find_instrument(name) raise KeyError(f"'{name}' is not an element of {self.name}.")
[docs] def add_element( self, element: DeviceElement, ) -> None: """ Add an element to the elements collection. Parameters ---------- element The element to add. Raises ------ ValueError If a element with a duplicated name is added to the collection. TypeError If :code:`element` is not an instance of the base element. """ if element.name in self.elements(): raise ValueError(f"'{element.name}' has already been added.") if not isinstance(element, DeviceElement): raise TypeError(f"{repr(element)} is not a DeviceElement.") self.elements().append(element.name) # list gets updated in place self._instrument_references[element.name] = element
[docs] def remove_element(self, name: str) -> None: """ Removes an element by name. Parameters ---------- name The element name. """ self.elements().remove(name) # list gets updated in place
[docs] def get_edge(self, name: str) -> Instrument: """ Returns an edge by name. Parameters ---------- name The edge name. Returns ------- : The edge. Raises ------ KeyError If key ``name`` is not present in ``self.edges``. """ if name in self.edges(): return self.find_instrument(name) raise KeyError(f"'{name}' is not an edge of {self.name}.")
[docs] def add_edge(self, edge: Edge) -> None: """ Add the edges. Parameters ---------- edge The edge name connecting the elements. Has to follow the convention 'element_0'-'element_1' """ if edge.name in self.edges(): raise ValueError(f"'{edge.name}' has already been added.") if not isinstance(edge, Edge): raise TypeError(f"{repr(edge)} is not an Edge.") self.edges().append(edge.name) self._instrument_references[edge.name] = edge
[docs] def remove_edge(self, edge_name: str) -> None: """ Remove an edge by name. Parameters ---------- edge_name The edge name. """ self.edges().remove(edge_name) # list gets updated in place