Skip to content

Sampling counters

This section describes the implementation of sampling counters and sampling counter controllers.

If you read this lines for the first time, please have a look to the counter basics first

The SamplingCounterController is designed for counters data that can be measured multiple times during the count time associated to one scan step. It requires that the method used to read data from the hardware is almost instantaneous and that it returns only one measure per call.

A typical example could be reading a current, a voltage or a temperature.

Depending on the sampling mode, data reading is performed only once per scan steps or repeated as many time as possible during the given count time. The returned counter value can be the mean value or statistical measurements.

Sampling counters provide the raw_read property to immediately read data from the hardware device at any time, even outside the context of a scan.

SamplingCounter class

Base class for all sampling counters. It inherits from the Counter base class. It is compatible with the HasMetadataForScan and HasMetadataForDataset protocols.

Signature (from bliss.common.counter)

class SamplingCounter(Counter):
    def __init__(
        self,
        name,
        controller,
        conversion_function=None,
        mode=SamplingMode.MEAN,
        unit=None,
    ):
  • name: a name for this counter (str)

  • controller: the counter controller that will manage this counter (mandatory)

  • conversion_function: a function to convert counter’s data on the fly (optional)

  • mode: the sampling mode (default is “MEAN”)

  • unit: the counter’s data unit (optional)

Sampling modes

@enum.unique
class SamplingMode(enum.IntEnum):
    """SamplingCounter modes:
    * MEAN: emit the mathematical average
    * STATS: in addition to MEAN, use iterative algorithms to emit std,min,max,N etc.
    * SAMPLES: in addition to MEAN, emit also individual samples as 1D array
    * SINGLE: emit the first value (if possible: call read only once)
    * LAST: emit the last value
    * INTEGRATE: emit MEAN multiplied by counting time
    """

    MEAN = enum.auto()
    STATS = enum.auto()
    SAMPLES = enum.auto()
    SINGLE = enum.auto()
    LAST = enum.auto()
    INTEGRATE = enum.auto()
    INTEGRATE_STATS = enum.auto()

More information about sampling modes here

Inherited properties (see Counter class)

    @property
    def _counter_controller(self):
        """ Return the CounterController owning this counter """

    @property
    def name(self):
        """ Return the counter name """

    @property
    def fullname(self):
        """ Return fullname as '<counter_controller_name>:<counter_name>' """

    @property
    def unit(self):
        """ Return counter data unit (default is None) """

    @unit.setter
    def unit(self, unit):

    @property
    def conversion_function(self):
        """ Return a data conversion function (default is `_identity`) """

    @conversion_function.setter
    def conversion_function(self, func):

Specific properties

    @property
    def mode(self):
        """ Return the current sampling mode """

    @mode.setter
    def mode(self, value):
        """ Set the current sampling mode """

    @property
    def statistics(self):
        """ Return the statistics of the last measurement """

    @property
    def raw_read(self):
        """ Perform one measurement (no sampling) and return the value """

Customizable properties and methods

    @property
    def dtype(self):
        """ Return counter data type """
        return float

    @property
    def shape(self):
        """ Return counter data shape """
        return ()

    def dataset_metadata(self) -> dict:
        """ Return metadata dictionary for ICAT """
        return {"name": self.name}

    def scan_metadata(self) -> dict:
        """ Return scan metadata dictionary """
        return dict()

    def __info__(self):
        """ Return a description of the counter as a string """
        info_str = super().__info__()
        info_str += f" mode  = {SamplingMode(self.mode).name} ({self.mode})\n"
        return info_str

SamplingCounterController class

The sampling counter controller object is a container for sampling counters associated to one equipment. It inherits from the CounterController base class.

Signature (from bliss.controllers.counter)

class SamplingCounterController(CounterController):
    def __init__(self, name, master_controller=None, register_counters=True):
  • name: a name for this counter controller (str)

  • master_controller: a counter controller instance acting as a master of this one

  • register_counters: a flag to tell if counters registration should be made by this object

Inherited properties and methods (see CounterController class)

    @property
    def name(self):
        """ Return counter controller name """

    @property
    def fullname(self):
        """ Return '<master_controller_fullname>:<name>' 
            If master controller is None, it just returns <name>
        """

    @property
    def _master_controller(self):
        """ Return the master counter controller object (default is None) """

    @property
    def counters(self):
        """ Return owned counters as a counter_namespace """

Specific properties

    @property
    def max_sampling_frequency(self):
        """get maximum sampling frequency in acquisition loop (Hz) (None -> no limit)"""

    @max_sampling_frequency.setter
    def max_sampling_frequency(self, freq):
        """Set maximum sampling frequency"""

