|
1 | 1 | import os |
2 | 2 |
|
3 | | -from hdfmap import NexusMap, NexusLoader, load_hdf, create_nexus_map |
4 | | -from hdfmap.eval_functions import dataset2data, dataset2str |
5 | | -import h5py |
6 | | -import numpy as np |
7 | | -import datetime |
| 3 | +from hdfmap import load_hdf, create_nexus_map |
8 | 4 |
|
9 | | -from mmg_toolbox.utils.misc_functions import DataHolder, shorten_string |
| 5 | +from mmg_toolbox.nexus.nexus_scan import NexusDataHolder, NexusScan |
10 | 6 | from mmg_toolbox.utils.file_functions import get_scan_number, replace_scan_number |
11 | | -from mmg_toolbox.beamline_metadata.hdfmap_generic import HdfMapMMGMetadata as Md |
12 | | -from mmg_toolbox.nexus.nexus_functions import get_dataset_value |
13 | | -from mmg_toolbox.nexus.instrument_model import NXInstrumentModel |
14 | | -from mmg_toolbox.xas.nxxas_loader import load_xas_scans, SpectraContainer |
15 | 7 |
|
16 | 8 |
|
17 | | -class NexusScan(NexusLoader): |
18 | | - """ |
19 | | - Light-weight NeXus file reader |
20 | | -
|
21 | | - Example: |
22 | | - scan = NexusScan('scan.nxs') |
23 | | - scan('scan_command') -> returns value |
24 | | -
|
25 | | - :param nxs_filename: path to nexus file |
26 | | - :param hdf_map: NexusMap object or None |
27 | | - """ |
28 | | - MAX_STR_LEN: int = 100 |
29 | | - |
30 | | - def __init__(self, nxs_filename: str, hdf_map: NexusMap | None = None): |
31 | | - super().__init__(nxs_filename, hdf_map) |
32 | | - |
33 | | - from mmg_toolbox.utils.fitting import ScanFitManager, poisson_errors |
34 | | - self.fit = ScanFitManager(self) |
35 | | - self._error_function = poisson_errors |
36 | | - from mmg_toolbox.plotting.scan_plot_manager import ScanPlotManager |
37 | | - self.plot = ScanPlotManager(self) |
38 | | - |
39 | | - def __repr__(self): |
40 | | - return f"NexusScan('{self.filename}')" |
41 | | - |
42 | | - def scan_number(self) -> int: |
43 | | - return get_scan_number(self.filename) |
44 | | - |
45 | | - def title(self) -> str: |
46 | | - return f"#{self.scan_number()}" |
47 | | - |
48 | | - def label(self) -> str: |
49 | | - return f"#{self.scan_number()}" |
50 | | - |
51 | | - def load_hdf(self) -> h5py.File: |
52 | | - """Load the Hdf file""" |
53 | | - return load_hdf(self.filename) |
54 | | - |
55 | | - def datasets(self, *args) -> list[h5py.Dataset]: |
56 | | - """Return HDF5 datasets from NeXus file (leaves file in open state)""" |
57 | | - with self.load_hdf() as hdf: |
58 | | - return [hdf[self.map.combined[name]] for name in args] |
59 | | - |
60 | | - def arrays(self, *args, units: str = '', default: np.ndarray = np.array([np.nan])) -> list[np.ndarray]: |
61 | | - """Return Numpy arrays""" |
62 | | - with self.load_hdf() as hdf: |
63 | | - return [ |
64 | | - get_dataset_value(self.map.combined[name], hdf, units=units, default=default) |
65 | | - for name in args |
66 | | - ] |
67 | | - |
68 | | - def values(self, *args, value_func=np.mean, |
69 | | - units: str = '', default: np.ndarray = np.array(np.nan)) -> list[np.floating]: |
70 | | - """Return float values""" |
71 | | - with self.load_hdf() as hdf: |
72 | | - return [ |
73 | | - value_func(get_dataset_value(self.map.combined[name], hdf, units=units, default=default)) |
74 | | - for name in args |
75 | | - ] |
76 | | - |
77 | | - def times(self, *args) -> list[datetime.datetime]: |
78 | | - """Return datetime object""" |
79 | | - with self.load_hdf() as hdf: |
80 | | - return [dataset2data(hdf[self.map.combined[name]]) for name in args] |
81 | | - |
82 | | - def strings(self, *args, units=False) -> list[str]: |
83 | | - """Return string value""" |
84 | | - with self.load_hdf() as hdf: |
85 | | - return [dataset2str(hdf[self.map.combined[name]], units=units) for name in args] |
86 | | - |
87 | | - def image(self, index: int | tuple | slice | None = None) -> np.ndarray: |
88 | | - """Return image or selection from default detector""" |
89 | | - with self.load_hdf() as hdf: |
90 | | - return self.map.get_image(hdf, index) |
91 | | - |
92 | | - def table(self, delimiter=', ', string_spec='', format_spec='f', default_decimals=8) -> str: |
93 | | - """Return data table""" |
94 | | - with self.load_hdf() as hdf: |
95 | | - return self.map.create_scannables_table(hdf, delimiter, string_spec, format_spec, default_decimals) |
96 | | - |
97 | | - #TODO: Remove this? |
98 | | - def get_plot_data(self, x_axis: str = 'axes0', y_axis: str = 'signal0') -> dict: |
99 | | - with self.load_hdf() as hdf: |
100 | | - cmd = self.map.eval(hdf, Md.cmd) |
101 | | - if len(cmd) > self.MAX_STR_LEN: |
102 | | - cmd = shorten_string(cmd) |
103 | | - xdata = self.map.eval(hdf, x_axis) |
104 | | - ydata = self.map.eval(hdf, y_axis) |
105 | | - yerror = self._error_function(ydata) |
106 | | - x_lab, y_lab = self.map.generate_ids(x_axis, y_axis, modify_missing=False) |
107 | | - return { |
108 | | - 'x': xdata, |
109 | | - 'y': ydata, |
110 | | - 'yerror': yerror, |
111 | | - 'xlabel': x_lab, |
112 | | - 'ylabel': y_lab, |
113 | | - 'title': f"#{self.scan_number()}\n{cmd}" |
114 | | - } |
115 | | - |
116 | | - def xas_scan(self) -> SpectraContainer: |
117 | | - """Load XAS Spectra""" |
118 | | - return load_xas_scans(self.filename)[0] |
119 | | - |
120 | | - def instrument_model(self) -> NXInstrumentModel: |
121 | | - """return instrument model""" |
122 | | - with self.load_hdf() as hdf: |
123 | | - return NXInstrumentModel(hdf) |
124 | | - |
125 | | - |
126 | | -class NexusDataHolder(DataHolder, NexusScan): |
| 9 | +def read_nexus_file(filename: str, flatten_scannables: bool = True) -> NexusDataHolder: |
127 | 10 | """ |
128 | | - Nexus data holder class |
129 | | - - Automatically reads scannable and metadata from file |
130 | | - - acts like the old .dat DataHolder class |
131 | | - - has additional functions to read data from NeXus file |
132 | | -
|
133 | | - Example: |
134 | | - scan = NexusDataHolder('12345.nxs') |
135 | | - scan.eta -> returns array |
136 | | - scan.metadata.metadata -> returns value |
137 | | - scan('signal') -> evaluate expression |
138 | | -
|
139 | | - :param filename: path to Nexus file |
140 | | - :param hdf_map: NexusMap object or None to generate |
141 | | - :param flatten_scannables: if True, flattens all scannable arrays to 1D |
| 11 | + Read Nexus file as DataHolder |
142 | 12 | """ |
143 | | - filename: str |
144 | | - map: NexusMap |
145 | | - metadata: DataHolder |
146 | | - |
147 | | - def __init__(self, filename: str | None, hdf_map: NexusMap | None = None, flatten_scannables: bool = True): |
148 | | - NexusScan.__init__(self, filename, hdf_map) |
149 | | - |
150 | | - with load_hdf(filename) as hdf: |
151 | | - metadata = self.map.get_metadata(hdf) |
152 | | - scannables = self.map.get_scannables(hdf, flatten=flatten_scannables) |
153 | | - DataHolder.__init__(self, **scannables) |
154 | | - self.metadata = DataHolder(**metadata) |
155 | | - |
156 | | - def __repr__(self): |
157 | | - return f"NexusDataHolder('{self.filename}')" |
| 13 | + return NexusDataHolder(filename, flatten_scannables=flatten_scannables) |
158 | 14 |
|
159 | 15 |
|
160 | | -def read_nexus_file(filename: str, flatten_scannables: bool = True) -> NexusDataHolder: |
| 16 | +def read_nexus_files(*filenames: str) -> list[NexusScan]: |
161 | 17 | """ |
162 | | - Read Nexus file as DataHolder |
| 18 | + Read Nexus files as NexusScan |
163 | 19 | """ |
164 | | - return NexusDataHolder(filename, flatten_scannables=flatten_scannables) |
| 20 | + hdf_map = create_nexus_map(filenames[0]) |
| 21 | + return [NexusScan(f, hdf_map) for f in filenames] |
165 | 22 |
|
166 | 23 |
|
167 | 24 | def find_matching_scans(filename: str, match_field: str = 'scan_command', |
|
0 commit comments