Commit c5d8a6d4 authored by Wout De Nolf's avatar Wout De Nolf Committed by Matias Guijarro

[writer] add test with lima master

parent c9555308
Pipeline #21213 canceled with stages
in 1 minute and 9 seconds
This diff is collapsed.
...@@ -36,14 +36,14 @@ def test_nxw_amesh_base_nopolicy(nexus_writer_base_nopolicy): ...@@ -36,14 +36,14 @@ def test_nxw_amesh_base_nopolicy(nexus_writer_base_nopolicy):
@nxw_test_utils.writer_stdout_on_exception @nxw_test_utils.writer_stdout_on_exception
def _test_nxw_amesh(session=None, tmpdir=None, writer=None, **kwargs): def _test_nxw_amesh(session=None, tmpdir=None, writer=None, **kwargs):
masters = "robx", "roby" positioners = [["robx"], ["roby"]]
scan_shape = 4, 5 scan_shape = 4, 5
scan = scans.amesh( scan = scans.amesh(
session.env_dict[masters[0]], session.env_dict[positioners[0][0]],
0, 0,
1, 1,
scan_shape[0] - 1, scan_shape[0] - 1,
session.env_dict[masters[1]], session.env_dict[positioners[1][0]],
0, 0,
1, 1,
scan_shape[1] - 1, scan_shape[1] - 1,
...@@ -53,5 +53,5 @@ def _test_nxw_amesh(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -53,5 +53,5 @@ def _test_nxw_amesh(session=None, tmpdir=None, writer=None, **kwargs):
nxw_test_utils.run_scan(scan) nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan, masters=masters, scan_shape=scan_shape, **kwargs scan, positioners=positioners, scan_shape=scan_shape, **kwargs
) )
...@@ -36,13 +36,13 @@ def test_nxw_ascan_base_nopolicy(nexus_writer_base_nopolicy): ...@@ -36,13 +36,13 @@ def test_nxw_ascan_base_nopolicy(nexus_writer_base_nopolicy):
@nxw_test_utils.writer_stdout_on_exception @nxw_test_utils.writer_stdout_on_exception
def _test_nxw_ascan(session=None, tmpdir=None, writer=None, **kwargs): def _test_nxw_ascan(session=None, tmpdir=None, writer=None, **kwargs):
masters = ("robx",) positioners = [["robx"]]
scan_shape = (10,) scan_shape = (10,)
scan = scans.ascan( scan = scans.ascan(
session.env_dict[masters[0]], 0, 1, scan_shape[0] - 1, .1, run=False session.env_dict[positioners[0][0]], 0, 1, scan_shape[0] - 1, .1, run=False
) )
nxw_test_utils.run_scan(scan) nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan, masters=masters, scan_shape=scan_shape, **kwargs scan, positioners=positioners, scan_shape=scan_shape, **kwargs
) )
...@@ -39,4 +39,6 @@ def _test_nxw_ct(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -39,4 +39,6 @@ def _test_nxw_ct(session=None, tmpdir=None, writer=None, **kwargs):
scan = scans.ct(.1, run=False, save=True) scan = scans.ct(.1, run=False, save=True)
nxw_test_utils.run_scan(scan) nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data(scan, **kwargs) nxw_test_data.assert_scan_data(
scan, positioners=[["elapsed_time", "epoch"]], **kwargs
)
# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
# Copyright (c) 2015-2019 Beamline Control Unit, ESRF
# Distributed under the GNU LGPLv3. See LICENSE for more info.
import nxw_test_utils
import nxw_test_data
def test_nxw_limact(nexus_writer_config):
_test_nxw_limact(**nexus_writer_config)
def test_nxw_limact_alt(nexus_writer_config_alt):
_test_nxw_limact(**nexus_writer_config_alt)
def test_nxw_limact_nopolicy(nexus_writer_config_nopolicy):
_test_nxw_limact(**nexus_writer_config_nopolicy)
def test_nxw_limact_base(nexus_writer_base):
_test_nxw_limact(**nexus_writer_base)
def test_nxw_limact_base_alt(nexus_writer_base_alt):
_test_nxw_limact(**nexus_writer_base_alt)
def test_nxw_limact_base_nopolicy(nexus_writer_base_nopolicy):
_test_nxw_limact(**nexus_writer_base_nopolicy)
def _test_nxw_limact(session=None, tmpdir=None, writer=None, **kwargs):
lima = session.env_dict["lima_simulator"]
scan = session.env_dict["limact"](lima, 0.1)
nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
if kwargs["config"]:
detectors = ["^lima_simulator$"]
else:
detectors = ["^lima_simulator_image$"]
nxw_test_data.assert_scan_data(
scan,
scan_shape=(1,),
positioners=[[]],
detectors=detectors,
hastimer=False,
**kwargs
)
...@@ -40,4 +40,6 @@ def _test_nxw_loopscan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -40,4 +40,6 @@ def _test_nxw_loopscan(session=None, tmpdir=None, writer=None, **kwargs):
scan = scans.loopscan(scan_shape[0], .1, run=False) scan = scans.loopscan(scan_shape[0], .1, run=False)
nxw_test_utils.run_scan(scan) nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data(scan, scan_shape=scan_shape, **kwargs) nxw_test_data.assert_scan_data(
scan, scan_shape=scan_shape, positioners=[["elapsed_time", "epoch"]], **kwargs
)
...@@ -81,8 +81,8 @@ def _test_aloopscan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -81,8 +81,8 @@ def _test_aloopscan(session=None, tmpdir=None, writer=None, **kwargs):
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan, scan,
subscan=1, subscan=1,
masters=(mot,),
scan_shape=(npoints1,), scan_shape=(npoints1,),
positioners=[[mot]],
detectors=detectors1, detectors=detectors1,
master_name="subscan1tmr", master_name="subscan1tmr",
**kwargs **kwargs
...@@ -91,6 +91,7 @@ def _test_aloopscan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -91,6 +91,7 @@ def _test_aloopscan(session=None, tmpdir=None, writer=None, **kwargs):
scan, scan,
subscan=2, subscan=2,
scan_shape=(npoints2,), scan_shape=(npoints2,),
positioners=[["elapsed_time", "epoch"]],
detectors=detectors2, detectors=detectors2,
master_name="subscan2tmr", master_name="subscan2tmr",
**kwargs **kwargs
...@@ -136,6 +137,7 @@ def _test_limatimescan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -136,6 +137,7 @@ def _test_limatimescan(session=None, tmpdir=None, writer=None, **kwargs):
scan, scan,
subscan=1, subscan=1,
scan_shape=(npoints1,), scan_shape=(npoints1,),
positioners=[["elapsed_time", "epoch"]],
detectors=detectors1, detectors=detectors1,
master_name="subscan1tmr", master_name="subscan1tmr",
**kwargs **kwargs
...@@ -144,6 +146,7 @@ def _test_limatimescan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -144,6 +146,7 @@ def _test_limatimescan(session=None, tmpdir=None, writer=None, **kwargs):
scan, scan,
subscan=2, subscan=2,
scan_shape=(npoints2,), scan_shape=(npoints2,),
positioners=[["elapsed_time", "epoch"]],
detectors=detectors2, detectors=detectors2,
master_name="subscan2tmr", master_name="subscan2tmr",
**kwargs **kwargs
......
...@@ -22,4 +22,6 @@ def _test_nxw_notes(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -22,4 +22,6 @@ def _test_nxw_notes(session=None, tmpdir=None, writer=None, **kwargs):
scan.add_comment(note) scan.add_comment(note)
nxw_test_utils.run_scan(scan) nxw_test_utils.run_scan(scan)
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data(scan, notes=notes, **kwargs) nxw_test_data.assert_scan_data(
scan, notes=notes, positioners=[["elapsed_time", "epoch"]], **kwargs
)
...@@ -17,7 +17,6 @@ def test_nxw_parallel(nexus_writer_config): ...@@ -17,7 +17,6 @@ def test_nxw_parallel(nexus_writer_config):
@nxw_test_utils.writer_stdout_on_exception @nxw_test_utils.writer_stdout_on_exception
def _test_nxw_parallel(session=None, tmpdir=None, writer=None, **kwargs): def _test_nxw_parallel(session=None, tmpdir=None, writer=None, **kwargs):
session.scan_saving.technique = "none"
detectors = ( detectors = (
"diode2alias", "diode2alias",
"diode3", "diode3",
...@@ -40,5 +39,9 @@ def _test_nxw_parallel(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -40,5 +39,9 @@ def _test_nxw_parallel(session=None, tmpdir=None, writer=None, **kwargs):
nxw_test_utils.wait_scan_data_finished(lst, writer=writer) nxw_test_utils.wait_scan_data_finished(lst, writer=writer)
for npoints, (scan, detector) in enumerate(zip(lst, detectors), 10): for npoints, (scan, detector) in enumerate(zip(lst, detectors), 10):
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan, scan_shape=(npoints,), detectors=[detector], **kwargs scan,
scan_shape=(npoints,),
positioners=[["elapsed_time", "epoch"]],
detectors=[detector],
**kwargs
) )
...@@ -58,7 +58,6 @@ def _test_nxw_readers( ...@@ -58,7 +58,6 @@ def _test_nxw_readers(
reset_dispatcher = False reset_dispatcher = False
try: try:
detector = "diode3" detector = "diode3"
session.scan_saving.technique = "none"
if mode == "a" and not enable_file_locking: if mode == "a" and not enable_file_locking:
# Readers will not crash the scan (not sure why) but corrupt the file # Readers will not crash the scan (not sure why) but corrupt the file
scan_shape = (100,) scan_shape = (100,)
...@@ -96,7 +95,11 @@ def _test_nxw_readers( ...@@ -96,7 +95,11 @@ def _test_nxw_readers(
nxw_test_utils.assert_scan_data_exists([scan]) nxw_test_utils.assert_scan_data_exists([scan])
nxw_test_utils.assert_scan_data_not_corrupt([scan]) nxw_test_utils.assert_scan_data_not_corrupt([scan])
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan, scan_shape=scan_shape, detectors=[detector], **kwargs scan,
scan_shape=scan_shape,
positioners=[["elapsed_time", "epoch"]],
detectors=[detector],
**kwargs
) )
finally: finally:
if reset_dispatcher: if reset_dispatcher:
......
...@@ -69,10 +69,18 @@ def _test_nxw_scangroup(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -69,10 +69,18 @@ def _test_nxw_scangroup(session=None, tmpdir=None, writer=None, **kwargs):
[scan1, scan2, scan_seq.sequence.scan, scan_grp.scan], writer=writer [scan1, scan2, scan_seq.sequence.scan, scan_grp.scan], writer=writer
) )
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan1, scan_shape=(npoints,), detectors=["diode3"], **kwargs scan1,
scan_shape=(npoints,),
positioners=[["elapsed_time", "epoch"]],
detectors=["diode3"],
**kwargs
) )
nxw_test_data.assert_scan_data( nxw_test_data.assert_scan_data(
scan2, scan_shape=(npoints,), detectors=["diode4"], masters=("robx",), **kwargs scan2,
scan_shape=(npoints,),
positioners=[["robx"]],
detectors=["diode4"],
**kwargs
) )
nxw_test_data.assert_scangroup_data(scan_seq.sequence, **kwargs) nxw_test_data.assert_scangroup_data(scan_seq.sequence, **kwargs)
nxw_test_data.assert_scangroup_data(scan_grp, **kwargs) nxw_test_data.assert_scangroup_data(scan_grp, **kwargs)
...@@ -93,7 +93,9 @@ def _test_nxw_timescan(session=None, tmpdir=None, writer=None, **kwargs): ...@@ -93,7 +93,9 @@ def _test_nxw_timescan(session=None, tmpdir=None, writer=None, **kwargs):
# Verify data # Verify data
nxw_test_utils.wait_scan_data_finished([scan], writer=writer) nxw_test_utils.wait_scan_data_finished([scan], writer=writer)
nxw_test_data.assert_scan_data(scan, scan_shape=(0,), **kwargs) nxw_test_data.assert_scan_data(
scan, scan_shape=(0,), positioners=[["elapsed_time", "epoch"]], **kwargs
)
# TODO: no proper cleanup by Bliss # TODO: no proper cleanup by Bliss
dispatcher.reset() dispatcher.reset()
...@@ -95,6 +95,13 @@ def limatimescan(lima, npoints1, expo1, detectors1, npoints2, expo2, detectors2) ...@@ -95,6 +95,13 @@ def limatimescan(lima, npoints1, expo1, detectors1, npoints2, expo2, detectors2)
return Scan(chain, "limatimescan", save=True) return Scan(chain, "limatimescan", save=True)
def limact(lima, expo):
"""Simple lima image
"""
chain = _lime_ct_chain(name, lima, expo)
return Scan(chain, "limact", save=True)
def _timer_chain(name, npoints, expo, detectors): def _timer_chain(name, npoints, expo, detectors):
chain = AcquisitionChain() chain = AcquisitionChain()
timer_master = SoftwareTimerMaster(expo, npoints=npoints, name=name + "tmr") timer_master = SoftwareTimerMaster(expo, npoints=npoints, name=name + "tmr")
...@@ -137,6 +144,21 @@ def _lima_chain(name, lima, npoints, expo, detectors): ...@@ -137,6 +144,21 @@ def _lima_chain(name, lima, npoints, expo, detectors):
return chain return chain
def _lime_ct_chain(name, lima, expo):
chain = AcquisitionChain()
scan_params = {"npoints": 1, "type": "loopscan"}
acq_params = {
"acq_nb_frames": 10,
"acq_expo_time": expo / 10,
"acq_mode": "SINGLE",
"acq_trigger_mode": "INTERNAL_TRIGGER",
"prepare_once": True,
"start_once": False,
}
_add_detectors(chain, None, [lima.image], scan_params, acq_params)
return chain
def _add_detectors(chain, master, detectors, scan_params, acq_params): def _add_detectors(chain, master, detectors, scan_params, acq_params):
builder = ChainBuilder(detectors) builder = ChainBuilder(detectors)
for top_node in builder.get_top_level_nodes(): for top_node in builder.get_top_level_nodes():
...@@ -149,4 +171,7 @@ def _add_detectors(chain, master, detectors, scan_params, acq_params): ...@@ -149,4 +171,7 @@ def _add_detectors(chain, master, detectors, scan_params, acq_params):
scan_params, acq_params scan_params, acq_params
) )
cnode.set_parameters(acq_params=node_acq_params, ctrl_params=None) cnode.set_parameters(acq_params=node_acq_params, ctrl_params=None)
chain.add(master, top_node) if master is None:
chain.add(top_node)
else:
chain.add(master, slave=top_node)
Markdown is supported
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