Coverage for larch/io/rixs_aps_gsecars.py: 0%
68 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 13-ID-E @ APS
6===========================================
8.. note: RIXS stands for Resonant Inelastic X-ray Scattering
10.. note: 13-ID-E is GSECARS-CAT
12"""
13import os
14import time
15import glob
16import numpy as np
18from silx.io.dictdump import dicttoh5
20from larch.utils.logging import getLogger
21_logger = getLogger('io_rixs_gsecars')
24def _parse_header(fname):
25 """Get parsed header
27 Return
28 ------
29 header : dict
30 {
31 'columns': list of strings,
32 'Analyzer.Energy': float,
33 }
34 """
35 with open(fname) as f:
36 lines = f.read().splitlines()
37 header_lines = [line[2:] for line in lines if line[0] == '#']
38 header = {}
39 for line in header_lines:
40 if 'Analyzer.Energy' in line:
41 ene_line = line.split()
42 break
43 else:
44 ene_line = ['Analyzer.Energy:', '0', '', '||', '', '13XRM:ANA:Energy.VAL'] #: expected line
45 header['Analyzer.energy'] = float(ene_line[1])
46 header['columns'] = header_lines[-1].split('\t')
47 return header
50def get_rixs_13ide(sample_name, scan_name, rixs_no='001', data_dir='.',
51 out_dir=None, counter_signal='ROI1', counter_norm=None, interp_ene_in=True, save_rixs=False):
52 """Build RIXS map as X,Y,Z 1D arrays
54 Parameters
55 ----------
56 sample_name : str
57 scan_name : str
58 rixs_no : str, optional
59 length 3 string, ['001']
60 data_dir : str, optional
61 path to the data ['.']
62 out_dir : str, optional
63 path to save the data [None -> data_dir]
64 counter_signal : str
65 name of the data column to use as signal
66 counter_norm : str
67 name of the data column to use as normaliztion
68 interp_ene_in: bool
69 perform interpolation ene_in to the energy step of ene_out [True]
70 save_rixs : bool
71 if True -> save outdict to disk (in 'out_dir')
73 Returns
74 -------
75 outdict : dict
76 {
77 '_x': array, energy in
78 '_y': array, energy out
79 '_z': array, signal
80 'writer_name': str,
81 'writer_version': str,
82 'writer_timestamp': str,
83 'filename_all' : list,
84 'filename_root': str,
85 'name_sample': str,
86 'name_scan': str,
87 'counter_all': str,
88 'counter_signal': str,
89 'counter_norm': str,
90 'ene_grid': float,
91 'ene_unit': str,
92 }
94 """
95 _writer = 'get_rixs_13ide'
96 _writer_version = "1.5.1" #: used for reading back in RixsData.load_from_h5()
97 _writer_timestamp = '{0:04d}-{1:02d}-{2:02d}_{3:02d}{4:02d}'.format(*time.localtime())
99 if out_dir is None:
100 out_dir = data_dir
101 fnstr = "{0}_{1}".format(scan_name, sample_name)
102 grepstr = "{0}*.{1}".format(fnstr, rixs_no)
103 fnames = glob.glob(os.path.join(data_dir, grepstr))
104 enes = np.sort(np.array([_parse_header(fname)['Analyzer.energy'] for fname in fnames]))
105 estep = round(np.average(enes[1:]-enes[:-1]), 2)
107 fname0 = fnames[0]
108 header = _parse_header(fname0)
109 cols = header['columns']
110 ix = cols.index('Energy') or 0
111 iz = cols.index(counter_signal)
112 i0 = cols.index(counter_norm)
114 if interp_ene_in:
115 dat = np.loadtxt(fname0)
116 x0 = dat[:, ix]
117 xnew = np.arange(x0.min(), x0.max()+estep, estep)
119 for ifn, fname in enumerate(fnames):
120 dat = np.loadtxt(fname)
121 x = dat[:, ix]
122 y = np.ones_like(x) * enes[ifn]
123 if counter_norm is not None:
124 z = dat[:, iz] / dat[:, i0]
125 else:
126 z = dat[:, iz]
127 if interp_ene_in:
128 y = np.ones_like(xnew) * enes[ifn]
129 z = np.interp(xnew, x, z)
130 x = xnew
131 if ifn == 0:
132 _xcol = x
133 _ycol = y
134 _zcol = z
135 else:
136 _xcol = np.append(x, _xcol)
137 _ycol = np.append(y, _ycol)
138 _zcol = np.append(z, _zcol)
139 _logger.info("Loaded scan {0}: {1} eV".format(ifn+1, enes[ifn]))
141 outdict = {
142 '_x': _xcol,
143 '_y': _ycol,
144 '_z': _zcol,
145 'writer_name': _writer,
146 'writer_version': _writer_version,
147 'writer_timestamp': _writer_timestamp,
148 'filename_root': fnstr,
149 'filename_all': fnames,
150 'counter_all': cols,
151 'counter_signal': counter_signal,
152 'counter_norm': counter_norm,
153 'sample_name': sample_name,
154 'scan_name': scan_name,
155 'ene_grid': estep,
156 'ene_unit': 'eV',
157 }
159 if save_rixs:
160 fnout = "{0}_rixs.h5".format(fnstr)
161 dicttoh5(outdict, os.path.join(out_dir, fnout))
162 _logger.info("RIXS saved to {0}".format(fnout))
164 return outdict
167if __name__ == '__main__':
168 pass