test_nxw_scangroup.py 3.03 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
43
44
45
46
47
48
49


@nxw_test_utils.writer_stdout_on_exception
def _test_nxw_scangroup(session=None, tmpdir=None, writer=None, **kwargs):
    session.scan_saving.add("technique", "none")
    npoints = 10
    detector1 = session.env_dict["diode3"]
    detector2 = session.env_dict["diode4"]
    motor = session.env_dict["robx"]

    seq = Sequence()
50
51
    seq.add_custom_channel(AcquisitionChannel("customdata", float, ()))
    seq.add_custom_channel(AcquisitionChannel("diode34", float, ()))
52
53
54
55
56
57
58
59
    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))
60
            gevent.joinall([g1, g2], raise_error=True)
61
62
63
64
65
66
67
            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()
68
69

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