Coverage for larch/io/rixs_esrf_id26.py: 0%
63 statements
« prev ^ index » next coverage.py v7.6.0, created at 2024-10-16 21:04 +0000
« prev ^ index » next coverage.py v7.6.0, created at 2024-10-16 21:04 +0000
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
4"""
5RIXS data reader for beamline ID26 @ ESRF
6=========================================
8RIXS stands for Resonant Inelastic X-ray Scattering
10**NOTE**: The current implementation is for RIXS files collected at ID26 in
11Spec format, that is, before 2019. Currently, ESRF/BLISS format is used at
12ID26. Updating this function to the new format should be straightforward by
13providing an example of data, but is not yet done/supported.
15"""
16import os
17import time
18import numpy as np
19from larch.io.specfile_reader import DataSourceSpecH5, _str2rng, _mot2array
20from silx.io.dictdump import dicttoh5
21from larch.utils.logging import getLogger
23_logger = getLogger("io_rixs_id26")
24_logger.setLevel("INFO")
27def _parse_header(fname):
28 """Get parsed header
30 Return
31 ------
32 header : dict
33 """
34 raise NotImplementedError
37def get_rixs_id26(
38 fname,
39 scans=None,
40 sample_name=None,
41 mode="rixs",
42 mot_axis2="Spec.Energy",
43 counter_signal="zap_det_dtc",
44 counter_mon="arr_I02sum",
45 interp_ene_in=True,
46 out_dir=None,
47 save_rixs=False,
48):
49 """Build RIXS map as X,Y,Z 1D arrays
51 Parameters
52 ----------
54 fname : str
55 path to the (Spec) data file (TODO: implement BLISS/HDF5)
57 scans : str or list of strings or list of ints, optional [None -> all scans in the file]
58 list of scans to load (the string is parsed by larch.io.specfile_reader._str2rng)
60 sample_name : str, optional ['UNKNOWN_SAMPLE']
61 name of the sample measured
63 mode : str, optional ['rixs']
64 RIXS acqusition mode (affects 'mot_axis2')
65 - 'rixs' -> incoming energy scans
66 - 'rixs_et' -> emitted energy scans
68 mot_axis2 : str ['Spec.Energy']
69 name of the counter to use as second axis
71 counter_signal : str ['zap_det_dtc']
72 name of the counter to use as signal
74 counter_mon : str ['arr_I02sum']
75 name of the counter to use as incoming beam monitor
77 interp_ene_in: bool
78 perform interpolation ene_in to the energy step of ene_out [True]
80 out_dir : str, optional
81 path to save the data [None -> data_dir]
83 save_rixs : bool
84 if True -> save outdict to disk (in 'out_dir')
86 Returns
87 -------
89 outdict : dict
90 {
91 '_x': array, energy in
92 '_y': array, energy out
93 '_z': array, signal
94 'mode': str
95 'scans': list, scans
96 'writer_name': str,
97 'writer_version': str,
98 'writer_timestamp': str,
99 'counter_signal': str, counter_signal,
100 'counter_mon': str, counter_mon,
101 'mon_axis2': str, mot_axis2,
102 'sample_name': str, sample_name,
103 'ene_unit': "eV",
104 'rixs_header': None,
105 'data_dir': str, data_dir,
106 'out_dir': str, out_dir,
107 'fname_ine': str, full path raw data
108 'fname_out': str, full path
109 }
110 """
111 _writer = "get_rixs_id26"
112 _writer_version = "1.5.2" #: used for reading back in RixsData.load_from_h5()
113 _writer_timestamp = "{0:04d}-{1:02d}-{2:02d}_{3:02d}{4:02d}".format(
114 *time.localtime()
115 )
116 if sample_name is None:
117 sample_name = "SAMPLE_UNKNOWN"
119 data_dir = os.path.join(os.sep, *fname.split(os.sep)[1:-1])
120 _logger.debug(f"data_dir: {data_dir}")
121 if out_dir is None:
122 out_dir = data_dir
123 ds = DataSourceSpecH5(fname)
125 if isinstance(scans, str):
126 scans = _str2rng(scans)
127 assert isinstance(scans, list), "scans should be a list"
129 mode = mode.lower()
130 assert mode in ("rixs", "rixs_et"), "RIXS mode not valid"
132 _counter = 0
133 for scan in scans:
134 try:
135 ds.set_scan(scan)
136 xscan, sig, lab, attrs = ds.get_curve(counter_signal, mon=counter_mon)
137 except Exception:
138 _logger.error(f"cannot load scan {scan}!")
139 continue
140 # keV -> eV
141 escan = xscan * 1000
142 estep = ds.get_motor_position(mot_axis2) * 1000
143 if mode == "rixs":
144 x = escan
145 y = _mot2array(estep, escan)
146 if mode == "rixs_et":
147 x = _mot2array(estep, escan)
148 y = escan
149 if _counter == 0:
150 xcol = x
151 ycol = y
152 zcol = sig
153 else:
154 xcol = np.append(xcol, x)
155 ycol = np.append(ycol, y)
156 zcol = np.append(zcol, sig)
157 _counter += 1
158 _logger.info(f"Loaded scan {scan}: {estep:.1f} eV")
160 outdict = {
161 "_x": xcol,
162 "_y": ycol,
163 "_z": zcol,
164 "mode": mode,
165 "scans": scans,
166 "writer_name": _writer,
167 "writer_version": _writer_version,
168 "writer_timestamp": _writer_timestamp,
169 "counter_signal": counter_signal,
170 "counter_mon": counter_mon,
171 "mon_axis2": mot_axis2,
172 "sample_name": sample_name,
173 "ene_unit": "eV",
174 "rixs_header": None,
175 "data_dir": data_dir,
176 "out_dir": out_dir,
177 }
179 if save_rixs:
180 fnstr = fname.split("/")[-1].split(".")[0]
181 fnout = "{0}_rixs.h5".format(fnstr)
182 fname_out = os.path.join(out_dir, fnout)
183 dicttoh5(outdict, fname_out)
184 outdict["fname_in"] = fname
185 outdict["fname_out"] = fname_out
186 _logger.info("RIXS saved to {0}".format(fnout))
188 return outdict
191if __name__ == "__main__":
192 pass