Commit 3faa1d4a authored by Cyril Guilloud's avatar Cyril Guilloud
Browse files

Merge branch 'issue-142' into 'master'

Add external trigger support (gate) for the mercury controller

See merge request !474
parents b04bfbdd b3835bb5
......@@ -17,10 +17,6 @@ DetectorType = enum.Enum(
'FALCONX1 FALCONX4 FALCONX8 XMAP MERCURY MERCURY4 MICRO_DXP DXP_2X '
'MAYA2000 MUSST_MCA MCA8000D DSA1000 MULTIMAX')
AcquisitionMode = enum.Enum(
'AcquisitionMode',
'SINGLE MULTIPLE')
TriggerMode = enum.Enum(
'TriggerMode',
'SOFTWARE EXTERNAL GATE')
......@@ -31,7 +27,8 @@ PresetMode = enum.Enum(
Stats = collections.namedtuple(
'Stats',
'realtime livetime triggers events icr ocr deadtime')
'realtime livetime triggers events icr ocr deadtime '
'underflows overflows')
# Base class
......@@ -85,13 +82,6 @@ class BaseMCA(object):
def set_preset_mode(self, mode, *args):
raise NotImplementedError
@property
def supported_acquisition_modes(self):
raise NotImplementedError
def set_acquisition_mode(self, mode):
raise NotImplementedError
@property
def supported_trigger_modes(self):
raise NotImplementedError
......@@ -108,11 +98,21 @@ class BaseMCA(object):
def set_spectrum_range(self, first, last):
raise NotImplementedError
# Acquisition
# Acquisition number (number of points in acquisition)
def prepare_acquisition(self):
@property
def acquisition_number(self):
raise NotImplementedError
def set_acquisition_number(self, value):
raise NotImplementedError
@property
def multiple_acquisition(self):
return self.acquisition_number > 1
# Acquisition
def start_acquisition(self):
raise NotImplementedError
......@@ -131,13 +131,16 @@ class BaseMCA(object):
# Extra logic
def run_single_acquisition(self, acquisition_time=1., polling_time=0.1):
# Prepare
# Acquisition number
self.set_acquisition_number(1)
# Trigger mode
self.set_trigger_mode(None)
# Preset mode
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()
if realtime:
......@@ -148,3 +151,27 @@ class BaseMCA(object):
# Stop and return data
self.stop_acquisition()
return self.get_acquisition_data(), self.get_acquisition_statistics()
def run_external_acquisition(self, acquistion_time=None, polling_time=0.1):
# Acquisition number
self.set_acquisition_number(1)
# Trigger mode
mode = TriggerMode.EXTERNAL if acquistion_time else TriggerMode.GATE
if mode not in self.supported_trigger_modes:
raise ValueError('{} is not supported'.format(mode))
self.set_trigger_mode(mode)
# Preset mode
if acquistion_time:
self.set_preset_mode(PresetMode.REALTIME, acquistion_time)
else:
self.set_preset_mode(None)
# Start and wait
self.start_acquisition()
get_realtime = lambda: self.get_acquisition_statistics()[0].realtime
previous, current = 0., get_realtime()
while current == 0. or previous != current:
time.sleep(polling_time)
previous, current = current, get_realtime()
# Stop and return data
self.stop_acquisition()
return self.get_acquisition_data(), self.get_acquisition_statistics()
......@@ -6,7 +6,7 @@ from numbers import Number
import zerorpc
import msgpack_numpy
from .mca import BaseMCA, Brand, DetectorType, PresetMode, Stats
from .mca import BaseMCA, Brand, DetectorType, PresetMode, Stats, TriggerMode
# Patch msgpack
msgpack_numpy.patch()
......@@ -107,12 +107,38 @@ class Mercury(BaseMCA):
grouped_channels = self._proxy.get_grouped_channels()
assert grouped_channels == ((0, ), )
# Acquisition
# Acquisition number
@property
def acquisition_number(self):
mapping = int(self._proxy.get_acquisition_value('mapping_mode', 0))
if mapping == 0:
return 1
# Should be:
# self._proxy.get_acquisition_value('num_map_pixels', 0)
raise NotImplementedError
def set_acquisition_number(self, value):
# Invalid argument
if value < 1:
raise ValueError('Acquisition number should be strictly positive')
# Single mode
if value == 1:
self._proxy.set_acquisition_value('mapping_mode', 0)
self._proxy.apply_acquisition_values()
return
# Multiple mode
# Should be:
# self._proxy.set_acquisition_value('mapping_mode', 1)
# self._proxy.set_acquisition_value('num_map_pixels, value)
# self._proxy.apply_acquisition_values()
raise NotImplementedError
def prepare_acquisition(self):
pass
# Acquisition
def start_acquisition(self):
# Make sure the acquisition is stopped first
self._proxy.stop_run()
self._proxy.start_run()
def stop_acquisition(self):
......@@ -160,13 +186,15 @@ class Mercury(BaseMCA):
if mode is None:
mode = PresetMode.NONE
# Check arguments
if mode not in self.supported_preset_modes:
raise ValueError('{!s} preset mode not supported'.format(mode))
if mode == PresetMode.NONE and value is not None:
raise TypeError(
'P1reset value should be None when no preset mode is set')
'Preset 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
# Get hardware values
ptype, pcast = {
PresetMode.NONE: (0, lambda x: 0),
PresetMode.REALTIME: (1, float),
......@@ -178,3 +206,21 @@ class Mercury(BaseMCA):
self._proxy.set_acquisition_value('preset_type', ptype)
self._proxy.set_acquisition_value('preset_value', pvalue)
self._proxy.apply_acquisition_values()
@property
def supported_trigger_modes(self):
return [TriggerMode.SOFTWARE,
TriggerMode.GATE]
def set_trigger_mode(self, mode):
# Cast arguments
if mode is None:
mode = TriggerMode.SOFTWARE
# Check arguments
if mode not in self.supported_trigger_modes:
raise ValueError('{!s} trigger mode not supported'.format(mode))
# Get hardware value
value = 0 if mode == TriggerMode.GATE else 1
# Configure
self._proxy.set_acquisition_value('gate_ignore', value)
self._proxy.apply_acquisition_values()
"""Test module for MCA base class."""
import pytest
from bliss.controllers.mca import BaseMCA, Brand, DetectorType, PresetMode
from bliss.controllers.mca import BaseMCA, Brand, DetectorType
from bliss.controllers.mca import PresetMode, TriggerMode, Stats
def test_mca_enums():
......@@ -46,10 +47,9 @@ def test_base_mca():
methods = {
mca.finalize: (),
mca.set_preset_mode: ('some_mode',),
mca.set_acquisition_mode: ('some_mode',),
mca.set_acquisition_number: (12,),
mca.set_trigger_mode: ('some_mode',),
mca.set_spectrum_range: ('some', 'range'),
mca.prepare_acquisition: (),
mca.start_acquisition: (),
mca.stop_acquisition: (),
mca.is_acquiring: (),
......@@ -66,8 +66,8 @@ def test_base_mca():
'detector_brand',
'detector_type',
'element_count',
'acquisition_number',
'supported_preset_modes',
'supported_acquisition_modes',
'supported_trigger_modes',
'calibration_type']
......@@ -78,6 +78,7 @@ def test_base_mca():
def test_base_mca_logic(mocker):
stats = Stats(*range(1, 10))
class TestMCA(BaseMCA):
......@@ -86,15 +87,22 @@ def test_base_mca_logic(mocker):
def set_preset_mode(self, mode):
assert mode in (None, PresetMode.NONE)
supported_trigger_modes = [TriggerMode.SOFTWARE, TriggerMode.GATE]
def set_trigger_mode(self, mode):
assert mode is None or mode in self.supported_trigger_modes
def set_acquisition_number(self, value):
assert value == 1
acquisition_number = 1
def initialize_attributes(self):
pass
def initialize_hardware(self):
pass
def prepare_acquisition(self):
pass
def start_acquisition(self):
pass
......@@ -108,15 +116,21 @@ def test_base_mca_logic(mocker):
return [[3, 2, 1]]
def get_acquisition_statistics(self):
return ['some_stats']
return [stats]
# Create a test mca
config = {}
mca = TestMCA('incomplete', config)
assert mca.name == 'incomplete'
assert mca._config is config
assert mca.multiple_acquisition is False
# Run a single acquisition
sleep = mocker.patch('time.sleep')
assert mca.run_single_acquisition(3.) == ([[3, 2, 1]], ['some_stats'])
assert mca.run_single_acquisition(3.) == ([[3, 2, 1]], [stats])
sleep.assert_called_once_with(3.)
# Run an external acquisition
sleep = mocker.patch('time.sleep')
assert mca.run_external_acquisition() == ([[3, 2, 1]], [stats])
sleep.assert_called_once_with(.1)
......@@ -2,11 +2,12 @@
import pytest
from bliss.controllers.mca import Brand, DetectorType, Stats, PresetMode
from bliss.controllers.mca import Brand, DetectorType, Stats
from bliss.controllers.mca import PresetMode, TriggerMode
def test_get_mercury_from_config(beacon, mocker):
# Mocking
@pytest.fixture
def mercury(beacon, mocker):
beacon.reload()
m = mocker.patch('zerorpc.Client')
client = m.return_value
......@@ -21,21 +22,31 @@ def test_get_mercury_from_config(beacon, mocker):
# Emulate running behavior
def mock_not_running():
client.is_running.return_value = False
client.mock_not_running = mock_not_running
# Instantiating the mercury
mercury = beacon.get('mercury-test')
mercury = beacon.get('mercury1')
assert mercury._proxy is client
m.assert_called_once_with('tcp://welisa.esrf.fr:8000')
yield mercury
def test_mercury_instanciation(mercury):
client = mercury._proxy
client.init.assert_called_once_with(
'C:\\\\blissadm\\\\mercury', 'mercury_src.ini')
assert mercury.current_configuration == 'mercury_src.ini'
assert mercury.configured
# Infos
def test_mercury_infos(mercury):
assert mercury.detector_brand == Brand.XIA
assert mercury.detector_type == DetectorType.MERCURY
assert mercury.element_count == 1
# Configuration
def test_mercury_configuration(mercury):
client = mercury._proxy
assert mercury.available_configurations == ['default.ini']
client.get_config_files.assert_called_once_with(
'C:\\\\blissadm\\\\mercury')
......@@ -43,26 +54,76 @@ def test_get_mercury_from_config(beacon, mocker):
client.get_config.assert_called_once_with(
'C:\\\\blissadm\\\\mercury', 'mercury_src.ini')
# PresetMode
def test_mercury_preset_mode(mercury):
client = mercury._proxy
# First test
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()
# Error tests
with pytest.raises(ValueError):
mercury.set_preset_mode(3)
with pytest.raises(TypeError):
mercury.set_preset_mode(PresetMode.NONE, 1)
with pytest.raises(TypeError):
mercury.set_preset_mode(PresetMode.REALTIME, None)
# Acquisition
def test_mercury_trigger_mode(mercury):
client = mercury._proxy
# First test
mercury.set_trigger_mode(None)
client.set_acquisition_value.assert_called_once_with('gate_ignore', 1)
client.apply_acquisition_values.assert_called_once_with()
# Second test
client.set_acquisition_value.reset_mock()
client.apply_acquisition_values.reset_mock()
mercury.set_trigger_mode(TriggerMode.GATE)
client.set_acquisition_value.assert_called_once_with('gate_ignore', 0)
client.apply_acquisition_values.assert_called_once_with()
# Error tests
with pytest.raises(ValueError):
mercury.set_trigger_mode(3)
def test_mercury_acquisition_number(mercury):
client = mercury._proxy
# Test setter
mercury.set_acquisition_number(1)
client.set_acquisition_value.assert_called_once_with('mapping_mode', 0)
client.apply_acquisition_values.assert_called_once_with()
# Test getter
client.get_acquisition_value.return_value = 0.
mercury.acquisition_number == 1
client.get_acquisition_value.assert_called_once_with('mapping_mode', 0)
# Error tests
with pytest.raises(ValueError):
mercury.set_acquisition_number(0)
def test_mercury_acquisition(mercury, mocker):
client = mercury._proxy
sleep = mocker.patch('time.sleep')
sleep.side_effect = lambda x: mock_not_running()
sleep.side_effect = lambda x: client.mock_not_running()
client.get_spectrums.return_value = {0: [3, 2, 1]}
client.get_statistics.return_value = {0: range(7)}
stats = Stats(*range(7))
client.get_statistics.return_value = {0: range(9)}
stats = Stats(*range(9))
assert mercury.run_single_acquisition(3.) == ([[3, 2, 1]], [stats])
sleep.assert_called_once_with(0.1)
# Load configuration
def test_mercury_configuration_error(mercury):
client = mercury._proxy
client.init.side_effect = IOError('File not found!')
with pytest.raises(IOError):
mercury.load_configuration('i-dont-exist')
......@@ -70,22 +131,24 @@ def test_get_mercury_from_config(beacon, mocker):
assert mercury.current_configuration is None
assert mercury.current_configuration_values is None
# Finalize
def test_mercury_finalization(mercury):
client = mercury._proxy
mercury.finalize()
client.close.assert_called_once_with()
def test_get_mercury_from_wrong_config(beacon, mocker):
def test_mercury_from_wrong_beacon_config(beacon, mocker):
# ZeroRPC error
beacon.reload()
m = mocker.patch('zerorpc.Client')
m.side_effect = IOError('Cannot connect!')
with pytest.raises(IOError):
beacon.get('mercury-test')
beacon.get('mercury1')
# Handel error
m = mocker.patch('zerorpc.Client')
client = m.return_value
client.init.side_effect = IOError('File not found!')
with pytest.raises(IOError):
beacon.get('mercury-test')
beacon.get('mercury1')
name: mercury-test
name: mercury1
module: xia
class: Mercury
plugin: bliss
......
name: opiom1
plugin: bliss
module: opiom
class: Opiom
serial:
url: rfc2217://ldp2.esrf.fr:28000
program: default
from bliss import setup_globals
def run_gate(a=1., b=1., nb=100):
o = setup_globals.opiom1
o.comm_ack('CNT 1 RESET')
args = int(a * 1000 * 2000), int(b * 1000 * 2000), int(nb)
o.comm_ack('CNT 1 CLK2 PULSE {:d} {:d} {:d}'.format(*args))
o.comm_ack('CNT 1 START')
class: Session
name: freddy
default: True
setup-file: ./mercury.py
config-objects:
- mercury1
- opiom1
\ No newline at end of file
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