Source code for qblox_instruments.ieee488_2.cluster_dummy_transport

# ----------------------------------------------------------------------------
# Description    : Transport layer (abstract, IP, file, dummy)
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# ----------------------------------------------------------------------------


# -- include -----------------------------------------------------------------

import json

from typing import Dict, List, Optional, Union, Iterable
from qblox_instruments import ClusterType
from qblox_instruments.ieee488_2 import (
    DummyTransport,
    QcmQrmDummyTransport,
    DummyBinnedAcquisitionData,
    DummyScopeAcquisitionData,
)


# -- class -------------------------------------------------------------------


[docs] class ClusterDummyTransport(DummyTransport): """ Class to replace Cluster device with dummy device to support software stack testing without hardware. The class implements all mandatory, required and Cluster specific SCPI calls. Call reponses are largely artifically constructed to be inline with the call's functionality (e.g. `*IDN?` returns valid, but artificial IDN data.) To assist development, the Q1ASM assembler has been completely implemented. Please have a look at the call's implentation to know what to expect from its response. """ # ------------------------------------------------------------------------
[docs] def __init__(self, dummy_cfg: Dict[Union[str, int], ClusterType]): """ Create Cluster dummy transport class. Parameters ---------- dummy_cfg : Dict Dictionary of dummy module types (e.g. Cluster QCM, Cluster QRM). Each key of the dictionary is a slot index with a dummy type specification of type :class:`~qblox_instruments.ClusterType`. Returns ---------- Raises ---------- """ # Initialize base class super().__init__( dummy_cfg["0"] if "0" in dummy_cfg else ClusterType._CLUSTER_MM ) # Set number of slots self._num_slots = 20 # Configure module dummy transports self._modules = {str(slot_idx): None for slot_idx in range(self._num_slots)} for slot_idx in dummy_cfg: if str(slot_idx) in self._modules: self._modules[str(slot_idx)] = QcmQrmDummyTransport(dummy_cfg[slot_idx]) # Set command list self._cmds["STATus:QUEStionable:LED:BRIGHTness?"] = self._get_led_brightness self._cmds["STATus:GENeral:STATE?"] = self._get_system_state self._cmds["*MODS?"] = self._get_mods_info self._cmds["BP:MODules?"] = self._get_modules_present
# ------------------------------------------------------------------------ def _execute_cmd( self, cmd_parts: list, cmd_params: list, cmd_args: list, bin_in: Optional[bytes] = None, ) -> None: """ Execute associated command method found in command dictionary. If the command is intended for a slot as indicated by the first command part being "SLOT#", the command is executed by the associated module. Parameters ---------- cmd_parts : list Reformated command sections cmd_params : list Command parameters cmd_args : list Command arguments bin_in : Optional[bytes] Binary data that needs to be send by the command. Returns ---------- Raises ---------- """ def is_fanout_command(cmd_parts: List[str]) -> bool: """ Check whether the command can fan out to all slots. Not all commands have this behavior. """ cmd_str = ":".join(cmd_parts) return cmd_str in { "SLOT:SEQuencer:ARM", "SLOT:SEQuencer:START", "SLOT:SEQuencer:STOP", "SLOT:SEQuencer:CLR:FLAGS", } # If command is intended for a slot, remove the slot specification # and execute the command on the module instead. if cmd_parts[0] == "SLOT#": cmd_parts.pop(0) slot = cmd_params.pop(0) if slot in self._modules: mod = self._modules[slot] mod._execute_cmd(cmd_parts, cmd_params, cmd_args, bin_in) # Copy results and errors from module. self._data_out = mod._data_out self._bin_out = mod._bin_out for error in mod._system_error: self._system_error.append(error) mod._system_error = [] else: self._system_error.append(f"Module in slot {slot} is not available.") # If command is intended for all slots, execute it on all modules. elif is_fanout_command(cmd_parts): cmd_parts.pop(0) for mod in self._modules.values(): if mod is None: continue mod._execute_cmd(cmd_parts, cmd_params, cmd_args, bin_in) # Copy results and errors from module. self._data_out = mod._data_out self._bin_out = mod._bin_out for error in mod._system_error: self._system_error.append(error) mod._system_error = [] else: super()._execute_cmd(cmd_parts, cmd_params, cmd_args, bin_in) # ------------------------------------------------------------------------ def _get_led_brightness( self, cmd_params: list, cmd_args: list, bin_in: bytes ) -> None: """ Get LED brightness Parameters ---------- cmd_params : list Command parameters indicated by '#' in the command. cmd_args : list Command arguments. bin_in : bytes Binary input data. Returns ---------- Raises ---------- """ self._data_out = "HIGH" # ------------------------------------------------------------------------ def _get_system_state( self, cmd_params: list, cmd_args: list, bin_in: bytes ) -> None: """ Get system status. Parameters ---------- cmd_params : list Command parameters indicated by '#' in the command. cmd_args : list Command arguments. bin_in : bytes Binary input data. Returns ---------- Raises ---------- """ super()._get_system_state(cmd_params, cmd_args, bin_in) for slot_idx in self._modules: if self._modules[slot_idx] is not None: self._modules[slot_idx]._get_system_state(cmd_params, cmd_args, bin_in) flags = self._modules[slot_idx]._data_out.split(";")[1] flags = flags.split(",")[:-1] flags = [f"SLOT {slot_idx} {flag}" for flag in flags] self._data_out += ",".join(flags) if len(flags) > 0: self._data_out += "," # ------------------------------------------------------------------------ def _get_mods_info(self, cmd_params: list, cmd_args: list, bin_in: bytes) -> None: """ Get information about the modules in the Cluster (i.e. IDN and RF indication). Parameters ---------- cmd_params : list Command parameters indicated by '#' in the command. cmd_args : list Command arguments. bin_in : bytes Binary input data. Returns ---------- Raises ---------- """ mod_info = {} for slot_idx in self._modules: if self._modules[slot_idx] is not None: self._modules[slot_idx]._get_idn([], [], None) mod_info["SLOT " + slot_idx] = { "IDN": self._modules[slot_idx]._data_out, "RF": self._modules[slot_idx].is_rf_type, } self._bin_out = self._encode_bin(json.dumps(mod_info).encode("utf-8")) # ------------------------------------------------------------------------ def _get_modules_present( self, cmd_params: list, cmd_args: list, bin_in: bytes ) -> None: """ Get modules present in the Cluster. Parameters ---------- cmd_params : list Command parameters indicated by '#' in the command. cmd_args : list Command arguments. bin_in : bytes Binary input data. Returns ---------- Raises ---------- """ mod_present = 0 for slot_idx in self._modules: if self._modules[slot_idx] is not None: mod_present |= 1 << (int(slot_idx) - 1) self._data_out = mod_present # ------------------------------------------------------------------------
[docs] def delete_dummy_binned_acquisition_data( self, slot_idx: int, sequencer: Optional[int] = None, acq_index_name: Optional[str] = None, ): """ Delete all dummy binned acquisition data for the dummy. Parameters ---------- slot_idx : int Slot of the hardware you want to set the data to on a cluster. sequencer : Optional[int] Sequencer. acq_index_name : Optional[str] Acquisition index name. Returns ---------- Raises ---------- """ self._modules[str(slot_idx)].delete_dummy_binned_acquisition_data( sequencer, acq_index_name )
# ------------------------------------------------------------------------
[docs] def set_dummy_binned_acquisition_data( self, slot_idx: int, sequencer: int, acq_index_name: str, data: Iterable[Union[DummyBinnedAcquisitionData, None]], ): """ Set dummy binned acquisition data for the dummy. Parameters ---------- slot_idx : int Slot of the hardware you want to set the data to on a cluster. sequencer : int Sequencer. acq_index_name : str Acquisition index name. data : Iterable[Union[DummyBinnedAcquisitionData, None]] Dummy data for the binned acquisition. An iterable of all the bin values. Returns ---------- Raises ---------- """ self._modules[str(slot_idx)].set_dummy_binned_acquisition_data( sequencer, acq_index_name, data )
# ------------------------------------------------------------------------
[docs] def delete_dummy_scope_acquisition_data(self, slot_idx: int): """ Delete dummy scope acquisition data for the dummy. Parameters ---------- slot_idx : int Slot of the hardware you want to set the data to on a cluster. Returns ---------- Raises ---------- """ self._modules[str(slot_idx)].delete_dummy_scope_acquisition_data()
# ------------------------------------------------------------------------
[docs] def set_dummy_scope_acquisition_data( self, slot_idx: int, data: DummyScopeAcquisitionData ): """ Set dummy scope acquisition data for the dummy. Parameters ---------- slot_idx : int Slot of the hardware you want to set the data to on a cluster. data : DummyScopeAcquisitionData Dummy data for the scope acquisition. Returns ---------- Raises ---------- """ self._modules[str(slot_idx)].set_dummy_scope_acquisition_data(data)