Counters

A counter represents an experimental parameter which can be measured during a scan or a count. Counters are passed to the Scan object in order to select the experimental parameters that will be measured or, in other words, the data that will be produced.

The data reading is not performed by the counter itself but by the CounterController that owns the counter. This allows to perform grouped data reading at the controller level, when multiple counters are owned by the same device.

The Counter object is a placeholder to store information about the counter and its associated data.

For general definitions and usage, see the Counters section.

Standard counters

Counter

The base class for all counters (mandatory).

It defines the common properties and methods shared by all counters:

• name: the counter short name (mandatory)
• _counter_controller: the CounterController which own this counter (mandatory)
• dtype: the data type associated to this counter (numpy dtype) (default is float)
• shape: the shape of the data associated to this counter (() for 0D)
• unit: the data unit (default is None)
• fullname: the concatenation of its name and controller name in the form controller_name:counter_name
• conversion_function: a function to convert data (default is None)
• scan_metadata(): return the metadata dictionary that will be stored in the scan data file (default is None)
• __ info __(): return a text description of the counter (displayed while typing the name of the counter in the shell).

SamplingCounter (SC)

Designed for instantaneous data reading (like reading the actual value of a sensor/channel). Depending on its mode property, the reading is performed once or repeated as many time as possible during the given counting time to provide averaged or statistical measurements. Also, sampling counters provide the raw_read property to read data at any time (even outside the context of a scan).

IntegratingCounter (IC)

Designed for data that are integrated over the counting time and buffered by the counting device. A polling mechanism ( Reading loop) empty the device buffer and retrieves the data chunks. This kind of counter can be bound to a master device that propagates its counting time to the integrating counters that depend on it.

CounterControllers

A Counter is always managed by a CounterController (see mycounter._counter_controller).

A CounterController is a container for counters of the same kind, that usually share the same hardware.

• CounterController (CC) is the base class for all counter controllers (mandatory):

  It defines the common properties and methods shared by all counter controllers:

  • name: the counter controller name (mandatory)
  • _master_controller: an optional master counter controller on top of this one (optional).
  • fullname: the concatenation of its name and master controller fullname in the form master_fullname:controller_name
  • counters: the list of managed counters as a counter_namespace(self._counters)
  • create_counter(counter_class, args, kwargs): dedicated method to create a counter and register it into this CC (self._counters[name]).
  • create_chain_node(): create the default chain node associated to this CC (over-writable).
  • get_acquisition_object(acq_params, ctrl_params, parent_acq_params): return the AcquisitionObject associated to this controller (must be implemented in the child class).
  • get_default_chain_parameters(scan_params, acq_params): return the default acquisition parameters to be use in the context of a standard step-by-step scan (must be implemented in the child class).
  • get_current_parameters(): return an exhaustive dict of the current controller parameters (default: return None).
  • apply_parameters(): load controller parameters into hardware controller (called at the beginning of each scan) (default: pass)

• SamplingCounterController (SCC): dedicated base class to manage SamplingCounters:

  • get_acquisition_object: returns SamplingCounterAcquisitionSlave (over-writable)
  • get_default_chain_parameters: returns count_time and npoints (over-writable)

• IntegratingCounterController (ICC): dedicated base class to manage IntegratingCounters:

  • get_acquisition_object: returns IntegratingCounterAcquisitionSlave (over-writable)
  • get_default_chain_parameters: returns count_time and npoints (over-writable)

Group read

Both ICC and SCC provide mechanism to perform group reads in order to read many counters at once, if they belong to a common controller able to read all channels at once.

For example, all ROI counters of one camera are managed by one counter controller which can retrieve data of all rois at once via one command.

