test_nxw_scangroup.py 3.11 KB
Newer Older
1
2
3
4
# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
Benoit Formet's avatar
Benoit Formet committed
5
# Copyright (c) 2015-2020 Beamline Control Unit, ESRF
6
7
8
9
10
11
12
# Distributed under the GNU LGPLv3. See LICENSE for more info.

import gevent
import numpy
from bliss.common import scans
from bliss.scanning.group import Sequence, Group
from bliss.scanning.chain import AcquisitionChannel
13
14
from tests.nexus_writer.helpers import nxw_test_utils
from tests.nexus_writer.helpers import nxw_test_data
15
16
17
18
19
20
21


def test_nxw_scangroup(nexus_writer_config):
    _test_nxw_scangroup(**nexus_writer_config)


def test_nxw_scangroup_alt(nexus_writer_config_alt):
22
    _test_nxw_scangroup(**nexus_writer_config_alt)
23
24
25


def test_nxw_scangroup_nopolicy(nexus_writer_config_nopolicy):
26
    _test_nxw_scangroup(**nexus_writer_config_nopolicy)
27
28
29


def test_nxw_scangroup_base(nexus_writer_base):
30
    _test_nxw_scangroup(**nexus_writer_base)
31
32
33


def test_nxw_scangroup_base_alt(nexus_writer_base_alt):
34
    _test_nxw_scangroup(**nexus_writer_base_alt)
35
36
37


def test_nxw_scangroup_base_nopolicy(nexus_writer_base_nopolicy):
38
    _test_nxw_scangroup(**nexus_writer_base_nopolicy)
39
40
41
42


@nxw_test_utils.writer_stdout_on_exception
def _test_nxw_scangroup(session=None, tmpdir=None, writer=None, **kwargs):
43
44
45
46
    try:
        session.scan_saving.dataset.all.definition = "none"
    except AttributeError:
        pass  # data policy disabled
47
48
49
50
51
52
    npoints = 10
    detector1 = session.env_dict["diode3"]
    detector2 = session.env_dict["diode4"]
    motor = session.env_dict["robx"]

    seq = Sequence()
53
54
    seq.add_custom_channel(AcquisitionChannel("customdata", float, ()))
    seq.add_custom_channel(AcquisitionChannel("diode34", float, ()))
55
56
57
58
59
60
61
62
    with gevent.Timeout(30):
        with seq.sequence_context() as scan_seq:
            scan1 = scans.loopscan(npoints, .1, detector1, run=False)
            scan2 = scans.ascan(motor, 0, 1, npoints - 1, .1, detector2, run=False)
            g1 = nxw_test_utils.run_scan(scan1, runasync=True)
            seq.custom_channels["customdata"].emit(numpy.arange(npoints // 2))
            g2 = nxw_test_utils.run_scan(scan2, runasync=True)
            seq.custom_channels["customdata"].emit(numpy.arange(npoints // 2, npoints))
63
            gevent.joinall([g1, g2], raise_error=True)
64
65
66
67
68
69
70
            diode34 = scan1.get_data()["diode3"] + scan2.get_data()["diode4"]
            seq.custom_channels["diode34"].emit(diode34)
            scan_seq.add(scan1)
            scan_seq.add(scan2)
        scan_grp = Group(scan1, scan2)
        scan_seq.wait_all_subscans()
        scan_grp.wait_all_subscans()
71
72

    nxw_test_utils.wait_scan_data_finished(
73
        [scan1, scan2, scan_seq.sequence.scan, scan_grp.scan], writer=writer
74
75
    )
    nxw_test_data.assert_scan_data(
76
77
78
79
80
        scan1,
        scan_shape=(npoints,),
        positioners=[["elapsed_time", "epoch"]],
        detectors=["diode3"],
        **kwargs
81
82
    )
    nxw_test_data.assert_scan_data(
83
84
85
86
        scan2,
        scan_shape=(npoints,),
        positioners=[["robx"]],
        detectors=["diode4"],
87
        softtimer="detector",
88
        **kwargs
89
90
91
    )
    nxw_test_data.assert_scangroup_data(scan_seq.sequence, **kwargs)
    nxw_test_data.assert_scangroup_data(scan_grp, **kwargs)