Commit b04bfbdd authored by Cyril Guilloud's avatar Cyril Guilloud
Browse files

Merge branch 'issue-141' into 'master'

Add preset and statistics support for the mercury controller

See merge request !473
parents c3175ae4 e40b4d5d
......@@ -3,6 +3,7 @@
# Imports
import time
import enum
import collections
# Enums
......@@ -18,16 +19,19 @@ DetectorType = enum.Enum(
AcquisitionMode = enum.Enum(
'AcquisitionMode',
'MCA_NORMAL MCA_MAPPING SCA_MAPPING LIST_MODE SCA_HARD')
'SINGLE MULTIPLE')
TriggerMode = enum.Enum(
'TriggerMode',
'INTERNAL SOFTWARE EXTERNAL_GATE EXTERNAL_MULTI_GATE '
'EXTERNAL_MULTI_TRIGGER')
'SOFTWARE EXTERNAL GATE')
PresetMode = enum.Enum(
'PresetMode',
'NONE REAL_TIME LIVE_TIME')
'NONE REALTIME LIVETIME EVENTS TRIGGERS')
Stats = collections.namedtuple(
'Stats',
'realtime livetime triggers events icr ocr deadtime')
# Base class
......@@ -115,17 +119,32 @@ class BaseMCA(object):
def stop_acquisition(self):
raise NotImplementedError
def get_acquisition_status(self):
def is_acquiring(self):
raise NotImplementedError
def get_acquisition_data(self):
raise NotImplementedError
def get_acquisition_statistics(self):
raise NotImplementedError
# Extra logic
def run_single_acquisition(self, acquisition_time=1.):
def run_single_acquisition(self, acquisition_time=1., polling_time=0.1):
# Prepare
realtime = PresetMode.REALTIME in self.supported_preset_modes
if realtime:
self.set_preset_mode(PresetMode.REALTIME, acquisition_time)
else:
self.set_preset_mode(None)
self.prepare_acquisition()
# Start and wait
self.start_acquisition()
time.sleep(acquisition_time)
if realtime:
while self.is_acquiring():
time.sleep(polling_time)
else:
time.sleep(acquisition_time)
# Stop and return data
self.stop_acquisition()
return self.get_acquisition_data()
return self.get_acquisition_data(), self.get_acquisition_statistics()
"""Controller classes for XIA multichannel analyzer"""
# Imports
from numbers import Number
import zerorpc
import msgpack_numpy
from .mca import BaseMCA, Brand, DetectorType
from .mca import BaseMCA, Brand, DetectorType, PresetMode, Stats
# Patch msgpack
msgpack_numpy.patch()
......@@ -111,16 +113,23 @@ class Mercury(BaseMCA):
pass
def start_acquisition(self):
self._proxy.start_run(0)
self._proxy.start_run()
def stop_acquisition(self):
self._proxy.stop_run(0)
self._proxy.stop_run()
def get_acquisition_status(self):
return ""
def is_acquiring(self):
return self._proxy.is_running()
def get_acquisition_data(self):
return [self._proxy.get_run_data(0)]
spectrums = self._proxy.get_spectrums()
nb = len(spectrums)
return [spectrums[i] for i in range(nb)]
def get_acquisition_statistics(self):
stats = self._proxy.get_statistics()
nb = len(stats)
return [Stats(*stats[i]) for i in range(nb)]
# Infos
......@@ -135,3 +144,37 @@ class Mercury(BaseMCA):
@property
def element_count(self):
return 1
# Modes
@property
def supported_preset_modes(self):
return [PresetMode.NONE,
PresetMode.REALTIME,
PresetMode.LIVETIME,
PresetMode.EVENTS,
PresetMode.TRIGGERS]
def set_preset_mode(self, mode, value=None):
# Cast arguments
if mode is None:
mode = PresetMode.NONE
# Check arguments
if mode == PresetMode.NONE and value is not None:
raise TypeError(
'P1reset value should be None when no preset mode is set')
if mode != PresetMode.NONE and not isinstance(value, Number):
raise TypeError(
'Preset value should be a number when a preset mode is set')
# Get hw values
ptype, pcast = {
PresetMode.NONE: (0, lambda x: 0),
PresetMode.REALTIME: (1, float),
PresetMode.LIVETIME: (2, float),
PresetMode.EVENTS: (3, int),
PresetMode.TRIGGERS: (4, int)}[mode]
pvalue = pcast(value)
# Configure
self._proxy.set_acquisition_value('preset_type', ptype)
self._proxy.set_acquisition_value('preset_value', pvalue)
self._proxy.apply_acquisition_values()
"""Test module for MCA base class."""
import pytest
from bliss.controllers.mca import BaseMCA, Brand, DetectorType
from bliss.controllers.mca import BaseMCA, Brand, DetectorType, PresetMode
def test_mca_enums():
......@@ -52,8 +52,9 @@ def test_base_mca():
mca.prepare_acquisition: (),
mca.start_acquisition: (),
mca.stop_acquisition: (),
mca.get_acquisition_status: (),
mca.get_acquisition_data: ()}
mca.is_acquiring: (),
mca.get_acquisition_data: (),
mca.get_acquisition_statistics: ()}
# Test methods
for method, args in methods.items():
......@@ -80,6 +81,11 @@ def test_base_mca_logic(mocker):
class TestMCA(BaseMCA):
supported_preset_modes = [PresetMode.NONE]
def set_preset_mode(self, mode):
assert mode in (None, PresetMode.NONE)
def initialize_attributes(self):
pass
......@@ -95,12 +101,15 @@ def test_base_mca_logic(mocker):
def stop_acquisition(self):
pass
def get_acquisition_status(self):
return ""
def is_running(self):
return False
def get_acquisition_data(self):
return [[3, 2, 1]]
def get_acquisition_statistics(self):
return ['some_stats']
# Create a test mca
config = {}
mca = TestMCA('incomplete', config)
......@@ -109,5 +118,5 @@ def test_base_mca_logic(mocker):
# Run a single acquisition
sleep = mocker.patch('time.sleep')
assert mca.run_single_acquisition(3.) == [[3, 2, 1]]
assert mca.run_single_acquisition(3.) == ([[3, 2, 1]], ['some_stats'])
sleep.assert_called_once_with(3.)
......@@ -2,7 +2,7 @@
import pytest
from bliss.controllers.mca import Brand, DetectorType
from bliss.controllers.mca import Brand, DetectorType, Stats, PresetMode
def test_get_mercury_from_config(beacon, mocker):
......@@ -16,12 +16,18 @@ def test_get_mercury_from_config(beacon, mocker):
client.get_grouped_channels.return_value = ((0, ), )
client.get_config_files.return_value = ['default.ini']
client.get_config.return_value = {'my': 'config'}
client.is_running.return_value = True
# Emulate running behavior
def mock_not_running():
client.is_running.return_value = False
# Instantiating the mercury
mercury = beacon.get('mercury-test')
m.assert_called_once_with('tcp://welisa.esrf.fr:8000')
client.init.assert_called_once_with(
'C:\\\\blissadm\\\\mercury', 'default.ini')
'C:\\\\blissadm\\\\mercury', 'mercury_src.ini')
assert mercury.current_configuration == 'mercury_src.ini'
assert mercury.configured
# Infos
......@@ -35,20 +41,38 @@ def test_get_mercury_from_config(beacon, mocker):
'C:\\\\blissadm\\\\mercury')
assert mercury.current_configuration_values == {'my': 'config'}
client.get_config.assert_called_once_with(
'C:\\\\blissadm\\\\mercury', 'default.ini')
'C:\\\\blissadm\\\\mercury', 'mercury_src.ini')
# PresetMode
mercury.set_preset_mode(None)
assert client.set_acquisition_value.call_args_list == \
[(('preset_type', 0),), (('preset_value', 0),)]
client.apply_acquisition_values.assert_called_once_with()
with pytest.raises(TypeError):
mercury.set_preset_mode(PresetMode.NONE, 1)
with pytest.raises(TypeError):
mercury.set_preset_mode(PresetMode.REALTIME, None)
# Acquisition
client.get_run_data.return_value = [3, 2, 1]
sleep = mocker.patch('time.sleep')
assert mercury.run_single_acquisition(3.) == [[3, 2, 1]]
sleep.assert_called_once_with(3.)
assert mercury.get_acquisition_status() == '' # TODO
sleep.side_effect = lambda x: mock_not_running()
client.get_spectrums.return_value = {0: [3, 2, 1]}
client.get_statistics.return_value = {0: range(7)}
stats = Stats(*range(7))
assert mercury.run_single_acquisition(3.) == ([[3, 2, 1]], [stats])
sleep.assert_called_once_with(0.1)
# Load configuration
client.init.side_effect = IOError('File not found!')
with pytest.raises(IOError):
mercury.load_configuration('i-dont-exist')
assert not mercury.configured
assert mercury.current_configuration is None
assert mercury.current_configuration_values is None
# Finalize
mercury.finalize()
client.close.assert_called_once_with()
def test_get_mercury_from_wrong_config(beacon, mocker):
......
......@@ -4,5 +4,5 @@ class: Mercury
plugin: bliss
url: tcp://welisa.esrf.fr:8000
configuration_directory: C:\\blissadm\\mercury
default_configuration: default.ini
default_configuration: mercury_src.ini
detector_mapping: sdd3elts
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment