Integrating counters

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

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

The IntegratingCounterController is designed for counters data not immediately available and buffered by the controller. It means that during acquisition, a polling mechanism checks periodically if the controller has data. When data are available it may contain multiple measurements that can be returned as a block of a variable size (see get_values).

Usually, it works in association with a master controller. A typical usage example could be counters associated to time integrated data or counters working on data such as ROI statistics in an image.

IntegratingCounter class

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

Signature (from bliss.common.counter)

class IntegratingCounter(Counter):
    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)

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):

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 = 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

IntegratingCounterController class

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

Usually, this object declares a master counter controller but it can work without.

Signature (from bliss.controllers.counter)

class IntegratingCounterController(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 """

Customizable methods (optional)

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

        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 IntegratingCounterAcquisitionSlave(
            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"]

        params = {"count_time": count_time}

        if self._master_controller is None:
            try:
                npoints = acq_params["npoints"]
            except KeyError:
                npoints = scan_params["npoints"]

            params["npoints"] = npoints

        return params

To be overridden methods (mandatory)

    def get_values(self, from_index, *counters):
        """
        Return the counter values for the list of given counters.
        For each counter, data is a list of values (one per measurement).
        All counters must retrieve the same number of data!

        args:
          - from_index: an integer corresponding to the index of the 
            measurement from which new data should be retrieved
          - counters: the list of counters for which measurements should be retrieved.

        example: 
            tmp = [self.get_available_measurements(cnt, from_index) for cnt in counters]
            dmin = min([len(cnt_data) for cnt_data in tmp])
            return [cnt_data[:dmin] for cnt_data in tmp]
        """
        raise NotImplementedError

IntegratingCounterAcquisitionSlave class

The default acquisition object associated to integrating counter controllers.

This class already implements the reading method of its base class. In that method it calls get_values(from_index, *counters) (detailed above) to obtain the list of available measurements for a given list of counters, since a given index.

Usually it is not necessary to override that class, except if the prepare_device, start_device and stop_device methods must be overridden.

If it has a master controller, the npoints, prepare_once and start_once attributes are automatically set to the values of it is master’s acquisition object.

Signature (from bliss.scanning.acquisition.counter)

class IntegratingCounterAcquisitionSlave(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 (float)

• npoints: the scan steps number (integer)

About npoints, prepare_once and start_once

This class sets npoints, prepare_once and start_once to the same values as its master counter controller (whatever the values received at the initialization). If it has no master prepare_once = start_once = False. Note that the scan count_time is accessible (as a property) but not used in this base class implementation.

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 of a MCA device with spectrum and ROIs counters

import time
import gevent
import numpy

from bliss.common.counter import IntegratingCounter
from bliss.controllers.counter import CounterController, IntegratingCounterController
from bliss.scanning.chain import AcquisitionMaster
from bliss.scanning.chain import TRIGGER_MODE_ENUM

class MyMcaDevice:
    """ A mockup of an hardware device measuring data from multiple channels as a spectrum """

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

        # attributes to simulate data acquisition
        self._random_generator = numpy.random.default_rng()
        self._acq_task = None
        self._data_buffer = []
        self._roi_data_buffer = {}

    def add_roi(self, name, roi):
        self.rois[name] = roi
        self._roi_data_buffer[name] = []

    def acquire_spectrum(self, num_of_acq=1):
        for i in range(num_of_acq):
            time.sleep(self.integration_time)
            sdata = self._random_generator.random(self.spectrum_size)*self.integration_time
            self._data_buffer.append(sdata)
            for name, roi in self.rois.items():
                self._roi_data_buffer[name].append(numpy.sum(sdata[roi[0]:roi[1]]))

    def start_acq(self, num_of_acq=1):
        if self._acq_task:
            raise RuntimeError(f"an acquisition task is already running {self._acq_task}")
        self._acq_task = gevent.spawn(self.acquire_spectrum, num_of_acq)

    def get_data(self, from_index):
        """ return spectrum data from a given index """
        return self._data_buffer[from_index:]

    def get_roi_data(self, from_index, *roi_names):
        """ return ROIs data from a given index """
        return [self._roi_data_buffer[name][from_index:] for name in roi_names]

    def clean_buffer(self):
        self._data_buffer = []
        for name in self._roi_data_buffer.keys():
            self._roi_data_buffer[name] = []

    def stop_acq(self):
        self._acq_task.kill()

class MyMcaMasterCounterController(CounterController):
    """ A master counter controller providing an AcquisitionMaster for spectrum acquisition """
    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 MyMcaAcquisitionMaster(self, **acq_params)

    def get_default_chain_parameters(self, scan_params, acq_params):
        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 MyMcaAcquisitionMaster(AcquisitionMaster):
    """ An AcquisitionMaster to drive spectrum acquisition """
    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

        self.t0 = time.perf_counter()

    def prepare(self):
        self.device.hw_device.integration_time = self.integration_time
        self.device.hw_device.clean_buffer()

    def start(self):
        pass

    def stop(self):
        self.device.hw_device.stop_acq()

    def trigger(self):
        self.trigger_slaves()
        self.device.hw_device.start_acq()

    def wait_ready(self):
        if self.device.hw_device._acq_task:
            self.device.hw_device._acq_task.join()

class MySpectrumCounter(IntegratingCounter):
    """ A 1D counter for spectrum counter data """
    def __init__(self, name, 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

    @property
    def dtype(self):
        return float

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

    def dataset_metadata(self) -> dict:
        """ Metadata for ICAT """
        return {"name": self.name, "spectrum_size":self.shape[0]}

    def scan_metadata(self) -> dict:
        metadata = super().scan_metadata()
        metadata.update({"spectrum_size": self.shape[0],})
        return metadata

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

class MySpectrumCounterController(IntegratingCounterController):
    """ An IntegratingCounterController to handle spectrum counter data """
    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_values(self, from_index, *counters):
        data = self.hw_device.get_data(from_index)
        return [data] # it has only one counter


class MyROICounter(IntegratingCounter):
    """ A counter for ROI data """
    def __init__(self, name, roi, 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
        if len(roi) != 2:
            raise ValueError(f"ROI value should be a list of 2 integer values")
        self.roi = roi
        self.hw_device.add_roi(name, roi)

    @property
    def dtype(self):
        return float

    @property
    def shape(self):
        return ()

    def dataset_metadata(self) -> dict:
        """ Metadata for ICAT """
        return {"name": self.name, "roi":self.roi}

    def scan_metadata(self) -> dict:
        metadata = super().scan_metadata()
        metadata.update({"roi": self.roi,})
        return metadata

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

class MyROICounterController(IntegratingCounterController):
    """ An IntegratingCounterController to handle ROI counters data """
    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_values(self, from_index, *counters):
        cnts_data = self.hw_device.get_roi_data(from_index, *[cnt.name for cnt in counters])
        return cnts_data

In this example MySpectrumCounterController and MyROICounterController use the default IntegratingCounterAcquisitionSlave.

During the whole scan, these integrating acquisition slaves will just poll data asynchronously, calling the get_values method defined above, until they receive the expected number of measurements (npoints). The value for npoints is automatically obtained from their master (i.e. MyMcaAcquisitionMaster). In the case of standard step by step scan, the master obtains the npoints value directly from the scan parameters (see MyMcaMasterCounterController.get_default_chain_parameters).

Usage

hw_device = MyMcaDevice()
McaMasterCC = MyMcaMasterCounterController(hw_device, "McaMaster")
SpectrumCC = MySpectrumCounterController(hw_device, "SpectrumCC", McaMasterCC)
RoisCC = MyROICounterController(hw_device, "RoisCC", McaMasterCC)
sp_cnt = MySpectrumCounter("spectrum", SpectrumCC, unit='counts')
roi_cnt1 = MyROICounter("r1", (0, 3), RoisCC, unit='counts')
roi_cnt2 = MyROICounter("r2", (3, 7), RoisCC, unit='counts')
roi_cnt2 = MyROICounter("r3", (7, 13), RoisCC, unit='counts')

loopscan(10, 0.1, sp_cnt, RoisCC) # count with spectrums and ROIs