test_nxw_scangroup.py 2.96 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
13
14
15
16
17
18
19
20
21
# 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
import nxw_test_utils
import nxw_test_data


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
50
51


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