Counters basics

A counter represents an experimental value that can be measured during an experiment with a given equipment.

In BLISS, Counter objects are passed to scan commands in order to describe what are the experimental values that will be measured and the type of data that will be produced.

Counters are always associated to a CounterController object which is the one who makes the link with the equipment and who defines methods used to perform counters data acquisition.

Counters of the same kind associated to one equipment are grouped under one counter controller instance (standard counters overview).

Counter class

Mandatory base class for all counters. It stores information about the counter itself and associated data. For compatibility with metadata production it inherits from the HasMetadataForScan (HDF5) and HasMetadataForDataset (ICAT) protocols.

Signature (from bliss.common.counter)

class Counter(HasMetadataForScan, HasMetadataForDataset):

    def __init__(self, name, controller, conversion_function=None, 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)

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

During its instantiation the counter is automatically added to the counters list of its controller

Properties

    @property
    def _counter_controller(self):
        """ Return the CounterController instance managing 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):
        """ Set the counter data unit (string) """

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

    @conversion_function.setter
    def conversion_function(self, func):
        """ Set the data conversion function (callable) """

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 (ICAT) """
        return {"name": self.name}

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

    def __info__(self):
        """ Return a description of the counter as a string """
        info_str = f"{self.__class__.__name__}:\n"
        info_str += f" name  = {self.name}\n"
        info_str += f" dtype = {self.dtype.__name__}\n"
        info_str += f" shape = {len(self.shape)}D\n"
        info_str += f" unit  = {self.unit}\n"
        if self.conversion_function is not _identity:
            info_str += f" conversion_function = {self.conversion_function}\n"
        return info_str

CounterController class

The counter controller object is a container for counters of the same kind that are associated to one equipment. The purposes of this object are:

• define methods to access counters data produced by an equipment

• provide acquisition object(s) that should be used in a scanning procedure

• reference a master counter controller (optional)

Signature (from bliss.controllers.counter)

class CounterController(CounterContainer):
    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