Customizable methods (optional)

    def get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
        """
        Return a SamplingCounterAcquisitionSlave.

        args:
         - `acq_params`: parameters for the acquisition object (dict)
         - `ctrl_params`: parameters for the controller (dict)
         - `parent_acq_params`: acquisition parameters of the master (if any)

        """
        return SamplingCounterAcquisitionSlave(
            self, ctrl_params=ctrl_params, **acq_params
        )

    def get_default_chain_parameters(self, scan_params, acq_params):
        """
        Return necessary acquisition parameters in the context of step by step scans.

        args:
         - `scan_params`: parameters of the scan (dict)
         - `acq_params`: parameters for the acquisition (dict)

        return: a dictionary of acquisition parameters

        In the context of a step-by-step scan, `acq_params` is usually empty 
        and the returned dict must be deduced from `scan_params`. 

        However, in the case of a customized DEFAULT_CHAIN, `acq_params` may 
        be not empty and these parameters must override the default ones.

        """
        try:
            count_time = acq_params["count_time"]
        except KeyError:
            count_time = scan_params["count_time"]

        try:
            npoints = acq_params["npoints"]
        except KeyError:
            npoints = scan_params["npoints"]

        params = {"count_time": count_time, "npoints": npoints}

        return params

    def read_all(self, *counters):
        """
        Return the values of the given counters as a list.
        If possible this method should optimize the reading of all counters at once.
        """
        values = []
        for cnt in counters:
            values.append(self.read(cnt))
        return values        

To be overridden methods (mandatory)

    def read(self, counter):
        """Return the value of the given counter"""
        raise NotImplementedError

If all counters can be read in one call, it is better to override the read_all method instead

SamplingCounterAcquisitionSlave class

The default acquisition object associated to sampling counter controllers. Usually it is not required to override that class, except if the prepare_device, start_device and stop_device needs to be overridden.

This class implements all the necessary acquisition object methods to perform sampled measurements.

Signature (from bliss.scanning.acquisition.counter)

class SamplingCounterAcquisitionSlave(BaseCounterAcquisitionSlave):
    def __init__(self, controller, ctrl_params=None, 
        count_time=None, npoints=1):
  • controller: the associated CounterController instance

  • ctrl_params: optional controller parameters

  • count_time: the scan step counting time (a float or a list of npoints numbers)

  • npoints: the scan steps number (integer)

About prepare_once and start_once

This class sets prepare_once = True and start_once = npoints > 0, then npoints = max(1, npoints).

Inherited properties and methods (see BaseCounterAcquisitionSlave class)

    @staticmethod
    def get_param_validation_schema():
        """ An acquisition parameters validation scheme"""
        acq_params_schema = {
            "count_time": {"type": "numeric"},
            "npoints": {"type": "int"},
        }

        schema = {"acq_params": {"type": "dict", "schema": acq_params_schema}}
        return schema

    @property
    def npoints(self):
        return self.__npoints

    @property
    def count_time(self):
        return self.__count_time 

Customizable methods (optional)

    def prepare_device(self):
        pass

    def start_device(self):
        pass

    def stop_device(self):
        pass

Example

Implementation example for a device measuring current and voltage

import numpy

from bliss.common.counter import SamplingCounter
from bliss.controllers.counter import SamplingCounterController

class MyPowerSupplyDevice:
    """ A mockup of an hardware device measuring current and voltage"""

    def __init__(self) -> None:            
        # simulate data generation
        self._random_generator = numpy.random.default_rng()

    def read_data(self, cmd):
        if cmd == "VLT":
            return self._random_generator.random()*1000
        elif cmd == "CUR":
            return self._random_generator.random()


class MyPowerSupplyCounter(SamplingCounter):
    """ A sampling counter """
    def __init__(self, name, channel, controller, 
                conversion_function=None, mode="MEAN", unit=None):
        super().__init__(name, controller, conversion_function, mode,  unit)

        # a reference to the hw device
        self.hw_device = self._counter_controller.hw_device

        # a reference to the hw device channel associated to this counter
        self.channel = channel

    def dataset_metadata(self) -> dict: # optional
        """ Return metadata dictionary for ICAT """
        return {"name": self.name, "channel":self.channel}

    def scan_metadata(self) -> dict: # optional
        """ Return scan metadata dictionary """
        metadata = super().scan_metadata()
        metadata.update({"channel":self.channel,})
        return metadata

    def __info__(self): # optional
        info_str = super().__info__()
        info_str += f" channel: {self.channel}\n"
        return info_str


class MyPowerSupplyCounterController(SamplingCounterController):
    """ A sampling counter controller class """
    def __init__(self, hw_device, name,  master_controller=None, register_counters=True):
        super().__init__(name, master_controller, register_counters)

        # a reference to the hardware device
        self.hw_device = hw_device

    def read(self, counter):
        """Return the value of the given counter"""
        return self.hw_device.read_data(counter.channel)

In this example MyPowerSupplyCounterController uses the default SamplingCounterAcquisitionSlave

Usage

hw_device = MyPowerSupplyDevice()
supplyCC = MyPowerSupplyCounterController(hw_device, "power-supply")
cnt1 = MyPowerSupplyCounter("voltage", "VLT", supplyCC, mode="SINGLE", unit='V')
cnt2 = MyPowerSupplyCounter("current", "CUR", supplyCC, mode="MEAN", unit='mA')
loopscan(10, 0.1, supplyCC) # counting with all counters of supplyCC
loopscan(10, 0.1, cnt1, cnt2) # or explicitely pass counter objects