Source code for qcodes.instrument_drivers.Keysight.keysightb1500.KeysightB1500_module

import re
from collections import namedtuple
from typing import TYPE_CHECKING, cast

import numpy as np
from typing_extensions import TypedDict, Unpack, deprecated

from qcodes.instrument import InstrumentBaseKWArgs, InstrumentChannel
from qcodes.utils import QCoDeSDeprecationWarning

from . import constants
from .constants import ChannelName, ChNr, MeasurementStatus, ModuleKind, SlotNr
from .message_builder import MessageBuilder

    import qcodes.instrument_drivers.Keysight.keysightb1500

_FMTResponse = namedtuple('_FMTResponse', 'value status channel type')

class MeasurementNotTaken(Exception):

def fmt_response_base_parser(raw_data_val: str) -> _FMTResponse:
    Parse the response from SPA for `FMT 1,0` format  into a named tuple
    with names, value (value of the data), status (Normal or with compliance
    error such as C, T, V), channel (channel number of the output data such
    as CH1,CH2), type (current 'I' or voltage 'V'). This parser is tested
    for FMT1,0 and FMT1,1 response.

        raw_data_val: Unparsed (raw) data for the instrument.

    values_separator = ','
    data_val = []
    data_status = []
    data_channel = []
    data_datatype = []

    for str_value in raw_data_val.split(values_separator):
        status = str_value[0]
        channel_id = constants.ChannelName[str_value[1]].value

        datatype = str_value[2]
        value = float(str_value[3:])


    data = _FMTResponse(data_val, data_status, data_channel, data_datatype)
    return data

