# ----------------------------------------------------------------------------
# Description : Native interface constants and enums
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# ----------------------------------------------------------------------------
# -- include -----------------------------------------------------------------
import copy
import json
import logging
import re
import warnings
from collections import namedtuple
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Literal, Optional, Union
from typing_extensions import Self
from qblox_instruments.constants import NUM_SLOTS # Originally declared here
logger = logging.getLogger(__name__)
# -- definitions -------------------------------------------------------------
# State enum base class
[docs]
class StateEnum(Enum):
"""
State enum base class that arranges child enum string representations.
"""
def __repr__(self) -> str:
return "<{}.{}>".format(str(type(self)).split("'")[1], self.name)
def __str__(self) -> str:
return str(self.name)
def __eq__(self, other: Any) -> bool:
if type(self) is type(other):
return str(self) == str(other)
elif other in [str(val) for val in type(self)]:
return str(self) == other
else:
raise KeyError(f"{other} is not of type {type(self)}")
def __key__(self) -> str:
return str(self)
def __hash__(self) -> int:
return hash(self.__key__())
[docs]
class DeprecatedStateEnum(StateEnum):
"""
State enum class that throws deprecation warning.
"""
[docs]
def __init__(self, _warning_message) -> None:
self.warning_message = _warning_message
def _deprecation_warning(self) -> None:
warnings.warn(
f"{self.warning_message}",
FutureWarning,
stacklevel=2,
)
def __str__(self) -> str:
self._deprecation_warning()
return StateEnum.__str__(self)
def __repr__(self) -> str:
self._deprecation_warning()
return StateEnum.__repr__(self)
def __eq__(self, other: Any) -> bool:
self._deprecation_warning()
return StateEnum.__eq__(self, other)
def __key__(self) -> str:
self._deprecation_warning()
return StateEnum.__key__(self)
def __hash__(self) -> int:
self._deprecation_warning()
return StateEnum.__hash__(self)
# It will be deprecated
# State tuple base class
[docs]
class StateTuple:
"""
State tuple base class that arranges child tuple string representations.
"""
[docs]
def __init__(self, _warning_message) -> None:
self.warning_message = _warning_message
def _deprecation_warning(self) -> None:
warnings.warn(
f"{self.warning_message}",
FutureWarning,
stacklevel=2,
)
def __str__(self) -> str:
# Status, flags and slot_flags are inherited from the child class
# using virtual inheritance, so we retrieve these attributes through
# getattr to not upset Pylint
status = getattr(self, "status")
flags = getattr(self, "flags")
flags = ", ".join([str(flag) for flag in flags]) if len(flags) > 0 else "NONE"
pretty_str = f"Status: {status}, Flags: {flags}"
if hasattr(self, "slot_flags"):
slot_flags = getattr(self, "slot_flags")
pretty_str += f", Slot flags: {slot_flags}"
self._deprecation_warning()
return pretty_str
# State tuple base class
[docs]
class SystemStatusTuple:
"""
System Status tuple base class that arranges child tuple string representations.
"""
def __str__(self) -> str:
# Status, flags and slot_flags are inherited from the child class
# using virtual inheritance, so we retrieve these attributes through
# getattr to not upset Pylint
status = getattr(self, "status")
flags = getattr(self, "flags")
flags = ", ".join([str(flag) for flag in flags]) if len(flags) > 0 else "NONE"
pretty_str = f"Status: {status}, Flags: {flags}"
if hasattr(self, "slot_flags"):
slot_flags = getattr(self, "slot_flags")
pretty_str += f", Slot flags: {slot_flags}"
return pretty_str
[docs]
class StatusTuple:
"""
Status tuple base class that arranges child tuple string representations.
"""
def __str__(self) -> str:
# getattr to not upset Pylint
state = getattr(self, "state")
status = getattr(self, "status")
exit_code = getattr(self, "exit_code")
info_flags = getattr(self, "info_flags")
warn_flags = getattr(self, "warn_flags")
err_flags = getattr(self, "err_flags")
log = getattr(self, "log")
flags = [info_flags, warn_flags, err_flags]
for type_idx, type_flags in enumerate(flags):
if len(type_flags) > 0:
flags[type_idx] = ", ".join([str(flag) for flag in type_flags])
else:
flags[type_idx] = "NONE"
pretty_str = (
f"Status: {status}, "
f"State: {state}, "
f"Exit Code: {exit_code}, "
f"Info Flags: {flags[0]}, "
f"Warning Flags: {flags[1]}, "
f"Error Flags: {flags[2]}, "
f"Log: {log}"
)
return pretty_str
# All System status enum
[docs]
class SystemStatuses(StateEnum):
"""
System status enum.
"""
BOOTING = "System is booting."
INITIALIZING = "System is initializing modules."
OKAY = "System is okay."
RESOLVED = "An error indicated by the flags occurred, but has been resolved."
ERROR = "An error indicated by the flags is occurring."
CRIT_ERROR = "A critical error indicated by the flags is occurring"
# System status flags enum
[docs]
class SystemStatusFlags(StateEnum):
"""
System status flags enum.
"""
PLL_UNLOCKED = "PLL is unlocked."
TEMPERATURE_OUT_OF_RANGE = "Temperature is out of range."
CRIT_TEMPERATURE_OUT_OF_RANGE = "Temperature is critically out of range."
MODULE_NOT_CONNECTED = "Module is not connected."
MODULE_FIRM_OR_HARDWARE_INCOMPATIBLE = "Module firmware is incompatible."
FEEDBACK_NETWORK_CALIBRATION_FAILED = "The feedback network calibration failed."
HARDWARE_COMPONENT_FAILED = "Hardware component failed."
TRIGGER_NETWORK_MISSED_EXT_TRIGGER = "Trigger Network Missed External Trigger."
SMF_RX_OVERFLOW = "Synchronous Message Feedback receive overflow"
SMF_PACKET_ERROR = "Synchronous Message Feedback packet error"
DDC_ADC_OVERRANGE = "Digital Down Converter ADC Overrange"
DUC_DAC_OVERFLOW = "Digital Up Converter DAC Overflow"
RAMPDOWN_OCCURRED = "Safety rampdown was triggered."
MULTI_TILE_SYNC_FAILURE = "Multi tile sync (MTS) failed."
RF_MODULE_IN_DC_CLUSTER = "RF module slotted into DC cluster"
UNKNOWN = "An unknown flag was reported by the instrument."
# Namedtuple representing the slot status flags
[docs]
class SystemStatusSlotFlags(
namedtuple(
"SystemStatusSlotFlags",
[f"slot{slot}" for slot in range(1, NUM_SLOTS + 1)],
)
):
"""
Tuple containing lists of Cluster slot status flag enums of type
:class:`~qblox_instruments.native.definitions.SystemStatusFlags`. Each Cluster slot has its
own status flag list attribute named `slot<X>`.
"""
__name__ = "SystemStatusSlotFlags"
__slots__ = ()
def __new__(cls, slot_flags: Optional[dict] = None) -> "SystemStatusSlotFlags":
slot_flags = slot_flags or {} # Avoid mutable default argument
slot_flag_lists = [[] for _ in range(NUM_SLOTS)]
for slot in range(0, NUM_SLOTS):
slot_str = f"slot{slot + 1}"
if slot_str in slot_flags:
slot_flag_lists[slot] = slot_flags[slot_str]
return super().__new__(cls, *slot_flag_lists)
def __repr__(self) -> str:
slot_str_list = []
for slot in range(0, NUM_SLOTS):
if len(self[slot]) > 0:
slot_str_list.append(f"slot{slot + 1}={self[slot]}") # noqa: PERF401
return f"{self.__name__}({', '.join(slot_str_list)})"
def __str__(self) -> str:
slot_str_list = []
for slot in range(0, NUM_SLOTS):
for flag in self[slot]:
slot_str_list.append(f"SLOT{slot + 1}_{flag}") # noqa: PERF401
if len(slot_str_list) > 0:
return ", ".join(slot_str_list)
else:
return "NONE"
# Namedtuple representing the system status
[docs]
class SystemStatus(
namedtuple("SystemStatus", ["status", "flags", "slot_flags"]), SystemStatusTuple
):
"""
System status tuple returned by :func:`!get_system_status`. The tuple
contains a system status enum of type
:class:`~qblox_instruments.native.definitions.SystemStatuses`, a list of associated system
status flag enums of type
:class:`~qblox_instruments.native.definitions.SystemStatusFlags` and a tuple of type
:class:`~qblox_instruments.native.definitions.SystemStatusSlotFlags` containing Cluster slot
status flags.
"""
status: SystemStatuses
flags: list[SystemStatusFlags]
slot_flags: SystemStatusSlotFlags
[docs]
@classmethod
def from_string(cls, raw_state: str) -> Self:
"""
Convert general system status (old format) to this tuple.
Parameters
----------
raw_state : str
The raw general system status string (in semicolon-separated format).
Returns
-------
SystemStatus
Tuple containing general system status and corresponding flags.
"""
state_elem_list = re.sub(r" |-", "_", raw_state).split(";")
state_flag_list = state_elem_list[-1].split(",")[:-1] if state_elem_list[-1] != "" else []
# Fail gracefully if the raw state returned by SCPI isn't in the semicolon-separated
# format that we expect.
if state_elem_list[0] not in SystemStatuses.__members__:
raise ValueError(f"Malformed system status string: {state_elem_list[0]}")
# Split system status flags from slot status flags
system_flags = []
slot_flags = {}
for flag in state_flag_list:
flag_parts = flag.split("_")
if flag_parts[0] != "SLOT":
system_flags.append(
SystemStatusFlags[flag]
if flag in SystemStatusFlags.__members__
else SystemStatusFlags.UNKNOWN
)
else:
slot = "slot" + flag_parts[1]
flag_name = "_".join(flag_parts[2:])
flag_ = (
SystemStatusFlags[flag_name]
if flag_name in SystemStatusFlags.__members__
else SystemStatusFlags.UNKNOWN
)
if slot not in slot_flags:
slot_flags[slot] = [flag_]
else:
slot_flags[slot].append(flag_)
return SystemStatus(
SystemStatuses[state_elem_list[0]],
system_flags,
SystemStatusSlotFlags(slot_flags),
)
SystemStatus.status.__doc__ = """
System status enum of type :class:`~qblox_instruments.native.definitions.SystemStatuses`.
"""
SystemStatus.flags.__doc__ = """
List of system status flag enums of type
:class:`~qblox_instruments.native.definitions.SystemStatusFlags`.
"""
SystemStatus.slot_flags.__doc__ = """
Tuple of type :class:`~qblox_instruments.native.definitions.SystemStatusSlotFlags containing
Cluster slot status flags
"""
# Sequencer states enum
[docs]
class SequencerStates(StateEnum):
"""
Sequencer state enum.
"""
IDLE = "Sequencer waiting to be armed and started."
ARMED = "Sequencer is armed and ready to start."
RUNNING = "Sequencer is running."
Q1_STOPPED = "Classical part of the sequencer has stopped; waiting for real-time part to stop."
STOPPED = "Sequencer has completely stopped."
# Sequencer statuses enum
[docs]
class SequencerStatuses(StateEnum):
"""
Sequencer status enum.
"""
OKAY = "OKAY"
WARNING = "WARNING"
ERROR = "ERROR"
# Sequencer status flags enum
[docs]
class SequencerStatusFlags(StateEnum):
"""
Sequencer status flags enum.
"""
DISARMED = "Sequencer was disarmed."
FORCED_STOP = "Sequencer was stopped while still running."
SEQUENCE_PROCESSOR_Q1_ILLEGAL_INSTRUCTION = (
"Classical sequencer part executed an unknown instruction."
)
SEQUENCE_PROCESSOR_RT_EXEC_ILLEGAL_INSTRUCTION = (
"Real-time sequencer part executed an unknown instruction."
)
SEQUENCE_PROCESSOR_RT_EXEC_COMMAND_UNDERFLOW = (
"Real-time sequencer part command queue underflow."
)
SEQUENCE_PROCESSOR_DATA_OVERFLOW = (
"The FIFO between the Event Processor and the Sequence Processor filled up."
)
SEQUENCE_PROCESSOR_EVENT_OVERFLOW = (
"Event Processor cannot keep up with the sustained event rate."
)
AWG_WAVE_PLAYBACK_INDEX_INVALID_PATH_0 = "AWG path 0 tried to play an unknown waveform."
AWG_WAVE_PLAYBACK_INDEX_INVALID_PATH_1 = "AWG path 1 tried to play an unknown waveform."
ACQ_WEIGHT_PLAYBACK_INDEX_INVALID_PATH_0 = "Acquisition path 0 tried to play an unknown weight."
ACQ_WEIGHT_PLAYBACK_INDEX_INVALID_PATH_1 = "Acquisition path 1 tried to play an unknown weight."
ACQ_SCOPE_DONE_PATH_0 = "Scope acquisition for path 0 has finished."
ACQ_SCOPE_OUT_OF_RANGE_PATH_0 = "Scope acquisition data for path 0 was out-of-range."
ACQ_SCOPE_OVERWRITTEN_PATH_0 = "Scope acquisition data for path 0 was overwritten."
ACQ_SCOPE_DONE_PATH_1 = "Scope acquisition for path 1 has finished."
ACQ_SCOPE_OUT_OF_RANGE_PATH_1 = "Scope acquisition data for path 1 was out-of-range."
ACQ_SCOPE_OVERWRITTEN_PATH_1 = "Scope acquisition data for path 1 was overwritten."
ACQ_SCOPE_DONE_PATH_2 = "Scope acquisition for path 2 has finished."
ACQ_SCOPE_OUT_OF_RANGE_PATH_2 = "Scope acquisition data for path 2 was out-of-range."
ACQ_SCOPE_OVERWRITTEN_PATH_2 = "Scope acquisition data for path 2 was overwritten."
ACQ_SCOPE_DONE_PATH_3 = "Scope acquisition for path 3 has finished."
ACQ_SCOPE_OUT_OF_RANGE_PATH_3 = "Scope acquisition data for path 3 was out-of-range."
ACQ_SCOPE_OVERWRITTEN_PATH_3 = "Scope acquisition data for path 3 was overwritten."
ACQ_BINNING_DONE = "Acquisition binning completed."
ACQ_BINNING_FIFO_ERROR = "Acquisition binning encountered internal FIFO error."
ACQ_BINNING_COMM_ERROR = "Acquisition binning encountered internal communication error."
ACQ_BINNING_OUT_OF_RANGE = "Acquisition binning data out-of-range."
ACQ_INDEX_INVALID = "Acquisition tried to process an invalid acquisition."
ACQ_BIN_INDEX_INVALID = "Acquisition tried to process an invalid bin."
TRIGGER_NETWORK_CONFLICT = "Trigger network has encountered a conflict."
TRIGGER_NETWORK_MISSED_INTERNAL_TRIGGER = "Trigger network missed an internal trigger."
OUTPUT_OVERFLOW = "Output overflow."
CLOCK_INSTABILITY = "Clock source instability occurred."
ACQ_INTEGRATOR_OUT_OF_RANGE_PATH_0 = (
"Acquisition integration input data for path 0 was out-of-range."
)
ACQ_INTEGRATOR_OUT_OF_RANGE_PATH_1 = (
"Acquisition integration input data for path 1 was out-of-range."
)
DIO_COMMAND_OVERFLOW = "DIO_COMMAND_OVERFLOW"
DIO_DELAY_OUT_OF_ORDER = "DIO_DELAY_OUT_OF_ORDER"
DIO_UNSUPPORTED_PULSE_WIDTH = "DIO_UNSUPPORTED_PULSE_WIDTH"
DIO_TIMETAG_DEADLINE_MISSED = "DIO_TIMETAG_DEADLINE_MISSED"
DIO_TIME_DELTA_INVALID = "DIO_TIME_DELTA_INVALID"
DIO_COUNT_INVALID = "DIO_COUNT_INVALID"
DIO_THRESHOLD_INVALID = "DIO_THRESHOLD_INVALID"
DIO_INTERNAL_ERROR = "DIO_INTERNAL_ERROR"
SYNCHRONOUS_MESSAGE_FEEDBACK_TRANSMIT_OVERFLOW = (
"LINQ Fabric cannot keep up with transmissions from the Sequence Processor."
)
SYNCHRONOUS_MESSAGE_FEEDBACK_PACKET_NOT_COMPLETED = (
"Event Processor tried to send a LINQ header without finishing the previous message."
)
SYNCHRONOUS_MESSAGE_FEEDBACK_DATA_WITHOUT_HEADER = (
"Event Processor tried to send LINQ data without first sending a header for it."
)
UNKNOWN = "UNKNOWN"
[docs]
class SequencerStatus(
namedtuple(
"SequencerStatus",
["status", "state", "exit_code", "info_flags", "warn_flags", "err_flags", "log"],
),
StatusTuple,
):
"""
Sequencer status tuple returned by :func:`!get_sequencer_status`. The tuple
contains a sequencer status, state, flags and log. The tuple contains:
a sequencer status enum of type :class:`~qblox_instruments.native.definitions.SequencerStatuses`,
a sequencer state enum of type :class:`~qblox_instruments.native.definitions.SequencerStates`,
a sequencer exit code of type :class`int`,
a list of associated info flags enums of type :class:`~qblox_instruments.native.definitions.SequencerStatusFlags`,
a list of associated warning flags enums of type :class:`~qblox_instruments.native.definitions.SequencerStatusFlags`,
a list of associated error flags enums of type :class:`~qblox_instruments.native.definitions.SequencerStatusFlags`,
a list of informative log message of type :class:`str`.
""" # noqa: E501
[docs]
@classmethod
def from_scpi_str(cls, s: str) -> "SequencerStatus":
"""
Helper function to parse the output of sequencer status cmd.
Parameters
----------
s : str
Full string from command response.
Returns
-------
SequencerStatus
"""
status_list = s.replace(" ", "_").split(";")
# STATUS;STATE;[CODE;][INFO_FLAGS;WARN_FLAGS;ERR_FLAGS;LOG]
status = status_list[0] # They are always present
state = status_list[1] # They are always present
if len(status_list) > 3:
if len(status_list) > 6:
exit_code = int(status_list[2])
i = 3
else:
exit_code = 0
i = 2
info_flag_list = (
status_list[i + 0].replace("-", "_").split(",")[:-1] if status_list[i + 0] else []
)
warn_flag_list = (
status_list[i + 1].replace("-", "_").split(",")[:-1] if status_list[i + 1] else []
)
err_flag_list = (
status_list[i + 2].replace("-", "_").split(",")[:-1] if status_list[i + 2] else []
)
log = status_list[i + 3].replace("-", "_") if status_list[i + 3] else []
else:
exit_code = 0
info_flag_list = []
warn_flag_list = []
err_flag_list = []
log = []
return cls(
SequencerStatuses[status],
SequencerStates[state],
exit_code,
[SequencerStatusFlags[flag] for flag in info_flag_list],
[SequencerStatusFlags[flag] for flag in warn_flag_list],
[SequencerStatusFlags[flag] for flag in err_flag_list],
log,
)
SequencerStatus.status.__doc__ = """
Sequencer status enum of type :class:`~qblox_instruments.native.definitions.SequencerStatuses`.
"""
SequencerStatus.state.__doc__ = """
Sequencer state enum of type :class:`~qblox_instruments.native.definitions.SequencerStates`.
"""
SequencerStatus.exit_code.__doc__ = """
Sequencer exit code of type :class:`int`.
"""
SequencerStatus.info_flags.__doc__ = """
List of sequencer status flag enums of type
:class:`~qblox_instruments.native.definitions.SequencerStatusFlags`.
"""
SequencerStatus.warn_flags.__doc__ = """
List of sequencer status flag enums of type
:class:`~qblox_instruments.native.definitions.SequencerStatusFlags`.
"""
SequencerStatus.err_flags.__doc__ = """
List of sequencer status flag enums of type
:class:`~qblox_instruments.native.definitions.SequencerStatusFlags`.
"""
SequencerStatus.log.__doc__ = """
List of log message with more detailed information in case of WARNING status.
"""
SCOPE_ACQUISITION_BIT_WIDTH_QRC = 15
"""
Raw scope data retrieved through SCPI is a
signed integer with this bit width on QRC.
"""
SCOPE_ACQUISITION_BIT_WIDTH_QRM = 12
"""
Raw scope data retrieved through SCPI is a
signed integer with this bit width on QRM and QRM-RF.
"""
BINNED_ACQUISITION_BIT_WIDTH_QRM = 12
"""
Raw binned data retrieved through SCPI is
a signed integer with this bit width on QRM and QRM-rf.
"""
BINNED_ACQUISITION_SCALING_QRC = 2
"""
Scaling factor for binned acquisition data for QRC.
"""
# Maximum program length allowed
MAX_PROGRAM_LENGTH = 10 * (128 * 1024 * 8 + 1024)
# Enum for channel type.
[docs]
class ChannelType(Enum):
AWG = 0
ACQ = 1
[docs]
def get_allowed_channels(
is_qrc: bool, direction: ChannelType, sequencer_id: int
) -> Union[set[int], Literal[True]]:
"""
Returns the set of allowed channels.
If returns `True`, all channels are allowed.
"""
if is_qrc and direction == ChannelType.AWG:
return [
{0, 1, 2, 3, 4, 5},
{0, 1, 2, 3, 6, 7},
{0, 1, 2, 3, 8, 9},
{0, 1, 2, 3, 10, 11},
{0, 1, 2, 3, 4, 5},
{0, 1, 2, 3, 6, 7},
{0, 1, 2, 3, 8, 9},
{0, 1, 2, 3, 10, 11},
{4, 5, 6, 7, 8, 9, 10, 11},
{4, 5, 6, 7, 8, 9, 10, 11},
{4, 5, 6, 7, 8, 9, 10, 11},
{4, 5, 6, 7, 8, 9, 10, 11},
][sequencer_id]
elif is_qrc and direction == ChannelType.ACQ:
return [
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
{0, 1, 2, 3},
set(),
set(),
set(),
set(),
][sequencer_id]
else:
return True
# JSON schema to validate sequence dictionaries with, without any required fields.
PARTIAL_QXM_SEQUENCE_JSON_SCHEMA = {
"title": "Sequence container",
"description": (
"Contains all waveforms, weights and acquisitions and a program required for a sequence."
),
"type": "object",
"properties": {
"program": {
"description": "Sequencer assembly program in string format.",
"type": "string",
},
"waveforms": {
"description": "Waveform dictionary containing one or multiple AWG waveform(s).",
"type": "object",
},
"weights": {
"description": "Weight dictionary containing one or multiple acquisition weights(s).",
"type": "object",
},
"acquisitions": {
"description": (
"Acquisition dictionary containing information about one "
"or multiple acquisition(s)."
),
"type": "object",
},
},
}
# JSON schema to validate QCM sequence dictionaries with
QCM_SEQUENCE_JSON_SCHEMA = copy.deepcopy(PARTIAL_QXM_SEQUENCE_JSON_SCHEMA)
QCM_SEQUENCE_JSON_SCHEMA["required"] = ["program", "waveforms"]
# JSON schema to validate QRM sequence dictionaries with
QRM_SEQUENCE_JSON_SCHEMA = copy.deepcopy(QCM_SEQUENCE_JSON_SCHEMA)
QRM_SEQUENCE_JSON_SCHEMA["required"] = [
"program",
"waveforms",
"weights",
"acquisitions",
]
# JSON schema to validate waveform and weight dictionaries with
WAVE_JSON_SCHEMA = {
"title": "Waveform/weight container",
"description": "Waveform/weight dictionary for a single waveform.",
"type": "object",
"required": ["data"],
"properties": {
"index": {"description": "Optional waveform index number.", "type": "number"},
},
}
# JSON schema to validate acquisition dictionaries with
ACQ_JSON_SCHEMA = {
"title": "Acquisition container",
"description": "Acquisition dictionary for a single acquisition.",
"type": "object",
"required": ["num_bins"],
"properties": {
"num_bins": {"description": "Number of bins in acquisition.", "type": "number"},
"index": {"description": "Optional waveform index number.", "type": "number"},
},
}
# JSON schema to validate sequence dictionaries with
# TODO QTM, add more fields here for V2
QTM_SEQUENCE_JSON_SCHEMA = {
"title": "Sequence container",
"description": "Contains all acquisitions and a program required for a sequence.",
"type": "object",
"required": ["program"],
"properties": {
"program": {
"description": "Sequencer assembly program in string format.",
"type": "string",
},
"acquisitions": {
"description": (
"Acquisition dictionary containing information about one "
"or multiple acquisition(s)."
),
"type": "object",
},
},
}
ClusterHealthLevel = SystemStatuses
HardwareFlag = SystemStatusFlags
[docs]
@dataclass(frozen=True)
class InstrumentState:
active_flags: list[HardwareFlag | str] = field(default_factory=list)
resolved_flags: list[HardwareFlag | str] = field(default_factory=list)
@staticmethod
def _parse_hw_flag(raw_flag: str) -> HardwareFlag | str:
"""
Parse a hardware flag into its corresponding enum.
If the flag is not a member of the enum, return it as a raw string.
"""
flag_name = re.sub(r"[ -]", "_", raw_flag)
if flag_name in HardwareFlag.__members__:
return HardwareFlag[flag_name]
else:
logger.warning(
"Unknown system status flag received: '%s' (normalized: '%s')", raw_flag, flag_name
)
# Current behaviour is to return the raw flag value for non-registered flags.
return raw_flag
@classmethod
def _from_flag_dict(cls, flag_dict: dict[str, list[str]]) -> Self:
"""
Convert a dict containing active and resolved flags into an instance of this dataclass.
The format is as described in :meth:`ClusterState._get_json_str`.
"""
return cls(
active_flags=[cls._parse_hw_flag(flag) for flag in flag_dict.get("ACTIVE", [])],
resolved_flags=[cls._parse_hw_flag(flag) for flag in flag_dict.get("RESOLVED", [])],
)
[docs]
@dataclass(frozen=True)
class ClusterState:
health_level: ClusterHealthLevel | str
system: InstrumentState
modules: dict[int, InstrumentState] = field(default_factory=dict)
@classmethod
def _from_json_str(cls, raw_state: str) -> Self:
"""
Convert general system status (new format) into an instance of this dataclass.
The input should be a JSON-serialized string with the following structure:
.. code-block:: text
{
"CLUSTER": {
"STATE": "OKAY/ERROR/...",
"ACTIVE": ["ACTIVE_FLAG_1", "ACTIVE_FLAG_2"],
"RESOLVED": ["RESOLVED_FLAG_1", "RESOLVED_FLAG_2"]
},
"SLOTS": {
"SLOT 1": {
"ACTIVE": ["ACTIVE_FLAG_1", "ACTIVE_FLAG_2"],
"RESOLVED": ["RESOLVED_FLAG_1", "RESOLVED_FLAG_2"]
},
"SLOT 2": { ... },
"SLOT N": { ... }
}
}
Parameters
----------
raw_state : str
The raw general system status string (in json-serialized format).
Returns
-------
ClusterState
Dataclass containing general system status and corresponding flags.
"""
state_dict = json.loads(raw_state)
try:
health_level = state_dict["CLUSTER"].get("STATE", "UNKNOWN")
system_state = InstrumentState._from_flag_dict(state_dict["CLUSTER"])
modules_state = {
int(k.split(" ")[1]): InstrumentState._from_flag_dict(v)
for k, v in state_dict["SLOTS"].items()
}
except (IndexError, KeyError, ValueError) as e:
raise ValueError(f"Malformed cluster state JSON string: {state_dict}") from e
return ClusterState(health_level=health_level, system=system_state, modules=modules_state)
