# Repository: https://gitlab.com/qblox/packages/software/qblox-scheduler
# Licensed according to the LICENSE file on the main branch
#
# Copyright 2020-2025, Quantify Consortium
# Copyright 2025, Qblox B.V.
"""Module containing Qblox InstrumentCoordinator Components."""
from __future__ import annotations
import asyncio
import logging
import os
import re
from abc import ABC, abstractmethod
from collections import namedtuple
from dataclasses import dataclass
from functools import partial, wraps
from math import isnan
from typing import (
TYPE_CHECKING,
Any,
Generic,
TypeVar,
Union,
cast,
)
from uuid import uuid4
import numpy as np
from qblox_instruments import (
Cluster,
ConfigurationManager,
SequencerStates,
SequencerStatus,
)
from qblox_instruments.qcodes_drivers.time import SyncRef
from xarray import DataArray, Dataset
from qblox_scheduler.analysis.data_handling import OutputDirectoryManager
from qblox_scheduler.backends.qblox import constants, driver_version_check
from qblox_scheduler.backends.qblox.enums import (
ChannelMode,
LoCalEnum,
SidebandCalEnum,
TimetagTraceType,
)
from qblox_scheduler.backends.qblox.helpers import (
single_scope_mode_acquisition_raise,
)
from qblox_scheduler.backends.qblox.operation_handling.bin_mode_compat import (
QRM_COMPATIBLE_BIN_MODES,
QTM_COMPATIBLE_BIN_MODES,
IncompatibleBinModeError,
)
from qblox_scheduler.backends.qblox.qblox_acq_index_manager import (
BinnedAcqControlFlowNode,
BinnedAcqInfo,
QbloxAcquisitionBinMappingTimetagTrace,
QbloxAcquisitionHardwareMapping,
QbloxAcquisitionHardwareMappingBinned,
QbloxAcquisitionIndex,
)
from qblox_scheduler.backends.types.common import ThresholdedTriggerCountMetadata
from qblox_scheduler.backends.types.qblox import (
AnalogModuleSettings,
AnalogSequencerSettings,
BaseModuleSettings,
ClusterModuleDescription,
ClusterSettings,
ExternalTriggerSyncSettings,
QCMDescription,
QCMRFDescription,
QRCDescription,
QRMDescription,
QRMRFDescription,
QSMDescription,
QTMDescription,
RFModuleSettings,
SequencerSettings,
TimetagModuleSettings,
TimetagSequencerSettings,
)
from qblox_scheduler.backends.types.qblox.settings import DCModuleSettings
from qblox_scheduler.enums import BinMode, TimeRef, TriggerCondition
from qblox_scheduler.instrument_coordinator.components import base
from qblox_scheduler.instrument_coordinator.components.base import instrument_to_component_name
from qblox_scheduler.instrument_coordinator.utility import (
add_acquisition_coords_binned,
add_acquisition_coords_nonbinned,
check_already_existing_acquisition,
lazy_async_set,
lazy_set,
parameter_value_same_as_cache,
search_settable_param,
)
from qblox_scheduler.quantify_utils import without
from qblox_scheduler.schedules.schedule import AcquisitionChannelsData
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Hashable, Iterable
from contextlib import suppress
with suppress(ImportError):
from qblox_instruments.qcodes_drivers.async_parameter import AsyncParameter
from qblox_instruments.qcodes_drivers.module import Module
from qcodes.instrument.instrument_base import InstrumentBase
from qblox_scheduler.schedules.schedule import CompiledSchedule
[docs]
logger = logging.getLogger(__name__)
logger.setLevel(logging.WARNING)
# Prevent unsupported qblox-instruments version from crashing this submodule
driver_version_check.verify_qblox_instruments_version()
@dataclass(frozen=True)
[docs]
class _StaticHardwareProperties:
"""Dataclass for storing configuration differences across Qblox devices."""
[docs]
settings_type: type[BaseModuleSettings]
"""The settings dataclass to use that the hardware needs to configure to."""
[docs]
number_of_sequencers: int
"""The number of sequencers the hardware has available."""
[docs]
number_of_output_channels: int
"""The number of physical output channels that can be used."""
"""The number of physical input channels that can be used."""
[docs]
number_of_scope_acq_channels: int
"""The number of scope acquisition channels."""
@dataclass(frozen=True)
[docs]
class _StaticAnalogModuleProperties(_StaticHardwareProperties):
"""Dataclass for storing configuration differences across Qblox devices."""
[docs]
settings_type: type[AnalogModuleSettings]
"""The settings dataclass to use that the hardware needs to configure to."""
"""Specifies if an internal lo source is available."""
@dataclass(frozen=True)
@dataclass(frozen=True)
[docs]
class _StaticDCModuleProperties(_StaticHardwareProperties):
"""Dataclass for storing configuration differences across Qblox devices."""
[docs]
settings_type: type[DCModuleSettings]
"""The settings dataclass to use that the hardware needs to configure to."""
[docs]
_QCM_BASEBAND_PROPERTIES = _StaticAnalogModuleProperties(
settings_type=AnalogModuleSettings,
has_internal_lo=False,
number_of_sequencers=6,
number_of_output_channels=4,
number_of_input_channels=0,
number_of_scope_acq_channels=0,
)
[docs]
_QRM_BASEBAND_PROPERTIES = _StaticAnalogModuleProperties(
settings_type=AnalogModuleSettings,
has_internal_lo=False,
number_of_sequencers=6,
number_of_output_channels=2,
number_of_input_channels=2,
number_of_scope_acq_channels=2,
)
[docs]
_QCM_RF_PROPERTIES = _StaticAnalogModuleProperties(
settings_type=RFModuleSettings,
has_internal_lo=True,
number_of_sequencers=6,
number_of_output_channels=2,
number_of_input_channels=0,
number_of_scope_acq_channels=0,
)
[docs]
_QRM_RF_PROPERTIES = _StaticAnalogModuleProperties(
settings_type=RFModuleSettings,
has_internal_lo=True,
number_of_sequencers=6,
number_of_output_channels=1,
number_of_input_channels=1,
number_of_scope_acq_channels=2,
)
[docs]
_QRC_PROPERTIES = _StaticAnalogModuleProperties(
settings_type=RFModuleSettings,
has_internal_lo=True,
number_of_sequencers=12,
number_of_output_channels=6,
number_of_input_channels=2,
number_of_scope_acq_channels=4,
)
[docs]
_QTM_PROPERTIES = _StaticTimetagModuleProperties(
settings_type=TimetagModuleSettings,
number_of_sequencers=8,
number_of_output_channels=8,
number_of_input_channels=8,
number_of_scope_acq_channels=0,
)
[docs]
_QSM_PROPERTIES = _StaticDCModuleProperties(
settings_type=DCModuleSettings,
number_of_sequencers=0,
number_of_output_channels=8,
number_of_input_channels=8,
number_of_scope_acq_channels=0,
)
[docs]
_HardwarePropertiesT_co = TypeVar(
"_HardwarePropertiesT_co", bound=_StaticHardwareProperties, covariant=True
)
[docs]
class _ModuleComponentBase(
base.InstrumentCoordinatorComponentBase, Generic[_HardwarePropertiesT_co]
):
"""Qblox InstrumentCoordinator component base class."""
[docs]
_hardware_properties: _HardwarePropertiesT_co
def __init__(self, instrument: Module) -> None:
super().__init__(instrument)
# The base class `InstrumentCoordinatorComponentBase` expects `instrument` to
# be a subclass of `Instrument`, _This_ class expects `instrument` to be a
# `Module` (for legacy reasons?), which does not subclass `Instrument` but
# `InstrumentChannel`, and is therefore not globally findable. Ergo, we store a
# reference here directly.
[docs]
self._instrument_module = instrument
[docs]
self._seq_name_to_idx_map = {
f"seq{idx}": idx for idx in range(self._hardware_properties.number_of_sequencers)
}
# TODO make a data model for this and replace the dictionary by a class.
[docs]
self._nco_frequency_changed: dict[int, bool] = {}
"""
Private attribute for automatic mixer calibration. The keys are sequencer
indices. The `prepare` method resets this to an empty dictionary.
"""
# See the comment on self._instrument_module in __init__. Base class is incorrectly
# overridden, so we silence pyright.
@property
[docs]
def instrument(self) -> Module: # type: ignore
"""Returns a reference to the module instrument."""
return self._instrument_module
[docs]
def _check_parameter_settable(
self,
instrument: InstrumentBase,
parameter_name: str,
val: Any, # noqa: ANN401
) -> bool:
# TODO: Bias-tee parameters for RTP are not officially supported in
# qblox-instruments. This hack is at a customer's request.
try:
search_settable_param(instrument=instrument, nested_parameter_name=parameter_name)
except ValueError as e:
if (
(re.search(r".*(out|marker)[0-3]_ihp_config", parameter_name) and val == "bypassed")
or re.search(r".*out[0-3]_ihp_tau_constant", parameter_name)
or re.search(r".*out[0-3]_exp[0-3]_tau_constant", parameter_name)
or parameter_name == "real_mode_en"
):
return False
raise e
return True
[docs]
def _set_parameter(
self,
instrument: InstrumentBase,
parameter_name: str,
val: Any, # noqa: ANN401, disallow Any as type
) -> None:
"""
Set the parameter directly or using the lazy set.
Parameters
----------
instrument
The instrument or instrument channel that holds the parameter to set,
e.g. `self.instrument` or `self.instrument[f"sequencer{idx}"]`.
parameter_name
The name of the parameter to set.
val
The new value of the parameter.
"""
if not self._check_parameter_settable(instrument, parameter_name, val):
return
if self.force_set_parameters():
instrument.parameters[parameter_name].set(val)
else:
lazy_set(instrument, parameter_name, val)
[docs]
async def _set_async_parameter(
self,
instrument: InstrumentBase,
parameter_name: str,
val: Any, # noqa: ANN401, disallow Any as type
) -> None:
"""
Set the parameter directly, asynchronously, or using the lazy set.
Parameters
----------
instrument
The instrument or instrument channel that holds the parameter to set,
e.g. `self.instrument` or `self.instrument[f"sequencer{idx}"]`.
parameter_name
The name of the parameter to set.
val
The new value of the parameter.
"""
if not self._check_parameter_settable(instrument, parameter_name, val):
return
if self.force_set_parameters():
parameter = cast("AsyncParameter", instrument.parameters[parameter_name])
await parameter.async_set(val)
else:
await lazy_async_set(instrument, parameter_name, val)
@property
[docs]
def is_running(self) -> bool:
"""
Finds if any of the sequencers is currently running.
Returns
-------
:
True if any of the sequencers reports the `SequencerStates.RUNNING` status.
"""
for seq_idx in range(self._hardware_properties.number_of_sequencers):
seq_status = self.instrument.get_sequencer_status(seq_idx)
if seq_status.state is SequencerStates.RUNNING:
return True
return False
[docs]
async def is_running_async(self) -> bool:
if not hasattr(self.instrument, "_get_sequencer_statuses_async"):
return self.is_running
seq_indices = list(range(self._hardware_properties.number_of_sequencers))
seq_statuses = await self.instrument._get_sequencer_statuses_async(seq_indices)
return any(seq_status.state is SequencerStates.RUNNING for seq_status in seq_statuses)
[docs]
def wait_done(self, timeout_sec: int = 10) -> None:
"""
Blocks the instrument until all the sequencers are done running.
Parameters
----------
timeout_sec
The timeout in seconds. N.B. the instrument takes the timeout in minutes
(int), therefore it is rounded down to whole minutes with a minimum of 1.
"""
timeout_min = timeout_sec // 60
if timeout_min == 0:
timeout_min = 1
for idx in range(self._hardware_properties.number_of_sequencers):
state: SequencerStatus = self.instrument.get_sequencer_status(
sequencer=idx, timeout=timeout_min
)
for flag in state.info_flags:
logger.info("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
for flag in state.warn_flags:
logger.warning("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
for flag in state.err_flags:
logger.error("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
[docs]
async def wait_done_async(self, timeout_sec: int = 10) -> None:
"""
Blocks the instrument until all the sequencers are done running.
Parameters
----------
timeout_sec
The timeout in seconds. N.B. the instrument takes the timeout in minutes
(int), therefore it is rounded down to whole minutes with a minimum of 1.
"""
if not hasattr(self.instrument, "_wait_for_sequencers_async"):
return self.wait_done(timeout_sec)
timeout_min = timeout_sec // 60
if timeout_min == 0:
timeout_min = 1
seq_indices = list(range(self._hardware_properties.number_of_sequencers))
statuses = await self.instrument._wait_for_sequencers_async(
seq_indices,
timeout=timeout_min,
)
for idx, status in enumerate(statuses):
for flag in status.info_flags:
logger.info("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
for flag in status.warn_flags:
logger.warning("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
for flag in status.err_flags:
logger.error("[%s|seq%d] %s - %s", self.name, idx, flag, flag.value)
[docs]
def get_hardware_log(
self,
compiled_schedule: CompiledSchedule,
) -> dict | None:
"""
Retrieve the hardware log of the Qblox instrument associated to this component.
This log does not include the instrument serial number and firmware version.
Parameters
----------
compiled_schedule
Compiled schedule to check if this component is referenced in.
Returns
-------
:
A dict containing the hardware log of the Qblox instrument, in case the
component was referenced; else None.
"""
if self.instrument.name not in compiled_schedule.compiled_instructions:
return None
return _download_log(_get_configuration_manager(_get_instrument_ip(self)))
[docs]
def prepare( # pyright: ignore[reportIncompatibleMethodOverride]
self,
program: dict[str, dict],
acq_channels_data: AcquisitionChannelsData,
repetitions: int,
) -> None:
"""Store program containing sequencer settings."""
self._configure(program, acq_channels_data, repetitions)
self._upload(program)
[docs]
async def prepare_async(
self,
program: dict[str, dict],
acq_channels_data: AcquisitionChannelsData,
repetitions: int,
) -> None:
self._configure(program, acq_channels_data, repetitions)
await self._upload_async(program)
[docs]
def disable_sync(self) -> None:
"""Disable sync for all sequencers."""
for idx in range(self._hardware_properties.number_of_sequencers):
# Prevent hanging on next run if instrument is not used.
self._set_parameter(self.instrument.sequencers[idx], "sync_en", False)
[docs]
async def disable_sync_async(self) -> None:
"""Disable sync for all sequencers."""
await asyncio.gather(
*[
self._set_async_parameter(self.instrument.sequencers[idx], "sync_en", False)
for idx in range(self._hardware_properties.number_of_sequencers)
]
)
[docs]
def stop(self) -> None:
"""Stops all execution."""
self.disable_sync()
self.instrument.stop_sequencer()
[docs]
async def stop_async(self) -> None:
"""Stops all execution."""
await self.disable_sync_async()
await self.instrument._stop_sequencer_async()
[docs]
def _upload(self, program: dict[str, dict]) -> None:
pass
[docs]
async def _upload_async(self, program: dict[str, dict]) -> None:
pass
[docs]
async def retrieve_acquisition_async(self) -> Dataset | None:
"""Gets and returns acquisition data."""
[docs]
def _get_program_settings(self, program: dict[str, dict]) -> Any: # noqa: ANN401, disallow Any as type
if (settings := program.get("settings")) is not None and isinstance(settings, dict):
settings = self._hardware_properties.settings_type.from_dict(settings)
return settings
@abstractmethod
[docs]
def _upload_to_sequencer(self, seq_idx: int, settings: SequencerSettings) -> None:
self._set_parameter(self.instrument.sequencers[seq_idx], "sequence", settings.sequence)
[docs]
async def _upload_to_sequencer_async(self, seq_idx: int, settings: SequencerSettings) -> None:
await self._set_async_parameter(
self.instrument.sequencers[seq_idx], "sequence", settings.sequence
)
[docs]
def arm_all_sequencers_in_program(self) -> None:
"""Arm all the sequencers that are part of the program."""
for seq_name in self._program.get("sequencers", {}):
if seq_name in self._seq_name_to_idx_map:
seq_idx = self._seq_name_to_idx_map[seq_name]
self.instrument.arm_sequencer(sequencer=seq_idx)
[docs]
async def arm_all_sequencers_in_program_async(self) -> None:
"""Arm all the sequencers that are part of the program."""
seq_indices = [
self._seq_name_to_idx_map[name]
for name in self._program.get("sequencers", {})
if name in self._seq_name_to_idx_map
]
await self.instrument._arm_sequencers_async(seq_indices)
[docs]
def start(self) -> None:
"""Clear data, arm sequencers and start sequencers."""
self.clear_data()
self.arm_all_sequencers_in_program()
self._start_armed_sequencers()
[docs]
async def start_async(self) -> None:
"""Clear data, arm sequencers and start sequencers."""
await self.clear_data_async()
await self.arm_all_sequencers_in_program_async()
await self._start_armed_sequencers_async()
[docs]
def _start_armed_sequencers(self) -> None:
"""Start execution of the schedule: start armed sequencers."""
for idx in range(self._hardware_properties.number_of_sequencers):
state = self.instrument.get_sequencer_status(idx)
if state.state is SequencerStates.ARMED:
self.instrument.start_sequencer(idx)
[docs]
async def _start_armed_sequencers_async(self) -> None:
"""Start execution of the schedule: start armed sequencers."""
for idx in range(self._hardware_properties.number_of_sequencers):
state = await self.instrument._get_sequencer_status_async(idx)
if state.state is SequencerStates.ARMED:
await self.instrument._start_sequencer_async(idx)
[docs]
def clear_data(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
return None
[docs]
async def clear_data_async(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
return None
[docs]
class _AnalogModuleComponent(_ModuleComponentBase):
"""Qblox InstrumentCoordinator component base class."""
[docs]
_hardware_properties: _StaticAnalogModuleProperties
def __init__(self, instrument: Module) -> None:
super().__init__(instrument)
if instrument.is_rf_type is not self._hardware_properties.has_internal_lo:
raise RuntimeError(
f"{self.__class__.__name__} not compatible with the "
"provided instrument. Please confirm whether your device "
"is a Qblox RF or baseband module (having or not having an "
"internal LO)."
)
@abstractmethod
[docs]
def _determine_channel_map_parameters(
self, settings: AnalogSequencerSettings
) -> dict[str, str]:
"""Returns a dictionary with the channel map parameters for this module."""
channel_map_parameters = {}
self._determine_output_channel_map_parameters(settings, channel_map_parameters)
return channel_map_parameters
[docs]
def _determine_output_channel_map_parameters(
self, settings: AnalogSequencerSettings, channel_map_parameters: dict[str, str]
) -> dict[str, str]:
"""Adds the outputs to the channel map parameters dict."""
for channel_idx in range(self._hardware_properties.number_of_output_channels):
param_setting = "off"
# For baseband, output indices map 1-to-1 to channel map indices
if (
len(settings.connected_output_indices) > 0
and channel_idx in settings.connected_output_indices
and channel_idx in settings.connected_output_indices
):
if ChannelMode.COMPLEX in settings.channel_name:
param_setting = "I" if (channel_idx % 2 == 0) else "Q"
elif ChannelMode.REAL in settings.channel_name:
param_setting = "I"
channel_map_parameters[f"connect_out{channel_idx}"] = param_setting
return channel_map_parameters
[docs]
def _upload(self, program: dict[str, dict]) -> None:
super()._upload(program)
for seq_name, settings in program["sequencers"].items():
if isinstance(settings, dict):
sequencer_settings = AnalogSequencerSettings.from_dict(settings)
else:
sequencer_settings = settings
if seq_name in self._seq_name_to_idx_map:
seq_idx = self._seq_name_to_idx_map[seq_name]
else:
raise KeyError(
f"Invalid program. Attempting to access non-existing sequencer "
f'with name "{seq_name}".'
)
self._upload_to_sequencer(seq_idx, sequencer_settings)
[docs]
async def _upload_async(self, program: dict[str, dict]) -> None:
await super()._upload_async(program)
for seq_name, settings in program["sequencers"].items():
if isinstance(settings, dict):
sequencer_settings = AnalogSequencerSettings.from_dict(settings)
else:
sequencer_settings = settings
if seq_name in self._seq_name_to_idx_map:
seq_idx = self._seq_name_to_idx_map[seq_name]
else:
raise KeyError(
f"Invalid program. Attempting to access non-existing sequencer "
f'with name "{seq_name}".'
)
await self._upload_to_sequencer_async(seq_idx, sequencer_settings)
[docs]
class _QCMComponent(_AnalogModuleComponent):
"""QCM specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QCM_BASEBAND_PROPERTIES
def __init__(self, instrument: Module) -> None:
if not instrument.is_qcm_type:
raise TypeError(
f"Trying to create _QCMComponent from non-QCM instrument "
f'of type "{type(instrument)}".'
)
super().__init__(instrument)
[docs]
def retrieve_acquisition(self) -> None:
"""
Retrieves the previous acquisition.
Returns
-------
:
QCM returns None since the QCM has no acquisition.
"""
return None
[docs]
async def retrieve_acquisition_async(self) -> None:
"""
Retrieves the previous acquisition.
Returns
-------
:
QCM returns None since the QCM has no acquisition.
"""
return None
[docs]
class _AnalogReadoutComponent(_AnalogModuleComponent):
"""Qblox InstrumentCoordinator readout component base class."""
def __init__(self, instrument: Module) -> None:
if not (instrument.is_qrm_type or instrument.is_qrc_type):
raise TypeError(
f"Trying to create _AnalogReadoutComponent from non-QRM or QRC instrument "
f'of type "{type(instrument)}".'
)
super().__init__(instrument)
[docs]
self._acquisition_manager: _QRMAcquisitionManager | None = None
"""Holds all the acquisition related logic."""
[docs]
def retrieve_acquisition(self) -> Dataset | None:
"""
Retrieves the latest acquisition results.
Returns
-------
:
The acquired data.
"""
if self._acquisition_manager:
return self._acquisition_manager.retrieve_acquisition()
else:
return None
[docs]
async def retrieve_acquisition_async(self) -> Dataset | None:
"""
Retrieves the latest acquisition results.
Returns
-------
:
The acquired data.
"""
if self._acquisition_manager:
return await self._acquisition_manager.retrieve_acquisition_async()
else:
return None
[docs]
def _determine_channel_map_parameters(
self, settings: AnalogSequencerSettings
) -> dict[str, str]:
"""Returns a dictionary with the channel map parameters for this module."""
channel_map_parameters = {}
self._determine_output_channel_map_parameters(settings, channel_map_parameters)
self._determine_input_channel_map_parameters(settings, channel_map_parameters)
return channel_map_parameters
[docs]
def _determine_scope_mode_acquisition_sequencer_and_qblox_acq_index(
self,
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
) -> tuple[int, int] | None:
"""
Finds the sequencer and qblox_acq_index that performs the raw trace acquisition.
Raises an error if multiple scope mode acquisitions are present per sequencer.
Note, that compiler ensures there is at most one scope mode acquisition,
however the user is able to freely modify the compiler program,
so we make sure this requirement is still satisfied. See
:func:`~qblox_scheduler.backends.qblox.analog.AnalogModuleCompiler._ensure_single_scope_mode_acquisition_sequencer`.
Returns
-------
:
The sequencer and qblox_acq_channel for the trace acquisition, if there is any,
otherwise None.
"""
sequencer_and_qblox_acq_index = None
for acq_channel, acq_channel_data in acq_channels_data.items():
if acq_channel_data.protocol == "Trace":
for sequencer_name, sequencer_hardware_mapping in acq_hardware_mapping.items():
# It's in the format "seq{n}", so we cut it.
sequencer_id = self._seq_name_to_idx_map[sequencer_name]
for (
hardware_mapping_acq_channel,
qblox_acq_index,
) in sequencer_hardware_mapping.non_binned.items():
if acq_channel == hardware_mapping_acq_channel:
assert isinstance(qblox_acq_index, int)
if (
sequencer_and_qblox_acq_index is not None
and sequencer_and_qblox_acq_index[0] != sequencer_id
):
single_scope_mode_acquisition_raise(
sequencer_0=sequencer_id,
sequencer_1=sequencer_and_qblox_acq_index[0],
module_name=self.name,
)
sequencer_and_qblox_acq_index = sequencer_id, qblox_acq_index
return sequencer_and_qblox_acq_index
[docs]
def clear_data(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
if self._acquisition_manager:
self._acquisition_manager.delete_acquisition_data()
[docs]
async def clear_data_async(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
if self._acquisition_manager:
await self._acquisition_manager.delete_acquisition_data_async()
[docs]
class _QRMComponent(_AnalogReadoutComponent):
"""QRM specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QRM_BASEBAND_PROPERTIES
[docs]
class _RFComponent(_AnalogModuleComponent):
"""Mix-in for RF-module-specific InstrumentCoordinatorComponent behaviour."""
[docs]
def _determine_output_channel_map_parameters(
self, settings: AnalogSequencerSettings, channel_map_parameters: dict[str, str]
) -> dict[str, str]:
"""Adds the outputs to the channel map parameters dict."""
expected_output_indices = {
i: (2 * i, 2 * i + 1)
for i in range(self._hardware_properties.number_of_output_channels)
}
for channel_idx in range(self._hardware_properties.number_of_output_channels):
param_setting = "off"
if (
ChannelMode.DIGITAL not in settings.channel_name
and len(settings.connected_output_indices) > 0
and tuple(settings.connected_output_indices)
== tuple(expected_output_indices[channel_idx])
):
param_setting = "IQ"
channel_map_parameters[f"connect_out{channel_idx}"] = param_setting
return channel_map_parameters
[docs]
def _get_connected_lo_idx_for_sequencer(
self, sequencer_settings: AnalogSequencerSettings
) -> list[int]:
"""
Looks at the connected _output_ ports of the sequencer (if any) to determine
which LO this sequencer's output is coupled to.
"""
channel_map_parameters = self._determine_output_channel_map_parameters(
sequencer_settings, channel_map_parameters={}
)
connected_lo_idx = [
channel_idx
for channel_idx in range(self._hardware_properties.number_of_output_channels)
if channel_map_parameters.get(f"connect_out{channel_idx}") == "IQ"
]
return connected_lo_idx
@abstractmethod
[docs]
class _QCMRFComponent(_RFComponent, _QCMComponent):
"""QCM-RF specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QCM_RF_PROPERTIES
[docs]
class _QRMRFComponent(_RFComponent, _QRMComponent):
"""QRM-RF specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QRM_RF_PROPERTIES
[docs]
class _QRCComponent(_RFComponent, _AnalogReadoutComponent):
"""QRC specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QRC_PROPERTIES
[docs]
class _QTMComponent(_ModuleComponentBase):
"""QTM specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QTM_PROPERTIES
def __init__(self, instrument: Module) -> None:
if not instrument.is_qtm_type:
raise TypeError(
f"Trying to create _QTMComponent from non-QTM instrument "
f'of type "{type(instrument)}".'
)
super().__init__(instrument)
[docs]
self._acquisition_manager: _QTMAcquisitionManager | None = None
"""Holds all the acquisition related logic."""
[docs]
def retrieve_acquisition(self) -> Dataset | None:
"""
Retrieves the latest acquisition results.
Returns
-------
:
The acquired data.
"""
if self._acquisition_manager:
return self._acquisition_manager.retrieve_acquisition()
else:
return None
[docs]
async def retrieve_acquisition_async(self) -> Dataset | None:
"""
Retrieves the latest acquisition results.
Returns
-------
:
The acquired data.
"""
if self._acquisition_manager:
return await self._acquisition_manager.retrieve_acquisition_async()
else:
return None
# No global settings yet.
[docs]
def _upload(self, program: dict[str, dict]) -> None:
super()._upload(program)
for seq_name, settings in program["sequencers"].items():
if seq_name in self._seq_name_to_idx_map:
seq_idx = self._seq_name_to_idx_map[seq_name]
else:
raise KeyError(
f"Invalid program. Attempting to access non-existing sequencer "
f'with name "{seq_name}".'
)
self._upload_to_sequencer(seq_idx, settings)
[docs]
async def _upload_async(self, program: dict[str, dict]) -> None:
await super()._upload_async(program)
for seq_name, settings in program["sequencers"].items():
if seq_name in self._seq_name_to_idx_map:
seq_idx = self._seq_name_to_idx_map[seq_name]
else:
raise KeyError(
f"Invalid program. Attempting to access non-existing sequencer "
f'with name "{seq_name}".'
)
await self._upload_to_sequencer_async(seq_idx, settings)
[docs]
def clear_data(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
if self._acquisition_manager:
self._acquisition_manager.delete_acquisition_data()
[docs]
async def clear_data_async(self) -> None:
"""Clears remaining data on the module. Module type specific function."""
if self._acquisition_manager:
await self._acquisition_manager.delete_acquisition_data_async()
[docs]
class _QSMComponent(_ModuleComponentBase):
"""QSM specific InstrumentCoordinator component."""
[docs]
_hardware_properties = _QSM_PROPERTIES
@property
[docs]
def is_running(self) -> bool:
"""Do nothing, the QSM has no sequencers."""
return False
[docs]
def wait_done(self, timeout_sec: int = 10) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
def disable_sync(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
async def disable_sync_async(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
def arm_all_sequencers_in_program(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
async def arm_all_sequencers_in_program_async(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
def start(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
async def start_async(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
def stop(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
async def stop_async(self) -> None:
"""Do nothing, the QSM has no sequencers."""
pass
[docs]
def retrieve_acquisition(self) -> Dataset | None:
"""Gets and returns acquisition data."""
pass
[docs]
async def retrieve_acquisition_async(self) -> Dataset | None:
"""Gets and returns acquisition data."""
pass
[docs]
_ReadoutModuleComponentT = Union[_AnalogReadoutComponent, _QTMComponent]
[docs]
class _AcquisitionManagerBase(ABC):
"""
Utility class that handles the acquisitions performed with a module.
An instance of this class is meant to exist only for a single prepare-start-
retrieve_acquisition cycle to prevent stateful behavior.
Parameters
----------
parent
Reference to the parent QRM IC component.
acq_channels_data
Provides a summary of the used acquisition protocol, bin mode, acquisition channels,
acquisition indices per channel, and repetitions.
acq_hardware_mapping
Acquisition hardware mapping.
acquisition_duration
The duration of each acquisition for each sequencer.
seq_name_to_idx_map
All available sequencer names to their ids in a dict.
repetitions
How many times the schedule repeats.
"""
def __init__(
self,
parent: _ReadoutModuleComponentT,
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
acquisition_duration: dict[str, int],
seq_name_to_idx_map: dict[str, int],
repetitions: int,
) -> None:
[docs]
self._acq_channels_data = acq_channels_data
[docs]
self._acq_hardware_mapping = acq_hardware_mapping
[docs]
self._acq_duration = acquisition_duration
[docs]
self._seq_name_to_idx_map = seq_name_to_idx_map
[docs]
self._repetitions = repetitions
@property
[docs]
def instrument(self) -> Module:
"""Returns the QRM driver from the parent IC component."""
return self.parent.instrument
@staticmethod
@abstractmethod
[docs]
def _check_bin_mode_compatible(
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
) -> None:
pass
@abstractmethod
[docs]
def _protocol_to_acq_function_map_non_binned(self, protocol: str) -> Callable:
"""
Mapping from acquisition protocol name to the function that processes the raw
acquisition data.
"""
@abstractmethod
[docs]
def _protocol_to_acq_async_function_map_non_binned(self, protocol: str) -> Callable:
"""
Mapping from acquisition protocol name to the async function that processes the raw
acquisition data.
"""
@abstractmethod
[docs]
def _protocol_to_bin_function(self, protocol: str) -> Callable:
pass
@abstractmethod
[docs]
def _retrieve_acquisition_binned_recursive(
self,
node: BinnedAcqControlFlowNode | BinnedAcqInfo,
qblox_acq_bin: int,
qblox_acq_index: int,
hardware_retrieved_acquisitions: dict,
acq_duration: int,
acq_index_offset: int,
total_average_repetitions: int,
dataset_dict: dict[Hashable, tuple[list, list]],
) -> int:
"""
Adds the data to dataset_dict (updates the argument) from hardware_retrieved_acquisitions,
so that the data includes the acquisition channel and acquisition index
based on the data in the node.
Note: it updates the dataset_dict argument.
The dataset_dict is a dict with acq_channel->(acq_index, acquisition data).
This simple data format makes sure that we can process the data quickly,
and later we can merge this into an xarray.Dataset, which is time consuming.
"""
if isinstance(node, BinnedAcqControlFlowNode):
append_repetitions = (
node.repetitions
if ((node.repetitions is not None) and (node.bin_mode is BinMode.APPEND))
else 1
)
average_repetitions = (
node.repetitions
if ((node.repetitions is not None) and (node.bin_mode is BinMode.AVERAGE))
else 1
)
for i in range(append_repetitions):
for child in node.children:
qblox_acq_bin = self._retrieve_acquisition_binned_recursive(
node=child,
qblox_acq_bin=qblox_acq_bin,
qblox_acq_index=qblox_acq_index,
hardware_retrieved_acquisitions=hardware_retrieved_acquisitions,
acq_duration=acq_duration,
acq_index_offset=append_repetitions * acq_index_offset + i,
total_average_repetitions=total_average_repetitions * average_repetitions,
dataset_dict=dataset_dict,
)
else:
assert isinstance(node, BinnedAcqInfo)
self._assert_acquisition_data_exists(
hardware_retrieved_acquisitions, qblox_acq_index, node.acq_channel
)
protocol = self._acq_channels_data[node.acq_channel].protocol
data = self._protocol_to_bin_function(protocol)(
qblox_acq_index,
qblox_acq_bin,
node.thresholded_trigger_count_metadata,
hardware_retrieved_acquisitions,
acq_duration,
acq_channel=node.acq_channel,
total_average_repetitions=total_average_repetitions,
)
if node.acq_channel not in dataset_dict:
dataset_dict[node.acq_channel] = ([], [])
acq_indices = node.acq_index.acq_index
assert isinstance(acq_indices, list) # For binned acquisitions it is a list.
acq_index = acq_indices[acq_index_offset]
dataset_dict[node.acq_channel][0].append(acq_index)
dataset_dict[node.acq_channel][1].append(data)
qblox_acq_bin += 1
return qblox_acq_bin
[docs]
def _retrieve_acquisition_binned(
self,
binned_mapping: QbloxAcquisitionHardwareMappingBinned,
hardware_retrieved_acquisitions: dict,
acq_duration: int,
) -> Dataset:
dataset_dict = {}
self._retrieve_acquisition_binned_recursive(
node=binned_mapping.tree,
qblox_acq_bin=binned_mapping.qblox_acq_bin_offset,
qblox_acq_index=binned_mapping.qblox_acq_index,
hardware_retrieved_acquisitions=hardware_retrieved_acquisitions,
acq_duration=acq_duration,
acq_index_offset=0,
total_average_repetitions=1,
dataset_dict=dataset_dict,
)
dataset = Dataset()
for acq_channel in dataset_dict:
acq_indices = dataset_dict[acq_channel][0]
acq_data = dataset_dict[acq_channel][1]
acq_index_dim_name = self._acq_channels_data[acq_channel].acq_index_dim_name
data_array = DataArray(
acq_data,
dims=[acq_index_dim_name],
coords={
acq_index_dim_name: acq_indices,
},
attrs=self._acq_channel_attrs(
self._acq_channels_data[acq_channel].protocol, acq_index_dim_name
),
)
coords = self._acq_channels_data[acq_channel].coords
add_acquisition_coords_binned(data_array, coords, acq_index_dim_name)
dataset = dataset.merge({acq_channel: data_array}, compat="no_conflicts", join="outer")
return dataset
[docs]
def retrieve_acquisition(self) -> Dataset:
"""
Retrieves all the acquisition data in the correct format.
Returns
-------
:
The acquisitions with the protocols specified in the `acquisition_metadata`.
Each `xarray.DataArray` in the `xarray.Dataset` corresponds to one `acq_channel`.
The ``acq_channel`` is the name of each `xarray.DataArray` in the `xarray.Dataset`.
Each `xarray.DataArray` is a two-dimensional array, with ``acq_index`` and
Each `xarray.DataArray` is a two-dimensional array,
with ``acq_index`` and ``repetition`` as dimensions.
"""
dataset = Dataset()
self._check_retrieval_conditions()
for sequencer_name, seq_hardware_mapping in self._acq_hardware_mapping.items():
if (
len(seq_hardware_mapping.binned) == 0
and len(seq_hardware_mapping.non_binned) == 0
and seq_hardware_mapping.timetagtrace is None
):
continue
# Retrieve the raw data from the readout sequencer.
hardware_retrieved_acquisitions = self.instrument.get_acquisitions(
self._seq_name_to_idx_map[sequencer_name]
)
# Merge binned data.
dataset = self._merge_retrieved_binned_data(
dataset,
sequencer_name,
seq_hardware_mapping.binned,
hardware_retrieved_acquisitions,
)
# Merge unbinned data.
self._check_retrieved_unbinned_data(
seq_hardware_mapping, hardware_retrieved_acquisitions
)
dataset = self._merge_retrieved_unbinned_data(
dataset,
sequencer_name,
seq_hardware_mapping.non_binned,
hardware_retrieved_acquisitions,
)
# Merge timetagtrace data.
if seq_hardware_mapping.timetagtrace is not None:
seq_idx = self._seq_name_to_idx_map[sequencer_name]
scope_data = self.instrument.io_channels[seq_idx].get_scope_data()
dataset = self._merge_timetag_trace_data(
hardware_retrieved_acquisitions,
seq_hardware_mapping.timetagtrace,
scope_data,
dataset,
)
return dataset
[docs]
async def retrieve_acquisition_async(self) -> Dataset:
"""
Retrieves all the acquisition data in the correct format.
Returns
-------
:
The acquisitions with the protocols specified in the `acquisition_metadata`.
Each `xarray.DataArray` in the `xarray.Dataset` corresponds to one `acq_channel`.
The ``acq_channel`` is the name of each `xarray.DataArray` in the `xarray.Dataset`.
Each `xarray.DataArray` is a two-dimensional array, with ``acq_index`` and
Each `xarray.DataArray` is a two-dimensional array,
with ``acq_index`` and ``repetition`` as dimensions.
"""
dataset = Dataset()
self._check_retrieval_conditions()
for sequencer_name, seq_hardware_mapping in self._acq_hardware_mapping.items():
if (
len(seq_hardware_mapping.binned) == 0
and len(seq_hardware_mapping.non_binned) == 0
and seq_hardware_mapping.timetagtrace is None
):
continue
# Retrieve the raw data from the readout sequencer.
hardware_retrieved_acquisitions = await self.instrument._get_acquisitions_async(
self._seq_name_to_idx_map[sequencer_name]
)
# Merge binned data.
dataset = self._merge_retrieved_binned_data(
dataset,
sequencer_name,
seq_hardware_mapping.binned,
hardware_retrieved_acquisitions,
)
# Merge unbinned data.
self._check_retrieved_unbinned_data(
seq_hardware_mapping, hardware_retrieved_acquisitions
)
dataset = await self._merge_retrieved_unbinned_data_async(
dataset,
sequencer_name,
seq_hardware_mapping.non_binned,
hardware_retrieved_acquisitions,
)
# Merge timetagtrace data.
if seq_hardware_mapping.timetagtrace is not None:
seq_idx = self._seq_name_to_idx_map[sequencer_name]
scope_data = await self.instrument.io_channels[seq_idx].get_scope_data_async()
dataset = self._merge_timetag_trace_data(
hardware_retrieved_acquisitions,
seq_hardware_mapping.timetagtrace,
scope_data,
dataset,
)
return dataset
[docs]
def _check_retrieval_conditions(self) -> None:
self._check_bin_mode_compatible(self._acq_channels_data, self._acq_hardware_mapping)
[docs]
def _merge_retrieved_binned_data(
self,
dataset: Dataset,
sequencer_name: str,
seq_hardware_mapping_binned: list[QbloxAcquisitionHardwareMappingBinned],
hardware_retrieved_acquisitions: dict,
) -> Dataset:
for binned_mapping in seq_hardware_mapping_binned:
new_dataset = self._retrieve_acquisition_binned(
binned_mapping,
hardware_retrieved_acquisitions,
acq_duration=self._acq_duration[sequencer_name],
)
dataset = dataset.merge(new_dataset, join="outer", compat="no_conflicts")
return dataset
[docs]
def _check_retrieved_unbinned_data(
self,
seq_hardware_mapping: QbloxAcquisitionHardwareMapping,
hardware_retrieved_acquisitions: dict,
) -> None:
for (
acq_channel,
seq_channel_hardware_mapping,
) in seq_hardware_mapping.non_binned.items():
assert isinstance(seq_channel_hardware_mapping, QbloxAcquisitionIndex)
qblox_acq_index = seq_channel_hardware_mapping
self._assert_acquisition_data_exists(
hardware_retrieved_acquisitions, qblox_acq_index, acq_channel
)
[docs]
def _merge_retrieved_unbinned_data(
self,
dataset: Dataset,
sequencer_name: str,
seq_hardware_mapping_non_binned: dict[Hashable, QbloxAcquisitionIndex],
hardware_retrieved_acquisitions: dict,
) -> Dataset:
for (
acq_channel,
seq_channel_hardware_mapping,
) in seq_hardware_mapping_non_binned.items():
acquisition_function: Callable = self._protocol_to_acq_function_map_non_binned(
self._acq_channels_data[acq_channel].protocol
)
# The `acquisition_function` formats the the acquisitions
# returned by the readout sequencer.
formatted_acquisitions = acquisition_function(
hardware_retrieved_acquisitions=hardware_retrieved_acquisitions,
acq_channel=acq_channel,
seq_channel_hardware_mapping=seq_channel_hardware_mapping,
acq_duration=self._acq_duration[sequencer_name],
sequencer_name=sequencer_name,
)
formatted_acquisitions_dataset = Dataset({acq_channel: formatted_acquisitions})
check_already_existing_acquisition(
new_dataset=formatted_acquisitions_dataset, current_dataset=dataset
)
dataset = dataset.merge(
formatted_acquisitions_dataset, join="outer", compat="no_conflicts"
)
return dataset
[docs]
async def _merge_retrieved_unbinned_data_async(
self,
dataset: Dataset,
sequencer_name: str,
seq_hardware_mapping_non_binned: dict[Hashable, QbloxAcquisitionIndex],
hardware_retrieved_acquisitions: dict,
) -> Dataset:
for (
acq_channel,
seq_channel_hardware_mapping,
) in seq_hardware_mapping_non_binned.items():
acquisition_function: Callable = self._protocol_to_acq_async_function_map_non_binned(
self._acq_channels_data[acq_channel].protocol
)
# The `acquisition_function` formats the the acquisitions
# returned by the readout sequencer.
formatted_acquisitions = await acquisition_function(
hardware_retrieved_acquisitions=hardware_retrieved_acquisitions,
acq_channel=acq_channel,
seq_channel_hardware_mapping=seq_channel_hardware_mapping,
acq_duration=self._acq_duration[sequencer_name],
sequencer_name=sequencer_name,
)
formatted_acquisitions_dataset = Dataset({acq_channel: formatted_acquisitions})
check_already_existing_acquisition(
new_dataset=formatted_acquisitions_dataset, current_dataset=dataset
)
dataset = dataset.merge(
formatted_acquisitions_dataset, join="outer", compat="no_conflicts"
)
return dataset
[docs]
def delete_acquisition_data(self) -> None:
"""
Delete acquisition data from sequencers that have associated hardware acquisition mapping.
To be called before starting the sequencers, so that old data does not get retrieved more
than once.
"""
for seq_name, mapping in self._acq_hardware_mapping.items():
if len(mapping.binned) or len(mapping.non_binned) or mapping.timetagtrace is not None:
self.instrument.delete_acquisition_data(
sequencer=self._seq_name_to_idx_map[seq_name], all=True
)
[docs]
async def delete_acquisition_data_async(self) -> None:
"""
Delete acquisition data from sequencers that have associated hardware acquisition mapping.
To be called before starting the sequencers, so that old data does not get retrieved more
than once.
"""
seq_indices = [
self._seq_name_to_idx_map[seq_name]
for seq_name, mapping in self._acq_hardware_mapping.items()
if len(mapping.binned) or len(mapping.non_binned)
]
await asyncio.gather(
*[
self.instrument._delete_acquisition_data_async(sequencer=seq_idx, all=True)
for seq_idx in seq_indices
]
)
[docs]
def _assert_acquisition_data_exists(
self,
hardware_retrieved_acquisitions: dict,
qblox_acq_index: int,
acq_channel: Hashable,
) -> None:
"""Assert that the qblox_acq_index is in the acquisition data."""
qblox_acq_name = self._qblox_acq_index_to_qblox_acq_name(qblox_acq_index)
if qblox_acq_name not in hardware_retrieved_acquisitions:
raise KeyError(
f"The acquisition data retrieved from the hardware does not contain "
f"data for acquisition channel {acq_channel} (referred to by Qblox "
f"acquisition index {qblox_acq_index}).\n"
f"{hardware_retrieved_acquisitions=}"
)
@staticmethod
[docs]
def _acq_channel_attrs(
protocol: str,
acq_index_dim_name: str,
) -> dict:
return {"acq_protocol": protocol, "acq_index_dim_name": acq_index_dim_name}
@classmethod
[docs]
def _get_bin_data(cls, hardware_retrieved_acquisitions: dict, qblox_acq_index: int = 0) -> dict:
"""Returns the bin entry of the acquisition data dict."""
qblox_acq_name = cls._qblox_acq_index_to_qblox_acq_name(qblox_acq_index)
channel_data = hardware_retrieved_acquisitions[qblox_acq_name]
if channel_data["index"] != qblox_acq_index:
raise RuntimeError(
f"Name does not correspond to a valid acquisition for name {qblox_acq_name}, "
f"which has index {channel_data['index']}."
)
return channel_data["acquisition"]["bins"]
@staticmethod
[docs]
def _qblox_acq_index_to_qblox_acq_name(qblox_acq_index: int) -> str:
"""Returns the name of the acquisition from the qblox_acq_index."""
return str(qblox_acq_index)
[docs]
def _get_trigger_count_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None, # noqa: ARG002, unused parameter
hardware_retrieved_acquisitions: dict,
acq_duration: int, # noqa: ARG002, unused parameter
acq_channel: Hashable,
sum_multiply_repetitions: bool,
total_average_repetitions: int,
) -> int:
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
# "avg_cnt" is the key for QRM modules, "count" for QTM modules. Note
# that QTM modules also return a "avg_cnt" key, that should not be used!
counts = bin_data["count"] if "count" in bin_data else bin_data["avg_cnt"]
if np.isnan(counts[qblox_acq_bin]):
return -1
else:
return (
self._repetitions * total_average_repetitions * counts[qblox_acq_bin]
if (
sum_multiply_repetitions
and (self._acq_channels_data[acq_channel].bin_mode == BinMode.AVERAGE_APPEND)
)
else round(counts[qblox_acq_bin])
)
[docs]
def _get_trigger_count_distribution_data(
self,
*,
hardware_retrieved_acquisitions: dict,
acq_channel: str,
seq_channel_hardware_mapping: QbloxAcquisitionIndex,
acq_duration: int, # noqa: ARG002, unused parameter
sequencer_name: str, # noqa: ARG002, unused parameter
) -> DataArray:
"""
Retrieves the trigger count acquisition data associated with `acq_channel`.
Parameters
----------
hardware_retrieved_acquisitions
The acquisitions dict as returned by the sequencer.
acq_channel
The acquisition channel.
seq_channel_hardware_mapping
Acquisition hardware mapping for the sequencer and channel.
acq_duration
Desired maximum number of samples for the scope acquisition.
sequencer_name
Sequencer.
sum_multiply_repetitions
Multiplies data by repetitions for the AVERAGE_APPEND bin mode.
Returns
-------
data : xarray.DataArray
The acquired trigger count data.
Notes
-----
- For BinMode.DISTRIBUTION, `data` contains the distribution of counts.
- For BinMode.APPEND, `data` contains the raw trigger counts.
"""
if self._acq_channels_data[acq_channel].bin_mode == BinMode.DISTRIBUTION:
# For TriggerCount distribution acquisition protocol,
# the mapping is of QbloxAcquisitionIndex type,
# guaranteed by the QbloxAcquisitionIndexManager.
assert isinstance(seq_channel_hardware_mapping, int)
qblox_acq_index = seq_channel_hardware_mapping
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
def _convert_from_cumulative(
cumulative_values: list[int],
) -> dict[int, int]:
"""
Return the distribution of counts from a cumulative distribution.
Note, the cumulative distribution is in reverse order.
The cumulative_values list can contain any number of integers and NaNs.
"""
result = {}
last_cumulative_value = 0
for count, current_cumulative_value in reversed(list(enumerate(cumulative_values))):
if (not isnan(current_cumulative_value)) and (
last_cumulative_value != current_cumulative_value
):
result[count + 1] = current_cumulative_value - last_cumulative_value
last_cumulative_value = current_cumulative_value
return result
result = _convert_from_cumulative(bin_data["avg_cnt"])
acq_index_dim_name = self._acq_channels_data[acq_channel].acq_index_dim_name
counts = list(result.keys())[::-1]
coords = self._acq_channels_data[acq_channel].coords
assert isinstance(coords, dict)
data_array = DataArray(
list(result.values())[::-1],
dims=[acq_index_dim_name],
coords={
acq_index_dim_name: range(len(counts)),
f"counts_{acq_channel}": (acq_index_dim_name, counts),
},
attrs=self._acq_channel_attrs(
self._acq_channels_data[acq_channel].protocol, acq_index_dim_name
),
)
add_acquisition_coords_nonbinned(data_array, coords, acq_index_dim_name)
return data_array
else:
# In principle unreachable due to _check_bin_mode_compatible, but included for
# completeness.
raise AssertionError("This should not be reachable due to _check_bin_mode_compatible.")
[docs]
def _get_thresholded_trigger_count_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None,
hardware_retrieved_acquisitions: dict,
acq_duration: int,
acq_channel: Hashable,
sum_multiply_repetitions: bool,
total_average_repetitions: int,
) -> int:
assert isinstance(thresholded_trigger_count_metadata, ThresholdedTriggerCountMetadata)
count = self._get_trigger_count_bin(
qblox_acq_index=qblox_acq_index,
qblox_acq_bin=qblox_acq_bin,
thresholded_trigger_count_metadata=thresholded_trigger_count_metadata,
hardware_retrieved_acquisitions=hardware_retrieved_acquisitions,
acq_duration=acq_duration,
acq_channel=acq_channel,
sum_multiply_repetitions=sum_multiply_repetitions,
total_average_repetitions=total_average_repetitions,
)
count = round(count)
if count == -1:
return -1
elif thresholded_trigger_count_metadata.condition == TriggerCondition.GREATER_THAN_EQUAL_TO:
return int(count >= thresholded_trigger_count_metadata.threshold)
elif thresholded_trigger_count_metadata.condition == TriggerCondition.LESS_THAN:
return int(count < thresholded_trigger_count_metadata.threshold)
else:
raise ValueError(
f"Unknown trigger condition {thresholded_trigger_count_metadata.condition}"
)
[docs]
class _QRMAcquisitionManager(_AcquisitionManagerBase):
"""
Utility class that handles the acquisitions performed with the QRM and QRC.
An instance of this class is meant to exist only for a single prepare-start-
retrieve_acquisition cycle to prevent stateful behavior.
Parameters
----------
parent
Reference to the parent QRM or QRC IC component.
acq_channels_data
Provides a summary of the used acquisition protocol, bin mode, acquisition channels,
acquisition indices per channel.
acq_hardware_mapping
Acquisition hardware mapping.
acquisition_duration
The duration of each acquisition for each sequencer.
seq_name_to_idx_map
All available sequencer names to their ids in a dict.
scope_mode_sequencer_and_qblox_acq_index
The sequencer and qblox acq_index of the scope mode acquisition if there's any.
sequencers
Sequencer data.
"""
def __init__(
self,
parent: _AnalogReadoutComponent,
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
acquisition_duration: dict[str, int],
seq_name_to_idx_map: dict[str, int],
repetitions: int,
scope_mode_sequencer_and_qblox_acq_index: tuple[int, int] | None = None,
sequencers: dict[str, dict] | None = None,
) -> None:
super().__init__(
parent=parent,
acq_channels_data=acq_channels_data,
acq_hardware_mapping=acq_hardware_mapping,
acquisition_duration=acquisition_duration,
seq_name_to_idx_map=seq_name_to_idx_map,
repetitions=repetitions,
)
[docs]
self._scope_mode_sequencer_and_qblox_acq_index = scope_mode_sequencer_and_qblox_acq_index
[docs]
self._sequencers = sequencers
[docs]
def _protocol_to_bin_function(self, protocol: str) -> Callable:
match protocol:
case "WeightedIntegratedSeparated" | "NumericalSeparatedWeightedIntegration":
return self._get_integration_weighted_separated_bin
case "NumericalWeightedIntegration":
return self._get_integration_real_bin
case "SSBIntegrationComplex":
return self._get_integration_amplitude_bin
case "ThresholdedAcquisition" | "WeightedThresholdedAcquisition":
return self._get_thresholded_bin
case "TriggerCount":
return partial(self._get_trigger_count_bin, sum_multiply_repetitions=False)
case "ThresholdedTriggerCount":
return partial(
self._get_thresholded_trigger_count_bin, sum_multiply_repetitions=False
)
case _:
raise AssertionError(
"This should not be reachable due to _check_bin_mode_compatible."
)
[docs]
def _protocol_to_acq_function_map_non_binned(self, protocol: str) -> Callable:
match protocol:
case "Trace":
return self._get_scope_data
case "TriggerCount":
return self._get_trigger_count_distribution_data
case _:
raise AssertionError(
"This should not be reachable due to _check_bin_mode_compatible."
)
[docs]
def _protocol_to_acq_async_function_map_non_binned(self, protocol: str) -> Callable:
# No further data retrieval needed for QRM.
sync_fn = self._protocol_to_acq_function_map_non_binned(protocol)
@wraps(sync_fn)
async def async_wrapper(*args, **kwargs) -> Any: # noqa: ANN401, Any return type
return await asyncio.to_thread(sync_fn, *args, **kwargs)
return async_wrapper
@staticmethod
[docs]
def _check_bin_mode_compatible_binned(
acq_channels_data: AcquisitionChannelsData,
node: BinnedAcqControlFlowNode | BinnedAcqInfo,
) -> None:
if isinstance(node, BinnedAcqControlFlowNode):
for c in node.children:
_QRMAcquisitionManager._check_bin_mode_compatible_binned(acq_channels_data, c)
else:
acq_channel = node.acq_channel
if (
acq_channels_data[acq_channel].bin_mode
not in QRM_COMPATIBLE_BIN_MODES[acq_channels_data[acq_channel].protocol]
):
raise IncompatibleBinModeError(
module_type="QRM",
protocol=acq_channels_data[acq_channel].protocol,
bin_mode=acq_channels_data[acq_channel].bin_mode,
)
@staticmethod
[docs]
def _check_bin_mode_compatible(
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
) -> None:
for seq_acq_hardware_mapping in acq_hardware_mapping.values():
for acq_hardware_mapping_binned in seq_acq_hardware_mapping.binned:
_QRMAcquisitionManager._check_bin_mode_compatible_binned(
acq_channels_data, acq_hardware_mapping_binned.tree
)
for acq_channel in seq_acq_hardware_mapping.non_binned.keys() | (
{seq_acq_hardware_mapping.timetagtrace[0]}
if seq_acq_hardware_mapping.timetagtrace
else {}
):
if (
acq_channels_data[acq_channel].bin_mode
not in QRM_COMPATIBLE_BIN_MODES[acq_channels_data[acq_channel].protocol]
):
raise IncompatibleBinModeError(
module_type="QRM",
protocol=acq_channels_data[acq_channel].protocol,
bin_mode=acq_channels_data[acq_channel].bin_mode,
)
[docs]
def retrieve_acquisition(self) -> Dataset:
"""
Retrieves all the acquisition data in the correct format.
Returns
-------
:
The acquisitions with the protocols specified in the `acq_channels_data`.
Each `xarray.DataArray` in the `xarray.Dataset` corresponds to one `acq_channel`.
The ``acq_channel`` is the name of each `xarray.DataArray` in the `xarray.Dataset`.
Each `xarray.DataArray` is a two-dimensional array,
with ``acq_index`` and ``repetition`` as dimensions.
"""
self._store_scope_acquisition()
return super().retrieve_acquisition()
[docs]
async def retrieve_acquisition_async(self) -> Dataset:
"""
Retrieves all the acquisition data in the correct format.
Returns
-------
:
The acquisitions with the protocols specified in the `acq_channels_data`.
Each `xarray.DataArray` in the `xarray.Dataset` corresponds to one `acq_channel`.
The ``acq_channel`` is the name of each `xarray.DataArray` in the `xarray.Dataset`.
Each `xarray.DataArray` is a two-dimensional array,
with ``acq_index`` and ``repetition`` as dimensions.
"""
await self._store_scope_acquisition_async()
return await super().retrieve_acquisition_async()
[docs]
def _store_scope_acquisition(self) -> None:
"""
Calls :code:`store_scope_acquisition` function on the Qblox instrument.
This will ensure that the correct sequencer will store the scope acquisition
data on the hardware, so it will be filled out when we call :code:`get_acquisitions`
on the Qblox instrument's sequencer corresponding to the scope acquisition.
"""
if (maybe_scope_name := self._get_scope_acquisition_name()) is None:
return
sequencer_index, qblox_acq_name = maybe_scope_name
self.instrument.store_scope_acquisition(sequencer_index, qblox_acq_name)
[docs]
async def _store_scope_acquisition_async(self) -> None:
"""
Calls :code:`store_scope_acquisition` function on the Qblox instrument.
This will ensure that the correct sequencer will store the scope acquisition
data on the hardware, so it will be filled out when we call :code:`get_acquisitions`
on the Qblox instrument's sequencer corresponding to the scope acquisition.
"""
if (maybe_scope_name := self._get_scope_acquisition_name()) is None:
return
sequencer_index, qblox_acq_name = maybe_scope_name
await self.instrument._store_scope_acquisition_async(sequencer_index, qblox_acq_name)
[docs]
def _get_scope_acquisition_name(self) -> tuple[int, str] | None:
if self._scope_mode_sequencer_and_qblox_acq_index is None:
return
sequencer_index = self._scope_mode_sequencer_and_qblox_acq_index[0]
if sequencer_index not in self._seq_name_to_idx_map.values():
raise ValueError(
f"Attempting to retrieve scope mode data from sequencer "
f"{sequencer_index}. A QRM only has the following sequencer indices: "
f"{list(self._seq_name_to_idx_map.values())}."
)
qblox_acq_index = self._scope_mode_sequencer_and_qblox_acq_index[1]
qblox_acq_name = self._qblox_acq_index_to_qblox_acq_name(qblox_acq_index)
return sequencer_index, qblox_acq_name
[docs]
def _get_scope_data(
self,
*,
hardware_retrieved_acquisitions: dict,
acq_channel: str,
seq_channel_hardware_mapping: QbloxAcquisitionIndex,
acq_duration: int,
sequencer_name: str,
) -> DataArray:
"""
Retrieves the scope mode acquisition associated with an `acq_channel`.
Parameters
----------
hardware_retrieved_acquisitions
The acquisitions dict as returned by the sequencer.
acq_channel
The acquisition channel.
seq_channel_hardware_mapping
Acquisition hardware mapping for the sequencer and channel.
acq_duration
Desired maximum number of samples for the scope acquisition.
sequencer_name
Sequencer.
Returns
-------
:
The scope mode data.
"""
if acq_duration < 0 or acq_duration > constants.MAX_SAMPLE_SIZE_SCOPE_ACQUISITIONS:
raise ValueError(
"Attempting to retrieve sample of size "
f"{acq_duration}, but only integer values "
f"0,...,{constants.MAX_SAMPLE_SIZE_SCOPE_ACQUISITIONS} "
f"are allowed."
)
# For scope acquisition protocols, the mapping is of QbloxAcquisitionIndex type,
# guaranteed by the QbloxAcquisitionIndexManager and coords is a dict.
assert isinstance(seq_channel_hardware_mapping, int)
coords = self._acq_channels_data[acq_channel].coords
assert isinstance(coords, dict)
qblox_acq_index = seq_channel_hardware_mapping
qblox_acq_name = self._qblox_acq_index_to_qblox_acq_name(qblox_acq_index)
scope_data = hardware_retrieved_acquisitions[qblox_acq_name]["acquisition"]["scope"]
path0 = scope_data["path0"]
path1 = scope_data["path1"]
if (self._sequencers is not None) and (
connected_sequencer := self._sequencers.get(sequencer_name)
) is not None:
connected_input_indices = getattr(connected_sequencer, "connected_input_indices", None)
if connected_input_indices == (2, 3):
path0 = scope_data["path2"]
path1 = scope_data["path3"]
elif connected_input_indices != (0, 1):
logger.warning(
"Invalid input indices are connected. Scope data might be invalid. "
"Connected input indices are %s. "
"Valid indices are (0, 1) and (2, 3).",
connected_input_indices,
)
if path0["out-of-range"] or path1["out-of-range"]:
logger.warning(
"The scope mode data of %s with acq_channel=%s was out-of-range.",
self.parent.name,
acq_channel,
)
# NB hardware already divides by avg_count for scope mode
scope_data_i = np.array(path0["data"][:acq_duration])
scope_data_q = np.array(path1["data"][:acq_duration])
# If using a dummy setup and there's no scope data set, then return NaN's
if self.parent.instrument.is_dummy and (len(scope_data_i) == len(scope_data_q) == 0):
scope_data_i = np.asarray([float("nan")] * acq_duration)
scope_data_q = np.asarray([float("nan")] * acq_duration)
acq_index_dim_name = self._acq_channels_data[acq_channel].acq_index_dim_name
trace_index_dim_name = f"trace_index_{acq_channel}"
data_array = DataArray(
data=(scope_data_i + scope_data_q * 1j).reshape((1, -1)),
dims=[acq_index_dim_name, trace_index_dim_name],
coords={
acq_index_dim_name: [0],
trace_index_dim_name: list(range(acq_duration)),
},
attrs=self._acq_channel_attrs(
self._acq_channels_data[acq_channel].protocol, acq_index_dim_name
),
)
add_acquisition_coords_nonbinned(data_array, coords, acq_index_dim_name)
return data_array
[docs]
def _get_integration_weighted_separated_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None, # noqa: ARG002, unused parameter
hardware_retrieved_acquisitions: dict,
acq_duration: int, # noqa: ARG002, unused parameter
acq_channel: Hashable, # noqa: ARG002, unused parameter
total_average_repetitions: int, # noqa: ARG002, unused parameter
) -> complex:
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
i_data = bin_data["integration"]["path0"][qblox_acq_bin]
q_data = bin_data["integration"]["path1"][qblox_acq_bin]
return i_data + q_data * 1j
[docs]
def _get_integration_amplitude_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None, # noqa: ARG002, unused parameter
hardware_retrieved_acquisitions: dict,
acq_duration: int,
acq_channel: Hashable, # noqa: ARG002, unused parameter
total_average_repetitions: int, # noqa: ARG002, unused parameter
) -> complex:
if acq_duration is None:
raise RuntimeError(
"Retrieving data failed. Expected the integration length to be defined,"
" but it is `None`."
)
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
i_data = bin_data["integration"]["path0"][qblox_acq_bin]
q_data = bin_data["integration"]["path1"][qblox_acq_bin]
return (i_data + q_data * 1j) / acq_duration
[docs]
def _get_integration_real_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None, # noqa: ARG002, unused parameter
hardware_retrieved_acquisitions: dict,
acq_duration: int, # noqa: ARG002, unused parameter
acq_channel: Hashable, # noqa: ARG002, unused parameter
total_average_repetitions: int, # noqa: ARG002, unused parameter
) -> complex:
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
i_data = bin_data["integration"]["path0"][qblox_acq_bin]
q_data = bin_data["integration"]["path1"][qblox_acq_bin]
return complex(i_data + q_data)
[docs]
def _get_thresholded_bin(
self,
qblox_acq_index: int,
qblox_acq_bin: int,
thresholded_trigger_count_metadata: ThresholdedTriggerCountMetadata | None, # noqa: ARG002, unused parameter
hardware_retrieved_acquisitions: dict,
acq_duration: int,
acq_channel: Hashable, # noqa: ARG002, unused parameter
total_average_repetitions: int, # noqa: ARG002, unused parameter
) -> int:
if acq_duration is None:
raise RuntimeError(
"Retrieving data failed. Expected the integration length to be defined,"
" but it is `None`."
)
bin_data = self._get_bin_data(hardware_retrieved_acquisitions, qblox_acq_index)
n = bin_data["threshold"][qblox_acq_bin]
return n if not isnan(n) else -1
[docs]
class _QTMAcquisitionManager(_AcquisitionManagerBase):
"""
Utility class that handles the acquisitions performed with the QTM.
An instance of this class is meant to exist only for a single prepare-start-
retrieve_acquisition cycle to prevent stateful behavior.
Parameters
----------
parent
Reference to the parent QRM IC component.
acq_channels_data
Provides a summary of the used acquisition protocol, bin mode, acquisition channels,
acquisition indices per channel, and repetitions, for each sequencer.
acquisition_duration
The duration of each acquisition for each sequencer.
seq_name_to_idx_map
All available sequencer names to their ids in a dict.
"""
[docs]
def _protocol_to_bin_function(self, protocol: str) -> Callable:
match protocol:
case "Timetag":
return self._get_timetag_bin
case "TriggerCount":
return partial(self._get_trigger_count_bin, sum_multiply_repetitions=True)
case "ThresholdedTriggerCount":
return partial(
self._get_thresholded_trigger_count_bin, sum_multiply_repetitions=True
)
case "DualThresholdedTriggerCount":
return partial(self._get_trigger_count_bin, sum_multiply_repetitions=True)
case _:
raise AssertionError(
"This should not be reachable due to _check_bin_mode_compatible."
)
[docs]
def _protocol_to_acq_function_map_non_binned(self, protocol: str) -> Callable:
match protocol:
case "TriggerCount":
return self._get_trigger_count_distribution_data
case "Trace":
return self._get_digital_trace_data
case _:
raise AssertionError(
"This should not be reachable due to _check_bin_mode_compatible."
)
[docs]
def _protocol_to_acq_async_function_map_non_binned(self, protocol: str) -> Callable:
match protocol:
case "Trace":
return self._get_digital_trace_data_async
case _:
sync_fn = self._protocol_to_acq_function_map_non_binned(protocol)
@wraps(sync_fn)
async def async_wrapper(*args, **kwargs) -> Any: # noqa: ANN401, Any return type
return await asyncio.to_thread(sync_fn, *args, **kwargs)
return async_wrapper
@staticmethod
[docs]
def _check_bin_mode_compatible_binned(
acq_channels_data: AcquisitionChannelsData,
node: BinnedAcqControlFlowNode | BinnedAcqInfo,
) -> None:
if isinstance(node, BinnedAcqControlFlowNode):
for c in node.children:
_QTMAcquisitionManager._check_bin_mode_compatible_binned(acq_channels_data, c)
else:
acq_channel = node.acq_channel
if (
acq_channels_data[acq_channel].bin_mode
not in QTM_COMPATIBLE_BIN_MODES[acq_channels_data[acq_channel].protocol]
):
raise IncompatibleBinModeError(
module_type="QTM",
protocol=acq_channels_data[acq_channel].protocol,
bin_mode=acq_channels_data[acq_channel].bin_mode,
)
@staticmethod
[docs]
def _check_bin_mode_compatible(
acq_channels_data: AcquisitionChannelsData,
acq_hardware_mapping: dict[
str,
QbloxAcquisitionHardwareMapping,
],
) -> None:
for seq_acq_hardware_mapping in acq_hardware_mapping.values():
for acq_hardware_mapping_binned in seq_acq_hardware_mapping.binned:
_QTMAcquisitionManager._check_bin_mode_compatible_binned(
acq_channels_data, acq_hardware_mapping_binned.tree
)
for acq_channel in seq_acq_hardware_mapping.non_binned.keys() | (
{seq_acq_hardware_mapping.timetagtrace[0]}
if seq_acq_hardware_mapping.timetagtrace
else {}
):
if (
acq_channels_data[acq_channel].bin_mode
not in QTM_COMPATIBLE_BIN_MODES[acq_channels_data[acq_channel].protocol]
):
raise IncompatibleBinModeError(
module_type="QTM",
protocol=acq_channels_data[acq_channel].protocol,
bin_mode=acq_channels_data[acq_channel].bin_mode,
)
[docs]
def _get_digital_trace_data(
self,
*,
hardware_retrieved_acquisitions: dict, # noqa: ARG002, unused argument
acq_channel: str,
seq_channel_hardware_mapping: QbloxAcquisitionIndex, # noqa: ARG002, unused argument
acq_duration: int,
sequencer_name: str,
) -> DataArray:
# We ignore the hardware_retrieved_acquisitions,
# and use data from from the io channel.
seq_idx = self._seq_name_to_idx_map[sequencer_name]
scope_data = np.array(self.instrument.io_channels[seq_idx].get_scope_data()[:acq_duration])
return self._add_digital_trace_data(acq_channel, acq_duration, scope_data)
[docs]
async def _get_digital_trace_data_async(
self,
*,
hardware_retrieved_acquisitions: dict, # noqa: ARG002, unused argument
acq_channel: str,
seq_channel_hardware_mapping: QbloxAcquisitionIndex, # noqa: ARG002, unused argument
acq_duration: int,
sequencer_name: str,
) -> DataArray:
# We ignore the hardware_retrieved_acquisitions,
# and use data from from the io channel.
seq_idx = self._seq_name_to_idx_map[sequencer_name]
scope_data = np.array(
(await self.instrument.io_channels[seq_idx]._get_scope_data_async())[:acq_duration]
)
return self._add_digital_trace_data(acq_channel, acq_duration, scope_data)
[docs]
def _add_digital_trace_data(
self,
acq_channel: str,
acq_duration: int,
scope_data: np.ndarray[Any, np.dtype[np.float64]],
) -> DataArray:
coords = self._acq_channels_data[acq_channel].coords
assert isinstance(coords, dict)
acq_index_dim_name = self._acq_channels_data[acq_channel].acq_index_dim_name
trace_index_dim_name = f"trace_index_{acq_channel}"
data_array = DataArray(
scope_data.reshape((1, -1)),
dims=[acq_index_dim_name, trace_index_dim_name],
coords={
acq_index_dim_name: [0],
trace_index_dim_name: list(range(acq_duration)),
},
attrs=self._acq_channel_attrs(
self._acq_channels_data[acq_channel].protocol, acq_index_dim_name
),
)
add_acquisition_coords_nonbinned(data_array, coords, acq_index_dim_name)
return data_array
[docs]
_LoopReturnT = TypeVar("_LoopReturnT")
[docs]
class ClusterComponent(base.InstrumentCoordinatorComponentBase):
"""
Class that represents an instrument coordinator component for a Qblox cluster.
New instances of the ClusterComponent will automatically add installed
modules using name `"<cluster_name>_module<slot>"`.
Parameters
----------
instrument
Reference to the cluster driver object.
"""
@dataclass
[docs]
class _Program:
[docs]
module_programs: dict[str, Any]
[docs]
settings: ClusterSettings
def __init__(self, instrument: Cluster) -> None:
super().__init__(instrument)
[docs]
self._cluster_modules: dict[str, _ModuleComponentBase] = {}
[docs]
self._program: ClusterComponent._Program | None = None
[docs]
self.cluster = instrument
[docs]
self._supports_async = hasattr(self.cluster, "_run_in_loop")
# Important: a tuple with only False may not occur as a key, because new
# unsupported module types may return False on all is_..._type functions.
# is_qcm_type, is_qrm_type, is_rf_type, is_qtm_type, is_qrc_type, is_qsm_type
module_type_map: dict[ComponentTypeProperties, type[_ModuleComponentBase]] = {
ComponentTypeProperties(True, False, False, False, False, False): _QCMComponent,
ComponentTypeProperties(True, False, True, False, False, False): _QCMRFComponent,
ComponentTypeProperties(False, True, False, False, False, False): _QRMComponent,
ComponentTypeProperties(False, True, True, False, False, False): _QRMRFComponent,
ComponentTypeProperties(False, False, True, False, True, False): _QRCComponent,
ComponentTypeProperties(False, False, False, True, False, False): _QTMComponent,
ComponentTypeProperties(False, False, False, False, False, True): _QSMComponent,
}
for instrument_module in instrument.modules:
try:
icc_class = module_type_map[
ComponentTypeProperties(
instrument_module.is_qcm_type,
instrument_module.is_qrm_type,
instrument_module.is_rf_type,
getattr(instrument_module, "is_qtm_type", False),
getattr(instrument_module, "is_qrc_type", False),
getattr(instrument_module, "is_qsm_type", False),
)
]
except KeyError:
continue
module_name = instrument_to_component_name(instrument_module.name)
if icc_class.exist(module_name):
icc_class.find_instrument(module_name).close()
self._cluster_modules[instrument_module.name] = icc_class(instrument_module)
[docs]
def _run_in_loop(self, cb: Awaitable[_LoopReturnT]) -> _LoopReturnT:
"""Run routine in async event loop."""
if not self._supports_async:
raise RuntimeError(
"Tried to run async routine without async support in qblox_instruments"
)
return self.cluster._run_in_loop(cb)
@property
[docs]
def is_running(self) -> bool:
"""Returns true if any of the modules are currently running."""
if self._supports_async:
return self._run_in_loop(self.is_running_async())
return any(comp.is_running for comp in self._cluster_modules.values())
[docs]
async def is_running_async(self) -> bool:
"""Returns true if any of the modules are currently running."""
return any(
await asyncio.gather(
*[comp.is_running_async() for comp in self._cluster_modules.values()]
)
)
[docs]
def _set_parameter(
self,
instrument: InstrumentBase,
parameter_name: str,
val: Any, # noqa: ANN401, disallow Any as type
) -> None:
"""
Set the parameter directly or using the lazy set.
Parameters
----------
instrument
The instrument or instrument channel that holds the parameter to set,
e.g. `self.instrument` or `self.instrument[f"sequencer{idx}"]`.
parameter_name
The name of the parameter to set.
val
The new value of the parameter.
"""
if self.force_set_parameters():
getattr(instrument, parameter_name).set(val)
else:
lazy_set(instrument, parameter_name, val)
[docs]
async def _set_async_parameter(
self,
instrument: InstrumentBase,
parameter_name: str,
val: Any, # noqa: ANN401, disallow Any as type
) -> None:
"""
Set the parameter directly or using the lazy set.
Parameters
----------
instrument
The instrument or instrument channel that holds the parameter to set,
e.g. `self.instrument` or `self.instrument[f"sequencer{idx}"]`.
parameter_name
The name of the parameter to set.
val
The new value of the parameter.
"""
if self.force_set_parameters():
parameter = cast("AsyncParameter", getattr(instrument, parameter_name))
await parameter.async_set(val)
else:
await lazy_async_set(instrument, parameter_name, val)
[docs]
def start(self) -> None:
"""Starts all the modules in the cluster."""
if self._supports_async:
return self._run_in_loop(self.start_async())
# Disarming all sequencers, to make sure the last
# `self.instrument.start_sequencer` only starts sequencers
# which are explicitly armed by the subsequent calls.
self.instrument.stop_sequencer()
if self._program is None:
# Nothing to start
return
# Arming all sequencers in the program.
for comp_name, comp in self._cluster_modules.items():
if comp_name in self._program.module_programs:
comp.clear_data()
comp.arm_all_sequencers_in_program()
# Starts all sequencers in the cluster, time efficiently.
if self._program.settings.sync_on_external_trigger is not None:
self._sync_on_external_trigger(self._program.settings.sync_on_external_trigger)
self.instrument.start_sequencer()
[docs]
async def start_async(self) -> None:
"""Starts all the modules in the cluster."""
# Disarming all sequencers, to make sure the last
# `self.instrument.start_sequencer` only starts sequencers
# which are explicitly armed by the subsequent calls.
await self.instrument._stop_sequencer_async()
if self._program is None:
# Nothing to start
return
# Arming all sequencers in the program.
async def arm_module(comp: _ModuleComponentBase) -> None:
await comp.clear_data_async()
await comp.arm_all_sequencers_in_program_async()
await asyncio.gather(
*[
arm_module(comp)
for comp_name, comp in self._cluster_modules.items()
if comp_name in self._program.module_programs
]
)
if self._program.settings.sync_on_external_trigger is not None:
await self._sync_on_external_trigger_async(
self._program.settings.sync_on_external_trigger
)
# Start all sequencers in the program.
await asyncio.gather(
*[
comp._start_armed_sequencers_async()
for comp_name, comp in self._cluster_modules.items()
if comp_name in self._program.module_programs
]
)
[docs]
def _sync_on_external_trigger(self, settings: ExternalTriggerSyncSettings) -> None:
module_idx = settings.slot - 1
channel_idx = settings.channel - 1
if module_idx != -1: # if not CMM
module: Module = self.instrument.modules[module_idx]
if not module.is_qtm_type:
raise ValueError(
f"Invalid module type {module.module_type} for sync_on_external_trigger. Must "
"be one of MM or QTM."
)
# See ClusterCompiler._validate_external_trigger_sync, which validates that these
# settings do not conflict, if they already exist in a QTMComponent.
self._set_parameter(
self.instrument.modules[module_idx].io_channels[channel_idx],
"mode",
"input",
)
if settings.input_threshold is None:
raise ValueError(
"Using `sync_on_external_trigger` with a QTM module, but there was no input "
"threshold specified in either `hardware_options.digitization_thresholds` or "
"`sync_on_external_trigger.input_threshold`."
)
self._set_parameter(
self.instrument.modules[module_idx].io_channels[channel_idx],
"analog_threshold",
settings.input_threshold,
)
sync_ref = SyncRef.ON if settings.sync_to_ref_clock else SyncRef.OFF
self.instrument.time.sync_ext_trigger(
slot=settings.slot,
channel=settings.channel,
trigger_timestamp=settings.trigger_timestamp,
timeout=settings.timeout,
format=settings.format,
edge_polarity=settings.edge_polarity,
sync_ref=sync_ref,
)
[docs]
async def _sync_on_external_trigger_async(self, settings: ExternalTriggerSyncSettings) -> None:
module_idx = settings.slot - 1
channel_idx = settings.channel - 1
if module_idx != -1: # if not CMM
module: Module = self.instrument.modules[module_idx]
if not module.is_qtm_type:
raise ValueError(
f"Invalid module type {module.module_type} for sync_on_external_trigger. Must "
"be one of MM or QTM."
)
# See ClusterCompiler._validate_external_trigger_sync, which validates that these
# settings do not conflict, if they already exist in a QTMComponent.
await self._set_async_parameter(
self.instrument.modules[module_idx].io_channels[channel_idx],
"mode",
"input",
)
if settings.input_threshold is None:
raise ValueError(
"Using `sync_on_external_trigger` with a QTM module, but there was no input "
"threshold specified in either `hardware_options.digitization_thresholds` or "
"`sync_on_external_trigger.input_threshold`."
)
await self._set_async_parameter(
self.instrument.modules[module_idx].io_channels[channel_idx],
"analog_threshold",
settings.input_threshold,
)
sync_ref = SyncRef.ON if settings.sync_to_ref_clock else SyncRef.OFF
await self.instrument.time._sync_ext_trigger_async(
slot=settings.slot,
channel=settings.channel,
trigger_timestamp=settings.trigger_timestamp,
timeout=settings.timeout,
format=settings.format,
edge_polarity=settings.edge_polarity,
sync_ref=sync_ref,
)
[docs]
def stop(self) -> None:
"""Stops all the modules in the cluster."""
if self._supports_async:
return self._run_in_loop(self.stop_async())
for comp in self._cluster_modules.values():
comp.disable_sync()
# Stops all sequencers in the cluster, time efficiently.
self.instrument.stop_sequencer()
[docs]
async def stop_async(self) -> None:
"""Stops all the modules in the cluster."""
await asyncio.gather(
*[comp.stop_async() for comp_name, comp in self._cluster_modules.items()]
)
[docs]
def _get_program_settings(
self,
program: dict[str, dict | ClusterSettings],
) -> ClusterSettings:
settings = program.get("settings", {})
if not isinstance(settings, ClusterSettings):
return ClusterSettings.from_dict(settings)
else:
return settings
[docs]
def prepare(
self,
program: dict[str, dict | ClusterSettings],
) -> None:
"""
Prepares the cluster component for execution of a schedule.
Parameters
----------
program
The compiled instructions to configure the cluster to.
"""
settings = program.get("settings", {})
if not isinstance(settings, ClusterSettings):
cluster_settings = ClusterSettings.from_dict(settings)
else:
cluster_settings = settings
self._configure(program, cluster_settings)
if self._supports_async:
self._run_in_loop(self._upload_async())
else:
self._upload()
[docs]
def _upload(self) -> None:
assert self._program is not None, "Program not set"
for name, comp_options in self._program.module_programs.items():
if name not in self._cluster_modules:
raise KeyError(
f"Attempting to prepare module {name} of cluster {self.name}, while"
f" module has not been added to the cluster component."
)
# See QTFY-848.
# InstrumentCoordinatorComponentBase.prepare expects only one argument,
# but we intend to change that in the future, so we can safely ignore the
# warning of reportCallIssue.
self._cluster_modules[name]._upload(comp_options) # pyright: ignore[reportCallIssue]
[docs]
async def _upload_async(self) -> None:
assert self._program is not None, "Program not set"
uploads = []
for name, comp_options in self._program.module_programs.items():
if name not in self._cluster_modules:
raise KeyError(
f"Attempting to prepare module {name} of cluster {self.name}, while"
f" module has not been added to the cluster component."
)
uploads.append(
self._cluster_modules[name]._upload_async(comp_options) # pyright: ignore[reportCallIssue]
)
await asyncio.gather(*uploads)
[docs]
def retrieve_acquisition(self) -> Dataset | None:
"""
Retrieves all the data from the instruments.
Returns
-------
:
The acquired data or ``None`` if no acquisitions have been performed.
"""
if self._program is None:
# No acquisitions
return
if self._supports_async:
return self._run_in_loop(self.retrieve_acquisition_async())
acquisitions = Dataset()
for comp_name, comp in self._cluster_modules.items():
if comp_name not in self._program.module_programs:
continue
comp_acq = comp.retrieve_acquisition()
if comp_acq is not None:
check_already_existing_acquisition(
new_dataset=comp_acq, current_dataset=acquisitions
)
acquisitions = acquisitions.merge(comp_acq, join="outer", compat="no_conflicts")
return acquisitions if len(acquisitions) > 0 else None
[docs]
async def retrieve_acquisition_async(self) -> Dataset | None:
"""
Retrieves all the data from the instruments.
Returns
-------
:
The acquired data or ``None`` if no acquisitions have been performed.
"""
if self._program is None:
# No acquisitions
return
comp_acquisitions = await asyncio.gather(
*[
comp.retrieve_acquisition_async()
for comp_name, comp in self._cluster_modules.items()
if comp_name in self._program.module_programs
]
)
acquisitions = Dataset()
for comp_acq in comp_acquisitions:
if comp_acq is not None:
check_already_existing_acquisition(
new_dataset=comp_acq, current_dataset=acquisitions
)
acquisitions = acquisitions.merge(comp_acq, join="outer", compat="no_conflicts")
return acquisitions if len(acquisitions) > 0 else None
[docs]
def wait_done(self, timeout_sec: int = 10) -> None:
"""
Blocks until all the components are done executing their programs.
Parameters
----------
timeout_sec
The time in seconds until the instrument is considered to have timed out.
"""
if self._supports_async:
return self._run_in_loop(self.wait_done_async(timeout_sec))
for comp in self._cluster_modules.values():
comp.wait_done(timeout_sec=timeout_sec)
[docs]
async def wait_done_async(self, timeout_sec: int = 10) -> None:
"""
Blocks until all the components are done executing their programs.
Parameters
----------
timeout_sec
The time in seconds until the instrument is considered to have timed out.
"""
await asyncio.gather(
*[comp.wait_done_async(timeout_sec) for comp in self._cluster_modules.values()]
)
[docs]
def get_hardware_log(
self,
compiled_schedule: CompiledSchedule,
) -> dict | None:
"""
Retrieve the hardware log of the Cluster Management Module and associated modules.
This log includes the module serial numbers and
firmware version.
Parameters
----------
compiled_schedule
Compiled schedule to check if this cluster is referenced in (and if so,
which specific modules are referenced in).
Returns
-------
:
A dict containing the hardware log of the cluster, in case the
component was referenced; else None.
"""
cluster = self.instrument
if cluster.name not in compiled_schedule.compiled_instructions:
return None
cluster_ip = _get_instrument_ip(self)
hardware_log = {
f"{cluster.name}_cmm": _download_log(
config_manager=_get_configuration_manager(cluster_ip),
is_cluster=True,
),
f"{cluster.name}_idn": str(cluster.get_idn()),
f"{cluster.name}_mods_info": str(cluster._get_mods_info()),
}
for module in cluster.modules:
if module.name in compiled_schedule.compiled_instructions[cluster.name]:
# Cannot fetch log from module.get_hardware_log here since modules are
# not InstrumentCoordinator components when using a cluster
module_ip = f"{cluster_ip}/{module.slot_idx}"
hardware_log[module.name] = _download_log(_get_configuration_manager(module_ip))
return hardware_log
[docs]
def get_module_descriptions(self) -> dict[int, ClusterModuleDescription]:
"""
Get the module types of this cluster, indexed by their position in the cluster.
Returns
-------
:
A dictionary containing the module types in this cluster,
indexed by their position in the cluster.
"""
module_types = {}
for module in self._cluster_modules.values():
index = module.instrument.slot_idx
if type(module) is _QTMComponent:
module_types[index] = QTMDescription()
elif type(module) is _QRMComponent:
module_types[index] = QRMDescription()
elif type(module) is _QCMComponent:
module_types[index] = QCMDescription()
elif type(module) is _QRMRFComponent:
module_types[index] = QRMRFDescription()
elif type(module) is _QCMRFComponent:
module_types[index] = QCMRFDescription()
elif type(module) is _QRCComponent:
module_types[index] = QRCDescription()
elif type(module) is _QSMComponent:
module_types[index] = QSMDescription()
return module_types
[docs]
def _get_instrument_ip(component: base.InstrumentCoordinatorComponentBase) -> str:
ip_config = component.instrument.get_ip_config()
if ip_config == "0":
raise ValueError(
f"Instrument '{component.instrument.name}' returned {ip_config=}."
f"Please make sure the physical instrument is connected and has a valid ip."
)
instrument_ip = ip_config
if "/" in instrument_ip:
instrument_ip = instrument_ip.split("/")[0]
return instrument_ip
[docs]
def _get_configuration_manager(instrument_ip: str) -> ConfigurationManager:
try:
config_manager = ConfigurationManager(instrument_ip)
except RuntimeError as error:
new_message = f"{error}\nNote: qblox-instruments might have changed ip formatting."
raise type(error)(new_message) from None
return config_manager
[docs]
def _download_log(
config_manager: ConfigurationManager,
is_cluster: bool | None = False,
) -> dict:
hardware_log = {}
sources = ["app", "system"]
if is_cluster:
sources.append("cfg_man")
for source in sources:
# uuid prevents unwanted deletion if file already exists
temp_log_file_name = os.path.join(
OutputDirectoryManager.get_datadir(), f"{source}_{uuid4()}"
)
config_manager.download_log(source=source, fmt="txt", file=temp_log_file_name)
if os.path.isfile(temp_log_file_name):
with open(temp_log_file_name, encoding="utf-8", errors="replace") as file:
log = file.read()
os.remove(temp_log_file_name)
hardware_log[f"{source}_log"] = log
else:
raise RuntimeError(
f"`ConfigurationManager.download_log` did not create a `{source}` file."
)
return hardware_log
