EXtra-data/extra_data/tests/conftest.py at 19f1efbf4feb4ee5903a93d584def42f1babb9de · European-XFEL/EXtra-data · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
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
import os.path as osp

import h5py
import numpy as np
import pytest
from tempfile import TemporaryDirectory

from . import make_examples


@pytest.fixture(scope='session', params=['0.5', '1.0'])
def format_version(request):
    return request.param


@pytest.fixture(scope='module')
def mock_agipd_data(format_version):
    # This one uses the older index format
    # (first/last/status instead of first/count)
    with TemporaryDirectory() as td:
        path = osp.join(td, 'CORR-R9999-AGIPD07-S00000.h5')
        make_examples.make_agipd_example_file(path, format_version=format_version)
        yield path


@pytest.fixture(scope='module')
def mock_lpd_data(format_version):
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R9999-LPD00-S00000.h5')
        make_examples.make_lpd_file(path, format_version=format_version)
        yield path


@pytest.fixture(scope='module')
def mock_fxe_control_data(format_version):
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R0450-DA01-S00001.h5')
        make_examples.make_fxe_da_file(path, format_version=format_version)
        yield path


@pytest.fixture(scope='module')
def mock_sa3_control_data(format_version):
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R0450-DA01-S00001.h5')
        make_examples.make_sa3_da_file(path, format_version=format_version)
        yield path


@pytest.fixture(scope='module')
def mock_spb_control_data_badname(format_version):
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R0309-DA01-S00000.h5')
        make_examples.make_data_file_bad_device_name(path, format_version=format_version)
        yield path


@pytest.fixture(scope='session')
def mock_fxe_raw_run(format_version):
    with TemporaryDirectory() as td:
        make_examples.make_fxe_run(td, format_version=format_version)
        yield td


@pytest.fixture(scope='session')
def mock_lpd_parallelgain_run():
    with TemporaryDirectory() as td:
        make_examples.make_lpd_parallelgain_run(td, format_version='1.0')
        yield td


@pytest.fixture(scope='session')
def mock_spb_proc_run(format_version):
    with TemporaryDirectory() as td:
        make_examples.make_spb_run(td, raw=False, format_version=format_version)
        yield td


@pytest.fixture(scope='session')
def mock_reduced_spb_proc_run(format_version):
    """Varying number of frames stored from AGIPD"""
    rng = np.random.RandomState(123)  # Fix seed
    with TemporaryDirectory() as td:
        make_examples.make_reduced_spb_run(td, raw=False, rng=rng,
                                           format_version=format_version)
        yield td


@pytest.fixture(scope='session')
def mock_spb_raw_run(format_version):
    with TemporaryDirectory() as td:
        make_examples.make_spb_run(td, format_version=format_version)
        yield td


@pytest.fixture(scope='session')
def mock_jungfrau_run():
    with TemporaryDirectory() as td:
        make_examples.make_jungfrau_run(td)
        yield td


@pytest.fixture(scope='session')
def mock_scs_run():
    with TemporaryDirectory() as td:
        make_examples.make_scs_run(td)
        yield td


@pytest.fixture(scope='session')
def empty_h5_file():
    with TemporaryDirectory() as td:
        path = osp.join(td, 'empty.h5')
        with h5py.File(path, 'w'):
            pass

        yield path

@pytest.fixture(scope='session')
def mock_empty_file():
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R0450-DA01-S00002.h5')
        make_examples.make_sa3_da_file(path, ntrains=0)
        yield path


@pytest.fixture(scope='function')
def mock_empty_dataset_file(format_version):
    with TemporaryDirectory() as td:
        path = osp.join(td, 'RAW-R0999-DA10-S00001.h5')
        make_examples.make_fxe_da_file(path, format_version=format_version)

        with h5py.File(path, 'a') as f:
            f['INSTRUMENT/SA1_XTD2_XGM/DOOCS/MAIN:output/data/intensityTD'].resize(0, axis=0)
            f['INSTRUMENT/SA1_XTD2_XGM/DOOCS/MAIN:output/data/trainId'].resize(0, axis=0)
            f['CONTROL/SA1_XTD2_XGM/DOOCS/MAIN/pulseEnergy/photonFlux/value'].resize(0, axis=0)
            f['CONTROL/SA1_XTD2_XGM/DOOCS/MAIN/beamPosition/ixPos/value'].resize(0, axis=0)

        yield path