# --------------------------------------------------------------------------
# Description : Cluster QCoDeS interface
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# --------------------------------------------------------------------------
# -- include -----------------------------------------------------------------
import warnings
from collections import defaultdict
from collections.abc import Iterable, Sequence
from functools import partial
from typing import Any, Callable, Optional, Union
from qcodes import Instrument, InstrumentChannel, Parameter
from qcodes import validators as vals
from qblox_instruments import DeviceInfo, SystemStatusFlags
from qblox_instruments.constants import NUM_SLOTS
from qblox_instruments.native import Cluster as ClusterNative
from qblox_instruments.native.helpers import ValidatedChannelConnection
from qblox_instruments.native.routing.cmm_router import SMFCMMRouter
from qblox_instruments.native.routing.module_router import ModuleSource, SMFModuleRouter
from qblox_instruments.qcodes_drivers.module import Module, get_item
from qblox_instruments.qcodes_drivers.sequencer import Sequencer
from qblox_instruments.qcodes_drivers.time import Time
from qblox_instruments.types import (
DebugLevel,
DummyConfiguration,
DummyModuleSpec,
ModuleConfigType,
)
# -- class -------------------------------------------------------------------
[docs]
class Cluster(ClusterNative, Instrument):
"""
Connects `QCoDeS <https://microsoft.github.io/Qcodes/>`_ to the Cluster native interface.
"""
# ------------------------------------------------------------------------
[docs]
def __init__(
self,
name: str,
identifier: Optional[str] = None,
port: Optional[int] = None,
debug: Optional[DebugLevel] = None,
dummy_cfg: Optional[Union[DummyConfiguration, DummyModuleSpec]] = None,
use_running_loop: bool = False,
) -> None:
"""
Creates Cluster QCoDeS class and adds all relevant instrument
parameters. These instrument parameters call the associated methods
provided by the native interface.
Parameters
----------
name : str
Instrument name.
identifier : Optional[str]
Instrument identifier. See :func:`~qblox_instruments.resolve()`.
If None, the instrument is identified by name.
port : Optional[int]
Override for the TCP port through which we should connect.
debug : Optional[DebugLevel]
Debug level. See :class:`~qblox_instruments.types.DebugLevel` for more
information. By default None, which means that for a connection to a dummy
cluster, `DebugLevel.ERROR_CHECK` will be used, and for a real cluster,
`DebugLevel.VERSION_CHECK`.
dummy_cfg : Optional[Union[DummyConfiguration, DummyModuleSpec]]
Configure as dummy using this configuration. For each slot that
needs to be occupied by a module add the slot index as key and
specify the type of module in the slot using the type
:class:`~qblox_instruments.ClusterType`.
use_running_loop : bool = False
Whether to use an existing event loop if it is already running;
use this only if you are supplying your own event loop and only invoking
asynchronous functions, or if the event loop is not running when invoking
synchronous command functions. Default to `False`.
"""
# For backwards compatibility: a boolean used to work, where True was
# interpreted as "no version check" and False as the default value.
if isinstance(debug, bool):
warnings.warn(
"Passing a boolean argument for the `debug` parameter is deprecated and could "
"cause unexpected behaviour in a future version.",
FutureWarning,
)
debug = DebugLevel.ERROR_CHECK if debug else None
# Initialize parent classes.
if identifier is None:
identifier = name
super().__init__(identifier, port, debug, dummy_cfg, use_running_loop=use_running_loop)
Instrument.__init__(self, name)
# Check for any errors that occurred during initialization
status = self.get_system_status()
if status.status == "ERROR":
for slot_idx, slot_err in enumerate(status.slot_flags, start=1):
if slot_err and any(
err == SystemStatusFlags.MODULE_FIRM_OR_HARDWARE_INCOMPATIBLE
for err in slot_err
):
if self._debug in (DebugLevel.ERROR_CHECK, DebugLevel.NO_CHECK):
warnings.warn(
f"Received a module incompatibility error in slot "
f"{slot_idx}: \n -> {status}"
)
else:
raise ConnectionError(
f"Received the following Error Status in slot "
f"{slot_idx}: \n -> {status}"
)
# Set number of slots
self._num_slots = NUM_SLOTS
# Initialize Routers
self._cmm_router = SMFCMMRouter()
self._module_router = SMFModuleRouter()
# Add QCoDeS parameters
self.add_parameter(
"led_brightness",
label="LED brightness",
docstring="Sets/gets frontpanel LED brightness.",
unit="",
vals=vals.Strings(),
val_mapping={
"off": "OFF",
"low": "LOW",
"medium": "MEDIUM",
"high": "HIGH",
},
set_parser=str,
get_parser=str,
set_cmd=self._set_led_brightness,
get_cmd=self._get_led_brightness,
)
self.add_parameter(
"reference_source",
label="Reference source.",
docstring="Sets/gets reference source ('internal' = internal "
"10 MHz, 'external' = external 10 MHz).",
unit="",
vals=vals.Bool(),
val_mapping={"internal": True, "external": False},
set_parser=bool,
get_parser=bool,
set_cmd=self._set_reference_source,
get_cmd=self._get_reference_source,
)
self.add_parameter(
"ext_trigger_input_delay",
label="Trigger input delay.",
docstring="Sets/gets the delay of the external input trigger in picoseconds.",
unit="ps",
vals=vals.Multiples(39, min_value=0, max_value=31 * 39),
set_parser=int,
get_parser=int,
set_cmd=self.set_trg_in_delay,
get_cmd=self.get_trg_in_delay,
)
self.add_parameter(
"ext_trigger_input_trigger_en",
label="Trigger input enable.",
docstring="Enable/disable the external input trigger.",
unit="",
vals=vals.Bool(),
set_parser=bool,
get_parser=bool,
set_cmd=self.set_trg_in_map_en,
get_cmd=self.get_trg_in_map_en,
)
self.add_parameter(
"ext_trigger_input_trigger_address",
label="Trigger address.",
docstring="Sets/gets the external input trigger address to which "
"the input trigger is mapped to the trigger network (T1 to "
"T15).",
unit="",
vals=vals.Numbers(1, 15),
set_parser=int,
get_parser=int,
set_cmd=self.set_trg_in_map_addr,
get_cmd=self.get_trg_in_map_addr,
)
for x in range(1, 16):
self.add_parameter(
f"trigger{x}_monitor_count",
label=f"Trigger monitor count for trigger address T{x}.",
docstring=f"Gets the trigger monitor count from trigger address T{x}.",
unit="",
get_cmd=partial(self.get_trigger_monitor_count, int(x)),
)
self.add_parameter(
"trigger_monitor_latest",
label="Latest monitor trigger for trigger address.",
docstring="Gets the trigger address which was triggered last (T1 to T15).",
unit="",
get_cmd=self.get_trigger_monitor_latest,
)
self.submodules.clear()
# Add modules for all slots
for slot_idx in range(1, self._num_slots + 1):
module = Module(self, f"module{slot_idx}", slot_idx)
self.add_submodule(f"module{slot_idx}", module)
# Add time-keeping functionality
if "time" not in self.submodules:
time = Time(self)
self.add_submodule("time", time)
# ------------------------------------------------------------------------
[docs]
def close(self) -> None:
"""
Irreversibly stop this instrument and free its resources.
Subclasses should override this if they have other specific resources to close.
"""
# The order of calls matter: we need to make sure that our Cluster connection
# termination logic completes *before* QCoDeS begins closing instruments,
# which could lead to jumbled garbage collection and asyncio exceptions.
ClusterNative.close(self)
Instrument.close(self)
# ------------------------------------------------------------------------
def __reinitialize_modules(self) -> None:
"""
Reinitialize modules based on the physical state of the slots.
"""
slot_info = self.get_json_description().get("modules") or {}
# Iterate over slot information to update or add modules
for slot_str, info in slot_info.items():
slot_id = int(slot_str)
# Skip if serial matches
if (
slot_id in self._mod_handles
and DeviceInfo.from_dict(info).serial == self._mod_handles[slot_id]["serial"]
):
continue
# Recreate module handle and add module
self.__remove_module(slot_id)
self.__add_module(slot_id)
# Remove entries if their slot_id is not in slot_info
for slot_id in list(self._mod_handles.keys()):
if str(slot_id) not in slot_info:
# Remove the module and add it again (a stub for empty slots)
self.__remove_module(slot_id)
self.__add_module(slot_id, create_handles=False)
# ------------------------------------------------------------------------
def __add_module(self, slot: int, create_handles: bool = True) -> None:
"""
Add the specific module for the provided slot.
Parameters
----------
slot: int
Slot index to add.
create_handles: bool
Whether to also create module-specific type and function reference handles.
"""
module = Module(self, f"module{slot}", slot)
# Add Module instance to self.submodules
self.add_submodule(f"module{slot}", module)
if create_handles:
self._create_mod_handles(slot)
# ------------------------------------------------------------------------
def __remove_module(self, slot: int) -> None:
"""
Remove traces of a disconnected module from the cluster.
Parameters
----------
slot: int
Slot index to remove.
"""
# Destroy module-specific handles, if present
self._destroy_mod_handles(slot)
# Remove Module instance from self.submodules
self.submodules.pop(f"module{slot}", None)
# ------------------------------------------------------------------------
@property
def modules(self) -> list[Module]:
"""
Get list of modules.
Returns
-------
list
List of modules.
"""
modules_list = [
submodule for submodule in self.submodules.values() if "module" in str(submodule)
]
return modules_list
# ------------------------------------------------------------------------
@property
def times(self) -> list[Time]:
"""
Get list of time blocks.
Returns
-------
list
List of digital time modules. There is only one, but we still need
to be able to iterate through it as if it was a list to not break
pytest.
"""
time_list = [
submodule for submodule in self.submodules.values() if "time" in str(submodule)
]
return time_list
# -------------------------------------------------------------------------
[docs]
def get_connected_modules(
self, filter_fn: Optional[Callable[[Module], bool]] = None
) -> dict[int, Module]:
"""
Get the currently connected modules for each occupied slot in the Cluster.
A selection of modules can be made by passing a filter function. For example:
.. code-block:: python
cluster.get_connected_modules(
filter_fn = lambda mod: mod.is_qrm_type and not mod.is_rf_type
)
Parameters
----------
filter_fn
Optional filter function that must return True for the modules that should
be included in the return value, and False otherwise.
Returns
-------
dict
Dictionary with key-value pairs consisting of slot numbers and corresponding
:class:`~.Module` objects. Only contains entries for modules that are
present and match the `filter_fn`.
"""
def checked_filter_fn(mod) -> bool:
if filter_fn is not None:
return filter_fn(mod)
return True
return {
mod.slot_idx: mod
for mod in self.modules
if mod.present() and mod.connected() and checked_filter_fn(mod)
}
# ------------------------------------------------------------------------
[docs]
def reset(self) -> None:
"""
Resets device, invalidates QCoDeS parameter cache and clears all
status and event registers (see
`SCPI <https://www.ivifoundation.org/downloads/SCPI/scpi-99.pdf>`_).
"""
self._reset()
# Reinitialize modules
self.__reinitialize_modules()
# Invalidate the QCoDeS cache
self._invalidate_qcodes_parameter_cache()
# ------------------------------------------------------------------------
def _hardware_reset(self) -> None:
"""
Revert reference clock source to internal and perform reset.
"""
self._scpi._hardware_reset() # pyright: ignore[reportAttributeAccessIssue]
# Reinitialize modules
self.__reinitialize_modules()
# Invalidate the QCoDeS cache
self._invalidate_qcodes_parameter_cache()
# ------------------------------------------------------------------------
[docs]
def disconnect_outputs(self, slot: int) -> None:
"""
Disconnects all outputs from the waveform generator paths of the
sequencers.
Parameters
----------
slot: int
Slot index
"""
self._disconnect_outputs(slot)
self._invalidate_qcodes_parameter_cache(slot)
# ------------------------------------------------------------------------
# ------------------------------------------------------------------------
[docs]
def connect_sequencer(self, slot: int, sequencer: int, *connections: str) -> None:
"""
Makes new connections between the indexed sequencer and some inputs
and/or outputs. This will fail if a requested connection already
existed, or if the connection could not be made due to a conflict with
an existing connection (hardware constraints). In such a case, the
channel map will not be affected.
Parameters
----------
slot: int
Slot index
sequencer : int
Sequencer index
*connections : str
Zero or more connections to make, each specified using a string.
The string should have the format `<direction><channel>` or
`<direction><I-channel>_<Q-channel>`. `<direction>` must be `in`
to make a connection between an input and the acquisition path,
`out` to make a connection from the waveform generator to an
output, or `io` to do both. The channels must be integer channel
indices. If only one channel is specified, the sequencer operates
in real mode; if two channels are specified, it operates in complex
mode.
Raises
------
RuntimeError
If the connection command could not be completed due to a conflict.
ValueError
If parsing of a connection fails.
"""
validated_connections = self.validate_connections(slot, sequencer, *connections)
self._sequencer_connect(slot, sequencer, *validated_connections)
self._invalidate_qcodes_parameter_cache(slot, sequencer)
[docs]
def validate_connections(
self, slot: int, sequencer: int, *connections: str
) -> list[ValidatedChannelConnection]:
module = self.modules[slot - 1]
return module.validate_connections(sequencer, *connections)
[docs]
def snapshot_base(
self,
update: Optional[bool] = False,
params_to_skip_update: Optional[Sequence[str]] = None,
) -> dict[Any, Any]:
"""
Override the cluster snapshot method to prevent it from sending unnecessary commands
to each module when polling its `connected` and `present` statuses.
We do this by first fetching all the slot numbers with a module connected/present,
then manually setting their QCoDeS parameter cache so their respective getter function
(which sends a SCPI command) will not be invoked.
See :meth:`InstrumentBase.snapshot_base` for the function signature.
"""
slots_with_module_connected = self._get_slots_with_module_connected()
slots_with_module_present = self._get_slots_with_module_present()
for module in self.modules:
module.parameters["connected"].cache.set(module.slot_idx in slots_with_module_connected)
module.parameters["present"].cache.set(module.slot_idx in slots_with_module_present)
params_to_skip_update = set(params_to_skip_update or []) | {"connected", "present"}
return super().snapshot_base(update, params_to_skip_update) # type: ignore
# ------------------------------------------------------------------------
def _invalidate_qcodes_parameter_cache(
self, slot: Optional[int] = None, sequencer: Optional[int] = None
) -> None:
"""
Marks the cache of all QCoDeS parameters in the module, including in
any sequencers the module might have, as invalid. Optionally,
a slot and a sequencer can be specified. This will invalidate the cache
of that slot or sequencer in that specific slot only instead of all
parameters.
Parameters
----------
slot : Optional[int]
Slot index of slot for which to invalidate the QCoDeS
parameters.
sequencer : Optional[int]
Sequencer index of sequencer for which to invalidate the QCoDeS
parameters.
"""
if slot is None:
self.invalidate_cache()
return
# Invalidate module parameters
self.modules[slot - 1]._invalidate_qcodes_parameter_cache(sequencer)
# ------------------------------------------------------------------------
def __getitem__(self, key: str) -> Union[InstrumentChannel, Parameter, Callable[..., Any]]:
"""
Get module or parameter using string based lookup.
Parameters
----------
key : str
Module, parameter or function to retrieve.
Returns
-------
Union[InstrumentChannel, Parameter, Callable[..., Any]]
Module, parameter or function.
Raises
------
KeyError
Module, parameter or function does not exist.
"""
return get_item(self, key)
# ------------------------------------------------------------------------
def __repr__(self) -> str:
"""
Returns simplified representation of class giving just the class,
name and connection.
Returns
-------
str
String representation of class.
"""
loc_str = ""
if hasattr(self._transport, "_socket"):
address, port = self._transport._socket.getpeername()
loc_str = f" at {address}:{port}"
return f"<{type(self).__name__}: {self.name}" + loc_str + ">"
[docs]
def clear_router(self) -> None:
"""
Clear SMF router configuration for ALL modules.
Raises
------
Exception
Invalid input parameter type.
"""
self._return_to_factory_router_config(0)
def _set_default_selfcast_max_id(self, max_selfcast_id: int) -> None:
"""
Sets the cluster-wide default range for automatic selfcast within each
sequencer for all modules.
Parameters
----------
max_selfcast_id : int
The maximum selfcast ID.
Raises
------
Exception
Invalid input parameter type.
"""
connected_modules = self.get_connected_modules(
filter_fn=lambda mod: mod.is_smf_capable
).values()
for module in connected_modules:
module._set_default_selfcast_max_id(max_selfcast_id)
def _get_default_selfcast_max_id(self) -> int:
"""
Get the cluster-wide default maximum selfcast ID.
Returns
-------
int
The cluster-wide default maximum selfcast ID.
"""
connected_modules = self.get_connected_modules(filter_fn=lambda mod: mod.is_smf_capable)
# Since all sequencers of all modules have the same max selfcast ID,
# we can just return the value for the first module.
return next(iter(connected_modules.values()))._get_default_selfcast_max_id()
def _validate_id(self, id_: list[int]) -> None:
"""
Check if the given event IDs do not overlap with selfcast IDs.
Parameters
----------
id_ : list[int]
The list of IDs (internally referred to as
event_id for the event processor).
Raises
------
ValueError
Invalid event IDs. They must be greater than the maximum selfcast ID.
"""
if max(id_) <= self._get_default_selfcast_max_id():
raise ValueError(
"Invalid event IDs. They must be greater than the maximum selfcast ID."
)
[docs]
def set_cmm_route(
self,
id_: Union[int, list[int]],
targets: Iterable[Union[Module, Sequencer]],
) -> None:
"""
Configures an inter-module routing configuration through the CMM.
Every module in the cluster is configured as a source, while the specified 'targets'
are configured as receivers. This enables inter-module communication across
the cluster.
Parameters
----------
id_ : list[int]
The list of IDs (internally referred to as
event_id for the event processor).
targets : Iterable[Union[Module, Sequencer]]
The list of modules and sequencers to configure as receivers.
Raises
------
ValueError
Target module is not connected to the cluster.
Exception
Invalid input parameter type.
"""
connected_modules = self.get_connected_modules(
filter_fn=lambda mod: mod.is_smf_capable
).values()
valid_targets = []
modules_to_config = defaultdict(list)
# 1. Filter and validate target modules/sequencers
for target in targets:
if isinstance(target, Module):
target_module = target
err_msg = "Target module is not connected to the cluster."
elif isinstance(target, Sequencer):
target_module = target.parent
err_msg = (
"Target sequencer belongs to a module that is not connected to the cluster."
)
else:
raise TypeError("Invalid target type. It should be a Module or Sequencer.")
# Skip target if its parent module is not SMF capable
if not target_module.is_smf_capable:
continue
# Ensure the target's parent module is connected
if target_module not in connected_modules:
raise ValueError(err_msg)
valid_targets.append(target)
# Group valid targets by their destination slot for later configuration
modules_to_config[target_module.slot_idx].append(target)
# Add 1024 offset for HW modules.
# This ID refers to the event_id for the event processor.
if isinstance(id_, int):
id_ = [id_]
self._validate_id(id_)
hw_event_id = [eid + 1024 for eid in id_]
# 2. Configure CMM Routing
# The CMM acts as the core of a star topology. Every connected SMF module
# is configured as a source, routing messages to all valid targets.
cmm_input_pairs = [(source, valid_targets) for source in connected_modules]
cmm_routes = self._cmm_router.get_compiled_routes(*cmm_input_pairs)
for route in cmm_routes:
route["event_id"] = hw_event_id
self._set_router_config(0, cmm_routes)
# 3. Configure Individual Module Routing
# For an all-to-all route setting, each module needs to know how to route
# incoming/outgoing messages.
for source_module in connected_modules:
slot = source_module.slot_idx
if slot in modules_to_config:
# Target module: Configure BOTH (scatter and gather) from PHY_A
# routing to the specific targets in this slot.
mod_routes = self._module_router.get_compiled_routes(
targets=modules_to_config[slot], source=ModuleSource.PHY_A
)
else:
# Non-target module: Configure to GATHER from PHY_A.
mod_routes = self._module_router.get_compiled_routes(
source=ModuleSource.PHY_A, config_type=ModuleConfigType.GATHER
)
for route in mod_routes:
route["event_id"] = hw_event_id
self._set_router_config(slot, mod_routes)
[docs]
def set_broadcast(self, id_: Union[int, list[int]]) -> None:
"""
Broadcasts every message with the specified id from all sequencers in all modules
to all other sequencers in the cluster via the CMM.
Parameters
----------
id_ : list[int]
The ID to broadcast (internally referred to as event_id for the event processor).
"""
modules = list(self.get_connected_modules().values())
self.set_cmm_route(id_, modules)
[docs]
def shutdown_module(self, slot: int) -> None:
"""
Shutdown specific connected module.
Parameters
----------
slot : int
Slot number of module to shut down.
"""
self._scpi.shutdown_module(slot)
self.__remove_module(slot)
[docs]
def shutdown_modules(self, slots: list[int]) -> None:
"""
Shutdown modules in the given slots.
Parameters
----------
slots : list[int]
Slot numbers of modules to shut down.
"""
self._scpi.shutdown_modules(slots)
for slot in slots:
self.__remove_module(slot)