• SamplingCounterController: user must implement at least the read method. If the controller is able to read multiple counters at once, overwrite the read_all method (the one called during acquisition). *counters is a list of one or more counters managed by this CC.

    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.
        """
        # overwrite if hardware can read all the given 'counters' at once
        values = []
        for cnt in counters:
            values.append(self.read(cnt))
        return values

    def read(self, counter):
        """ return the value of the given counter """
        # access hardware and read data corresponding to the given 'counter'
        raise NotImplementedError

• IntegratingCounterController: user must implement the get_values method. This method must retrieve the latest measurements available from the hardware since a given from_index. As the hardware may have buffered multiple measurements, this method must handle a list of data for each counter (data_list). In the returned per counter list, all data_list must have the same length (i.e. same number of measurements).

    def get_values(self, from_index, *counters):
        # access hardware and return latest data since 'from_index'
        # for the given 'counters'.
        # !!! returned (per counter) data lists must have the same length !!!
        raise NotImplementedError

Device controller and counters

While writing a new controller for an hardware device, it is recommended to declare the CounterController(s) as local attribute(s).

The counter controller class could be directly inherited but this would mix the counter controller methods with the device controller API.

The fact that the device controller may need multiple counter controllers must be kept in mind.

For example, a device controller may require a SamplingCounterController and an IntegratingCounterController.

CounterContainer protocol

In order to tell Bliss that a device controller can be used in a scan as a counting device, it must inherit from the CounterContainer protocol. Inheriting from this protocol implies that the class implements a .counters property which returns the list of managed counters (as a counter_namespace).

class MyDeviceController(CounterContainer):
    def __init__(self, name, config):
        self.name = name
        self._icc_taggs = ['xxx', 'yyy', 'zzz']
        self._scc = MySamplingCounterController(f"{self.name}_scc")
        self._icc = MyIntegratingCounterController(f"{self.name}_icc")

        for cnt_cfg in config.get('counters', []):
            tag = cnt_cfg['tag']
            if tag in self._icc_taggs:
                self._icc.create_counter(MyIntegratingCounter, cnt_cfg)
            else:
                self._scc.create_counter(MySamplingCounter, cnt_cfg)

    @autocomplete_property
    def counters(self):     # conform to the CounterContainer protocol
        cnts = list(self._scc.counters) + list(self._icc.counters)
        return counter_namespace(cnts)

Examples

Moco controller with sampling counters

# === The Counter class
class MocoCounter(SamplingCounter):
    def __init__(self, name, config, controller):
        self.role = config["role"]
        if self.role not in controller.moco.values.keys():
            raise RuntimeError(
                f"moco: counter {self.name} role {self.role} does not exists"
            )
        SamplingCounter.__init__(self, name, controller, mode=SamplingMode.LAST)

# === The CounterController class
class MocoCounterController(SamplingCounterController):
    def __init__(self, moco):
        self.moco = moco
        super().__init__(self.moco.name, register_counters=False)
        global_map.register(moco, parents_list=["counters"])

    def read_all(self, *counters):
        self.moco.moco_read_counters()
        values = []
        for cnt in counters:
            values.append(self.moco.values[cnt.role])
        return values

# === The top level controller class (exposed at user level)
class Moco(CounterContainer):
    def __init__(self, name, config_tree):
        self.values = {"outbeam": 0.0, "inbeam": 0.0,}
        self.name = name
        # Communication
        self._cnx = get_comm(config_tree, timeout=3)
        global_map.register(self, children_list=[self._cnx])
        # default config
        self._default_config = config_tree.get("default_config", None)
        # motor
        self.motor = None

        # Counters
        self.counters_controller = MocoCounterController(self)
        self.counters_controller.max_sampling_frequency = config_tree.get(
            "max_sampling_frequency"
        )
        counter_node = config_tree.get("counters")
        for config_dict in counter_node:
            counter_name = config_dict.get("counter_name")
            MocoCounter(counter_name, config_dict, self.counters_controller)

    @property
    def counters(self):
        return self.counters_controller.counters

    def moco_read_counters(self):
        ret_val = self.comm("xxx")

CT2 controller (as master) with integrating counters (as slave)

# === The Counter class
class CT2Counter(IntegratingCounter):
    def __init__(self, name, channel, controller, unit=None):
        self.channel = channel
        super().__init__(name, controller, unit=unit)

    def convert(self, data):
        return data

# === The CounterController class
class CT2CounterController(IntegratingCounterController):
    def __init__(self, name, master_controller):
        super().__init__(name=name, master_controller=master_controller)
        self.counter_indexes = {}

    def get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
        from bliss.scanning.acquisition.ct2 import CT2CounterAcquisitionSlave

        if "acq_expo_time" in parent_acq_params:
            acq_params.setdefault("count_time",
                                   parent_acq_params["acq_expo_time"])
        if "npoints" in parent_acq_params:
            acq_params.setdefault("npoints", parent_acq_params["npoints"])

        return CT2CounterAcquisitionSlave(self, ctrl_params=ctrl_params,
                                                **acq_params)

    def get_values(self, from_index, *counters):
        data = self._master_controller.get_data(from_index).T
        if not data.size:
            return len(counters) * (numpy.array(()),)
        result = [
            counter.convert(data[self.counter_indexes[counter]]) \
            for counter in counters
        ]
        return result

# === The main controller which is also a CounterController acting
# === as a master above the CT2CounterController (slave) class
class CT2Controller(CounterController):
    def __init__(self, device_config, name="ct2_cc", **kwargs):
        super().__init__(name=name, register_counters=False)

        # declare an ICC with 'self' as the '_master_controller'
        self._slave = CT2CounterController("ct2_counters_controller", self)

        # Add ct2 counters
        for channel in device_config.get("channels", ()):
            ct_name = channel.get("counter name", None)
            if ct_name:
                address = int(channel["address"])
                self._slave.create_counter(CT2Counter, ct_name, address)

    @property
    def counters(self):
        return self._slave.counters

    def get_acquisition_object(self, acq_params, ctrl_params, parent_acq_params):
        from bliss.scanning.acquisition.ct2 import (
            CT2AcquisitionMaster,
            CT2VarTimeAcquisitionMaster,
        )

        if isinstance(acq_params.get("acq_expo_time"), list):
            return CT2VarTimeAcquisitionMaster(
                self, ctrl_params=ctrl_params, **acq_params
            )
        else:
            return CT2AcquisitionMaster(self,
                                        ctrl_params=ctrl_params,
                                        **acq_params)

    def get_default_chain_parameters(self, scan_params, acq_params):
        # Extract scan parameters
        try:
            npoints = acq_params["npoints"]
        except KeyError:
            npoints = scan_params.get("npoints", 1)

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

        acq_point_period = acq_params.get("acq_point_period")
        acq_mode = acq_params.get("acq_mode", AcqMode.IntTrigMulti)
        prepare_once = acq_params.get("prepare_once", True)
        start_once = acq_params.get("start_once", True)
        read_all_triggers = acq_params.get("read_all_triggers", False)

        params = {}
        params["npoints"] = npoints
        params["acq_expo_time"] = acq_expo_time
        params["acq_point_period"] = acq_point_period
        params["acq_mode"] = acq_mode
        params["prepare_once"] = prepare_once
        params["start_once"] = start_once
        params["read_all_triggers"] = read_all_triggers

        return params