# ----------------------------------------------------------------------------
# Description : Cluster native interface
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# ----------------------------------------------------------------------------
import asyncio
import json
import logging
import re
import struct
import time
# -- include -----------------------------------------------------------------
import warnings
from collections.abc import Awaitable, Generator, Iterable, Iterator
from contextlib import contextmanager
from copy import deepcopy
from functools import partial
from typing import Any, ClassVar, Optional, TypeVar, Union, overload
from typing_extensions import Unpack
import numpy
from qblox_instruments.build import DeviceInfo
from qblox_instruments.exceptions import FirmwareCompatibilityError
from qblox_instruments.ieee488_2 import (
ClusterDummyTransport,
DummyBinnedAcquisitionData,
DummyScopeAcquisitionData,
Ieee488_2,
IpTransport,
make_sync_wrapper,
)
from qblox_instruments.native.definitions import (
BINNED_ACQUISITION_BIT_WIDTH_QRM,
BINNED_ACQUISITION_SCALING_QRC,
SCOPE_ACQUISITION_BIT_WIDTH_QRC,
SCOPE_ACQUISITION_BIT_WIDTH_QRM,
ChannelType,
ClusterState,
InstrumentState,
SequencerStates,
SequencerStatus,
SequencerStatusFlags,
SystemStatus,
get_allowed_channels,
)
from qblox_instruments.native.helpers import (
ChannelMapCache,
Ieee488_2Connection,
ValidatedChannelConnection,
check_io_channel_index,
check_is_valid_type,
check_program_length,
check_sequencer_index,
get_generic_json_config_val,
get_generic_json_config_val_async,
obtain_scpi_layer_from_transport,
pack_samples,
set_generic_json_config_val,
set_generic_json_config_val_async,
validate_acq,
validate_qcm_sequence,
validate_qrm_sequence,
validate_qtm_sequence,
validate_wave,
)
from qblox_instruments.pnp import resolve
from qblox_instruments.scpi import scpi_error_check
from qblox_instruments.scpi.cluster import slot_error_check_async
from qblox_instruments.types import (
AcquisitionEncodingConfig,
AverageCountDataFormat,
DebugLevel,
DummyConfiguration,
DummyModuleSpec,
InstrumentClass,
InstrumentType,
IQDataFormat,
OptionalSequence,
SampleEncoding,
Sequence,
ThresholdDataFormat,
TypeHandle,
ValidDataFormat,
)
T = TypeVar("T")
logger = logging.getLogger(__name__)
# -- class -------------------------------------------------------------------
[docs]
class Cluster:
"""
Class that provides the native API for the Cluster. It provides methods
to control all functions and features provided by the Cluster.
"""
use_configuration_cache: ClassVar[bool] = False
# ------------------------------------------------------------------------
[docs]
def __init__(
self,
identifier: str,
port: Optional[int] = None,
debug: Optional[DebugLevel] = None,
dummy_cfg: Optional[Union[DummyConfiguration, DummyModuleSpec]] = None,
use_configuration_cache: Optional[bool] = None,
use_running_loop: bool = False,
) -> None:
"""
Creates Cluster native interface object.
Parameters
----------
identifier : str
Instrument identifier. See :func:`~qblox_instruments.resolve()`
for more information.
port : Optional[int]
Instrument port. If None, this will be determined automatically.
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.MINIMAL_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`.
Raises
------
RuntimeError
Instrument cannot be reached due to invalid IP configuration.
ConnectionError
Instrument type is not supported.
"""
self._config_cache_enabled = False
# Create transport layer (dummy or socket interface)
self._dummy_config_present = False
if dummy_cfg is not None:
if not isinstance(dummy_cfg, DummyConfiguration):
dummy_cfg = DummyConfiguration(dummy_cfg)
self._dummy_config_present = True
self._transport = ClusterDummyTransport(dummy_cfg, use_running_loop=use_running_loop)
if debug is None:
debug = DebugLevel.ERROR_CHECK
else:
addr_info = resolve(identifier)
if addr_info.protocol != "ip":
raise RuntimeError(
f"Instrument cannot be reached due to invalid IP configuration. "
f"Use qblox-pnp tool to rectify; serial number is {addr_info.address}"
)
host = addr_info.address
if port is None:
port = addr_info.scpi_port
self._transport = IpTransport(host=host, port=port, use_running_loop=use_running_loop)
if debug is None:
debug = DebugLevel.MINIMAL_CHECK
self._debug = debug
# Obtain SCPI layer
self._cmm_dev_info, self._scpi = obtain_scpi_layer_from_transport(
self._transport, debug=debug
)
model = self._cmm_dev_info.model
self._type_handle = TypeHandle(model)
if not self._type_handle.is_mm_type:
raise ConnectionError(f"Unsupported instrument type detected ({self.instrument_type})")
self._scpi_transaction_connection_map: dict[int, Ieee488_2Connection] = {}
# Initialize state
self._create_mod_handles()
if use_configuration_cache is None:
use_configuration_cache = self.use_configuration_cache
self._init_config_cache(use_configuration_cache)
def __getattr__(self, name: str) -> Any:
if "_scpi" in self.__dict__ and hasattr(self._scpi, name):
return getattr(self._scpi, name)
return super().__getattr__(name)
def __dir__(self) -> list[str]:
"""
Return list of attribute names, including delegated SCPI methods.
This makes methods delegated via __getattr__ discoverable by dir()
and inspect.getmembers().
"""
result = set(super().__dir__())
result.update(dir(self._scpi))
return sorted(result)
def _run_in_loop(self, cb: Awaitable[T]) -> T:
return self._scpi._run_in_loop(cb)
[docs]
def close(self) -> None:
"""Close instrument and free resources."""
# Needs to explicitly exist to be resolved in superclass' `close()`.
if hasattr(self, "_scpi"):
self._run_in_loop(self._scpi.close())
@property
def is_dummy(self) -> bool:
"""
Return True if the cluster is configured as dummy.
Returns
-------
bool
Whether this is a dummy cluster.
"""
return self._dummy_config_present
def _create_mod_handles(self, slot: Optional[int] = None) -> None:
"""
Set up module-specific type and function reference handles for each
module slot or a specific slot if provided. This method initializes and
populates the `_mod_handles` dictionary with information about the modules.
It retrieves module information, checks for firmware version mismatches,
and sets up type handles and function references for the modules.
Parameters
----------
slot : Optional[int]
The slot to update. If None, updates all slots.
Raises
------
ConnectionError
If there is a mismatch between the application version of the CMM
and a module, and debug mode is not enabled. This requires a
firmware update for the entire cluster.
"""
# Set module specific type and FuncRefs handles
if slot is None:
# No specific slot provided, update all slots
self._mod_handles = {}
slot_info = self.get_json_description().get("modules") or {}
else:
# Only update the specified slot
self._mod_handles.pop(slot, None)
slot_info = (
{slot: self.get_json_description()["modules"][str(slot)]}
if str(slot) in self.get_json_description().get("modules") or {}
else {}
)
for slot_str, info in slot_info.items():
slot_id = int(slot_str)
mod_dev_info = DeviceInfo.from_dict(info)
# Module type handle
model = mod_dev_info.model
try:
mod_type_handle = TypeHandle(model)
except RuntimeError as e:
if self._debug == DebugLevel.NO_CHECK:
warnings.warn(str(e))
else:
raise e
if "is_rf" not in info:
warnings.warn(
f"Module in slot {slot_id} has responded with incomplete information "
f"(missing `is_rf` field) due to an incompatible firmware version. "
f"Please proceed with caution."
)
mod_type_handle._is_rf_type = bool(info.get("is_rf", False))
mod_type_handle._is_eom_type = bool(info.get("qtm_eom", False))
# Update module handles dictionary
self._mod_handles[slot_id] = {
"serial": mod_dev_info.serial,
"type_handle": mod_type_handle,
"features": info.get("features"),
}
# Register slot transport
if isinstance(self._transport, IpTransport):
slot_addr = resolve(f"{self._transport.host}/{slot_str}")
if slot_addr.protocol == "ip":
slot_transport = IpTransport(
slot_addr.address,
slot_addr.scpi_port,
loop_from=self._transport,
)
else:
raise RuntimeError(
f"Instrument cannot be reached due to invalid IP configuration. "
f"Use qblox-pnp tool to rectify; serial number is {slot_addr.address}"
)
elif isinstance(self._transport, ClusterDummyTransport):
slot_transport = self._transport.get_module_transport(slot_id)
else:
raise RuntimeError(f"Unsupported transport: {self._transport}")
self._scpi._add_slot(slot_id, slot_transport)
def _destroy_mod_handles(self, slot: int) -> None:
"""
Unset module-specific type and function reference handles for each
module slot or a specific slot if provided.
Parameters
----------
slot : int
The slot to update. For all slots, simply close the cluster.
"""
# Remove TypeHandle
self._mod_handles.pop(slot, None)
# Remove SCPI layer connections and locks
# NOTE: _remove_slot is async, so we need to run it in a loop
self._run_in_loop(self._scpi._remove_slot(slot))
# Remove SCPI transaction connection, if present
self._scpi_transaction_connection_map.pop(slot, None)
# Remove config cache entries, if enabled
if self._config_cache_enabled:
for cache_dict in (self._config_cache_channel_map, self._config_cache_sequencer):
for slot_idx, seq_idx in list(cache_dict.keys()):
if slot_idx == slot:
del cache_dict[(slot_idx, seq_idx)]
self._config_cache_pre_distortion.pop(slot)
def _init_config_cache(self, enabled: bool) -> None:
self._config_cache_enabled = enabled
self._clear_config_cache()
def _clear_config_cache(self) -> None:
self._config_cache_channel_map: dict[tuple[int, int], str] = {}
self._config_cache_sequencer: dict[tuple[int, int], str] = {}
self._config_cache_pre_distortion: dict[int, str] = {}
# ------------------------------------------------------------------------
@property
def _scpi_version(self) -> tuple[int, int]:
"""
Get the version of the currently loaded SCPI layer.
Returns
-------
tuple[int, int]
The loaded SCPI layer version in (major, minor) format.
"""
return self._cmm_dev_info.build.application_compat.scpi_version or (0, 1)
# ------------------------------------------------------------------------
@property
def instrument_class(self) -> InstrumentClass:
"""
Get instrument class (e.g. Cluster).
Returns
-------
InstrumentClass
Instrument class
"""
return self._type_handle.instrument_class
# ------------------------------------------------------------------------
@property
def instrument_type(self) -> InstrumentType:
"""
Get instrument type (e.g. MM, QRM, QCM).
Returns
-------
InstrumentType
Instrument type
"""
return self._type_handle.instrument_type
# ------------------------------------------------------------------------
def _present_at_init(self, slot: int) -> TypeHandle:
"""
Get an indication of module presence during initialization of this
object for a specific slot in the Cluster and return the associated
module type handle and function reference container if present.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
TypeHandle
Module type handle
Raises
------
KeyError
Module is not available.
"""
if slot in self._mod_handles:
return self._mod_handles[slot]["type_handle"]
else:
raise KeyError(f"Module at slot {slot} is not available.")
[docs]
def reconnect_slot(self, slot: int) -> None:
"""
Close and re-open connection to module at `slot`.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
"""
self._present_at_init(slot)
self._scpi._reconnect_slot(slot)
# ------------------------------------------------------------------------
def _module_type(self, slot: int) -> InstrumentType:
"""
Get indexed module's type (e.g. QRM, QCM, QTM, QDM, LINQ, QRC).
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
InstrumentType
Module type
"""
type_handle = self._present_at_init(slot)
return type_handle.instrument_type
# ------------------------------------------------------------------------
def _is_qcm_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QCM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool
True if module is of type QCM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qcm_type
# ------------------------------------------------------------------------
def _is_qrm_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QRM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type QRM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qrm_type
# ------------------------------------------------------------------------
def _is_qtm_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QTM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type QTM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qtm_type
# ------------------------------------------------------------------------
def _is_qdm_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QDM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type QDM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qdm_type
# ------------------------------------------------------------------------
def _is_eom_type(self, slot: int) -> bool:
"""
Return if indexed module is of type EOM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type EOM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_eom_type
# ------------------------------------------------------------------------
def _is_linq_type(self, slot: int) -> bool:
"""
Return if indexed module is of type LINQ.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type LINQ.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_linq_type
# ------------------------------------------------------------------------
def _is_qrc_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QRC.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type QRC.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qrc_type
# ------------------------------------------------------------------------
def _is_qsm_type(self, slot: int) -> bool:
"""
Return if indexed module is of type QSM.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module is of type QSM.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_qsm_type
# ------------------------------------------------------------------------
def _is_rf_type(self, slot: int) -> bool:
"""
Return if indexed module has RF functionality.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool:
True if module has RF functionality.
"""
type_handle = self._present_at_init(slot)
return type_handle.is_rf_type
# ------------------------------------------------------------------------
def _has_waveforms(self, slot: int, sequencer: int) -> bool:
"""
Return whether the module at the specified slot supports waveforms.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Returns
-------
bool
True if module supports waveforms.
"""
return self._is_qcm_type(slot) or self._is_qrm_type(slot) or self._is_qrc_type(slot)
# ------------------------------------------------------------------------
def _has_weights(self, slot: int, sequencer: int) -> bool:
"""
Return whether the module at the specified slot supports weights.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Returns
-------
bool
True if module supports weights.
"""
return self._is_qrm_type(slot) or (
self._is_qrc_type(slot)
and bool(get_allowed_channels(self._is_qrc_type(slot), ChannelType.ACQ, sequencer))
)
# ------------------------------------------------------------------------
def _has_acquisitions(self, slot: int, sequencer: int) -> bool:
"""
Return whether the module at the specified slot supports acquisitions.
Parameters
----------
slot : int
The slot index of the module being referred to.
Returns
-------
bool
True if module supports acquisitions.
"""
return (
self._is_qrm_type(slot)
or self._is_qtm_type(slot)
or (
self._is_qrc_type(slot)
and bool(get_allowed_channels(self._is_qrc_type(slot), ChannelType.ACQ, sequencer))
)
)
# ------------------------------------------------------------------------
def _get_scpi_commands(self) -> dict:
"""
Get SCPI commands and convert to dictionary.
Returns
-------
dict
Dictionary containing all available SCPI commands, corresponding
parameters, arguments and Python methods and finally a descriptive
comment.
"""
# Split function
def split(cmd_elem: str) -> list:
if cmd_elem not in ("None", ""):
return cmd_elem.split(",")
else:
return []
# Format command string
cmds = self._scpi._get_scpi_commands()
cmd_elem_list = cmds.split(";")[:-1]
cmd_list = numpy.reshape(cmd_elem_list, (int(len(cmd_elem_list) / 9), 9))
cmd_dict = {
cmd[0]: {
"scpi_in_type": split(cmd[1]),
"scpi_out_type": split(cmd[2]),
"python_func": cmd[3],
"python_in_type": split(cmd[4]),
"python_in_var": split(cmd[5]),
"python_out_type": split(cmd[6]),
"comment": cmd[8].replace("\t", "\n"),
}
for cmd in cmd_list
}
return cmd_dict
# ------------------------------------------------------------------------
[docs]
def get_idn(self) -> dict:
"""
Get device identity and build information and convert them to a
dictionary.
Returns
-------
dict
Dictionary containing manufacturer, model, serial number and build
information. The build information is subdivided into FPGA firmware,
kernel module software, application software and driver software build
information. Each of those consist of the version, build date,
build Git hash and Git build dirty indication.
"""
return DeviceInfo.from_idn(self._scpi._get_idn()).to_idn_dict()
# ------------------------------------------------------------------------
[docs]
def get_system_status(self) -> SystemStatus:
"""
Get general system status and convert it to a
:class:`~qblox_instruments.native.definitions.SystemStatus`.
Returns
-------
SystemStatus
Tuple containing general system status and corresponding flags.
"""
raw_state = self._scpi._get_system_state()
logger.debug("Raw response received by _get_system_state(): %s", raw_state)
return SystemStatus.from_string(raw_state)
# ------------------------------------------------------------------------
def _get_cluster_state(self) -> ClusterState:
"""
Get the current state of the cluster in a structured form.
The retrieved information includes:
- the general health level of the cluster;
- any flags (active and resolved) raised at system level;
- any flags (active and resolved) raised by the modules, on a per-slot basis.
.. note::
This feature requires firmware version **2.1.0** or higher.
Returns
-------
ClusterState
A dataclass containing status information about the components of the cluster.
Raises
------
FirmwareCompatibilityError
If the cluster does not support the command necessary to obtain data parsable
in the new structure.
"""
if self._scpi_version < (1, 2):
raise FirmwareCompatibilityError(min_version="2.1.0") from None
raw_state = self._scpi._get_system_state_json()
logger.debug("Raw response received by _get_system_state_json(): %s", raw_state)
return ClusterState._from_json_str(raw_state)
# ----------------------------------------------------------------------------
def _get_module_state(self, slot: int) -> InstrumentState:
"""
Get the current state of the given module in a structured form.
Parameters
----------
slot : int
The slot index of the module being referred to.
Returns
-------
InstrumentState
A dataclass containing active and resolved flags for the module.
Raises
------
FirmwareCompatibilityError
If the cluster does not support the command necessary to obtain data parsable
in the new structure.
"""
self._present_at_init(slot)
return self._get_cluster_state().modules.get(slot, InstrumentState())
# ----------------------------------------------------------------------------
def _set_acq_scope_config(self, slot: int, config: dict) -> None:
"""
Set configuration of the scope acquisition. The configuration consists of
multiple parameters in a C struct format. If an invalid sequencer index
is given or the configuration struct does not have the correct format, an
error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
config : dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
self._scpi._set_acq_scope_config(slot, config)
# ----------------------------------------------------------------------------
def _get_acq_scope_config(self, slot: int) -> dict:
"""
Get configuration of the scope acquisition. The configuration consists of
multiple parameters in a C struct format. If an invalid sequencer index is
given, an error is set in system error.
Returns
-------
slot : int
The slot index of the module being referred to.
dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
return self._scpi._get_acq_scope_config(slot)
# ------------------------------------------------------------------------
def _set_acq_scope_config_val(self, slot: int, keys: Any, val: Any) -> None:
"""
Set value of specific scope acquisition parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
val: Any
Value to set parameter to.
"""
self._present_at_init(slot)
set_generic_json_config_val(
lambda: self._get_acq_scope_config(slot),
lambda cfg: self._set_acq_scope_config(slot, cfg),
keys,
val,
)
# ------------------------------------------------------------------------
def _get_acq_scope_config_val(self, slot: int, keys: Any) -> Any:
"""
Get value of specific scope acquisition parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
return get_generic_json_config_val(
lambda: self._get_acq_scope_config(slot),
keys,
)
# ------------------------------------------------------------------------
@overload
def _set_io_channel_config(self, slot: int, config: list[dict[str, Any]]) -> None: ...
@overload
def _set_io_channel_config(self, slot: int, channel: int, config: dict[str, Any]) -> None: ...
def _set_io_channel_config(
self,
slot: int,
*_args,
**_kwargs,
) -> None:
"""
Set IO channel configuration. The configuration consists of
multiple parameters in a JSON format. If the configuration struct does not
have the correct format, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
config : dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
check_is_valid_type(self._is_qtm_type(slot) or self._is_qsm_type(slot))
num_args = len(_args) + len(_kwargs)
if num_args == 2: # Single channel
args_list = list(_args)
channel = _kwargs.get("channel") or args_list.pop(0)
config = _kwargs.get("config") or args_list.pop(0)
if self._is_qsm_type(slot):
self.modules[slot - 1].io_channels[channel]._validate_config(config)
return self._scpi._set_io_channel_config(slot, channel, config)
elif num_args == 1: # Multiple channels
config_list = _kwargs.get("config") or _args[0]
if self._is_qsm_type(slot):
for config_item in config_list:
try:
channel_idx = config_item["channel"]
except KeyError:
raise ValueError(
f"One of the configuration dicts is missing the 'channel' key: "
f"{config_item}"
)
self.modules[slot - 1].io_channels[channel_idx]._validate_config(config_item)
return self._set_io_output_config(slot, config_list)
else: # Wrong arguments
raise TypeError(
f"Invalid argument combination: {_args, _kwargs}.\n"
f"Please pass one of the following combinations:\n"
f" channel: int, config: dict\n"
f" config: list[dict]"
)
# ------------------------------------------------------------------------
def _set_output_normalized_amplitude(self, slot: int, channel: int, amplitude: float) -> None:
"""
Set IO Pulse output amplitude.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
Channel index.
amplitude : float
Normalized amplitude.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
self._scpi._set_output_normalized_amplitude(slot, channel, amplitude)
# ------------------------------------------------------------------------
def _get_output_normalized_amplitude(self, slot: int, channel: int) -> float:
"""
Get IO Pulse output amplitude.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
Channel index.
Returns
-------
amplitude : float
Normalized output amplitude
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
return self._scpi._get_output_normalized_amplitude(slot, channel)
# ------------------------------------------------------------------------
def _set_io_pulse_output_offset(self, slot: int, channel: int, offset: float) -> None:
"""
Set IO Pulse channel output offset.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
Channel index.
offset : float
I/O channel index.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
self._scpi._set_io_pulse_output_offset(slot, channel, offset)
# ------------------------------------------------------------------------
def _get_io_pulse_output_offset(self, slot: int, channel: int) -> float:
"""
Get IO Pulse channel output offset.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
Channel index.
Returns
-------
offset : float
output offset
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
return self._scpi._get_io_pulse_output_offset(slot, channel)
# ------------------------------------------------------------------------
def _set_io_pulse_width_config(self, slot: int, channel: int, config: dict) -> None:
"""
Set IO Pulse width. Config must be a dict containing coarse and fine settings.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O quad index.
config : dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
self._scpi._set_io_pulse_width(slot, channel, config)
# ------------------------------------------------------------------------
def _get_io_pulse_width_config(self, slot: int, channel: int) -> Any:
"""
Get IO Pulse width. Config must be a dict containing coarse and fine settings.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
Returns
-------
dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot) and self._is_eom_type(slot))
return self._scpi._get_io_pulse_width(slot, channel)
# --------------------------------------------------------------------------
[docs]
def set_io_pulse_width_config_val(self, slot: int, channel: int, keys: Any, val: Any) -> None:
"""
Set value of specific IO channel configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
keys : Union[List[str], str]
Configuration key to access, or hierarchical list thereof.
val: Any
Value to set parameter to.
"""
set_generic_json_config_val(
lambda: self._get_io_pulse_width_config(slot, channel),
lambda cfg: self._set_io_pulse_width_config(slot, channel, cfg),
keys,
val,
)
# --------------------------------------------------------------------------
[docs]
def get_io_pulse_width_config_val(self, slot: int, channel: int, keys: Any) -> Any:
"""
Get value of specific IO channel configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
keys : Union[List[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
return get_generic_json_config_val(lambda: self._get_io_channel_config(slot, channel), keys)
# ------------------------------------------------------------------------
@overload
async def _get_io_channel_config_async(self, slot: int) -> list[dict[str, Any]]: ...
@overload
async def _get_io_channel_config_async(self, slot: int, channel: int) -> dict[str, Any]: ...
async def _get_io_channel_config_async(
self, slot: int, channel: Optional[int] = None
) -> Union[dict[str, Any], list[dict[str, Any]]]:
"""
Get IO channel configuration. The configuration consists of
multiple parameters in a JSON format.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : Optional[int]
I/O channel index.
Returns
-------
dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
check_is_valid_type(self._is_qtm_type(slot) or self._is_qsm_type(slot))
if channel is None:
return await self._scpi._get_io_output_config_async(slot)
else:
return await self._scpi._get_io_channel_config_async(slot, channel)
_get_io_channel_config = make_sync_wrapper(_get_io_channel_config_async)
# ------------------------------------------------------------------------
def _get_io_channel_status(self, slot: int, channel: int) -> dict:
"""
Get IO channel status. The status consists of multiple values in a JSON
format.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
Returns
-------
dict
Status dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot))
return self._scpi._get_io_channel_status(slot, channel)
# ------------------------------------------------------------------------
async def _set_io_channel_config_val_async(
self, slot: int, channel: int, keys: Any, val: Any
) -> None:
"""
Set value of specific IO channel configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof.
channel : int
I/O channel index.
val: Any
Value to set parameter to.
"""
self._present_at_init(slot)
await set_generic_json_config_val_async(
partial(self._get_io_channel_config_async, slot, channel),
partial(self._set_io_channel_config_async, slot, channel),
keys,
val,
)
_set_io_channel_config_val = make_sync_wrapper(_set_io_channel_config_val_async)
# ------------------------------------------------------------------------
async def _get_io_channel_config_val_async(self, slot: int, channel: int, keys: Any) -> Any:
"""
Get value of specific IO channel configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
return await get_generic_json_config_val_async(
partial(self._get_io_channel_config_async, slot, channel),
keys,
)
_get_io_channel_config_val = make_sync_wrapper(_get_io_channel_config_val_async)
# ------------------------------------------------------------------------
def _get_io_channel_status_val(self, slot: int, channel: int, keys: Any) -> Any:
"""
Get value of specific IO channel status parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
channel : int
I/O channel index.
keys : Union[list[str], str]
Status key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
return get_generic_json_config_val(
lambda: self._get_io_channel_status(slot, channel),
keys,
)
# ----------------------------------------------------------------------------
def _set_quad_config(self, slot: int, quad: int, config: dict) -> None:
"""
Set quad configuration. The configuration consists of
multiple parameters in a JSON format. If the configuration struct does not
have the correct format, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
quad : int
I/O quad index.
config : dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot))
self._scpi._set_quad_config(slot, quad, config)
# ----------------------------------------------------------------------------
def _get_quad_config(self, slot: int, quad: int) -> Any:
"""
Get quad configuration. The configuration consists of
multiple parameters in a JSON format.
Parameters
----------
slot : int
The slot index of the module being referred to.
quad : int
I/O quad index.
Returns
-------
dict
Configuration dictionary.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot))
return self._scpi._get_quad_config(slot, quad)
# ------------------------------------------------------------------------
def _set_quad_config_val(self, slot: int, quad: int, keys: Any, val: Any) -> None:
"""
Set value of specific quad configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
quad : int
I/O quad index.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
val: Any
Value to set parameter to.
"""
self._present_at_init(slot)
set_generic_json_config_val(
lambda: self._get_quad_config(slot, quad),
lambda cfg: self._set_quad_config(slot, quad, cfg),
keys,
val,
)
# ------------------------------------------------------------------------
def _get_quad_config_val(self, slot: int, quad: int, keys: Any) -> Any:
"""
Get value of specific quad configuration parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
quad : int
I/O quad index.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
return get_generic_json_config_val(
lambda: self._get_quad_config(slot, quad),
keys,
)
# ----------------------------------------------------------------------------
def _can_precompile(self, slot: int, sequencer_index: int) -> tuple[bool, bool]:
features = self._mod_handles[slot].get("features")
if features is None or "pre_compile" not in features:
module_support = False
sequencer_support = False
else:
module_support = True
sequencer_support = sequencer_index in features["pre_compile"]
return (module_support, sequencer_support)
# ----------------------------------------------------------------------------
def _can_encode_partial_acquisitions(self, slot: int) -> bool:
features = self._mod_handles[slot].get("features")
return features is not None and "partial_acquisition" in features
# ----------------------------------------------------------------------------
[docs]
def get_sequencer_isa_version(self, slot: int, sequencer: int) -> tuple[int, int]:
"""
Return the sequencer ISA version.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Returns
-------
tuple[int, int]
Sequencer version, in (major, minor) format.
"""
self._present_at_init(slot)
if hasattr(self._scpi, "_get_isa_versions"):
version = self._scpi._get_isa_versions(slot, sequencer)
else:
version = {"major": 1, "minor": 0}
return version["major"], version["minor"]
# ----------------------------------------------------------------------------
async def _set_sequencer_program_async(
self, slot: int, sequencer: int, program: str, immediate_asm_install: bool = True
) -> None:
"""
Assemble and set Q1ASM program for the indexed sequencer. If assembling fails,
a RuntimeError is thrown with the assembler log attached.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
program : str
Q1ASM program.
Raises
------
RuntimeError
Assembly failed.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
# remove all comments
program = re.sub(r"#.*$", "", program, flags=re.MULTILINE)
# remove all double whitespace
program = re.sub(r"[^\S\r\n]+", " ", program)
# trim each line/remove empty lines
program = re.sub(r"^\s*", "", program, flags=re.MULTILINE)
program = re.sub(r"\s*$", "", program, flags=re.MULTILINE)
check_program_length(program)
try:
if immediate_asm_install:
await self._scpi._set_sequencer_program_async(slot, sequencer, program)
else:
module_supported, sequencer_supported = self._can_precompile(slot, sequencer)
scpi_layer_support = hasattr(self._scpi, "_assemble_sequencer_program")
if not module_supported or not scpi_layer_support:
raise ValueError("pre-compilation is not supported for this module")
if not sequencer_supported:
raise ValueError("pre-compilation is not supported for this sequencer index")
await self._scpi._assemble_sequencer_program_async(slot, sequencer, program)
except:
print(await self._scpi._get_assembler_log_async(slot))
raise
_set_sequencer_program = make_sync_wrapper(_set_sequencer_program_async)
# ----------------------------------------------------------------------------
async def _set_sequencer_config_async(self, slot: int, sequencer: int, config: dict) -> None:
"""
Set configuration of the indexed sequencer. The configuration consists
dictionary containing multiple parameters that will be converted into a
JSON object supported by the device.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
config : dict
Configuration dictionary.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
if self._config_cache_enabled:
self._config_cache_sequencer[slot, sequencer] = deepcopy(config)
await self._scpi._set_sequencer_config_async(slot, sequencer, config)
_set_sequencer_config = make_sync_wrapper(_set_sequencer_config_async)
# ----------------------------------------------------------------------------
async def _get_sequencer_config_async(self, slot: int, sequencer: int) -> dict:
"""
Get configuration of the indexed sequencer. The configuration consists
dictionary containing multiple parameters that will be converted from a
JSON object provided by the device.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Returns
-------
dict
Configuration dictionary.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
if self._config_cache_enabled and (slot, sequencer) in self._config_cache_sequencer:
return deepcopy(self._config_cache_sequencer[slot, sequencer])
result = await self._scpi._get_sequencer_config_async(slot, sequencer)
if self._config_cache_enabled:
self._config_cache_sequencer[slot, sequencer] = deepcopy(result)
return result
_get_sequencer_config = make_sync_wrapper(_get_sequencer_config_async)
# ------------------------------------------------------------------------
async def _set_sequencer_config_val_async(
self, slot: int, sequencer: int, keys: Any, val: Any
) -> None:
"""
Set value of specific sequencer parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
val : Any
Value to set parameter to.
"""
self._present_at_init(slot)
if self._scpi_version < (1, 1) and "real_mode_en" in keys:
raise NotImplementedError(
"'real_mode_en' is not supported on the firmware version of the connected cluster. "
"This feature is supported on firmware version >= 2.0.0."
)
try:
await set_generic_json_config_val_async(
partial(self._get_sequencer_config_async, slot, sequencer),
partial(self._set_sequencer_config_async, slot, sequencer),
keys,
val,
is_sequencer=True,
)
except TypeError as e:
typ = e.args[1]
raise TypeError(
f"{'.'.join(str(k) for k in keys)} should be of type {typ.__name__}: {e.__cause__}"
)
except KeyError:
raise KeyError(f"{'.'.join(str(k) for k in keys)} is an incomplete sequencer path")
_set_sequencer_config_val = make_sync_wrapper(_set_sequencer_config_val_async)
# ------------------------------------------------------------------------
async def _get_sequencer_config_val_async(self, slot: int, sequencer: int, keys: Any) -> Any:
"""
Get value of specific sequencer parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
if self._scpi_version < (1, 1) and "real_mode_en" in keys:
raise NotImplementedError(
"'real_mode_en' is not supported on the firmware version of the connected cluster. "
"This feature is supported on firmware version >= 2.0.0."
)
try:
return await get_generic_json_config_val_async(
partial(self._get_sequencer_config_async, slot, sequencer),
keys,
is_sequencer=True,
)
except KeyError as e:
key = e.args[1]
raise KeyError(
f"cfg_dict[{']['.join(str(e) for e in keys)}] is not a valid sequencer path, "
f"failed at {key}"
)
_get_sequencer_config_val = make_sync_wrapper(_get_sequencer_config_val_async)
# -------------------------------------------------------------------------
[docs]
def get_sequencer_registers(
self, slot: int, sequencer: int, registers: Optional[list[str]] = None
) -> dict[str, int]:
"""
Get registers of the indexed sequencer.
The register structure is a mapping of register name to register value.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
slot : int
slot index.
sequencer : int
Sequencer index.
registers: Optional[list[str]]
Optionally, a list of registers to fetch. If absent, all registers will be fetched.
Returns
-------
dict[str, int]
Current register values.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
self._present_at_init(slot)
if registers is not None:
return self._get_sequencer_registers(slot, sequencer, ",".join(registers))
else:
return self._get_sequencer_registers(slot, sequencer, "*")
# -------------------------------------------------------------------------
[docs]
def set_sequencer_registers(self, slot: int, sequencer: int, registers: dict[str, int]) -> None:
"""
Set registers of the indexed sequencer.
The register structure is a mapping of register name to register value.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
slot : int
slot index.
sequencer : int
Sequencer index.
registers: dict[str, int]
Optionally, a list of registers to fetch. If absent, all registers will be fetched.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
self._present_at_init(slot)
return self._set_sequencer_registers(slot, sequencer, registers)
# ------------------------------------------------------------------------
def _set_sequencer_channel_map(
self, slot: int, sequencer: int, sequencer_channel_map: Any
) -> None:
"""
Set channel map of the indexed sequencer. The channel map consists of a list
with two elements representing the components of the complex signal to be acquired,
each mapping to a list of DAC indices. Index i maps to output i+1.
If an invalid sequencer index is given or the channel map is not valid,
an error is set in system error.
Parameters
----------
slot : int
slot index.
sequencer : int
Sequencer index.
sequencer_channel_map : Any
Current Sequencer index Any.
Returns
-------
None
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
if self._config_cache_enabled:
self._config_cache_channel_map[slot, sequencer] = deepcopy(sequencer_channel_map)
self._scpi._set_sequencer_channel_map(slot, sequencer, sequencer_channel_map)
# -------------------------------------------------------------------------
def _get_sequencer_channel_map(self, slot: int, sequencer: int) -> Any:
"""
Get channel map of the indexed sequencer. The channel map consists of a list
with two elements representing the components of the complex signal to be acquired,
each mapping to a list of DAC indices. Index i maps to output i+1.
If an invalid sequencer index is given or the channel map is not valid,
an error is set in system error.
Parameters
----------
slot : int
slot index.
sequencer : int
Sequencer index.
Returns
-------
Any
Current Sequencer index.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
if self._config_cache_enabled and (slot, sequencer) in self._config_cache_channel_map:
return deepcopy(self._config_cache_channel_map[slot, sequencer])
result = self._scpi._get_sequencer_channel_map(slot, sequencer)
if self._config_cache_enabled:
self._config_cache_channel_map[slot, sequencer] = deepcopy(result)
return result
# ------------------------------------------------------------------------
def _set_sequencer_config_rotation_matrix(
self, slot: int, sequencer: int, phase_incr: float
) -> None:
"""
Sets the integration result phase rotation matrix in the acquisition path.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
phase_incr : float
Phase increment in degrees.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
cfg_dict = self._get_sequencer_config(slot, sequencer)
cfg_dict["acq"][0]["th_acq"]["rotation_matrix_a11"] = numpy.cos(
numpy.deg2rad(360 - phase_incr)
)
cfg_dict["acq"][0]["th_acq"]["rotation_matrix_a12"] = numpy.sin(
numpy.deg2rad(360 - phase_incr)
)
self._set_sequencer_config(slot, sequencer, cfg_dict)
# ------------------------------------------------------------------------
def _get_sequencer_config_rotation_matrix(self, slot: int, sequencer: int) -> float:
"""
Gets the integration result phase rotation matrix in the acquisition path.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Returns
-------
float
Phase increment in degrees.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
cfg = self._get_sequencer_config(slot, sequencer)
vector = (
cfg["acq"][0]["th_acq"]["rotation_matrix_a11"]
+ cfg["acq"][0]["th_acq"]["rotation_matrix_a12"] * 1j
)
phase_incr = numpy.angle(vector, deg=True)
if phase_incr == 0:
return 0
elif phase_incr >= 0:
return 360 - phase_incr
else:
return -1.0 * phase_incr
# ------------------------------------------------------------------------
def _set_sequencer_connect_out(
self, slot: int, sequencer: int, output: int, state: str
) -> None:
"""
Set whether the output of the indexed sequencer is connected to the given
output and if so with which path.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
output : int
Zero-based output index.
state : str | bool
- For baseband modules, one of:
- "off": the output is not connected.
- "I": the output is connected to path0/I.
- "Q": the output is connected to path1/Q.
- For RF modules, one of:
- "off" or False: the RF output is not connected.
- "IQ" or True: the RF output is connected.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
with ChannelMapCache(self, slot) as channel_map:
# Note that for RF modules, each connect_out parameter controls the
# configuration for two DACs, hardwired to the I and Q input of the
# RF mixer on the front end. It doesn't make sense to control the
# DACs individually in this case.
if self._is_rf_type(slot):
if state is False or state == "off":
channel_map.disconnect(ChannelType.AWG, sequencer, 0, output * 2)
channel_map.disconnect(ChannelType.AWG, sequencer, 0, output * 2 + 1)
channel_map.disconnect(ChannelType.AWG, sequencer, 1, output * 2)
channel_map.disconnect(ChannelType.AWG, sequencer, 1, output * 2 + 1)
elif state is True or state == "IQ":
channel_map.connect(ChannelType.AWG, sequencer, 0, output * 2)
channel_map.connect(ChannelType.AWG, sequencer, 1, output * 2 + 1)
else:
raise ValueError(f"invalid new connection state {state!r} for RF device")
elif state == "off":
channel_map.disconnect(ChannelType.AWG, sequencer, 0, output)
channel_map.disconnect(ChannelType.AWG, sequencer, 1, output)
elif state == "I":
channel_map.connect(ChannelType.AWG, sequencer, 0, output)
channel_map.disconnect(ChannelType.AWG, sequencer, 1, output)
elif state == "Q":
channel_map.disconnect(ChannelType.AWG, sequencer, 0, output)
channel_map.connect(ChannelType.AWG, sequencer, 1, output)
else:
raise ValueError(f"invalid new connection state {state!r} for baseband device")
# -------------------------------------------------------------------------
def _get_sequencer_connect_out(self, slot: int, sequencer: int, output: int) -> str:
"""
Returns whether the output of the indexed sequencer is connected to the
given output and if so with which path.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
output : int
Zero-based output index.
Returns
-------
str
- For baseband modules, one of:
- "off": the output is not connected.
- "I": the output is connected to path0/I.
- "Q": the output is connected to path1/Q.
- For RF modules, one of:
- "off": the RF output is not connected.
- "IQ": the RF output is connected.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
with ChannelMapCache(self, slot) as channel_map:
# Note that for RF modules, each connect_out parameter controls the
# configuration for two DACs, hardwired to the I and Q input of the
# RF mixer on the front end. It doesn't make sense to control the
# DACs individually in this case, and as such the user isn't given
# that level of control, but nevertheless the channel map state
# could hypothetically be in some weird in-between state. However,
# since we have to return something either way, just checking one
# of the paths should be good enough.
if self._is_rf_type(slot):
if channel_map.is_connected(ChannelType.AWG, sequencer, 0, output * 2):
return "IQ"
elif channel_map.is_connected(ChannelType.AWG, sequencer, 0, output):
return "I"
elif channel_map.is_connected(ChannelType.AWG, sequencer, 1, output):
return "Q"
return "off"
# ------------------------------------------------------------------------
def _set_sequencer_connect_acq(self, slot: int, sequencer: int, path: int, state: str) -> None:
"""
Set whether the input of the indexed sequencer's acquisition path is
connected to an external input and if so which.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
path : int
Path index: 0 for baseband path0/I, 1 for baseband path1/Q, ignored for RF.
state : str | bool
- One of:
- "off" or False: connection disabled.
- "in#": the acquisition input path is connected to external input #,
where # is a zero-based input index.
- True: if there is only one option other than off, True is allowed as alias.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
# Desugar state input.
if state is False:
input_ = None
elif state is True and self._is_rf_type(slot):
input_ = 0 # only 1 input
elif state == "off":
input_ = None
elif m := re.fullmatch(r"in(0|[1-9][0-9]*)", state):
input_ = int(m.group(1))
else:
raise ValueError(f"invalid new connection state {state!r}")
with ChannelMapCache(self, slot) as channel_map:
# Note that for RF modules, each connect_acq parameter controls the
# configuration for both paths of the acquisition engine, because
# each input maps to two ADCs.
if self._is_rf_type(slot):
channel_map.clear(ChannelType.ACQ, sequencer)
if input_ is not None:
channel_map.connect(ChannelType.ACQ, sequencer, 0, input_ * 2)
channel_map.connect(ChannelType.ACQ, sequencer, 1, input_ * 2 + 1)
else:
channel_map.clear_path(ChannelType.ACQ, sequencer, path)
if input_ is not None:
channel_map.connect(ChannelType.ACQ, sequencer, path, input_)
# -------------------------------------------------------------------------
def _get_sequencer_connect_acq(self, slot: int, sequencer: int, path: int) -> str:
"""
Get whether the input of the indexed sequencer's acquisition path is
connected to an external input and if so which.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
path : int
Path index: 0 for baseband path0/I, 1 for baseband path1/Q, ignored for RF.
Returns
-------
str
- One of:
- "off": connection disabled.
- "in#": the acquisition input path is connected to external input #,
where # is a zero-based input index.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
with ChannelMapCache(self, slot) as channel_map:
channels = list(channel_map.get_connected_channels(ChannelType.ACQ, sequencer, path))
if not channels:
return "off"
# If multiple inputs are connected to the same acquisition path in the
# channel map (an illegal configuration), do the same thing the firmware
# does, which is prioritizing lower-indexed channels because there is no
# good error path here.
channel = min(channels)
# Divide by two for RF modules to map from ADC channel to input, as there
# are two ADCs per input (I and Q). For baseband modules the mapping is
# one to one.
if self._is_rf_type(slot):
channel //= 2
return f"in{channel}"
# ----------------------------------------------------------------------------
def _get_output_latency(self, slot: int, output: int) -> float:
"""
Get the latency in output path. The output path can change depending on
the filter configuration of the output.
Parameters
----------
slot : int
The slot index of the module being referred to.
output: int
The output for which the latency should be returned.
Returns
-------
latency: float
Latency of the output path.
"""
self._present_at_init(slot)
return self._scpi._get_output_latency(slot, output)
# ----------------------------------------------------------------------------
def _set_pre_distortion_config(self, slot: int, config: dict) -> None:
"""
Set pre distortion configuration. The configuration consists
dictionary containing multiple parameters that will be converted into a
JSON object.
Parameters
----------
slot : int
The slot index of the module being referred to.
config : dict
Configuration dictionary.
"""
self._present_at_init(slot)
if self._config_cache_enabled:
self._config_cache_pre_distortion[slot] = deepcopy(config)
self._scpi._set_pre_distortion_config(slot, config)
# ----------------------------------------------------------------------------
def _get_pre_distortion_config(self, slot: int) -> dict:
"""
Get pre-distortion configuration. The configuration consists
dictionary containing multiple parameters that will be converted from a
JSON object.
Parameters
----------
slot : int
The slot index of the module being referred to.
Returns
-------
dict
Configuration dictionary.
"""
self._present_at_init(slot)
if self._config_cache_enabled and slot in self._config_cache_pre_distortion:
return deepcopy(self._config_cache_pre_distortion[slot])
result = self._scpi._get_pre_distortion_config(slot)
if self._config_cache_enabled:
self._config_cache_pre_distortion[slot] = deepcopy(result)
return result
# ------------------------------------------------------------------------
def _set_pre_distortion_config_val(self, slot: int, keys: Any, val: Any) -> None:
"""
Set value of specific pre-distortion filtering parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
val : Any
Value to set parameter to.
"""
self._present_at_init(slot)
set_generic_json_config_val(
lambda: self._get_pre_distortion_config(slot),
lambda cfg: self._set_pre_distortion_config(slot, cfg),
keys,
val,
)
# ------------------------------------------------------------------------
def _get_pre_distortion_config_val(self, slot: int, keys: Any) -> Any:
"""
Get value of specific pre-distortion filtering parameter.
Parameters
----------
slot : int
The slot index of the module being referred to.
keys : Union[list[str], str]
Configuration key to access, or hierarchical list thereof
Returns
-------
Any
Parameter value.
"""
self._present_at_init(slot)
return get_generic_json_config_val(
lambda: self._get_pre_distortion_config(slot),
keys,
)
# -------------------------------------------------------------------------
def _disconnect_outputs(self, slot: int) -> None:
"""
Disconnects all outputs from the sequencers.
Parameters
----------
slot : int
The slot index of the module being referred to.
"""
self._present_at_init(slot)
with ChannelMapCache(self, slot) as channel_map:
channel_map.clear(ChannelType.AWG)
# -------------------------------------------------------------------------
def _disconnect_inputs(self, slot: int) -> None:
"""
Disconnects all inputs from the sequencers.
Parameters
----------
slot : int
The slot index of the module being referred to.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
with ChannelMapCache(self, slot) as channel_map:
channel_map.clear(ChannelType.ACQ)
# -------------------------------------------------------------------------
def _iter_connections(self, slot: int) -> Iterator[tuple[int, str, str]]:
"""
Iterates over all enabled connections between ADCs, DACs, and
sequencers.
Parameters
----------
slot : int
The slot index of the module being referred to.
Returns
-------
Iterator[tuple[int, str, str]]
An iterator of connections. The four components of each connection are:
- the index of the sequencer for the connection;
- the connection point of the sequencer being connected to,
being one of `I`, `Q`, `acq_I`, or `acq_Q`;
- the external connection, being either `adc#` or `dac#`,
where `#` is the zero-based ADC or DAC index.
Note that these are ADC and DAC indices. For baseband modules,
these indices map one-to-one to the external SMA ports, but for RF
modules they don't: each pair of DACs or ADCs maps to a single RF
port, the I component being generated by ADC/DAC index 0/2/... and
the Q component being generated by ADC/DAC index 1/3/...
"""
self._present_at_init(slot)
return ChannelMapCache(self, slot).iter_connections()
# -------------------------------------------------------------------------
def _sequencer_connect(
self, slot: int, sequencer: int, *connections: ValidatedChannelConnection
) -> 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
The slot index of the module being referred to.
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.
"""
self._present_at_init(slot)
# Intentionally don't use the context manager: in case of an exception,
# do not make *any* changes.
channel_map = ChannelMapCache(self, slot)
for index, conn in enumerate(connections):
try:
# Try to apply the changes.
for direction in conn.directions:
channel_map.connect(direction, sequencer, 0, conn.i_channel, False)
if conn.q_channel is not None:
channel_map.connect(direction, sequencer, 1, conn.q_channel, False)
except (RuntimeError, ValueError) as e: # noqa: PERF203
raise e.__class__(f"connection command {conn!r} (index {index}): {e}")
# Everything seems to have worked: write new configuration to the
# instrument.
channel_map.flush()
# ------------------------------------------------------------------------
async def _arm_sequencer_async(
self, slot: Optional[int] = None, sequencer: Optional[int] = None
) -> None:
"""
Prepare the indexed sequencer to start by putting it in the armed state.
If no sequencer index is given, all sequencers are armed. Any sequencer
that was already running is stopped and rearmed. If an invalid sequencer
index is given, an error is set in system error.
Parameters
----------
slot : Optional[int]
The slot index of the module being referred to.
sequencer : Optional[int]
Sequencer index.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
if slot is not None:
self._present_at_init(slot)
slots = [slot]
else:
slots = [
mod.slot_idx for mod in self.modules if self._scpi._slot_present(mod.slot_idx)
] # Arm sequencers across all modules
if sequencer is not None:
check_sequencer_index(sequencer)
else:
sequencer = "" # Arm all sequencers within a module
if slot is not None and sequencer != "":
return await self._scpi._arm_sequencer_async(slot, sequencer)
scpi_cmd_prefix = f"SEQuencer{sequencer}"
# The SCPI command prefix is set by the native instrument layer so that
# it can select to arm a specific sequencer (e.g. "SLOT1:SEQuencer0") or
# all sequencers (e.g. "SLOT:SEQuencer")
# The actual SCPI call is wrapped in a function to make use of the
# scpi_error_check method.
@slot_error_check_async(minimal_check=True)
async def arm_sequencer_func(instrument: Any, slot: int) -> None:
async with instrument._slot_connection(slot) as slot_conn:
await slot_conn._write_async(f"{scpi_cmd_prefix}:ARM")
await asyncio.gather(*(arm_sequencer_func(self, slot) for slot in slots))
arm_sequencer = make_sync_wrapper(_arm_sequencer_async)
# ------------------------------------------------------------------------
async def _arm_sequencers_async(self, sequencers: list[tuple[int, int]]) -> None:
"""
Prepare the indexed sequencers to start by putting them in the armed state.
Any sequencer that was already running is stopped and rearmed.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to arm.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
checked_slots.add(slot)
check_sequencer_index(sequencer)
await asyncio.gather(
*(self._scpi._arm_sequencer_async(slot, sequencer) for slot, sequencer in sequencers)
)
arm_sequencers = make_sync_wrapper(_arm_sequencers_async)
# ------------------------------------------------------------------------
async def _start_sequencer_async(
self, slot: Optional[int] = None, sequencer: Optional[int] = None
) -> None:
"""
Start the indexed sequencer, thereby putting it in the running state.
If an invalid sequencer index is given or the indexed sequencer was not
yet armed, an error is set in system error. If no sequencer index is
given, all armed sequencers are started and any sequencer not in the armed
state is ignored. However, if no sequencer index is given and no
sequencers are armed, and error is set in system error.
Parameters
----------
slot : Optional[int]
The slot index of the module being referred to.
sequencer : Optional[int]
Sequencer index.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
if slot is not None:
self._present_at_init(slot)
slots = [slot]
else:
slots = [
mod.slot_idx for mod in self.modules if self._scpi._slot_present(mod.slot_idx)
] # Start sequencers across all modules
if sequencer is not None:
check_sequencer_index(sequencer)
else:
sequencer = "" # Start all sequencers within a module
if slot is not None and sequencer != "":
return await self._scpi._start_sequencer_async(slot, sequencer)
scpi_cmd_prefix = f"SEQuencer{sequencer}"
# The SCPI command prefix is set by the native instrument layer so that
# it can select to start a specific sequencer (e.g. "SLOT1:SEQuencer0") or
# all sequencers (e.g. "SLOT:SEQuencer")
# The actual SCPI call is wrapped in a function to make use of the
# scpi_error_check method.
@slot_error_check_async(minimal_check=True)
async def start_sequencer_func(instrument: Any, slot: int) -> None:
async with instrument._slot_connection(slot) as slot_conn:
await slot_conn._write_async(f"{scpi_cmd_prefix}:START")
await asyncio.gather(*(start_sequencer_func(self, slot) for slot in slots))
start_sequencer = make_sync_wrapper(_start_sequencer_async)
# ------------------------------------------------------------------------
async def _start_sequencers_async(self, sequencers: list[tuple[int, int]]) -> None:
"""
Start the indexed sequencers, thereby putting them in the running state.
If an invalid sequencer index is given or the indexed sequencer was not
yet armed, an error is set in system error.
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to start.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
checked_slots.add(slot)
check_sequencer_index(sequencer)
await asyncio.gather(
*(self._scpi._start_sequencer_async(slot, sequencer) for slot, sequencer in sequencers)
)
start_sequencers = make_sync_wrapper(_start_sequencers_async)
# ------------------------------------------------------------------------
async def _stop_sequencer_async(
self, slot: Optional[int] = None, sequencer: Optional[int] = None
) -> None:
"""
Stop the indexed sequencer, thereby putting it in the stopped state. If
an invalid sequencer index is given, an error is set in system error. If
no sequencer index is given, all sequencers are stopped.
Parameters
----------
slot : Optional[int]
The slot index of the module being referred to.
sequencer : Optional[int]
Sequencer index.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
if slot is not None:
self._present_at_init(slot)
slots = [slot]
else:
slots = [
mod.slot_idx for mod in self.modules if self._scpi._slot_present(mod.slot_idx)
] # Stop sequencers across all modules
if sequencer is not None:
check_sequencer_index(sequencer)
else:
sequencer = "" # Stop all sequencers within a module
if slot is not None and sequencer != "":
return await self._scpi._stop_sequencer_async(slot, sequencer)
scpi_cmd_prefix = f"SEQuencer{sequencer}"
# The SCPI command prefix is set by the native instrument layer so that
# it can select to stop a specific sequencer (e.g. "SLOT1:SEQuencer0") or
# all sequencers (e.g. "SLOT:SEQuencer")
# The actual SCPI call is wrapped in a function to make use of the
# scpi_error_check method.
@slot_error_check_async(minimal_check=True)
async def stop_sequencer_func(instrument: Any, slot: int) -> None:
async with instrument._slot_connection(slot) as slot_conn:
await slot_conn._write_async(f"{scpi_cmd_prefix}:STOP")
await asyncio.gather(*(stop_sequencer_func(self, slot) for slot in slots))
stop_sequencer = make_sync_wrapper(_stop_sequencer_async)
# ------------------------------------------------------------------------
async def _stop_sequencers_async(self, sequencers: list[tuple[int, int]]) -> None:
"""
Stop the indexed sequencers, thereby putting them in the stopped state.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to stop.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
checked_slots.add(slot)
check_sequencer_index(sequencer)
await asyncio.gather(
*(self._scpi._stop_sequencer_async(slot, sequencer) for slot, sequencer in sequencers)
)
stop_sequencers = make_sync_wrapper(_stop_sequencers_async)
# ------------------------------------------------------------------------
async def _clear_sequencer_flags_async(
self, slot: Optional[int] = None, sequencer: Optional[int] = None
) -> None:
"""
Clear flags.
Parameters
----------
slot : Optional[int]
The slot index of the module being referred to.
sequencer : int
Sequencer index.
"""
if slot is not None:
self._present_at_init(slot)
else:
slot = "" # Clear sequencer flags across all modules
if sequencer is not None:
check_sequencer_index(sequencer)
else:
sequencer = "" # Clear all sequencers flags within a module
if slot != "" and sequencer != "":
return await self._scpi._clear_sequencer_flags_async(slot, sequencer)
scpi_cmd_prefix = f"SLOT{slot}:SEQuencer{sequencer}"
# The SCPI command prefix is set by the native instrument layer so that
# it can select to clear a specific sequencer flag (e.g. "SLOT1:SEQuencer0") or
# all sequencers (e.g. "SLOT:SEQuencer")
# The actual SCPI call is wrapped in a function to make use of the
# scpi_error_check method.
@scpi_error_check
def clear_sequencer_flags_func(instrument: Any) -> None:
instrument._write(f"{scpi_cmd_prefix}:CLR:FLAGS")
clear_sequencer_flags_func(self)
clear_sequencer_flags = make_sync_wrapper(_clear_sequencer_flags_async)
# ------------------------------------------------------------------------
async def _get_sequencer_status_async(
self,
slot: int,
sequencer: int,
timeout: int = 0,
timeout_poll_res: float = 0.02,
) -> SequencerStatus:
"""
Get the sequencer status. If an invalid sequencer index is given, an error
is set in system error. If the timeout is set to zero, the function
returns the state immediately. If a positive non-zero timeout is set, the
function blocks until the sequencer completes. If the sequencer hasn't
stopped before the timeout expires, a TimeoutError is thrown.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
timeout : int
Timeout in minutes.
timeout_poll_res : float
Timeout polling resolution in seconds.
Returns
-------
SequencerStatus
Tuple containing sequencer status and corresponding flags.
Raises
------
TimeoutError
Timeout
"""
self._present_at_init(slot)
# Format status string
check_sequencer_index(sequencer)
full_status = await self._scpi._get_sequencer_state_async(slot, sequencer)
state_tuple = SequencerStatus.from_scpi_str(full_status)
elapsed_time = 0.0
start_time = time.time()
timeout = timeout * 60.0
while (
state_tuple.state in (SequencerStates.RUNNING, SequencerStates.Q1_STOPPED)
and elapsed_time < timeout
):
await asyncio.sleep(timeout_poll_res)
state_tuple = await self._get_sequencer_status_async(slot, sequencer)
elapsed_time = time.time() - start_time
if elapsed_time >= timeout:
raise TimeoutError(
f"Sequencer {sequencer} did not stop in timeout period "
f"of {int(timeout / 60)} minutes."
)
return state_tuple
get_sequencer_status = make_sync_wrapper(_get_sequencer_status_async)
# ------------------------------------------------------------------------
async def _get_sequencer_statuses_async(
self, sequencers: list[tuple[int, int]]
) -> list[SequencerStatus]:
"""
Get the status for multiple sequencers.
If an invalid sequencer index is given, an error is set in system error.
The status is returned immediately.
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to get status of.
Returns
-------
list[SequencerStatus[]
Tuple containing sequencer status and corresponding flags.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
checked_slots.add(slot)
check_sequencer_index(sequencer)
scpi_states = await asyncio.gather(
*(
self._scpi._get_sequencer_state_async(slot, sequencer)
for slot, sequencer in sequencers
)
)
return [SequencerStatus.from_scpi_str(s) for s in scpi_states]
get_sequencer_statuses = make_sync_wrapper(_get_sequencer_statuses_async)
# ------------------------------------------------------------------------
async def _wait_for_sequencers_async(
self, sequencers: list[tuple[int, int]], timeout: int, timeout_poll_res: float = 0.02
) -> list[SequencerStatus]:
"""
Wait for multiple sequencers to finish and return their respective statuses.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to stop.
Returns
-------
list[SequencerStatus[]
Tuple containing sequencer status and corresponding flags.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
checked_slots.add(slot)
check_sequencer_index(sequencer)
remaining_sequencers = sequencers[:]
end = time.time() + timeout * 60.0
statuses = {}
while remaining_sequencers and time.time() < end:
# Don't sleep the first time
if statuses:
await asyncio.sleep(timeout_poll_res)
# Wait for response up until our timeout
try:
remaining_statuses = await asyncio.wait_for(
self._get_sequencer_statuses_async(remaining_sequencers),
end - time.time(),
)
except TimeoutError:
break
# Remove sequencers which have finished from our remaining list
for (slot, sequencer), status in zip(remaining_sequencers[:], remaining_statuses):
statuses[slot, sequencer] = status
if status.state not in (SequencerStates.RUNNING, SequencerStates.Q1_STOPPED):
remaining_sequencers.remove((slot, sequencer))
# Did we reach the end while still having sequencers running?
if remaining_sequencers:
remaining_str = ", ".join(
f"slot {slot} sequencer {sequencer}" for slot, sequencer in remaining_sequencers
)
raise TimeoutError(
f"Sequencers {remaining_str} did not stop in timeout period of {timeout} minutes."
)
else:
return [statuses[slot, sequencer] for slot, sequencer in sequencers]
wait_for_sequencers = make_sync_wrapper(_wait_for_sequencers_async)
# ----------------------------------------------------------------------------
async def _get_acq_acquisitions_async(
self,
slot: int,
sequencer: int,
*,
as_numpy: bool = False,
acq_encoding: AcquisitionEncodingConfig = AcquisitionEncodingConfig.legacy_encoding(),
) -> dict:
"""
Get all acquisitions in acquisition list of indexed sequencer's
acquisition path. If an invalid sequencer index is given, an error is set
in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return path, threshold and avg_cnt data as NumPy arrays
instead of Python lists.
Returns
-------
dict
Dictionary with acquisitions.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
self._present_at_init(slot)
# SCPI call
is_data_encoded = acq_encoding is not None
if is_data_encoded and not self._can_encode_partial_acquisitions(slot):
raise ValueError("partial acquisition is not supported")
async with self._scpi._slot_connection(slot) as slot_conn:
if is_data_encoded:
enc_flag = acq_encoding.to_bit_flag()
command = f"SEQuencer{sequencer}:ACQ:PARTial:ALISt? {enc_flag}"
else:
command = f"SEQuencer{sequencer}:ACQ:ALISt?"
acq_list = await slot_conn._read_bin_async(command, False)
num_acq = struct.unpack("I", acq_list)[0]
if num_acq == 0:
await slot_conn._flush_line_end_async()
acquisition_dict = {}
for acq_it in range(0, num_acq):
# Get name and index
name = str(await slot_conn._read_bin_async("", False), "utf-8")
index = struct.unpack("I", await slot_conn._read_bin_async("", False))[0]
# Get data
if self._is_qtm_type(slot):
acq = await self._get_acq_data_async(
slot,
"",
acq_it >= (num_acq - 1),
slot_conn,
as_numpy=as_numpy,
is_data_encoded=is_data_encoded,
)
else:
acq = await self._get_acq_data_and_convert_async(
slot,
"",
acq_it >= (num_acq - 1),
slot_conn,
as_numpy=as_numpy,
is_data_encoded=is_data_encoded,
)
# Add to dictionary
acquisition_dict[name] = {"index": index, "acquisition": acq}
return acquisition_dict
_get_acq_acquisitions = make_sync_wrapper(_get_acq_acquisitions_async)
# ----------------------------------------------------------------------------
async def _get_acq_data_and_convert_async(
self,
slot: int,
init_command: str,
flush_line_end: bool,
connection: Ieee488_2,
*,
as_numpy: bool = False,
is_data_encoded: bool = False,
) -> dict:
"""
Get acquisition data and convert it to a dictionary.
Parameters
----------
slot : int
The slot index of the module being referred to.
init_command : Callable[[Optional[int], Optional[str]], bytes]
Command that performs the initial binary read.
flush_line_end : bool
Indication to flush final characters after final read.
Returns
-------
dict
Dictionary with data of single acquisition.
"""
acquisition_dict = {
"scope": {
"path0": {"data": [], "out-of-range": False, "avg_cnt": 0},
"path1": {"data": [], "out-of-range": False, "avg_cnt": 0},
},
"bins": {},
}
# QRC-related changes
if self._is_qrc_type(slot):
acquisition_dict["scope"].update(
{"path2": {"data": [], "out-of-range": False, "avg_cnt": 0}}
)
acquisition_dict["scope"].update(
{"path3": {"data": [], "out-of-range": False, "avg_cnt": 0}}
)
sample_width_scope = (
SCOPE_ACQUISITION_BIT_WIDTH_QRC
if self._is_qrc_type(slot)
else SCOPE_ACQUISITION_BIT_WIDTH_QRM
)
max_sample_value_scope = 2 ** (sample_width_scope - 1)
sample_width_binned = BINNED_ACQUISITION_BIT_WIDTH_QRM
max_sample_value_binned = 2 ** (sample_width_binned - 1)
binned_acquisition_scaling = (
BINNED_ACQUISITION_SCALING_QRC if self._is_qrc_type(slot) else 1
)
# Retrieve scope data
if self._is_qrc_type(slot):
(
path0_scope_raw,
path0_avg_cnt,
path0_oor,
path1_scope_raw,
path1_avg_cnt,
path1_oor,
path2_scope_raw,
path2_avg_cnt,
path2_oor,
path3_scope_raw,
path3_avg_cnt,
path3_oor,
) = await self._read_acquisition_raw_data_async(slot, init_command, connection)
else:
(
path0_scope_raw,
path0_avg_cnt,
path0_oor,
path1_scope_raw,
path1_avg_cnt,
path1_oor,
) = await self._read_acquisition_raw_data_async(slot, init_command, connection)
# Normalize scope data (Ignore division by 0)
with numpy.errstate(divide="ignore", invalid="ignore"):
path0_scope = numpy.where(
path0_avg_cnt > 0,
path0_scope_raw / max_sample_value_scope / path0_avg_cnt,
path0_scope_raw / max_sample_value_scope,
)
path1_scope = numpy.where(
path1_avg_cnt > 0,
path1_scope_raw / max_sample_value_scope / path1_avg_cnt,
path1_scope_raw / max_sample_value_scope,
)
if self._is_qrc_type(slot):
path2_scope = numpy.where(
path2_avg_cnt > 0,
path2_scope_raw / max_sample_value_scope / path2_avg_cnt,
path2_scope_raw / max_sample_value_scope,
)
path3_scope = numpy.where(
path3_avg_cnt > 0,
path3_scope_raw / max_sample_value_scope / path3_avg_cnt,
path3_scope_raw / max_sample_value_scope,
)
# Retrieve bin data and convert to long values
read_bin_function = (
self._read_bin_encoded_data_async if is_data_encoded else self._read_bin_raw_data_async
)
(
path0_raw,
path1_raw,
valid_raw,
avg_count_raw,
threshold_raw,
) = await read_bin_function(flush_line_end, connection)
# Specific data manipulation for QRM
path0_data = (
numpy.where(
valid_raw,
binned_acquisition_scaling * path0_raw / max_sample_value_binned**2,
numpy.nan,
)
if path0_raw is not None
else None
)
path1_data = (
numpy.where(
valid_raw,
binned_acquisition_scaling * path1_raw / max_sample_value_binned**2,
numpy.nan,
)
if path1_raw is not None
else None
)
thres_data = (
numpy.where(valid_raw, threshold_raw, numpy.nan) if threshold_raw is not None else None
)
avg_cnt_data = (
numpy.where(valid_raw, avg_count_raw, 0) if avg_count_raw is not None else None
)
# Set final results
acquisition_dict["scope"]["path0"]["data"] = (
path0_scope if as_numpy else path0_scope.tolist()
)
acquisition_dict["scope"]["path0"]["out-of-range"] = path0_oor
acquisition_dict["scope"]["path0"]["avg_cnt"] = path0_avg_cnt
acquisition_dict["scope"]["path1"]["data"] = (
path1_scope if as_numpy else path1_scope.tolist()
)
acquisition_dict["scope"]["path1"]["out-of-range"] = path1_oor
acquisition_dict["scope"]["path1"]["avg_cnt"] = path1_avg_cnt
if self._is_qrc_type(slot):
acquisition_dict["scope"]["path2"]["data"] = (
path2_scope if as_numpy else path2_scope.tolist()
)
acquisition_dict["scope"]["path2"]["out-of-range"] = path2_oor
acquisition_dict["scope"]["path2"]["avg_cnt"] = path2_avg_cnt
acquisition_dict["scope"]["path3"]["data"] = (
path3_scope if as_numpy else path3_scope.tolist()
)
acquisition_dict["scope"]["path3"]["out-of-range"] = path3_oor
acquisition_dict["scope"]["path3"]["avg_cnt"] = path3_avg_cnt
if path0_data is not None:
acquisition_dict["bins"].setdefault("integration", {})["path0"] = (
path0_data if as_numpy else path0_data.tolist()
)
if path1_data is not None:
acquisition_dict["bins"].setdefault("integration", {})["path1"] = (
path1_data if as_numpy else path1_data.tolist()
)
if thres_data is not None:
acquisition_dict["bins"]["threshold"] = thres_data if as_numpy else thres_data.tolist()
if avg_cnt_data is not None:
acquisition_dict["bins"]["avg_cnt"] = (
avg_cnt_data if as_numpy else avg_cnt_data.tolist()
)
return acquisition_dict
_get_acq_data_and_convert = make_sync_wrapper(_get_acq_data_and_convert_async)
# ----------------------------------------------------------------------------
# Next two functions are meant to be used only inside _get_acq_data_* functions
# because they are tied together of how firmware sends raw scope and bin data
# QTM fix end
# ----------------------------------------------------------------------------
async def _read_bin_raw_data_async(self, flush_line_end, connection: Ieee488_2 = None) -> tuple:
bins = await connection._read_bin_async("", flush_line_end)
packet_layout = numpy.dtype(
[
("valid", numpy.uint64),
("path0", numpy.int64),
("path1", numpy.int64),
("thres_raw", numpy.uint32),
("avg_cnt", numpy.uint32),
]
)
bin_data = numpy.frombuffer(bins, dtype=packet_layout)
# This prevents any copies from being made while
# still able to interpret the data normally
valid = bin_data["valid"]
path0_raw = bin_data["path0"]
path1_raw = bin_data["path1"]
thres_raw = bin_data["thres_raw"]
avg_cnt = bin_data["avg_cnt"]
# Normalize bin data (Ignore division by 0)
with numpy.errstate(divide="ignore", invalid="ignore"):
# numpy.where creates copies of the array, which is not desired.
# No alternatives that edit the array in-place exist as far as I know
path0_raw = numpy.where(avg_cnt > 0, path0_raw / avg_cnt, path0_raw)
path1_raw = numpy.where(avg_cnt > 0, path1_raw / avg_cnt, path1_raw)
thres_raw = numpy.where(avg_cnt > 0, thres_raw / avg_cnt, thres_raw)
return path0_raw, path1_raw, valid, avg_cnt, thres_raw
_read_bin_raw_data = make_sync_wrapper(_read_bin_raw_data_async)
# ----------------------------------------------------------------------------
async def _read_bin_encoded_data_async(
self,
flush_line_end: bool,
connection: Ieee488_2Connection,
) -> tuple:
# read encoding format for binary data
enc_flag_raw = await connection._read_bin_async("", False)
enc_bit_flag = int.from_bytes(enc_flag_raw[0:4], byteorder="little")
enc_config: AcquisitionEncodingConfig = AcquisitionEncodingConfig.from_bit_flag(
enc_bit_flag
)
samples_n_raw = await connection._read_bin_async("", False)
samples_n: int = int.from_bytes(samples_n_raw[0:4], byteorder="little", signed=False)
avg_count_skip = enc_config.avg_count_fitting == AverageCountDataFormat.SKIP
iq_skip = enc_config.iq_fitting == IQDataFormat.SKIP
threshold_skip = enc_config.threshold_fitting == ThresholdDataFormat.SKIP
# read valid array, if actually encoded
match enc_config.valid_fitting:
case ValidDataFormat.SINGLE_BIT:
should_flush = False
if iq_skip and avg_count_skip and threshold_skip:
should_flush = flush_line_end
valid_raw = await connection._read_bin_async("", should_flush)
# read path samples
if not iq_skip:
path_0_raw = await connection._read_bin_async("", False)
# if no threshold nor count is present, then the data ends here: flush
should_flush = avg_count_skip and threshold_skip and flush_line_end
path_1_raw = await connection._read_bin_async("", should_flush)
# read threshold bits, if actually encoded. If no avg_count, data ends here: flush
should_flush = avg_count_skip and flush_line_end
if not threshold_skip:
threshold_raw = await connection._read_bin_async("", should_flush)
# only average counts, if actually encoded
if not avg_count_skip:
average_count_raw = await connection._read_bin_async("", flush_line_end)
# decode data
data_type = None
match enc_config.iq_fitting:
case IQDataFormat.I8:
data_type = numpy.int8
case IQDataFormat.I16:
data_type = numpy.int16
case IQDataFormat.I32:
data_type = numpy.int32
case IQDataFormat.I64:
data_type = numpy.int64
case IQDataFormat.AVG_F32:
# average has been computed by EAPP, read float directly
data_type = numpy.float32
if data_type is not None:
path_0 = numpy.frombuffer(path_0_raw, dtype=data_type)
path_1 = numpy.frombuffer(path_1_raw, dtype=data_type)
else:
path_0 = None
path_1 = None
# read or compute validity
match enc_config.valid_fitting:
case ValidDataFormat.SINGLE_BIT:
valid_array = numpy.frombuffer(valid_raw, dtype=numpy.uint8)
bits = numpy.unpackbits(valid_array, bitorder="little")
valid = bits[:samples_n].astype(numpy.uint64)
# read normal thresholds if encoded, else generate an empty array
match enc_config.threshold_fitting:
case ThresholdDataFormat.SKIP:
threshold = None
case ThresholdDataFormat.U8:
threshold = numpy.frombuffer(threshold_raw, dtype=numpy.uint8)
case ThresholdDataFormat.U16:
threshold = numpy.frombuffer(threshold_raw, dtype=numpy.uint16)
case ThresholdDataFormat.U32:
threshold = numpy.frombuffer(threshold_raw, dtype=numpy.uint32)
case ThresholdDataFormat.AVG_F32:
threshold = numpy.frombuffer(threshold_raw, dtype=numpy.float32)
# read average_count if encoded, else generate an empty array
match enc_config.avg_count_fitting:
case AverageCountDataFormat.SKIP:
average_count = None
case AverageCountDataFormat.U8:
average_count = numpy.frombuffer(average_count_raw, dtype=numpy.uint8)
case AverageCountDataFormat.U16:
average_count = numpy.frombuffer(average_count_raw, dtype=numpy.uint16)
case AverageCountDataFormat.U32:
average_count = numpy.frombuffer(average_count_raw, dtype=numpy.uint32)
# normalize iq data if the computation has not been done by EAPP already
has_average = enc_config.avg_count_fitting != AverageCountDataFormat.SKIP
has_iq_data = data_type is not None
needs_iq_averaging = (
has_iq_data and (enc_config.iq_fitting != IQDataFormat.AVG_F32) and has_average
)
needs_threshold_averaging = (
enc_config.threshold_fitting
not in {ThresholdDataFormat.AVG_F32, ThresholdDataFormat.SKIP}
and has_average
)
with numpy.errstate(divide="ignore", invalid="ignore"):
if needs_iq_averaging:
path_0 = numpy.where(average_count > 0, path_0 / average_count, path_0)
path_1 = numpy.where(average_count > 0, path_1 / average_count, path_1)
if needs_threshold_averaging:
threshold = numpy.where(average_count > 0, threshold / average_count, threshold)
return path_0, path_1, valid, average_count, threshold
_read_bin_encoded_data = make_sync_wrapper(_read_bin_encoded_data_async)
# ----------------------------------------------------------------------------
async def _read_acquisition_raw_data_async(
self, slot, init_command, connection: Ieee488_2
) -> tuple:
async def _retrieve_scope_data(init: bool = False) -> tuple:
scope_data = await connection._read_bin_async(init_command if init else "", False)
path_scope_raw = numpy.frombuffer(scope_data, dtype=numpy.int32)
path_oor = struct.unpack("?", await connection._read_bin_async("", False))[0]
path_avg_cnt = struct.unpack("I", await connection._read_bin_async("", False))[0]
return path_scope_raw, path_oor, path_avg_cnt
# Retrieve scope data
path0_scope_raw, path0_oor, path0_avg_cnt = await _retrieve_scope_data(init=True)
path1_scope_raw, path1_oor, path1_avg_cnt = await _retrieve_scope_data()
if self._is_qrc_type(slot):
path2_scope_raw, path2_oor, path2_avg_cnt = await _retrieve_scope_data()
path3_scope_raw, path3_oor, path3_avg_cnt = await _retrieve_scope_data()
return (
(
path0_scope_raw,
path0_avg_cnt,
path0_oor,
path1_scope_raw,
path1_avg_cnt,
path1_oor,
path2_scope_raw,
path2_avg_cnt,
path2_oor,
path3_scope_raw,
path3_avg_cnt,
path3_oor,
)
if self._is_qrc_type(slot)
else (
path0_scope_raw,
path0_avg_cnt,
path0_oor,
path1_scope_raw,
path1_avg_cnt,
path1_oor,
)
)
_read_acquisition_raw_data = make_sync_wrapper(_read_acquisition_raw_data_async)
# ----------------------------------------------------------------------------
# QTM fix end
# ----------------------------------------------------------------------------
async def _get_acq_data_async(
self,
slot: int,
init_command: str,
flush_line_end: bool,
connection: Ieee488_2,
*,
as_numpy: bool = False,
is_data_encoded: bool = False,
) -> dict:
"""
Get acquisition data and convert it to a dictionary.
Parameters
----------
init_command : Callable[[Optional[int], Optional[str]], bytes]
Command that performs the initial binary read.
flush_line_end : bool
Indication to flush final characters after final read.
Returns
-------
dict
Dictionary with data of single acquisition.
"""
# TODO this is only needed here because of the consecutive calls to _read_bin
# which retrieves data from the socket.
# So scope_data is not used all in this function
(
_path0_scope_raw,
_path0_avg_cnt,
_path0_oor,
_path1_scope_raw,
_path1_avg_cnt,
_path1_oor,
) = await self._read_acquisition_raw_data_async(slot, init_command, connection)
# Retrieve bin data and convert to long values
read_bin_function = (
self._read_bin_encoded_data_async if is_data_encoded else self._read_bin_raw_data_async
)
path0_raw, path1_raw, valid, avg_cnt, thres_raw = await read_bin_function(
flush_line_end, connection
)
bins = {}
# Specific data manipulation for QTM
if path0_raw is not None:
path0_data = numpy.where(valid, path0_raw, numpy.nan)
bins["count"] = path0_data if as_numpy else path0_data.tolist()
if path1_raw is not None:
path1_data = numpy.where(valid, path1_raw, numpy.nan)
bins["timedelta"] = path1_data if as_numpy else path1_data.tolist()
if thres_raw is not None:
thres_data = numpy.where(valid, thres_raw, numpy.nan)
bins["threshold"] = thres_data if as_numpy else thres_data.tolist()
if avg_cnt is not None:
avg_cnt_data = numpy.where(valid, avg_cnt, 0)
bins["avg_cnt"] = avg_cnt_data if as_numpy else avg_cnt_data.tolist()
acquisition_dict = {"bins": bins}
return acquisition_dict
_get_acq_data = make_sync_wrapper(_get_acq_data_async)
# ------------------------------------------------------------------------
async def _add_waveforms_async(self, slot: int, sequencer: int, waveforms: dict) -> None:
"""
Add all waveforms in JSON compatible dictionary to the AWG waveform list
of indexed sequencer.
The dictionary must be structured as follows:
- name: waveform name.
- data: waveform samples in a range of 1.0 to -1.0.
- index: optional waveform index used by the sequencer Q1ASM program
to refer to the waveform.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
waveforms : dict
JSON compatible dictionary with one or more waveforms and weights.
Raises
------
KeyError
Missing waveform data of waveform in dictionary.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
for name, waveform in waveforms.items():
if "data" in waveform:
if "index" in waveform:
await self._add_awg_waveform_async(
slot,
sequencer,
name,
waveform["data"],
waveform["index"],
)
else:
await self._add_awg_waveform_async(slot, sequencer, name, waveform["data"])
else:
raise KeyError(f"Missing data key for {name} in AWG waveform dictionary")
_add_waveforms = make_sync_wrapper(_add_waveforms_async)
# ------------------------------------------------------------------------
def _detect_encoding(self, slot: int) -> SampleEncoding:
supported_features = self._mod_handles[slot]["features"]
if supported_features is None or "wave_encoding" not in supported_features:
return SampleEncoding.UNSUPPORTED
supports_i16_encoding = SampleEncoding.I16 in supported_features["wave_encoding"]
waveform_encoding = SampleEncoding.I16 if supports_i16_encoding else SampleEncoding.F32
return waveform_encoding
# ------------------------------------------------------------------------
async def _add_awg_waveform_async(
self,
slot: int,
sequencer: int,
name: str,
waveform: list[float],
index: Optional[int] = None,
) -> None:
"""
Add new waveform to AWG waveform list of indexed sequencer's AWG path. If
an invalid sequencer index is given or if the waveform causes the waveform
memory limit to be exceeded or if the waveform samples are out-of-range,
an error is set in the system error. The waveform names 'all' and 'ALL'
are reserved and adding waveforms with those names will also result in an
error being set in system error. The optional index argument is used to
specify an index for the waveform in the waveform list which is used by
the sequencer Q1ASM program to refer to the waveform. If no index is
given, the next available waveform index is selected (starting from 0).
If an invalid waveform index is given, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Waveform name.
waveform : list
List of floats in the range of 1.0 to -1.0 representing the waveform.
index : Optional[int]
Waveform index of the waveform in the waveform list.
"""
waveform_encoding = self._detect_encoding(slot)
if waveform_encoding == SampleEncoding.UNSUPPORTED:
# fall back to legacy API
await self._scpi._add_awg_waveform_async(slot, sequencer, name, len(waveform), False)
await self._scpi._set_awg_waveform_data_async(slot, sequencer, name, waveform)
else:
# use waveform encoding API
waveform_bytes = pack_samples(waveform_encoding, waveform)
await self._scpi._add_awg_waveform_encoded_async(
slot, sequencer, name, len(waveform), int(waveform_encoding)
)
await self._scpi._set_awg_waveform_packed_bytes_async(
slot, sequencer, name, waveform_bytes
)
if index is not None:
await self._scpi._set_awg_waveform_index_async(slot, sequencer, name, index)
_add_awg_waveform = make_sync_wrapper(_add_awg_waveform_async)
# ------------------------------------------------------------------------
async def _delete_waveform_async(
self,
slot: int,
sequencer: int,
name: str = "",
*,
delete_all: bool = False,
) -> None:
"""
Delete a waveform specified by name in the AWG waveform list of indexed
sequencer or delete all waveforms if `all` is True.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Waveform name
delete_all : bool
All waveforms
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
await self._scpi._delete_awg_waveform_async(slot, sequencer, "all" if delete_all else name)
_delete_waveform = make_sync_wrapper(_delete_waveform_async)
# ----------------------------------------------------------------------------
async def _get_awg_waveforms_async(
self, slot: int, sequencer: int, *, as_numpy: bool = False
) -> dict:
"""
Get all waveforms in waveform list of indexed sequencer's AWG path. If an
invalid sequencer index is given, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return data as NumPy arrays instead of Python lists.
Returns
-------
dict
Dictionary with waveforms.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
# SCPI call
async with self._scpi._slot_connection(slot) as slot_conn:
awg_wlist = await slot_conn._read_bin_async(f"SEQuencer{sequencer}:AWG:WLISt?", False)
num_waveforms = struct.unpack("I", awg_wlist)[0]
if num_waveforms == 0:
await slot_conn._flush_line_end_async()
waveform_dict = {}
for wave_it in range(0, num_waveforms):
# Get name and index
name = str(await slot_conn._read_bin_async("", False), "utf-8")
index = struct.unpack("I", await slot_conn._read_bin_async("", False))[0]
# Get data
raw_bytes = await slot_conn._read_bin_async("", wave_it >= (num_waveforms - 1))
data = numpy.frombuffer(raw_bytes, dtype=numpy.float32)
# Add to dictionary
waveform_dict[name] = {"index": index, "data": data if as_numpy else data.tolist()}
return waveform_dict
_get_awg_waveforms = make_sync_wrapper(_get_awg_waveforms_async)
# ----------------------------------------------------------------------------
async def _add_acq_weight_async(
self,
slot: int,
sequencer: int,
name: str,
weight: list[float],
index: Optional[int] = None,
) -> None:
"""
Add new weight to acquisition weight list of indexed sequencer's
acquisition path. If an invalid sequencer index is given or if the weight
causes the weight memory limit to be exceeded or if the weight samples are
out-of-range, an error is set in the system error. The weight names 'all'
and 'ALL' are reserved and adding weights with those names will also
result in an error being set in system error. The optional index argument
is used to specify an index for the weight in the weight list which is
used by the sequencer Q1ASM program to refer to the weight. If no index
is given, the next available weight index is selected (starting from 0).
If an invalid weight index is given, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Weight name.
weight : list
List of floats in the range of 1.0 to -1.0 representing the weight.
index : Optional[int]
Weight index of the weight in the weight list.
"""
weight_encoding = self._detect_encoding(slot)
if weight_encoding == SampleEncoding.UNSUPPORTED:
# fall back to legacy API
await self._scpi._add_acq_weight_async(slot, sequencer, name, len(weight), False)
await self._scpi._set_acq_weight_data_async(slot, sequencer, name, weight)
else:
# use waveform encoding API
weight_bytes = pack_samples(weight_encoding, weight)
await self._scpi._add_acq_weight_encoded_async(
slot, sequencer, name, len(weight), int(weight_encoding)
)
await self._scpi._set_acq_weight_packed_bytes_async(slot, sequencer, name, weight_bytes)
if index is not None:
await self._scpi._set_acq_weight_index_async(slot, sequencer, name, index)
_add_acq_weight = make_sync_wrapper(_add_acq_weight_async)
# ------------------------------------------------------------------------
async def _get_waveforms_async(
self, slot: int, sequencer: int, *, as_numpy: bool = False
) -> dict:
"""
Get all waveforms in the AWG waveform list of indexed sequencer.
The returned dictionary is structured as follows:
- name: waveform name.
- data: waveform samples in a range of 1.0 to -1.0.
- index: waveform index used by the sequencer Q1ASM program to refer
to the waveform.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return data as NumPy arrays instead of Python lists.
Returns
-------
dict
Dictionary with waveforms.
"""
self._present_at_init(slot)
check_sequencer_index(sequencer)
return await self._get_awg_waveforms_async(slot, sequencer, as_numpy=as_numpy)
get_waveforms = make_sync_wrapper(_get_waveforms_async)
# ------------------------------------------------------------------------
async def _add_weights_async(self, slot: int, sequencer: int, weights: dict) -> None:
"""
Add all weights in JSON compatible dictionary to the acquisition weight list
of indexed sequencer.
The dictionary must be structured as follows:
- name : weight name.
- data: weight samples in a range of 1.0 to -1.0.
- index: optional waveform index used by the sequencer Q1ASM program to refer
to the weight.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
weights : dict
JSON compatible dictionary with one or more weights.
Raises
------
KeyError
Missing weight data of weight in dictionary.
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
check_sequencer_index(sequencer)
for name, weight in weights.items():
if "data" in weight:
if "index" in weight:
await self._add_acq_weight_async(
slot,
sequencer,
name,
weight["data"],
weight["index"],
)
else:
await self._add_acq_weight_async(slot, sequencer, name, weight["data"])
else:
raise KeyError(f"Missing data key for {name} in acquisition weight dictionary")
_add_weights = make_sync_wrapper(_add_weights_async)
# ------------------------------------------------------------------------
async def _delete_weight_async(
self, slot: int, sequencer: int, name: str = "", *, delete_all: bool = False
) -> None:
"""
Delete a weight specified by name in the acquisition weight list of
indexed sequencer or delete all weights if `all` is True.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Weight name
delete_all : bool
All weights
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
check_sequencer_index(sequencer)
await self._delete_acq_weight_async(slot, sequencer, "all" if delete_all else name)
_delete_weight = make_sync_wrapper(_delete_weight_async)
# ------------------------------------------------------------------------
async def _get_weights_async(
self, slot: int, sequencer: int, *, as_numpy: bool = False
) -> dict:
"""
Get all weights in the acquisition weight lists of indexed sequencer.
The returned dictionary is structured as follows:
-name : weight name.
- data: weight samples in a range of 1.0 to -1.0.
- index: weight index used by the sequencer Q1ASM program to refer
to the weight.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return data as NumPy arrays instead of Python lists.
Returns
-------
dict
Dictionary with weights.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qrm_type(slot) or self._is_qrc_type(slot))
check_sequencer_index(sequencer)
return await self._get_acq_weights_async(slot, sequencer, as_numpy=as_numpy)
get_weights = make_sync_wrapper(_get_weights_async)
# ----------------------------------------------------------------------------
async def _get_acq_weights_async(
self, slot: int, sequencer: int, *, as_numpy: bool = False
) -> dict:
"""
Get all weights in weight list of indexed sequencer's acquisition path.
If an invalid sequencer index is given, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return data as NumPy arrays instead of Python lists.
Returns
-------
dict
Dictionary with weights.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
# SCPI call
async with self._scpi._slot_connection(slot) as slot_conn:
acq_wlist = await slot_conn._read_bin_async(f"SEQuencer{sequencer}:ACQ:WLISt?", False)
num_weights = struct.unpack("I", acq_wlist)[0]
if num_weights == 0:
await slot_conn._flush_line_end_async()
weight_dict = {}
for weight_it in range(0, num_weights):
# Get name and index
name = str(await slot_conn._read_bin_async("", False), "utf-8")
index = struct.unpack("I", await slot_conn._read_bin_async("", False))[0]
# Get data
raw_bytes = await slot_conn._read_bin_async("", weight_it >= (num_weights - 1))
data = numpy.frombuffer(raw_bytes, dtype=numpy.float32)
# Add to dictionary
weight_dict[name] = {"index": index, "data": data if as_numpy else data.tolist()}
return weight_dict
_get_acq_weights = make_sync_wrapper(_get_acq_weights_async)
# ----------------------------------------------------------------------------
async def _add_acq_acquisition_async(
self,
slot: int,
sequencer: int,
name: str,
num_bins: int,
index: Optional[int] = None,
) -> None:
"""
Add new acquisition to acquisition list of indexed sequencer's acquisition
path. If an invalid sequencer index is given or if the required
acquisition memory cannot be allocated, an error is set in system error.
The acquisition names 'all' and 'ALL' are reserved and adding those will
also result in an error being set in system error. If no index is given,
the next available weight index is selected (starting from 0). If an
invalid weight index is given, an error is set in system error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Acquisition name.
num_bins : int
Number of bins in acquisition. Maximum is 2^24.
index : Optional[int]
Waveform index of the acquisition in the acquisition list.
"""
await self._scpi._add_acq_acquisition_async(slot, sequencer, name, num_bins)
if index is not None:
await self._scpi._set_acq_acquisition_index_async(slot, sequencer, name, index)
_add_acq_acquisition = make_sync_wrapper(_add_acq_acquisition_async)
# ------------------------------------------------------------------------
async def _get_acquisition_status_async(
self,
slot: int,
sequencer: int,
timeout: int = 0,
timeout_poll_res: float = 0.02,
check_seq_state: bool = True,
) -> bool:
"""
Return acquisition binning completion status of the indexed sequencer. If
an invalid sequencer is given, an error is set in system error. If the
timeout is set to zero, the function returns the status immediately. If a
positive non-zero timeout is set, the function blocks until the acquisition
binning completes. If the acquisition hasn't completed before the timeout
expires, a TimeoutError is thrown. Note that when sequencer state checking
is enabled, the sequencer state is checked using get_sequencer_status with
the selected timeout period first and then the acquisition status is checked
with the same timeout period. This means that the total timeout period is
two times the set timeout period.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
timeout : int
Timeout in minutes.
timeout_poll_res : float
Timeout polling resolution in seconds.
check_seq_state : bool
Check if sequencer is done before checking acquisition status.
Returns
-------
bool
Indicates the acquisition binning completion status (False = uncompleted,
True = completed).
Raises
------
TimeoutError
Timeout
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
# Check if sequencer has stopped
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
if check_seq_state:
seq_status = await self._get_sequencer_status_async(
slot, sequencer, timeout, timeout_poll_res
)
if seq_status.state != SequencerStates.STOPPED:
return False
else:
seq_status = await self._get_sequencer_status_async(slot, sequencer)
# Get acquisition status
acq_status = SequencerStatusFlags.ACQ_BINNING_DONE in seq_status.info_flags
elapsed_time = 0.0
timeout = timeout * 60.0
while acq_status is False and elapsed_time < timeout:
await asyncio.sleep(timeout_poll_res)
seq_status = await self._get_sequencer_status_async(slot, sequencer)
acq_status = SequencerStatusFlags.ACQ_BINNING_DONE in seq_status.info_flags
elapsed_time += timeout_poll_res
if elapsed_time >= timeout:
raise TimeoutError(
f"Acquisitions on sequencer {sequencer} did not complete in timeout period "
f"of {int(timeout / 60)} minutes."
)
return acq_status
get_acquisition_status = make_sync_wrapper(_get_acquisition_status_async)
# ------------------------------------------------------------------------
async def _add_acquisitions_async(self, slot: int, sequencer: int, acquisitions: dict) -> None:
"""
Add all waveforms and weights in JSON compatible dictionary to AWG waveform
and acquisition weight lists of indexed sequencer.
The dictionary must be structured as follows:
- name: acquisition name.
- num_bins: number of bins in acquisition.
- index: optional acquisition index used by the sequencer Q1ASM program to refer
to the acquisition.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
acquisitions : dict
JSON compatible dictionary with one or more acquisitions.
Raises
------
KeyError
Missing dictionary key in acquisitions.
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
check_sequencer_index(sequencer)
for name, acquisition in acquisitions.items():
if "num_bins" in acquisition:
if "index" in acquisition:
await self._add_acq_acquisition_async(
slot,
sequencer,
name,
acquisition["num_bins"],
acquisition["index"],
)
else:
await self._add_acq_acquisition_async(
slot, sequencer, name, acquisition["num_bins"]
)
else:
raise KeyError(f"Missing num_bins key for {name} in acquisition dictionary")
_add_acquisitions = make_sync_wrapper(_add_acquisitions_async)
# ------------------------------------------------------------------------
async def _delete_acquisition_async(
self,
slot: int,
sequencer: int,
name: str = "",
*,
delete_all: bool = False,
) -> None:
"""
Delete an acquisition specified by name in the acquisition list of indexed
sequencer or delete all acquisitions if `all` is True.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Weight name
delete_all : bool
All acquisitions
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
check_sequencer_index(sequencer)
await self._delete_acq_acquisition_async(slot, sequencer, "all" if delete_all else name)
_delete_acquisition = make_sync_wrapper(_delete_acquisition_async)
# --------------------------------------------------------------------------
async def _delete_acquisition_data_async(
self,
slot: int,
sequencer: int,
name: str = "",
all: bool = False, # noqa: A002 (public method)
) -> None:
"""
Delete data from an acquisition specified by name in the acquisition list
of indexed sequencer or delete data in all acquisitions if `all` is True.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Weight name
all : bool
All acquisition data
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
check_sequencer_index(sequencer)
await self._delete_acq_acquisition_data_async(slot, sequencer, "all" if all else name)
delete_acquisition_data = make_sync_wrapper(_delete_acquisition_data_async)
# -------------------------------------------------------------------------
async def _set_acq_acquisition_data_async(self, slot: int, sequencer: int, name: str) -> None:
"""
Move scope mode (raw) acquisition data into acquisition in acquisition list of
indexed sequencer's acquisition path. If an invalid sequencer index is given or
if a non-existing acquisition name is given, an error is set in system error.
Parameters
----------
slot : int
slot index.
sequencer : int
Sequencer index.
name : str
Acquisition name.
"""
async with self._scpi._slot_connection(slot) as slot_conn:
await slot_conn._write_async(
f'SEQuencer{sequencer}:ACQ:ALISt:ACQuisition:DATA "{name}"'
)
_set_acq_acquisition_data = make_sync_wrapper(_set_acq_acquisition_data_async)
# -------------------------------------------------------------------------
async def _store_scope_acquisition_async(self, slot: int, sequencer: int, name: str) -> None:
"""
After an acquisition has completed, store the scope acquisition results
in the acquisition specified by name of the indexed sequencers. If an
invalid sequencer index is given an error is set in system error. To get
access to the acquisition results, the sequencer will be stopped when
calling this function.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Acquisition name.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
check_sequencer_index(sequencer)
await self._set_acq_acquisition_data_async(slot, sequencer, name)
store_scope_acquisition = make_sync_wrapper(_store_scope_acquisition_async)
# ------------------------------------------------------------------------
async def _get_acq_acquisition_data_async(
self,
slot: int,
sequencer: int,
name: str,
*,
as_numpy: bool = False,
acq_encoding: Optional[AcquisitionEncodingConfig] = None,
) -> dict:
"""
Get acquisition data of acquisition in acquisition list of indexed
sequencer's acquisition path. The acquisition scope and bin data is
normalized to a range of -1.0 to 1.0 taking both the bit widths of the
processing path and average count into consideration. For the binned
integration results, the integration length is not handled during
normalization and therefore these values have to be divided by their
respective integration lengths. If an invalid sequencer index is given or
if a non-existing acquisition name is given, an error is set in system
error.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
name : str
Acquisition name.
as_numpy : bool
Whether to return path, threshold and avg_cnt data as NumPy arrays
instead of Python lists.
Returns
-------
dict
Dictionary with data of single acquisition.
Raises
------
RuntimeError
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
self._present_at_init(slot)
is_data_encoded = acq_encoding is not None
if is_data_encoded and not self._can_encode_partial_acquisitions(slot):
raise ValueError("partial acquisition is not supported")
# SCPI call
check_sequencer_index(sequencer)
async with self._scpi._slot_connection(slot) as slot_conn:
if is_data_encoded:
encoding_flag = acq_encoding.to_bit_flag()
command = (
f"SEQuencer{sequencer}:ACQ:ALISt:ACQuisition:PARTial:DATA? "
f'"{name}",{encoding_flag}'
)
else:
command = f'SEQuencer{sequencer}:ACQ:ALISt:ACQuisition:DATA? "{name}"'
return await self._get_acq_data_and_convert_async(
slot,
command,
True,
slot_conn,
as_numpy=as_numpy,
is_data_encoded=is_data_encoded,
)
_get_acq_acquisition_data = make_sync_wrapper(_get_acq_acquisition_data_async)
# ------------------------------------------------------------------------
async def _get_acquisitions_async(
self,
slot: int,
sequencer: int,
*,
as_numpy: bool = False,
acq_encoding: AcquisitionEncodingConfig = AcquisitionEncodingConfig.legacy_encoding(),
) -> dict:
"""
Get all acquisitions in acquisition lists of indexed sequencer. The
acquisition scope and bin data is normalized to a range of -1.0 to 1.0
taking both the bit widths of the processing path and average count into
consideration. For the binned integration results, the integration length
is not handled during normalization and therefore these values have to be
divided by their respective integration lengths.
The returned dictionary is structured as follows:
- name: acquisition name
- index: acquisition index used by the sequencer Q1ASM program to refer
to the acquisition.
- acquisition: acquisition dictionary
- scope: Scope data
- path0: input path 0
- data: acquisition samples in a range of 1.0 to -1.0.
- out-of-range: out-of-range indication for the entire acquisition
(False = in-range, True = out-of-range).
- avg_cnt: number of averages.
- path1: input path 1
- data: acquisition samples in a range of 1.0 to -1.0.
- out-of-range: out-of-range indication for the entire acquisition
(False = in-range, True = out-of-range).
- avg_cnt: number of averages.
- bins: bin data
- integration: integration data
- path_0: input path 0 integration result bin list
- path_1: input path 1 integration result bin list
- threshold: threshold result bin list
- valid: list of valid indications per bin
- avg_cnt: list of number of averages per bin
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
as_numpy : bool
Whether to return path, threshold and avg_cnt data as NumPy arrays
instead of Python lists.
acq_encoding: AcquisitionEncodingConfig
The acquisition encoding to use in the partial acquisition download (if any).
If no encoding is requested, full acquisition download is executed.
Encoding presets are declared in AcquisitionEncodingConfig as class methods.
e.g AcquisitionEncodingConfig.iq_and_avg_count()
Returns
-------
dict
Dictionary with acquisitions.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
check_sequencer_index(sequencer)
return await self._get_acq_acquisitions_async(
slot, sequencer, as_numpy=as_numpy, acq_encoding=acq_encoding
)
get_acquisitions = make_sync_wrapper(_get_acquisitions_async)
# ------------------------------------------------------------------------
async def _get_all_acquisitions_async(
self,
sequencers: list[tuple[int, int]],
*,
as_numpy: bool = False,
acq_encoding: AcquisitionEncodingConfig = AcquisitionEncodingConfig.legacy_encoding(),
) -> list[dict]:
"""
Get all acquisitions in acquisition lists of indexed sequencers. The
acquisition scope and bin data is normalized to a range of -1.0 to 1.0
taking both the bit widths of the processing path and average count into
consideration. For the binned integration results, the integration length
is not handled during normalization and therefore these values have to be
divided by their respective integration lengths.
The returned dictionaries are structured as follows:
- name: acquisition name
- index: acquisition index used by the sequencer Q1ASM program to refer
to the acquisition.
- acquisition: acquisition dictionary
- scope: Scope data
- path0: input path 0
- data: acquisition samples in a range of 1.0 to -1.0.
- out-of-range: out-of-range indication for the entire acquisition
(False = in-range, True = out-of-range).
- avg_cnt: number of averages.
- path1: input path 1
- data: acquisition samples in a range of 1.0 to -1.0.
- out-of-range: out-of-range indication for the entire acquisition
(False = in-range, True = out-of-range).
- avg_cnt: number of averages.
- bins: bin data
- integration: integration data
- path_0: input path 0 integration result bin list
- path_1: input path 1 integration result bin list
- threshold: threshold result bin list
- valid: list of valid indications per bin
- avg_cnt: list of number of averages per bin
Parameters
----------
sequencers : list[tuple[int, int]]
List of (slot, sequencer) tuples of sequencers to acquire data from.
as_numpy : bool
Whether to return path, threshold and avg_cnt data as NumPy arrays
instead of Python lists.
Returns
-------
list[dict]
Dictionaries with acquisitions.
Raises
------
NotImplementedError
Functionality not available on any of the modules.
"""
checked_slots = set()
for slot, sequencer in sequencers:
if slot not in checked_slots:
self._present_at_init(slot)
check_is_valid_type(
self._is_qrm_type(slot) or self._is_qtm_type(slot) or self._is_qrc_type(slot)
)
checked_slots.add(slot)
check_sequencer_index(sequencer)
return await asyncio.gather(
*(
self._get_acq_acquisitions_async(
slot, sequencer, as_numpy=as_numpy, acq_encoding=acq_encoding
)
for slot, sequencer in sequencers
)
)
get_all_acquisitions = make_sync_wrapper(_get_all_acquisitions_async)
# ------------------------------------------------------------------------
[docs]
def scope_trigger_arm(self, slot: int) -> None:
"""
Arms the external scope trigger logic on a QTM, such that it will send
a trigger to scope acquisition blocks in the I/O channels when the trigger
condition is satisfied.
Parameters
----------
slot : int
The slot index of the module being referred to.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot))
self._scope_trigger_arm(slot)
# ------------------------------------------------------------------------
async def _get_scope_data_async(
self, slot: int, io_channel: int
) -> numpy.ndarray[Any, numpy.dtype[numpy.float64]]:
"""
Returns the QTM I/O channel scope data for the given slot and channel
acquired since the previous call.
Parameters
----------
slot : int
The slot index of the module being referred to.
io_channel : int
I/O channel you want to get the data for.
Returns
-------
Any
The acquired data. Empty if no data acquired since last call.
Raises
------
NotImplementedError
Functionality not available on this module.
"""
self._present_at_init(slot)
check_is_valid_type(self._is_qtm_type(slot))
check_io_channel_index(io_channel)
return await self._scpi._get_io_channel_scope_data_async(slot, io_channel)
get_scope_data = make_sync_wrapper(_get_scope_data_async)
# -------------------------------------------------------------------------
[docs]
def reset_io_channel_output(
self, slot: Optional[int] = None, io_channel: Optional[int] = None
) -> None:
"""
Resets the output for a specified QSM channel to zero.
Parameters
----------
slot : Optional[int]
slot index.
io_channel : Optional[int]
Source or measurement channel to manipulate. If None, resets all channels.
Raises
------
Exception
Invalid input parameter type.
"""
if slot is not None:
check_is_valid_type(self._is_qsm_type(slot))
else:
slot = ""
if io_channel is None:
io_channel = ""
scpi_cmd_prefix = f"SLOT{slot}:IO:CHANnel{io_channel}"
@scpi_error_check
def reset_io_channel_output_func(instrument: Any) -> None:
instrument._write(f"{scpi_cmd_prefix}:OUT:ZERO")
reset_io_channel_output_func(self)
# -------------------------------------------------------------------------
def _set_voltage_instant(self, slot: int, io_channel: int, voltage: float) -> None:
"""
Sets the voltage for a specified channel immediately, bypassing ramping constraints.
Parameters
----------
slot : int
slot index.
io_channel : int
Source or measurement channel to manipulate.
voltage : float
The desired voltage in volts.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
check_is_valid_type(self._is_qsm_type(slot))
return self._scpi._set_instant(slot, io_channel, voltage)
# -------------------------------------------------------------------------
def _set_voltage_wait(self, slot: int, io_channel: int, voltage: float) -> None:
"""
Sets the voltage for a specified channel and blocks execution
until the voltage stabilizes at the requested value.
Parameters
----------
slot : int
slot index.
io_channel : int
Source or measurement channel to manipulate.
voltage : float
The desired voltage in volts.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
check_is_valid_type(self._is_qsm_type(slot))
return self._scpi._set_voltage_wait(slot, io_channel, voltage)
# --------------------------------------------------------------------------
def _set_safe_voltage_range(self, slot: int, min_voltage: float, max_voltage: float) -> None:
"""
Set the safe voltage range for all channels of the QSM module at the specified slot.
Parameters
----------
slot : int
slot index.
min_voltage : float
The desired minimum voltage in volts.
max_voltage : float
The desired maximum voltage in volts.
"""
check_is_valid_type(self._is_qsm_type(slot))
for io_chan in self.modules[slot - 1].io_channels:
io_chan.set_safe_voltage_range(min_voltage, max_voltage)
# --------------------------------------------------------------------------
[docs]
def set_safe_voltage_range(self, min_voltage: float, max_voltage: float) -> None:
"""
Set the safe voltage range for all channels of all the QSM modules on the entire cluster.
Parameters
----------
min_voltage : float
The desired minimum voltage in volts.
max_voltage : float
The desired maximum voltage in volts.
"""
modules_to_update = [
self.modules[slot - 1]
for slot in self._get_slots_with_module_connected()
if self._is_qsm_type(slot)
]
for mod in modules_to_update:
for io_chan in mod.io_channels:
io_chan.set_safe_voltage_range(min_voltage, max_voltage)
# -------------------------------------------------------------------------
def _set_current_wait(self, slot: int, io_channel: int, current: float) -> None:
"""
Sets the current for a specified channel and blocks execution
until the current stabilizes at the requested value.
Parameters
----------
slot : int
slot index.
io_channel : int
Source or measurement channel to manipulate.
current : float
The desired current in amperes.
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
check_is_valid_type(self._is_qsm_type(slot))
return self._scpi._set_current_wait(slot, io_channel, current)
# --------------------------------------------------------------------------
def _set_safe_current_range(self, slot: int, min_current: float, max_current: float) -> None:
"""
Set the safe current range for all channels of the QSM module at the specified slot.
Parameters
----------
slot : int
slot index.
min_current : float
The desired minimum current in amperes.
max_current : float
The desired maximum current in amperes.
"""
check_is_valid_type(self._is_qsm_type(slot))
for io_chan in self.modules[slot - 1].io_channels:
io_chan.set_safe_current_range(min_current, max_current)
# --------------------------------------------------------------------------
[docs]
def set_safe_current_range(self, min_current: float, max_current: float) -> None:
"""
Set the safe current range for all channels of all the QSM modules on the entire cluster.
Parameters
----------
min_current : float
The desired minimum current in amperes.
max_current : float
The desired maximum current in amperes.
"""
modules_to_update = [
self.modules[slot - 1]
for slot in self._get_slots_with_module_connected()
if self._is_qsm_type(slot)
]
for mod in modules_to_update:
for io_chan in mod.io_channels:
io_chan.set_safe_current_range(min_current, max_current)
# --------------------------------------------------------------------------
[docs]
def request_rampdown(self, slot: Optional[int]) -> None:
"""
Resets the output for all QSM channels to zero, but it also sets an error flag
in the system and blocks the coarse DAC until such flag is cleared (reset).
Parameters
----------
slot : Optional[int]
slot index. If None, ramps down all connected QSM modules.
"""
if slot is not None:
check_is_valid_type(self._is_qsm_type(slot))
slots_to_update = [slot]
else:
slots_to_update = list(
filter(self._is_qsm_type, self._get_slots_with_module_connected())
)
for slot_idx in slots_to_update:
self._scpi._request_rampdown(slot_idx)
# --------------------------------------------------------------------------
[docs]
def clear_rampdown(self, slot: Optional[int]) -> None:
"""
Clear the error flag to be able to use QSM DACs again after triggering a rampdown.
Parameters
----------
slot : Optional[int]
slot index. If None, clears all connected QSM modules.
"""
if slot is not None:
check_is_valid_type(self._is_qsm_type(slot))
slots_to_update = [slot]
else:
slots_to_update = list(
filter(self._is_qsm_type, self._get_slots_with_module_connected())
)
for slot_idx in slots_to_update:
self._scpi._clear_rampdown(slot_idx)
# --------------------------------------------------------------------------
@overload
def _measure_current(self, slot: int, io_channels: int) -> float: ...
@overload
def _measure_current(self, slot: int, io_channels: Iterable[int]) -> list[float]: ...
def _measure_current(
self, slot: int, io_channels: Union[int, Iterable[int]]
) -> Union[float, list[float]]:
"""
Measures the current of the specified IO channel(s) on a specific slot.
Parameters
----------
slot : int
The slot index of the module to measure.
io_channels : Union[int, Iterable[int]]
The index (or indices) of the IO channel(s) to measure.
Returns
-------
Union[float, list[float]]
The measured current in Amperes.
Returns a single `float` if `io_channels` is an `int`.
Returns a `list[float]` if `io_channels` is an iterable.
"""
check_is_valid_type(self._is_qsm_type(slot))
if isinstance(io_channels, int):
single_input = True
channels = [io_channels]
else:
single_input = False
channels = list(io_channels)
currents = self._scpi._measure_current(slot, channels)
return currents[0] if single_input else currents
# ------------------------------------------------------------------------
def _validate_qsm_voltage_measurement(self, slot: int, io_channel: int) -> None:
"""
Ensure that voltage measurement is invoked on a QSM and on a supported I/O channel.
Parameters
----------
slot : int
The slot index of the module being referred to.
io_channel : int
I/O channel to manipulate.
Raises
------
NotImplementedError
If the specified slot is not a QSM.
ValueError
If the I/O channel number is not 0 or 4.
"""
check_is_valid_type(self._is_qsm_type(slot))
if io_channel not in (0, 4):
raise ValueError(
f"Voltage measurement is only implemented for I/O channels 0 and 4 "
f"on the QSM module at slot {slot}."
)
# ------------------------------------------------------------------------
def _measure_voltage(self, slot: int, io_channel: int) -> float:
"""
Returns the voltage measured for a specified channel.
Parameters
----------
slot : int
The slot index of the module being referred to.
io_channel : int
I/O channel to manipulate.
Returns
-------
float
The measured voltage for the specified channel in volts.
Raises
------
NotImplementedError
Functionality not available on this module.
ValueError
If the I/O channel number is not 0 or 4.
"""
self._validate_qsm_voltage_measurement(slot, io_channel)
return self._scpi._measure_voltage(slot, io_channel)
def _measure_voltage_samples(self, slot: int, io_channel: int) -> list[float]:
"""
Returns all measured voltage samples for a specified channel.
Parameters
----------
slot : int
The slot index of the module being referred to.
io_channel : int
I/O channel to manipulate.
Returns
-------
list[float]
The measured voltage samples in volts for a specified channel.
Raises
------
NotImplementedError
Functionality not available on this module.
ValueError
If the I/O channel number is not 0 or 4.
"""
self._validate_qsm_voltage_measurement(slot, io_channel)
return self._scpi._measure_voltage_samples(slot, io_channel)
def _measure_voltage_statistic(self, slot: int, io_channel: int) -> dict[str, int | float]:
"""
Returns the voltage measured, the total number of samples, and the standard deviation
for a specified channel.
Parameters
----------
slot : int
The slot index of the module being referred to.
io_channel : int
I/O channel to manipulate.
Returns
-------
dict[str, int | float]
The measured mean voltage in volts, the total sample count, and the standard deviation
for the specified channel.
Raises
------
NotImplementedError
Functionality not available on this module.
ValueError
If the I/O channel number is not 0 or 4.
"""
self._validate_qsm_voltage_measurement(slot, io_channel)
return self._scpi._measure_voltage_statistic(slot, io_channel)
# --------------------------------------------------------------------------
[docs]
def delete_dummy_binned_acquisition_data(
self,
slot_idx: int,
sequencer: Optional[int] = None,
acq_index_name: Optional[str] = None,
) -> None:
"""
Delete all dummy binned acquisition data for the dummy.
Parameters
----------
slot_idx : int
Slot of the hardware you want to set the data to on a cluster.
sequencer : Optional[int]
Sequencer.
acq_index_name : Optional[str]
Acquisition index name.
"""
self._transport.delete_dummy_binned_acquisition_data(slot_idx, sequencer, acq_index_name)
# --------------------------------------------------------------------------
[docs]
def set_dummy_binned_acquisition_data(
self,
slot_idx: int,
sequencer: int,
acq_index_name: str,
data: Iterable[Union[DummyBinnedAcquisitionData, None]],
) -> None:
"""
Set dummy binned acquisition data for the dummy.
Parameters
----------
slot_idx : int
Slot of the hardware you want to set the data to on a cluster.
sequencer : int
Sequencer.
acq_index_name : str
Acquisition index name.
data : Iterable[Union[DummyBinnedAcquisitionData, None]]
Dummy data for the binned acquisition.
An iterable of all the bin values.
"""
self._transport.set_dummy_binned_acquisition_data(slot_idx, sequencer, acq_index_name, data)
# --------------------------------------------------------------------------
[docs]
def delete_dummy_scope_acquisition_data(
self, slot_idx: int, sequencer: Union[int, None]
) -> None:
"""
Delete dummy scope acquisition data for the dummy.
Parameters
----------
slot_idx : int
Slot of the hardware you want to set the data to on a cluster.
sequencer : Union[int, None]
Sequencer.
"""
self._transport.delete_dummy_scope_acquisition_data(slot_idx)
# --------------------------------------------------------------------------
[docs]
def set_dummy_scope_acquisition_data(
self,
slot_idx: int,
sequencer: Union[int, None],
data: DummyScopeAcquisitionData,
) -> None:
"""
Set dummy scope acquisition data for the dummy.
Parameters
----------
slot_idx : int
Slot of the hardware you want to set the data to on a cluster.
sequencer : Union[int, None]
Sequencer.
data : DummyScopeAcquisitionData
Dummy data for the scope acquisition.
"""
self._transport.set_dummy_scope_acquisition_data(slot_idx, data)
# ------------------------------------------------------------------------
async def _set_sequence_async(
self,
slot: int,
sequencer: int,
sequence: Union[str, Sequence],
validation_enable: bool = True,
immediate_asm_install: bool = True,
) -> None:
"""
Set sequencer program, AWG waveforms, acquisition weights and acquisitions
from a JSON file or from a dictionary directly. The JSON file or
dictionary need to apply the schema specified by
`QCM_SEQUENCE_JSON_SCHEMA`, `QRM_SEQUENCE_JSON_SCHEMA`, `WAVE_JSON_SCHEMA`
and `ACQ_JSON_SCHEMA`.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
sequence : Union[str, Sequence]
Path to sequence file or dictionary.
validation_enable : bool
Enable JSON schema validation on sequence.
immediate_asm_install: bool
Whether the sequencer program should be installed immediately.
If True, the sequencer is stopped to install the program.
If False, program will be just compiled, installation will
happen on the next sequence arm.
Raises
------
fastjsonschema.JsonSchemaValueException
Invalid JSON object.
"""
# Set dictionary
if isinstance(sequence, dict):
sequence_dict = sequence
else:
with open(sequence, "r") as file:
sequence_dict = json.load(file)
await self._update_sequence_async(
slot,
sequencer,
sequence_dict,
validation_enable=validation_enable,
erase_existing=True,
immediate_asm_install=immediate_asm_install,
)
_set_sequence = make_sync_wrapper(_set_sequence_async)
# ------------------------------------------------------------------------
async def _update_sequence_async(
self,
slot: int,
sequencer: int,
sequence_dict: OptionalSequence,
validation_enable: bool,
erase_existing: bool,
immediate_asm_install: bool,
) -> None:
"""
Update sequencer program, AWG waveforms, acquisition weights and acquisitions
from a sequence dictionary. The update can be either cumulative or a replacement.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
sequence_dict : Sequence
Sequence dictionary.
validation_enable : bool
Enable JSON schema validation on sequence.
erase_existing : bool
Whether existing waveforms/weights/acquisitions should be erased
before adding new ones.
immediate_asm_install: bool
Whether the sequencer program should be installed immediately.
If True, the sequencer is stopped to install the program.
If False, program will be just compiled, installation will
happen on the next sequence arm.
Raises
------
fastjsonschema.JsonSchemaValueException
Invalid JSON object.
"""
self._present_at_init(slot)
# Validate dictionary
if validation_enable:
self._validate_sequence(slot, sequencer, sequence_dict)
if "program" in sequence_dict:
await self._set_sequencer_program_async(
slot,
sequencer,
sequence_dict["program"],
immediate_asm_install=immediate_asm_install,
)
if self._has_waveforms(slot, sequencer) and "waveforms" in sequence_dict:
if erase_existing:
await self._delete_waveform_async(slot, sequencer, delete_all=True)
if sequence_dict["waveforms"]:
await self._add_waveforms_async(slot, sequencer, sequence_dict["waveforms"])
if self._has_weights(slot, sequencer) and "weights" in sequence_dict:
if erase_existing:
await self._delete_weight_async(slot, sequencer, delete_all=True)
if sequence_dict["weights"]:
await self._add_weights_async(slot, sequencer, sequence_dict["weights"])
if self._has_acquisitions(slot, sequencer) and "acquisitions" in sequence_dict:
if not immediate_asm_install:
raise ValueError(
"Cannot modify existing acquisitions with delayed asm install."
"The features are incompatible."
)
elif erase_existing:
await self._delete_acquisition_async(slot, sequencer, delete_all=True)
if sequence_dict["acquisitions"]:
await self._add_acquisitions_async(slot, sequencer, sequence_dict["acquisitions"])
_update_sequence = make_sync_wrapper(_update_sequence_async)
# ------------------------------------------------------------------------
def _validate_sequence(
self,
slot: int,
sequencer: int,
sequence_dict: Sequence,
) -> None:
"""
Validate a sequence against the JSON schemas specified by
`QCM_SEQUENCE_JSON_SCHEMA`, `QRM_SEQUENCE_JSON_SCHEMA`, `WAVE_JSON_SCHEMA`
and `ACQ_JSON_SCHEMA`.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
sequence_dict : dict[str, Any]
Sequence dictionary.
"""
if self._is_qrm_type(slot):
validate_qrm_sequence(sequence_dict)
elif self._is_qcm_type(slot):
validate_qcm_sequence(sequence_dict)
elif self._is_qtm_type(slot):
validate_qtm_sequence(sequence_dict)
elif self._is_qrc_type(slot):
# NOTE: The only difference between QRM and QRC from a sequence json standpoint
# is that some, but not all sequencers allow "acquisitions" to be there.
validate_qrm_sequence(sequence_dict)
else:
raise TypeError("Device type not supported")
if self._has_waveforms(slot, sequencer):
for wave_data in sequence_dict["waveforms"].values():
validate_wave(wave_data)
if self._has_weights(slot, sequencer):
for wave_data in sequence_dict["weights"].values():
validate_wave(wave_data)
if self._has_acquisitions(slot, sequencer):
for acq_data in sequence_dict["acquisitions"].values():
validate_acq(acq_data)
# ------------------------------------------------------------------------
[docs]
def update_sequence(
self,
slot: int,
sequencer: int,
erase_existing: bool = False,
immediate_asm_install: bool = True,
**sequence_args: Unpack[Sequence],
) -> None:
r"""
Update sequencer program, AWG waveforms, acquisition weights and/or acquisitions
from a series of keyword arguments. By default, the update is cumulative
i.e. existing waveforms/weights/acquisitions are kept on the sequencer
and new ones are added (as long as indexes or names aren't reused).
For example:
.. code-block:: python
sequencer.update_sequence(
waveforms={"wave0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
weights={"weight0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
erase_existing=False
)
will add one waveform and one weight to the sequencer, without either removing
currently uploaded waveforms and weights, or clearing program or acquisitions.
If the `erase_existing` parameter is passed, then the update is destructive
i.e. existing program/waveforms/weights/acquisitions will be completely overwritten
with the new ones.
For example:
.. code-block:: python
sequencer.update_sequence(
program="play 0,0,100\nstop",
acquisitions={"acq0": {"index": 0, "num_bins": 1}},
erase_existing=True
)
will completely replace the program and the acquisitions on the sequencer,
but will not touch currently uploaded waveforms and weights.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
erase_existing : bool
Whether existing waveforms/weights/acquisitions should be erased
before adding new ones.
immediate_asm_install: bool
Whether the sequencer program should be installed immediately.
If True, the sequencer is stopped to install the program.
If False, program will be just compiled, installation will
happen on the next sequence arm.
**sequence_args : dict
Sequence dictionary in keyword arguments format.
Raises
------
fastjsonschema.JsonSchemaValueException
Invalid JSON object.
"""
self._update_sequence(
slot,
sequencer,
sequence_args,
validation_enable=False, # The sequence is partial and won't pass validation
erase_existing=erase_existing,
immediate_asm_install=immediate_asm_install,
)
# Invalidate QCoDeS cache because this call doesn't invoke a parameter
self.modules[slot - 1].sequencers[sequencer].sequence.cache.invalidate()
# ------------------------------------------------------------------------
async def _update_sequences_async(
self,
sequences: dict[tuple[int, int], OptionalSequence],
erase_existing: bool = False,
immediate_asm_install: bool = True,
) -> None:
r"""
Update multiple sequencer program, AWG waveforms, acquisition weights and/or acquisitions,
asynchronously. By default, the update is cumulative i.e. existing waveforms, weights,
and acquisitions are kept on the sequencer and new ones are added (as long as indexes
or names aren't reused).
For example:
.. code-block:: python
await sequencer._update_sequences_async({
(2, 0): {
"waveforms": {"wave0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
"weights": {"weight0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
},
(5, 1): {
"waveforms": {"wave0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
"weights": {"weight0": {"index": 0, "data": [0.0, 0.1, 0.2, 0.3]}},
},
erase_existing=False
)
will add one waveform and one weight to sequencer 0 at slot 2 and sequencer 1 at slot 5,
without either removing currently uploaded waveforms and weights,
or clearing program or acquisitions.
If the `erase_existing` parameter is passed, then the update is destructive
i.e. existing program/waveforms/weights/acquisitions will be completely overwritten
with the new ones.
For example:
.. code-block:: python
await sequencer._update_sequences_async({
(2, 0): {
"program": "play 1, 1, 100\nstop",
"acquisitions": {},
},
(5, 1): {
"program": "play 1, 1, 100\nstop",
"acquisitions": {},
},
erase_existing=True
)
will completely replace the program and the acquisitions on the sequencers,
but will not touch currently uploaded waveforms and weights.
Parameters
----------
sequences: dict[tuple[int, int], OptionalSequence]
Mapping of (slot, sequencer) tuples to sequence parts.
erase_existing : bool
Whether existing waveforms/weights/acquisitions should be erased
before adding new ones.
immediate_asm_install : bool
Whether the sequencer program should be installed immediately.
If True, the sequencer is stopped to install the program.
If False, program will be just compiled, installation will
happen on the next sequence arm.
Raises
------
fastjsonschema.JsonSchemaValueException
Invalid JSON object.
"""
await asyncio.gather(
*(
self._update_sequence_async(
slot,
sequencer,
sequence_args,
validation_enable=False,
erase_existing=erase_existing,
immediate_asm_install=immediate_asm_install,
)
for (slot, sequencer), sequence_args in sequences.items()
)
)
for slot, sequencer in sequences:
# Invalidate QCoDeS cache because this call doesn't invoke a parameter
self.modules[slot - 1].sequencers[sequencer].sequence.cache.invalidate()
update_sequences = make_sync_wrapper(_update_sequences_async)
# -----------------------------------------------------------------------
@staticmethod
def _is_module_present(slot: int, bitmask: int) -> bool:
"""
Given a slot number and module presence indicator in the form of a bitmask,
performs the necessary bitwise operations to determine whether that slot
has a module present.
Parameters
----------
slot : int
The slot index of the module being referred to.
bitmask : int
The module presence indicator as returned by the ``BP:MODules?`` command.
Returns
-------
bool
Whether the bitmask indicates presence for the slot.
"""
return (bitmask >> (slot - 1)) & 1 == 1
# ------------------------------------------------------------------------
def _get_slots_with_module_present(self) -> set[int]:
"""
Get all slot numbers that have a module present.
Returns
-------
set[int]
A set of slot numbers.
"""
bitmask = self._scpi._get_modules_present()
return {slot for slot in range(1, 20 + 1) if self._is_module_present(slot, bitmask)}
# ------------------------------------------------------------------------
def _get_slots_with_module_connected(self) -> set[int]:
"""
Get all slot numbers that have a module connected.
Returns
-------
set[int]
A set of slot numbers.
"""
mods_info = self._scpi._get_mods_info()
return {int(key.split()[-1]) for key in mods_info.keys()}
# ------------------------------------------------------------------------
def _get_modules_present(self, slot: int) -> bool:
"""
Get an indication of module presence for a specific slot in the Cluster.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool
Module presence.
"""
return self._is_module_present(slot, self._scpi._get_modules_present())
# ------------------------------------------------------------------------
def _get_modules_connected(self, slot: int) -> bool:
"""
Get an indication of module connection for a specific slot in the Cluster.
Parameters
----------
slot : int
Slot index ranging from 1 to 20.
Returns
-------
bool
Module connection
"""
return slot in self._get_slots_with_module_connected()
# ------------------------------------------------------------------------
[docs]
@contextmanager
def scpi_transaction(
self, slot: int, sequencer: int
) -> Generator[tuple[Ieee488_2Connection, str], None, None]:
"""
Initiate a "transaction" based on a slot/sequencer combination, establishing an ad-hoc
connection to be reused within the context manager so that all read/write operations
will be performed within that same connection.
Parameters
----------
slot : int
The slot index of the module being referred to.
sequencer : int
Sequencer index.
Yields
------
tuple[Ieee488_2Connection, str]
A tuple containing the connection object for the slot/sequencer combination
and a prefix for the SCPI commands that should be sent.
"""
prefix = f"SEQuencer{sequencer}:"
if self.is_dummy:
prefix = f"SLOT{slot}:" + prefix
if slot not in self._scpi_transaction_connection_map:
if self.is_dummy:
# We're using `ClusterDummyTransport`
transport = self._transport
else:
# We're on real hardware
host, _ = self._transport._socket.getpeername()
addr_info = resolve(f"{host}/{slot}")
transport = IpTransport(
addr_info.address,
addr_info.scpi_port,
loop_from=self._transport,
)
# Store a new connection into the appropriate slot
self._scpi_transaction_connection_map[slot] = Ieee488_2Connection(Ieee488_2(transport))
conn = self._scpi_transaction_connection_map[slot]
if self._debug in (DebugLevel.ERROR_CHECK, DebugLevel.VERSION_AND_ERROR_CHECK):
try:
yield conn, prefix
except OSError:
raise
except Exception as error:
conn._interface._check_error(error)
else:
yield conn, prefix
# ------------------------------------------------------------------------
[docs]
def toggle_all_lo(
self, slot_or_enable: Union[int, bool, None] = None, enable: Optional[bool] = None
) -> None:
"""
Turn ON or OFF all local oscillators present on all modules or a specific module.
Parameters
----------
slot_or_enable : int | bool | None
The slot index of the module being referred to.
enable : bool | None
Turn ON if True, OFF if False.
"""
# the signature used to be toggle_all_lo(self, enable: bool), the logic
# below is the add a slot index in the first argument in a non-breaking
# way
if isinstance(slot_or_enable, bool) and isinstance(enable, bool):
raise TypeError("toggle_all_lo() requires 1 boolean argument") # pyright: ignore[reportUnreachable, reportUnreachable]
if slot_or_enable is None or isinstance(slot_or_enable, bool):
rf_modules = self.get_connected_modules(
filter_fn=lambda mod: (mod.is_qcm_type or mod.is_qrm_type) and mod.is_rf_type
)
for module in rf_modules.values():
for param_name, param in module.parameters.items():
if re.compile(r"out\d+(_in\d+)?_lo_en").fullmatch(param_name):
param.set(bool(slot_or_enable) or bool(enable))
elif isinstance(slot_or_enable, int):
module = self.module[int(slot_or_enable) - 1]
for param_name, param in module.parameters.items():
if re.compile(r"out\d+(_in\d+)?_lo_en").fullmatch(param_name):
param.set(enable)
else:
raise TypeError("toggle_all_lo() missing 1 required positional argument: 'enable'") # pyright: ignore[reportUnreachable, reportUnreachable]
def _check_error(self, error: Optional[Exception] = None) -> None:
self._scpi._check_error(error)
# ------------------------------------------------------------------------
@staticmethod
def _frequency_plan_allowed_keys(frequency: int) -> set[str]:
# We only allow certain frequency parameters to be set and get for now.
# Only certified parameter ranges can be exposed to users, and
# for now only certain parameters were certified depending on the frequency.
if frequency <= 1800:
return {"Input DSA1", "Input DSA2"}
else:
return {"DAC Current (uA)", "Output DSA", "Input DSA1", "Input DSA2"}
# ------------------------------------------------------------------------
def _get_current_frequency_plan(self, slot: int) -> dict[str, Any]:
"""
Get current frequency plan in JSON format.
Parameters
----------
slot : int
slot index
Returns
-------
dict[str, Any]
Exported frequency plan in JSON format
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
check_is_valid_type(self._is_qrc_type(slot))
frequency_plan = self._scpi._get_current_frequency_plan(slot)
def filter_allowed_entries(frequency: str, data: dict) -> dict:
allowed_keys = self._frequency_plan_allowed_keys(int(frequency))
return {k: v for k, v in data.items() if k in allowed_keys}
return {
frequency: filter_allowed_entries(frequency, data)
for frequency, data in frequency_plan.items()
}
# ------------------------------------------------------------------------
def _set_user_freq_plan_patch(self, slot: int, save: bool, plan_json: Any) -> None:
"""
Set user frequency plan settings patch.
Parameters
----------
slot : int
slot index
save : bool
True to save it to persistent memory
plan_json : Any
frequency plan path in JSON format
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
check_is_valid_type(self._is_qrc_type(slot))
for frequency, data in plan_json.items():
allowed_keys = self._frequency_plan_allowed_keys(int(frequency))
for k in data.keys():
if k not in allowed_keys:
raise ValueError(
f"Setting user frequency plan failed. Key {k} is not allowed "
f"for frequency {frequency}. Only {allowed_keys} are allowed."
)
return self._scpi._set_user_freq_plan_patch(slot, save, plan_json)
