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

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3 

4""" 

5RIXS data reader for beamline 13-ID-E @ APS 

6=========================================== 

7 

8.. note: RIXS stands for Resonant Inelastic X-ray Scattering 

9 

10.. note: 13-ID-E is GSECARS-CAT 

11 

12""" 

13import os 

14import time 

15import glob 

16import numpy as np 

17 

18from silx.io.dictdump import dicttoh5 

19 

20from larch.utils.logging import getLogger 

21_logger = getLogger('io_rixs_gsecars') 

22 

23 

24def _parse_header(fname): 

25 """Get parsed header 

26 

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 

48 

49 

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 

53 

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') 

72 

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 } 

93 

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()) 

98 

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) 

106 

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) 

113 

114 if interp_ene_in: 

115 dat = np.loadtxt(fname0) 

116 x0 = dat[:, ix] 

117 xnew = np.arange(x0.min(), x0.max()+estep, estep) 

118 

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])) 

140 

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 } 

158 

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)) 

163 

164 return outdict 

165 

166 

167if __name__ == '__main__': 

168 pass