Module enrgdaq.daq.jobs.caen.toolbox

Classes

class DAQJobCAENToolbox (config: DAQJobCAENToolboxConfig,
**kwargs)

Expand source code

class DAQJobCAENToolbox(DAQJob):
    """
    Dumps register data of a CAEN digitizer using the caen-toolbox cli.
    Attributes:
        config (DAQJobCAENToolboxConfig): Configuration for the DAQ job.
        config_type (DAQJobCAENToolboxConfig): Configuration type for the DAQ job.
    """

    config_type = DAQJobCAENToolboxConfig
    config: DAQJobCAENToolboxConfig

    def __init__(self, config: DAQJobCAENToolboxConfig, **kwargs):
        super().__init__(config, **kwargs)
        # Check if 'caen-toolbox' cli is present
        if which("caen-toolbox") is None:
            raise Exception("caen-toolbox cli not found")

    def _dump_digitizer(self) -> dict[str, int]:
        temp_file = os.path.join(tempfile.gettempdir(), "reg-dump.csv")
        # Dump registers. Might not be the best way of constructing a shell command
        # but we don't need to worry about injection attacks here
        res = subprocess.run(
            [
                f"caen-toolbox {self.config.digitizer_type} dump {self.config.connection_string} {temp_file}"
            ],
            shell=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.STDOUT,
        )
        if res.returncode != 0:
            raise Exception(f"caen-toolbox dump failed: {res.stderr}")
        if not os.path.exists(temp_file):
            raise Exception("Register dump file not found")
        with open(temp_file, "r") as f:
            lines = f.readlines()
        raw_registers = {}
        # Load CSV file
        for line in lines:
            if line.startswith("#"):
                continue
            parts = line.split(",")
            raw_registers[parts[0]] = int(parts[1], 16)
        registers = {}
        for reg_label in self.config.register_labels:
            key, label = reg_label.register, reg_label.label
            registers[label] = raw_registers[key]
        return registers

    def start(self):
        while True:
            start_time = datetime.now()
            registers = self._dump_digitizer()
            self._put_message_out(
                DAQJobMessageStoreTabular(
                    store_config=self.config.store_config,
                    keys=["timestamp", *[x for x in registers]],
                    data=[
                        [
                            get_now_unix_timestamp_ms(),
                            *[registers[x] for x in registers],
                        ]
                    ],
                )
            )
            sleep_for(DAQ_JOB_CAEN_TOOLBOX_SLEEP_INTERVAL, start_time)

Dumps register data of a CAEN digitizer using the caen-toolbox cli.

Attributes

config : DAQJobCAENToolboxConfig

Configuration for the DAQ job.

config_type : DAQJobCAENToolboxConfig

Configuration type for the DAQ job.

Ancestors

• DAQJob

Class variables

var config : DAQJobCAENToolboxConfig

var config_type : Any

Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.

Attributes

digitizer_type : str

Type of the CAEN digitizer, e.g. dig1

connection_string : str

Connection string of the CAEN digitizer, e.g. -c OPTICAL_LINK -l 0

register_labels : dict[str, str]

Dictionary of register labels, e.g. {"0x11a8": "ADC_ch_1_Temperature"}

Methods

def start(self)

Expand source code

def start(self):
    while True:
        start_time = datetime.now()
        registers = self._dump_digitizer()
        self._put_message_out(
            DAQJobMessageStoreTabular(
                store_config=self.config.store_config,
                keys=["timestamp", *[x for x in registers]],
                data=[
                    [
                        get_now_unix_timestamp_ms(),
                        *[registers[x] for x in registers],
                    ]
                ],
            )
        )
        sleep_for(DAQ_JOB_CAEN_TOOLBOX_SLEEP_INTERVAL, start_time)

Inherited members

• DAQJob:
  • consume
  • handle_message

class DAQJobCAENToolboxConfig (store_config: DAQJobStoreConfig,
digitizer_type: str,
connection_string: str,
register_labels: list[RegisterLabel],
*,
verbosity: LogVerbosity = LogVerbosity.INFO,
remote_config: DAQRemoteConfig | None = <factory>,
daq_job_type: str)

Expand source code

class DAQJobCAENToolboxConfig(StorableDAQJobConfig):
    """
    Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.
    Attributes:
        digitizer_type (str): Type of the CAEN digitizer, e.g. `dig1`
        connection_string (str): Connection string of the CAEN digitizer, e.g. `-c OPTICAL_LINK -l 0`
        register_labels (dict[str, str]): Dictionary of register labels, e.g. `{"0x11a8": "ADC_ch_1_Temperature"}`
    """

    digitizer_type: str
    connection_string: str
    register_labels: list[RegisterLabel]

Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.

Attributes

digitizer_type : str

Type of the CAEN digitizer, e.g. dig1

connection_string : str

Connection string of the CAEN digitizer, e.g. -c OPTICAL_LINK -l 0

register_labels : dict[str, str]

Dictionary of register labels, e.g. {"0x11a8": "ADC_ch_1_Temperature"}

Ancestors

• StorableDAQJobConfig
• DAQJobConfig
• msgspec.Struct
• msgspec._core._StructMixin

Instance variables

var connection_string : str

Expand source code

class DAQJobCAENToolboxConfig(StorableDAQJobConfig):
    """
    Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.
    Attributes:
        digitizer_type (str): Type of the CAEN digitizer, e.g. `dig1`
        connection_string (str): Connection string of the CAEN digitizer, e.g. `-c OPTICAL_LINK -l 0`
        register_labels (dict[str, str]): Dictionary of register labels, e.g. `{"0x11a8": "ADC_ch_1_Temperature"}`
    """

    digitizer_type: str
    connection_string: str
    register_labels: list[RegisterLabel]

