test_hdf5.py 6.11 KB
Newer Older
1
2
3
4
# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
5
# Copyright (c) 2015-2019 Beamline Control Unit, ESRF
6
7
8
9
10
11
# Distributed under the GNU LGPLv3. See LICENSE for more info.

import pytest
from bliss import setup_globals
from bliss.common import scans
from bliss.common.axis import Axis
12
from bliss.common.utils import get_axes_positions_iter
13
14
15
16
from bliss.scanning.scan import Scan, ScanSaving
from bliss.scanning.chain import AcquisitionChain
from bliss.scanning.acquisition import timer
from bliss.scanning.acquisition.lima import LimaAcquisitionMaster
17
18
19
20
21
import numpy
import h5py
import time
import datetime
import os
22
23
24
import h5py


Sebastien Petitdemange's avatar
Sebastien Petitdemange committed
25
def h5dict(scan_file):
26
    with h5py.File(scan_file, "r") as f:
Sebastien Petitdemange's avatar
Sebastien Petitdemange committed
27
        items = []
28
        f.visititems(lambda *args: items.append(args))
Sebastien Petitdemange's avatar
Sebastien Petitdemange committed
29
30
31
        return {
            name: {key: val for key, val in obj.attrs.items()} for name, obj in items
        }
32

33

34
def test_hdf5_metadata(beacon, session):
35

36
37
38
    all_motors = dict(
        [
            (name, pos)
Geoffrey Mant's avatar
Geoffrey Mant committed
39
            for name, pos, _, _ in get_axes_positions_iter(on_error="ERR")
40
41
42
            if pos != "ERR"
        ]
    )
43

44
45
46
47
48
49
    roby = beacon.get("roby")
    diode = beacon.get("diode")

    s = scans.ascan(roby, 0, 10, 10, 0.01, diode, save=True, return_scan=True)
    assert s is setup_globals.SCANS[-1]

50
51
52
    iso_start_time = datetime.datetime.fromtimestamp(
        s.scan_info["start_timestamp"]
    ).isoformat()
53

54
55
    with h5py.File(s.writer.filename, "r") as f:
        dataset = f[s.node.name]
Vincent Michel's avatar
Vincent Michel committed
56
        assert dataset["title"].value == "ascan roby 0 10 10 0.01"
57
        assert dataset["start_time"].value.startswith(iso_start_time)
58
        assert dataset["measurement"]
59
        assert dataset["instrument"]
V. Armando Solé's avatar
V. Armando Solé committed
60
        for name, pos in dataset["instrument/positioners"].items():
61
62
            assert all_motors.pop(name) == pos.value
        assert len(all_motors) == 0
63
64


65
def test_hdf5_file_items(beacon, session):
66

67
68
69
70
    roby = beacon.get("roby")
    diode = beacon.get("diode")
    simu1 = beacon.get("simu1")

71
72
73
74
75
76
77
78
79
80
81
    s = scans.ascan(
        roby,
        0,
        5,
        5,
        0.001,
        diode,
        simu1.counters.spectrum_det0,
        save=True,
        return_scan=True,
    )
82

Sebastien Petitdemange's avatar
Sebastien Petitdemange committed
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
    scan_dict = h5dict(s.writer.filename)
    scan_name = s.node.name
    expected_dict = {
        f"{scan_name}": {"NX_class": "NXentry"},
        f"{scan_name}/instrument": {"NX_class": "NXinstrument"},
        f"{scan_name}/measurement": {"NX_class": "NXcollection"},
        f"{scan_name}/measurement/axis:roby": {},
        f"{scan_name}/measurement/diode:diode": {},
        f"{scan_name}/measurement/simu1:spectrum_det0": {},
        f"{scan_name}/measurement/timer:elapsed_time": {},
        f"{scan_name}/start_time": {},
        f"{scan_name}/title": {},
    }
    for key, val in expected_dict.items():
        assert key in scan_dict
        assert val.items() <= scan_dict[key].items()
