# SPDX-FileCopyrightText: © 2025 Qblox <https://qblox.com>
# SPDX-License-Identifier: BSD-4-Clause
# -- 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
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: dict | None = 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
) -> 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)