var digitizer_type : str

Expand source code

class DAQJobCAENToolboxConfig(StorableDAQJobConfig):
    """
    Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.
    Attributes:
        digitizer_type (str): Type of the CAEN digitizer, e.g. `dig1`
        connection_string (str): Connection string of the CAEN digitizer, e.g. `-c OPTICAL_LINK -l 0`
        register_labels (dict[str, str]): Dictionary of register labels, e.g. `{"0x11a8": "ADC_ch_1_Temperature"}`
    """

    digitizer_type: str
    connection_string: str
    register_labels: list[RegisterLabel]

var register_labels : list[RegisterLabel]

Expand source code

class DAQJobCAENToolboxConfig(StorableDAQJobConfig):
    """
    Configuration for the DAQ job that dumps register data of a CAEN digitizer using the caen-toolbox cli.
    Attributes:
        digitizer_type (str): Type of the CAEN digitizer, e.g. `dig1`
        connection_string (str): Connection string of the CAEN digitizer, e.g. `-c OPTICAL_LINK -l 0`
        register_labels (dict[str, str]): Dictionary of register labels, e.g. `{"0x11a8": "ADC_ch_1_Temperature"}`
    """

    digitizer_type: str
    connection_string: str
    register_labels: list[RegisterLabel]

class RegisterLabel (register: str, label: str)

Expand source code

class RegisterLabel(Struct):
    register: str
    label: str

A base class for defining efficient serializable objects.

Fields are defined using type annotations. Fields may optionally have default values, which result in keyword parameters to the constructor.

Structs automatically define __init__, __eq__, __repr__, and __copy__ methods. Additional methods can be defined on the class as needed. Note that __init__/__new__ cannot be overridden, but other methods can. A tuple of the field names is available on the class via the __struct_fields__ attribute if needed.

Additional class options can be enabled by passing keywords to the class definition (see example below). These configuration options may also be inspected at runtime through the __struct_config__ attribute.

Configuration

frozen: bool, default False Whether instances of this type are pseudo-immutable. If true, attribute assignment is disabled and a corresponding __hash__ is defined. order: bool, default False If True, __lt__, `le, gt, and ge`` methods will be generated for this type. eq: bool, default True If True (the default), an __eq__ method will be generated for this type. Set to False to compare based on instance identity alone. kw_only: bool, default False If True, all fields will be treated as keyword-only arguments in the generated __init__ method. Default is False. omit_defaults: bool, default False Whether fields should be omitted from encoding if the corresponding value is the default for that field. Enabling this may reduce message size, and often also improve encoding & decoding performance. forbid_unknown_fields: bool, default False If True, an error is raised if an unknown field is encountered while decoding structs of this type. If False (the default), no error is raised and the unknown field is skipped. tag: str, int, bool, callable, or None, default None Used along with tag_field for configuring tagged union support. If either are non-None, then the struct is considered "tagged". In this case, an extra field (the tag_field) and value (the tag) are added to the encoded message, which can be used to differentiate message types during decoding.

Set tag=True to enable the default tagged configuration (tag_field is "type", tag is the class name). Alternatively, you can provide a string (or less commonly int) value directly to be used as the tag (e.g. tag="my-tag-value").tag can also be passed a callable that takes the class qualname and returns a valid tag value (e.g. tag=str.lower). See the docs for more information. tag_field: str or None, default None The field name to use for tagged union support. If tag is non-None, then this defaults to "type". See the tag docs above for more information. rename: str, mapping, callable, or None, default None Controls renaming the field names used when encoding/decoding the struct. May be one of "lower", "upper", "camel", "pascal", or "kebab" to rename in lowercase, UPPERCASE, camelCase, PascalCase, or kebab-case respectively. May also be a mapping from field names to the renamed names (missing fields are not renamed). Alternatively, may be a callable that takes the field name and returns a new name or None to not rename that field. Default is None for no field renaming. repr_omit_defaults: bool, default False Whether fields should be omitted from the generated repr if the corresponding value is the default for that field. array_like: bool, default False If True, this struct type will be treated as an array-like type during encoding/decoding, rather than a dict-like type (the default). This may improve performance, at the cost of a more inscrutable message encoding. gc: bool, default True Whether garbage collection is enabled for this type. Disabling this may help reduce GC pressure, but will prevent reference cycles composed of only gc=False from being collected. It is the user's responsibility to ensure that reference cycles don't occur when setting gc=False. weakref: bool, default False Whether instances of this type support weak references. Defaults to False. dict: bool, default False Whether instances of this type will include a __dict__. Setting this to True will allow adding additional undeclared attributes to a struct instance, which may be useful for holding private runtime state. Defaults to False. cache_hash: bool, default False If enabled, the hash of a frozen struct instance will be computed at most once, and then cached on the instance for further reuse. For expensive hash values this can improve performance at the cost of a small amount of memory usage.

Examples

Here we define a new Struct type for describing a dog. It has three fields; two required and one optional.

>>> class Dog(Struct):
...     name: str
...     breed: str
...     is_good_boy: bool = True
...
>>> Dog('snickers', breed='corgi')
Dog(name='snickers', breed='corgi', is_good_boy=True)

Additional struct options can be set as part of the class definition. Here we define a new Struct type for a frozen Point object.

>>> class Point(Struct, frozen=True):
...     x: float
...     y: float
...
>>> {Point(1.5, 2.0): 1}  # frozen structs are hashable
{Point(x=1.5, y=2.0): 1}

Ancestors

• msgspec.Struct
• msgspec._core._StructMixin

Instance variables

var label : str

Expand source code

class RegisterLabel(Struct):
    register: str
    label: str

var register : str

Expand source code

class RegisterLabel(Struct):
    register: str
    label: str