Vincent Michel's avatar
Vincent Michel committed
99
100
101
102
103
104


def test_hdf5_values(beacon, session):
    roby = beacon.get("roby")
    diode = beacon.get("diode")
    s = scans.ascan(roby, 0, 10, 3, 0.01, diode, save=True, return_scan=True)
105
    scan_file = s.writer.filename
Vincent Michel's avatar
Vincent Michel committed
106
107
    data = s.get_data()["diode"]
    f = h5py.File(scan_file)
108
    dataset = f[s.node.name]["measurement"]["diode:diode"]
Vincent Michel's avatar
Vincent Michel committed
109
    assert list(dataset) == list(data)
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139


def test_subscan_in_hdf5(beacon, lima_simulator, dummy_acq_master, dummy_acq_device):
    chain = AcquisitionChain()
    master1 = timer.SoftwareTimerMaster(0.1, npoints=2, name="timer1")
    dummy1 = dummy_acq_device.get(None, "dummy1", npoints=1)
    master2 = timer.SoftwareTimerMaster(0.001, npoints=50, name="timer2")
    lima_sim = beacon.get("lima_simulator")
    lima_master = LimaAcquisitionMaster(lima_sim, acq_nb_frames=1, acq_expo_time=0.001)
    dummy2 = dummy_acq_device.get(None, "dummy2", npoints=1)
    chain.add(lima_master, dummy2)
    chain.add(master2, lima_master)
    chain.add(master1, dummy1)
    master1.terminator = False

    scan_saving = ScanSaving("test")
    scan = Scan(chain, "test", scan_saving=scan_saving)
    scan.run()

    scan_file = scan.writer.filename
    f = h5py.File(scan_file)

    scan_number, scan_name = scan.node.name.split("_", maxsplit=1)
    scan_index = 1
    subscan_name = f"{scan_number}{'.%d_' % scan_index}{scan_name}"

    assert f[scan.node.name]["measurement"]["timer1:elapsed_time"]
    assert f[subscan_name]["measurement"]["timer2:elapsed_time"]
    assert f[scan.node.name]["measurement"]["dummy1:nb"]
    assert f[subscan_name]["measurement"]["dummy2:nb"]
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202


def test_image_reference_in_hdf5(alias_session, scan_tmpdir):

    env_dict, session = alias_session

    # put scan file in a tmp directory
    env_dict["SCAN_SAVING"].base_path = str(scan_tmpdir)

    s = scans.ascan(env_dict["robyy"], 0, 1, 3, .1, env_dict["lima_simulator"])

    f = h5py.File(s.writer.filename)

    refs = numpy.array(f["1_ascan/measurement/lima_simulator:image"])

    assert numpy.array_equal(
        refs,
        numpy.array(
            [
                [
                    "lima_simulator_0000.edf",
                    "EDF",
                    "0",
                    "",
                    "scan0001/lima_simulator_0000.edf",
                ],
                [
                    "lima_simulator_0001.edf",
                    "EDF",
                    "0",
                    "",
                    "scan0001/lima_simulator_0001.edf",
                ],
                [
                    "lima_simulator_0002.edf",
                    "EDF",
                    "0",
                    "",
                    "scan0001/lima_simulator_0002.edf",
                ],
            ],
            dtype=object,
        ),
    )


def test_lima_instrument_entry(alias_session, scan_tmpdir):

    env_dict, session = alias_session

    # put scan file in a tmp directory
    env_dict["SCAN_SAVING"].base_path = str(scan_tmpdir)

    s = scans.ascan(env_dict["robyy"], 0, 1, 3, .1, env_dict["lima_simulator"])

    f = h5py.File(s.writer.filename)

    assert (
        "saving_frame_per_file"
        in f["1_ascan/instrument/lima_simulator/lima_parameters"]
    )
    assert "acq_mode" in f["1_ascan/instrument/lima_simulator/lima_parameters"]
    assert "height" in f["1_ascan/instrument/lima_simulator/roi_counters/r1"]