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