def parse_module_query_response(response: str) -> dict[SlotNr, str]:
    Extract installed module information from the given string and return the
    information as a dictionary.

        response: Response str to `UNT? 0` query.

        Dictionary from slot numbers to model name strings.
    pattern = r";?(?P<model>\w+),(?P<revision>\d+)"

    moduleinfo = re.findall(pattern, response)

    return {
        SlotNr(slot_nr): model
        for slot_nr, (model, rev) in enumerate(moduleinfo, start=1)
        if model != "0"

# pattern to match dcv experiment
_pattern_lrn = re.compile(

def parse_dcv_measurement_response(response: str) -> dict[str, str | float]:
    Extract status, channel number, value  and accompanying metadata from
    the string and return them as a dictionary.

        response: Response str to lrn_query For the MFCMU.

    match = re.match(_pattern_lrn, response)
    if match is None:
        raise ValueError(f"{response!r} didn't match {_pattern_lrn!r} pattern")

    dd = match.groupdict()
    d = cast(dict[str, str | float], dd)
    return d

# Pattern to match the spot measurement response against
_pattern = re.compile(

class SpotResponse(TypedDict):
    value: float
    status: MeasurementStatus
    channel: ChannelName
    dtype: str

def parse_spot_measurement_response(response: str) -> SpotResponse:
    Extract measured value and accompanying metadata from the string
    and return them as a dictionary.

        response: Response str to spot measurement query.

        Dictionary with measured value and associated metadata (e.g.
        timestamp, channel number, etc.)
    match = re.match(_pattern, response)
    if match is None:
        raise ValueError(f"{response!r} didn't match {_pattern!r} pattern")

    dd = match.groupdict()

    d = SpotResponse(

    return d

_DCORRResponse = namedtuple('_DCORRResponse', 'mode primary secondary')

def parse_dcorr_query_response(response: str) -> _DCORRResponse:
    Parse string response of ``DCORR?`` `command into a named tuple of
    :class:`constants.DCORR.Mode` and primary and secondary reference or
    calibration values.
    mode, primary, secondary = response.split(',')
    return _DCORRResponse(mode=constants.DCORR.Mode(int(mode)),

def fixed_negative_float(response: str) -> float:
    Keysight sometimes responds for ex. '-0.-1' as an output when you input
    '-0.1'. This function can convert such strings also to float.
    if len(response.split('.')) > 2:
        raise ValueError('String must of format `a` or `a.b`')

    parts = response.split('.')
    number = parts[0]
    decimal = parts[1] if len(parts) > 1 else '0'

    decimal = decimal.replace("-", "")

    output = ".".join([number, decimal])
    return float(output)

_dcorr_labels_units_map = {
    constants.DCORR.Mode.Cp_G: dict(
        primary=dict(label='Cp', unit='F'),
        secondary=dict(label='G', unit='S')
    constants.DCORR.Mode.Ls_Rs: dict(
        primary=dict(label='Ls', unit='H'),
        secondary=dict(label='Rs', unit='Ω'))

def format_dcorr_response(r: _DCORRResponse) -> str:
    Format a given response tuple ``_DCORRResponse`` from
    ``DCORR?`` command as a human-readable string.
    labels_units = _dcorr_labels_units_map[r.mode]
    primary = labels_units['primary']
    secondary = labels_units['secondary']

    result_str = (
        f"Mode: {}, "
        f"Primary {primary['label']}: {r.primary} {primary['unit']}, "
        f"Secondary {secondary['label']}: {r.secondary} {secondary['unit']}"
    return result_str

def get_name_label_unit_of_impedance_model(
        mode: constants.IMP.MeasurementMode
) -> tuple[tuple[str, str], tuple[str, str], tuple[str, str]]:
    params ='_')

    param1 = params[0]
    param2 = '_'.join(params[1:])

    label = (constants.IMP.Name[param1].value,

    unit = (constants.IMP.Unit[param1].value,

    name = (label[0].lower().replace(' ', '_'),
            label[1].lower().replace(' ', '_'))

    return name, label, unit

# TODO notes:
# - [ ] Instead of generating a Qcodes InstrumentChannel for each **module**,
#   it might make more sense to generate one for each **channel**

def get_measurement_summary(status_array: np.ndarray) -> str:
    unique_error_statuses = np.unique(status_array[status_array != "N"])
    if len(unique_error_statuses) > 0:
        summary = " ".join(
            constants.MeasurementStatus[err] for err in
        summary = constants.MeasurementStatus["N"]

    return summary

def convert_dummy_val_to_nan(param: _FMTResponse) -> None:
    Converts dummy value to NaN. Instrument may output dummy value (
    199.999E+99) if measurement data is over the measurement range. Or the
    sweep measurement was aborted by the automatic stop function or power
    compliance. Or if any abort condition is detected. Dummy data
    199.999E+99 will be returned for the data after abort."

        param: This must be of type named tuple _FMTResponse.

    for index, value in enumerate(param.value):
        param.value[index] = _convert_to_nan_if_dummy_value(param.value[index])

def _convert_to_nan_if_dummy_value(value: float) -> float:
    return float('nan') if value > 1e99 else value

[docs] class KeysightB1500Module(InstrumentChannel): """Base class for all modules of B1500 Parameter Analyzer When subclassing, - set ``MODULE_KIND`` attribute to the correct module kind :class:`~.constants.ModuleKind` that the module is. - populate ``channels`` attribute according to the number of channels that the module has. Args: parent: Mainframe B1500 instance that this module belongs to name: Name of the instrument instance to create. If `None` (Default), then the name is autogenerated from the instrument class. slot_nr: Slot number of this module (not channel number) """ MODULE_KIND: ModuleKind def __init__( self, parent: "qcodes.instrument_drivers.Keysight.keysightb1500.KeysightB1500", name: str | None, slot_nr: int, **kwargs: Unpack[InstrumentBaseKWArgs], ): # self.channels will be populated in the concrete module subclasses # because channel count is module specific self.channels: tuple[ChNr, ...] self.slot_nr = SlotNr(slot_nr) if name is None: number = len(parent.by_kind[self.MODULE_KIND]) + 1 name = self.MODULE_KIND.lower() + str(number) super().__init__(parent=parent, name=name, **kwargs) # Response parsing functions as static methods for user convenience parse_spot_measurement_response = parse_spot_measurement_response parse_module_query_response = parse_module_query_response
[docs] def enable_outputs(self) -> None: """ Enables all outputs of this module by closing the output relays of its channels. """ # TODO This always enables all outputs of a module, which is maybe not # desirable. (Also check the TODO item at the top about # InstrumentChannel per Channel instead of per Module. msg = MessageBuilder().cn(self.channels).message self.write(msg)
[docs] def disable_outputs(self) -> None: """ Disables all outputs of this module by opening the output relays of its channels. """ # TODO See enable_output TODO item msg = MessageBuilder().cl(self.channels).message self.write(msg)
[docs] def is_enabled(self) -> bool: """ Check if channels of this module are enabled. Returns: `True` if *all* channels of this module are enabled. `False`, otherwise. """ # TODO If a module has multiple channels, and only one is enabled, then # this will return false, which is probably not desirable. # Also check the TODO item at the top about InstrumentChannel per # Channel instead of per Module. msg = (MessageBuilder() .lrn_query(constants.LRN.Type.OUTPUT_SWITCH) .message ) response = self.ask(msg) activated_channels = re.sub(r"[^,\d]", "", response).split(",") is_enabled = set(self.channels).issubset( int(x) for x in activated_channels if x != '' ) return is_enabled
[docs] def clear_timer_count(self) -> None: """ This command clears the timer count. This command is effective for all measurement modes, regardless of the TSC setting. This command is not effective for the 4 byte binary data output format (FMT3 and FMT4). """ self.root_instrument.clear_timer_count(chnum=self.channels)
@deprecated("Use KeysightB1500Module", category=QCoDeSDeprecationWarning) class B1500Module(KeysightB1500Module): pass class StatusMixin: def __init__(self) -> None: self.names = tuple(['param1', 'param2']) def status_summary(self) -> dict[str, str]: return_dict: dict[str, str] = {} for name_index, name in enumerate(self.names): param_data: _FMTResponse = getattr(self, f"param{name_index+1}") status_array = param_data.status if status_array is None: self_full_name = getattr(self, "full_name", "this") raise MeasurementNotTaken( f"First run sweep measurement with {self_full_name} " f"parameter to obtain the data; then it will be possible " f"to obtain status summary for that data." ) summary = get_measurement_summary(status_array) return_dict[name] = summary return return_dict