Properties and methods

    @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 instance (default is None) """

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

Customizable methods (optional)

    def create_chain_node(self):
        """
        Return associated ChainNode object.
        """
        return ChainNode(self)

    def apply_parameters(self, parameters):
        """
        This method is called by the acquisition chain at the beginning of each scan.
        It receives a dictionary of parameters that could be applied to the controller
        before running a scan.

        This method is usefull when a controller state could have been modifed outside 
        BLISS by another process. By implementing this method and applying the given 
        parameters you can override any potential modifications made from the outside, 
        before running the scan.
        """
        pass

    def get_current_parameters(self):
        """Return an exhaustive dictionary of parameters characterizing the current
        state of a controller.
        """
        return None

To be overridden methods (mandatory)

    def get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
        """
        Return the acquisition object instance that should be used during a scanning procedure.
        The choice can be different depending on the received parameters.

        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: an AcquisitionObject instance
        """
        raise NotImplementedError

    def get_default_chain_parameters(self, scan_params, acq_params):
        """
        Return a dictionary of acquisition parameters that are necessary to
        instantiate an acquisition object in the context of a step-by-step scan.

        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.

        """
        raise NotImplementedError

More information in the scan builder mechanisms chapter

Scan builder mechanisms

In order to perform a scan in BLISS, it is necessary to build an acquisition chain from the list of counters passed to the scan command.

In the case of standard scans commands, this is done automatically by BLISS via the default chain. For other cases, this must be written by hand (see writing a custom scan).

In both cases, the construction of the acquisition chain relies on the ChainBuilder helper object. Staring from the counters list, it gathers all the associated counter controllers and from them, it obtains corresponding acquisition objects which will populate the acquisition chain.

The role of an acquisition object is to describe when and how to acquire and publish counters data during the lifetime of a scan. For each counter involved in the scan, the associated acquisition object creates an AcquisitionChannel which will emit the data toward REDIS.

Obtaining acquisition objects

The acquisition object is obtained from a counter controller by calling its get_acquisition_object(acq_params, ...) method. This method must return an instance of an AcquisitionObject (master or slave).

The acq_params argument of this method is a dictionary containing the acquisition parameters that are necessary for the instantiation of the acquisition object.

Depending on the value of some acquisition parameters it is possible to choose between different acquisition object.

Implementation example

def get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
    """
    Return the acquisition object instance that should be used during a scanning procedure.
    The choice can be different depending on the received parameters.

    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: an AcquisitionObject instance
    """

    trigger_mode = acq_params.pop("trigger_mode", "SOFTWARE")
    if trigger_mode == "SOFTWARE":
        return MyAcquisitionObject_SW(self, **acq_params)
    elif trigger_mode == "HARDWARE":
        return MyAcquisitionObject_HW(self, **acq_params)
    else:
        raise ValueError(f"Unknown trigger mode {trigger_mode}")

The ctrl_params are not detailed here and it can be assumed to be an empty dict

The parent_acq_params dict is usually empty except if this counter controller has a master

Default acquisition parameters

In the context of standard scans, the default chain will call the get_default_chain_parameters(scan_params, acq_params) method on counter controllers to obtain the proper acquisition parameters (acq_params) that should be passed to the get_acquisition_object method detailed above.

This method must return a dictionary containing all the necessary parameters that are required to instantiate the acquisition object returned by the get_acquisition_object method.

The get_default_chain_parameters method receives the scan_params and acq_params arguments. Usually the second is empty and the acquisition parameters that must be returned are guessed from the scan parameters.

For example, the number of measurements that an acquisition object must perform can be guessed from the number of scan points (steps). Also, the exposure time of a device can be guessed from the count time of the scan.

However, because the default chain can be customized with presets, it is possible that the second argument is not empty. In that case, the custom acquisition parameters must override the default.

For that reason, it is necessary to first try to find the parameter value in the acq_params argument and if it cannot be found there, the default value can be guessed from one of the scan parameters or just hardcoded.

Implementation example

def get_default_chain_parameters(self, scan_params, acq_params):
    """
    Return a dictionary of acquisition parameters that are necessary to
    instantiate an acquisition object in the context of a step-by-step scan.

    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.

    """
    params = {}

    # defaults guessed from scan_params
    params["npoints"] = acq_params.get("npoints", scan_params.get("npoints", 1))
    params["acq_expo_time"] = acq_params.get("acq_expo_time", scan_params["count_time"])

    # hardcoded defaults 
    params["prepare_once"] = acq_params.get("prepare_once", True)
    params["start_once"] = acq_params.get("start_once", True)

    return params

Master counter controllers

A counter controller with a master counter controller (see _master_controller property) will automatically bring its master when involved in a scanning procedure. The acquisition object of the counter controller will be placed just below the acquisition object of its master. Multiple counter controllers can share the same master.

About master and acquisition parameters

If a child of a master cannot found its own acquisition parameters, it will try to find them from the master’s parameters. So it is not necessary to set the acquisition parameters of children if the master have them already.

Example

Implementation of a device measuring spectrums on two channels

import numpy
import time
import gevent

from bliss.common.counter import Counter
from bliss.controllers.counter import CounterController
from bliss.scanning.chain import AcquisitionSlave
from bliss.scanning.chain import TRIGGER_MODE_ENUM

class MySpectroDevice:
    """ A mockup of an hardware device measuring spectrums on 2 channels A and B"""

    def __init__(self) -> None:
        self.integration_time = 1
        self.spectrum_size = 100

        # attributes to simulate data acquisition
        self._random_generator = numpy.random.default_rng()
        self._acq_tasks = {}

    def acquire_spectrum(self, channel):
        factor = max(int(self.integration_time*1000), 1)
        time.sleep(self.integration_time)
        if channel == "A":
            return self._random_generator.integers(0, 255*factor, self.spectrum_size)
        else:
            return self._random_generator.integers(100, 255*2*factor, self.spectrum_size)

    def start_acq(self):
        for task in self._acq_tasks.values():
            if task:
                raise RuntimeError(f"an acquisition task is already running {task}")

        self._acq_tasks = {chan: gevent.spawn(self.acquire_spectrum, chan) for chan in ["A", "B"]}

    def get_data(self, channel):
        return self._acq_tasks[channel].get()

    def stop_acq(self):
        gevent.killall(list(self._acq_tasks.values()))


class MyIntegerSpectrumCounter(Counter):
    """ A counter for spectrum (1D) of integer data """
    def __init__(self, name, channel, controller, conversion_function=None, unit=None):
        super().__init__(name, controller, conversion_function, 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

    @property
    def dtype(self):
        return int

    @property
    def shape(self):
        return (self.hw_device.spectrum_size, )

    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({"spectrum_size": self.shape[0],})
        return metadata

    def __info__(self): # optional
        info_str = super().__info__()
        info_str += f" spectrum size: {self.shape[0]}\n"
        return info_str


class MySpectrumCounterController(CounterController):
    """ A counter controller class for spectrum counters"""
    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 get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
        """return associated acquisition object"""
        return MySpectrumAcquisitionObject(self, **acq_params)

    def get_default_chain_parameters(self, scan_params, acq_params):
        """return default acquisition parameters in the context of a step by step scan"""
        params = {}

        # defaults guessed from scan_params
        params["npoints"] = acq_params.get("npoints", scan_params.get("npoints", 1))
        params["integration_time"] = acq_params.get("integration_time", scan_params["count_time"])

        # hardcoded defaults 
        params["prepare_once"] = acq_params.get("prepare_once", True)
        params["start_once"] = acq_params.get("start_once", False)

        return params

class MySpectrumAcquisitionObject(AcquisitionSlave):
    """ An acquisition object to handle spectrum measurement of MySpectroDevice """
    def __init__(self, 
        controller, # the associated counter controller 
        name=None, 
        npoints=1, 
        integration_time=1,
        trigger_type=TRIGGER_MODE_ENUM.SOFTWARE,
        prepare_once=True, 
        start_once=False, 
        ctrl_params=None):

        super().__init__(
            controller, 
            name=name, 
            npoints=npoints, 
            trigger_type=trigger_type,
            prepare_once=prepare_once, 
            start_once=start_once, 
            ctrl_params=ctrl_params
            )

        self.integration_time = integration_time

    def prepare(self):
        self.device.hw_device.integration_time = self.integration_time
        self._stop_flag = False

    def start(self):
        self.device.hw_device.start_acq()

    def stop(self):
        self._stop_flag = True
        self.device.hw_device.stop_acq()

    def trigger(self):
        pass

    def reading(self):
        data = [self.device.hw_device.get_data(cnt.channel) for cnt in self._counters]
        if not self._stop_flag:
            self.channels.update_from_iterable(data)

Find more about writing an acquisition object here

In this example, the acquisition object is set with start_once=False. In that case start is called at each scan step. Also, note that the reading method is spawned after each start if not already running.

Usage

hw_device = MySpectroDevice()
spectroCC = MySpectrumCounterController(hw_device, "spectro")
cnt1 = MyIntegerSpectrumCounter("spectrum-1", "A", spectroCC, unit='counts')
cnt2 = MyIntegerSpectrumCounter("spectrum-2", "B", spectroCC, unit='counts')
loopscan(10, 0.1, spectroCC) # counting with all counters of spectroCC
loopscan(10, 0.1, cnt1, cnt2) # or explicitely pass counter objects