Skip to content

experiment

Experiment Folder Monitor

Experiment

Experiment class Monitors data folders for scan files and provies a convenient way to load scan data using only a scan number.

E.G. from mmg_toolbox import Experiment exp = Experiment('path/to/folder1', 'path/to/folder2', instrument='i06') print(scan) scan = exp.scan(-1) # latest scan scan_list = exp.scans(range(12345, 12355)) data = exp.join_scan_data(range(-100, 0), data_fields=['cmd', 'Ta']) # returns dict of arrays

Parameters:

Name Type Description Default
folder_paths str

file directories containing .nxs files

 ()
instrument str | None

instrument name for configuration.

 None
Source code in mmg_toolbox/utils/experiment.py 
 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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
class Experiment:
    """
    Experiment class
    Monitors data folders for scan files and provies a
    convenient way to load scan data using only a scan number.

    E.G.
        from mmg_toolbox import Experiment
        exp = Experiment('path/to/folder1', 'path/to/folder2', instrument='i06')
        print(scan)
        scan = exp.scan(-1)  # latest scan
        scan_list = exp.scans(*range(12345, 12355))
        data = exp.join_scan_data(*range(-100, 0), data_fields=['cmd', 'Ta'])  # returns dict of arrays

    :param folder_paths: file directories containing .nxs files
    :param instrument: instrument name for configuration.
    """

    def __init__(self, *folder_paths: str, instrument: str | None = None):
        self.folder_paths = [os.path.dirname(f) if os.path.isfile(f) else f for f in folder_paths]
        self.scan_list = {}
        self._scan_list_update = None
        self.instrument = instrument or get_beamline_from_directory(folder_paths[0], None)
        self.config = beamline_config(self.instrument)
        from ..plotting.exp_plot_manager import ExperimentPlotManager
        self.plot = ExperimentPlotManager(self)

    def __repr__(self):
        paths = ', '.join(f"'{p}'" for p in self.folder_paths)
        return f"Experiment({paths}, instrument={self.instrument})"

    def __str__(self):
        self._update_scan_list()
        scan_numbers = self._scan_numbers()
        lines = ['Instrument: ' + self.instrument]
        lines.extend(self.folder_paths)
        if scan_numbers:
            lines.extend([
                f"    Files: {len(scan_numbers)}",
                f"    Scans: {scan_numbers[0]}-{scan_numbers[-1]}",
            ])
        else:
            lines.extend(["  No NeXus files found."])
        return '\n'.join(lines)

    def __getitem__(self, item: int | slice) -> NexusScan | list[NexusScan]:
        if isinstance(item, slice):
            scan_numbers = self.all_scan_numbers()[item]
            return self.scans(*scan_numbers)
        else:
            scan_numbers = self.all_scan_numbers()[item]
            return self.scans(scan_numbers)[0]

    def __len__(self) -> int:
        return len(self.all_scan_numbers())

    def _update_scan_list(self):
        mod_times = [last_folder_update(folder) for folder in self.folder_paths]
        folders = [
            folder for folder, time in zip(self.folder_paths, mod_times)
            if self._scan_list_update is None or time > self._scan_list_update
        ]
        self._scan_list_update = max(mod_times)  # datetime.now()?
        self.scan_list.update(scan_number_mapping(*folders))

    def _scan_numbers(self) -> list[int]:
        self._update_scan_list()
        return list(self.scan_list.keys())

    def add_data_paths(self, *folder_paths: str):
        """Add additional paths"""
        new_paths = list(
            path for path in folder_paths
            if path not in self.folder_paths and os.path.isdir(path)
        )
        self.folder_paths = self.folder_paths + new_paths
        self._update_scan_list()

    def all_scans(self) -> dict[int, str]:
        self._update_scan_list()
        return self.scan_list.copy()

    def all_scan_numbers(self) -> list[int]:
        self._update_scan_list()
        return list(self.scan_list.keys())

    def all_scan_files(self) -> list[str]:
        self._update_scan_list()
        return list(self.scan_list.values())

    def get_scan_filename(self, scan_file: int | str = -1) -> str:
        """Return the full filename of a scan number"""
        if isinstance(scan_file, int) or scan_file.isdigit():
            scan_file = int(scan_file)
            if scan_file < 1:
                scan_numbers = self._scan_numbers()
                return self.scan_list[scan_numbers[scan_file]]
            self._update_scan_list()
            if scan_file in self.scan_list:
                return self.scan_list[scan_file]
            scan_numbers = self._scan_numbers()
            return self.scan_list[scan_numbers[scan_file]]

        if os.path.isfile(scan_file):
            return os.path.abspath(scan_file)
        raise FileNotFoundError(f"scan file {scan_file} not found")

    def scan(self, scan_file: int | str = -1) -> NexusDataHolder:
        """read Nexus file as NexusDataHolder"""
        return NexusDataHolder(self.get_scan_filename(scan_file), config=self.config)

    def scans(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None) -> list[NexusScan]:
        """Read Nexus files as lazy NexusScan. All files use the same HdfMap, based on the first scan"""
        filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
        if not filenames:
            filenames = list(self.all_scans().values())
        if filenames and hdf_map is None:
            hdf_map = hdfmap.create_nexus_map(filenames[0])
        return [NexusScan(file, hdf_map, config=self.config) for file in filenames]

    def find_scans(self, *scan_files: int | str,  hdf_map: hdfmap.NexusMap | None = None, first_only: bool = False,
                   **matches: str | float | tuple[float, float]) -> list[NexusScan]:
        """
        Find scans files with matching metadata

            matches = {
                'name1': 'scan', # matches if 'scan' in file['name1']
                'name2': value, # matches if file['name2'] ~= value
                'name3': (value, tol), # matches if abs(file['name3'] - value) < tol
            }
            match_files = find_scans(*filenames, **matches)

        :param scan_files: set of scan numbers or filenames, if not given will search all scans in folder.
        :param hdf_map: if given, uses this hdfmap rather than generating one.
        :param first_only: if true, returns on the first result
        :param matches: keyword arguments for matching parameters
        :returns: list of scan files that match all requirements
        """
        filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
        if not filenames:
            filenames = list(self.all_scans().values())
        if hdf_map is None:
            hdf_map = hdfmap.create_nexus_map(filenames[0])
        matches = find_scans(*filenames, hdf_map=hdf_map, first_only=first_only, **matches)
        return self.scans(*matches, hdf_map=hdf_map)

    def join_scan_data(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None,
                       data_fields: list[str] | None = None, default: np.ndarray = np.array([0.0])) -> dict[str, list]:
        """
        Join data from scans
        """
        scans = self.scans(*scan_files, hdf_map=hdf_map)
        data_fields = [self.config[C.scan_description]] if data_fields is None else data_fields
        data = {name: [] for name in data_fields}
        for scan in scans:
            with scan.load_hdf() as hdf:
                for name in data_fields:
                    data[name].append(scan.map.eval(hdf, name, default=default))
        return data

    def get_all_data(self, *fields: str, default: np.ndarray = np.array([0.0])) -> dict[str, list]:
        """
        Return dict of data for all files
        """
        return self.join_scan_data(data_fields=list(fields), default=default)

    def generate_mesh(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None,
                      axes: str | tuple[str, str] = 'axes', signal: str = 'axes',
                      values: str | None = None) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
        """
        Generate 2D mesh from scan or scans

            x, y, z = generate_mesh(*range(-10, 0), axes='eta', signal='roi2_sum', values='Tsample')
            # or, if scan 12345 is a 2D grid scan
            x, y, z = generate_mesh(12345, axes=('sx', 'sy'), signal='roi2_sum')

        :param scan_files: multiple files or single 2D grid scan
        :param hdf_map: hdfmap.NeXus map, or None to generate
        :param axes: x-axis name, or for grid scans the names of ('xaxis', 'yaxis')
        :param signal: signal name
        :param values: name of the value that changes between multiple files
        :returns: X, Y, IMAGE rank 2 arrays
        """
        scans = self.scans(*scan_files, hdf_map=hdf_map)
        if len(scan_files) == 1:
            # Single 2D Grid scan
            scan = scans[0]
            try:
                x_axis, y_axis = axes
            except ValueError:
                raise Exception("axes should be specified as axes=('axes0', 'axes1')")
            with scan.load_hdf() as hdf:
                x_data = scan.map.eval(hdf, x_axis, default=0.0).squeeze()
                y_data = scan.map.eval(hdf, y_axis, default=0.0).squeeze()
                z_data = scan.map.eval(hdf, signal, default=0.0).squeeze()
            if x_data.size != y_data.size:
                raise Exception(f"arrays '{x_axis}'[{x_data.size}] and '{y_axis}'[{y_data.size}] have different sizes")
            if x_data.ndim == 1 and y_data.ndim == 1:
                y_data, x_data = np.meshgrid(y_data, x_data)
            if z_data.shape != x_data.shape or z_data.shape != y_data.shape:
                raise Exception(
                    f"{repr(scan)} '{x_axis}', '{y_axis}' and '{signal}' shapes are not consistent: " +
                    f"x: {x_data.shape}, y: {y_data.shape}, im: {z_data.shape}"
                )
            return x_data, y_data, z_data
        else:
            x_data, y_data, z_data = [], [], []
            for n, scan in enumerate(scans):
                with scan.load_hdf() as hdf:
                    x = np.reshape(scan.map.eval(hdf, axes, default=0.0), -1)
                    y = np.reshape(scan.map.eval(hdf, signal, default=0.0), -1)
                    val = np.reshape(scan.map.eval(hdf, values, default=0.0), -1) if values is not None else np.array([n])
                if val.size == 1:
                    val = np.tile(val, x.size)
                if y.size != x.size or val.size != x.size:
                    raise Exception(
                        f"{repr(scan)} '{axes}', '{signal}' and '{values}' shapes are not consistent: " +
                        f"x, y, val = ({x.size}, {y.size}, {val.size})"
                    )
                x_data.append(x)
                y_data.append(val)
                z_data.append(y)
            # create a regular sized array
            min_len = min(len(x) for x in x_data)
            # array size [len(scan_files), min_len]
            x_array = np.array([x[:min_len] for x in x_data])
            y_array = np.array([y[:min_len] for y in y_data])
            z_array = np.array([z[:min_len] for z in z_data])
            return x_array, y_array, z_array

    def scan_str(self, scan_file: int | str = 0, metadata_str: str | None = None,
                 hdf_map: hdfmap.NexusMap | None = None) -> str:
        """Read scan file and return metadata string"""
        if metadata_str is None:
            from ..tkguis.misc.beamline_metadata import META_STRING, BEAMLINE_META
            if self.instrument in BEAMLINE_META:
                metadata_str = BEAMLINE_META[self.instrument]
            else:
                metadata_str = META_STRING

        scan_file = self.get_scan_filename(scan_file)

        if hdf_map is None:
            hdf_map = hdfmap.create_nexus_map(scan_file)

        with hdfmap.load_hdf(self.get_scan_filename(scan_file)) as hdf:
            return hdf_map.format_hdf(hdf, metadata_str, raise_errors=True)

    def scans_str(self, *scan_files: int | str, metadata_str: str | None = None,
                  hdf_map: hdfmap.NexusMap | None = None) -> list[str]:
        """Return list of string descriptions for multiple files"""
        if metadata_str is None:
            metadata_str = " : {str(start_time):30} : " + self.config[C.scan_description]
        filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
        if hdf_map is None:
            hdf_map = hdfmap.create_nexus_map(filenames[0])
        folder_file = ['/'.join(filename.split(os.sep)[-2:]) for filename in filenames]
        return [
            name + self.scan_str(file, metadata_str, hdf_map)
            for file, name in zip(filenames, folder_file)
        ]

    def all_scans_str(self, metadata_str: str | None = None, hdf_map: hdfmap.NexusMap | None = None) -> str:
        """Return string descriptions for all files"""
        scan_files = self.all_scan_files()
        return '\n'.join(self.scans_str(*scan_files, metadata_str=metadata_str, hdf_map=hdf_map))

    def _generate_scans_title(self, *scans: NexusScan, metadata_str: str | None = None) -> str:
        """Generate title from multiple scan files"""
        if metadata_str is None:
            metadata_str = "\n" + self.config[C.scan_description]
        first_scan = scans[0]
        folder = first_scan.filename.split(os.sep)[-2]
        meta = first_scan.format(metadata_str)
        scan_numbers = [scan.scan_number() for scan in scans]
        number_range = numbers2string(scan_numbers)
        return f"{folder} {number_range} {meta}"

    def generate_scans_title(self, *scan_files: int | str, metadata_str: str | None = None,
                             hdf_map: hdfmap.NexusMap | None = None) -> str:
        """Generate title from multiple scan files"""
        scans = self.scans(*scan_files, hdf_map=hdf_map)
        return self._generate_scans_title(*scans, metadata_str=metadata_str)

    def load_xas(self, *scan_files: int | str, sample_name: str | None = '', element_edge: str | None = None,
                 mode: str | list[str] = 'all', dls_loader: bool = False, match_metadata: bool = True,
                 temp_tol: float = 1., field_tol: float = 0.1) -> list[SpectraContainer]:
        """
        Read XAS spectra - see xas.SpectraContainer

            spectra_list = exp.load_xas(12345, 12346, mode='TEY', match_metadata=True)

        :param scan_files: List of filenames or scan numbers in folder
        :param sample_name: sample name, e.g. 'sample1' or None to load from NeXus file
        :param element_edge: element edge, e.g. 'FeL3' or None to determine from energy range
        :param mode: detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file
        :param dls_loader: bool, if True uses explicit loading of metadata from DLS MMG beamlines
        :param match_metadata: bool, if True uses metadata to determine from sample name
        :param temp_tol: Tolerance for temperature comparison (default: 0.1 K)
        :param field_tol: Tolerance for field comparison (default: 0.1 T)
        :return: List of SpectraContainer objects containing XAS spectra
        """
        filenames = [self.get_scan_filename(file) for file in scan_files]
        kwargs = dict(sample_name=sample_name, element_edge=element_edge, mode=mode, dls_loader=dls_loader)
        if match_metadata and len(scan_files) > 1:
            return find_similar_measurements(*filenames, temp_tol=temp_tol, field_tol=field_tol, **kwargs)
        return load_xas_scans(*filenames, **kwargs)

    def xas_polarised_spectra(self, *scan_files: int | str, sample_name: str | None = '',
                              element_edge: str | None = None, mode: str | list[str] = 'all',
                              dls_loader: bool = False, match_metadata: bool = True,
                              temp_tol: float = 1., field_tol: float = 0.1) -> tuple[SpectraContainer, SpectraContainer]:
        """
        Read XAS spectra - see xas.SpectraContainer

            pol1, pol2 = exp.xas_polarised_spectra(12345, 12346, mode='TEY', match_metadata=True)

        :param scan_files: List of filenames or scan numbers in folder
        :param sample_name: sample name, e.g. 'sample1' or None to load from NeXus file
        :param element_edge: element edge, e.g. 'FeL3' or None to determine from energy range
        :param mode: detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file
        :param dls_loader: bool, if True uses explicit loading of metadata from DLS MMG beamlines
        :param match_metadata: bool, if True uses metadata to determine from sample name
        :param temp_tol: Tolerance for temperature comparison (default: 0.1 K)
        :param field_tol: Tolerance for field comparison (default: 0.1 T)
        :return: pol1, pol2 SpectraContainer objects containing averaged XAS spectra
        """
        spectra = self.load_xas(*scan_files, sample_name=sample_name, element_edge=element_edge,
                                mode=mode, dls_loader=dls_loader, match_metadata=match_metadata,
                                temp_tol=temp_tol, field_tol=field_tol)
        return average_polarised_scans(*spectra)

    def add_roi(self, name: str, cen_i: int | str, cen_j: int | str,
                wid_i: int = 30, wid_j: int = 30, image_name: str = 'IMAGE'):
        """
        Add an image ROI (region of interest) to scans
        The ROI operates on the default IMAGE dataset, loading only the required region from the file.
        The following expressions will be added, for use in self.eval etc.
            *name* -> returns the whole ROI array as a HDF5 dataset
            *name*_total -> returns the sum of each image in the ROI array
            *name*_max -> returns the max of each image in the ROI array
            *name*_min -> returns the min of each image in the ROI array
            *name*_mean -> returns the mean of each image in the ROI array
            *name*_bkg -> returns the background ROI array (area around ROI)
            *name*_rmbkg -> returns the total with background subtracted
            *name*_box -> returns the pixel positions of the ROI corners
            *name*_bkg_box -> returns the pixel positions of the background ROI

        :param name: string name of the ROI
        :param cen_i: central pixel index along first dimension, can be callable string
        :param cen_j: central pixel index along second dimension, can be callable string
        :param wid_i: full width along first dimension, in pixels
        :param wid_j: full width along second dimension, in pixels
        :param image_name: string name of the image
        """
        add_roi(self.config, name, cen_i, cen_j, wid_i, wid_j, image_name)

add_data_paths(*folder_paths)

Add additional paths

Source code in mmg_toolbox/utils/experiment.py 
86
87
88
89
90
91
92
93
def add_data_paths(self, *folder_paths: str):
    """Add additional paths"""
    new_paths = list(
        path for path in folder_paths
        if path not in self.folder_paths and os.path.isdir(path)
    )
    self.folder_paths = self.folder_paths + new_paths
    self._update_scan_list()

add_roi(name, cen_i, cen_j, wid_i=30, wid_j=30, image_name='IMAGE')

Add an image ROI (region of interest) to scans The ROI operates on the default IMAGE dataset, loading only the required region from the file. The following expressions will be added, for use in self.eval etc. name -> returns the whole ROI array as a HDF5 dataset name_total -> returns the sum of each image in the ROI array name_max -> returns the max of each image in the ROI array name_min -> returns the min of each image in the ROI array name_mean -> returns the mean of each image in the ROI array name_bkg -> returns the background ROI array (area around ROI) name_rmbkg -> returns the total with background subtracted name_box -> returns the pixel positions of the ROI corners name_bkg_box -> returns the pixel positions of the background ROI

Parameters:

Name Type Description Default
name str

string name of the ROI

 required
cen_i int | str

central pixel index along first dimension, can be callable string

 required
cen_j int | str

central pixel index along second dimension, can be callable string

 required
wid_i int

full width along first dimension, in pixels

 30
wid_j int

full width along second dimension, in pixels

 30
image_name str

string name of the image

 'IMAGE'
Source code in mmg_toolbox/utils/experiment.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def add_roi(self, name: str, cen_i: int | str, cen_j: int | str,
            wid_i: int = 30, wid_j: int = 30, image_name: str = 'IMAGE'):
    """
    Add an image ROI (region of interest) to scans
    The ROI operates on the default IMAGE dataset, loading only the required region from the file.
    The following expressions will be added, for use in self.eval etc.
        *name* -> returns the whole ROI array as a HDF5 dataset
        *name*_total -> returns the sum of each image in the ROI array
        *name*_max -> returns the max of each image in the ROI array
        *name*_min -> returns the min of each image in the ROI array
        *name*_mean -> returns the mean of each image in the ROI array
        *name*_bkg -> returns the background ROI array (area around ROI)
        *name*_rmbkg -> returns the total with background subtracted
        *name*_box -> returns the pixel positions of the ROI corners
        *name*_bkg_box -> returns the pixel positions of the background ROI

    :param name: string name of the ROI
    :param cen_i: central pixel index along first dimension, can be callable string
    :param cen_j: central pixel index along second dimension, can be callable string
    :param wid_i: full width along first dimension, in pixels
    :param wid_j: full width along second dimension, in pixels
    :param image_name: string name of the image
    """
    add_roi(self.config, name, cen_i, cen_j, wid_i, wid_j, image_name)

all_scans_str(metadata_str=None, hdf_map=None)

Return string descriptions for all files

Source code in mmg_toolbox/utils/experiment.py 
279
280
281
282
def all_scans_str(self, metadata_str: str | None = None, hdf_map: hdfmap.NexusMap | None = None) -> str:
    """Return string descriptions for all files"""
    scan_files = self.all_scan_files()
    return '\n'.join(self.scans_str(*scan_files, metadata_str=metadata_str, hdf_map=hdf_map))

find_scans(*scan_files, hdf_map=None, first_only=False, **matches)

Find scans files with matching metadata

matches = {
    'name1': 'scan', # matches if 'scan' in file['name1']
    'name2': value, # matches if file['name2'] ~= value
    'name3': (value, tol), # matches if abs(file['name3'] - value) < tol
}
match_files = find_scans(*filenames, **matches)

Parameters:

Name Type Description Default
scan_files int | str

set of scan numbers or filenames, if not given will search all scans in folder.

 ()
hdf_map NexusMap | None

if given, uses this hdfmap rather than generating one.

 None
first_only bool

if true, returns on the first result

 False
matches str | float | tuple[float, float]

keyword arguments for matching parameters

 {}

Returns:

Type Description
list[NexusScan]

list of scan files that match all requirements
Source code in mmg_toolbox/utils/experiment.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def find_scans(self, *scan_files: int | str,  hdf_map: hdfmap.NexusMap | None = None, first_only: bool = False,
               **matches: str | float | tuple[float, float]) -> list[NexusScan]:
    """
    Find scans files with matching metadata

        matches = {
            'name1': 'scan', # matches if 'scan' in file['name1']
            'name2': value, # matches if file['name2'] ~= value
            'name3': (value, tol), # matches if abs(file['name3'] - value) < tol
        }
        match_files = find_scans(*filenames, **matches)

    :param scan_files: set of scan numbers or filenames, if not given will search all scans in folder.
    :param hdf_map: if given, uses this hdfmap rather than generating one.
    :param first_only: if true, returns on the first result
    :param matches: keyword arguments for matching parameters
    :returns: list of scan files that match all requirements
    """
    filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
    if not filenames:
        filenames = list(self.all_scans().values())
    if hdf_map is None:
        hdf_map = hdfmap.create_nexus_map(filenames[0])
    matches = find_scans(*filenames, hdf_map=hdf_map, first_only=first_only, **matches)
    return self.scans(*matches, hdf_map=hdf_map)

generate_mesh(*scan_files, hdf_map=None, axes='axes', signal='axes', values=None)

Generate 2D mesh from scan or scans

x, y, z = generate_mesh(*range(-10, 0), axes='eta', signal='roi2_sum', values='Tsample')
# or, if scan 12345 is a 2D grid scan
x, y, z = generate_mesh(12345, axes=('sx', 'sy'), signal='roi2_sum')

Parameters:

Name Type Description Default
scan_files int | str

multiple files or single 2D grid scan

 ()
hdf_map NexusMap | None

hdfmap.NeXus map, or None to generate

 None
axes str | tuple[str, str]

x-axis name, or for grid scans the names of ('xaxis', 'yaxis')

 'axes'
signal str

signal name

 'axes'
values str | None

name of the value that changes between multiple files

 None

Returns:

Type Description
tuple[ndarray, ndarray, ndarray]

X, Y, IMAGE rank 2 arrays
Source code in mmg_toolbox/utils/experiment.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def generate_mesh(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None,
                  axes: str | tuple[str, str] = 'axes', signal: str = 'axes',
                  values: str | None = None) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    Generate 2D mesh from scan or scans

        x, y, z = generate_mesh(*range(-10, 0), axes='eta', signal='roi2_sum', values='Tsample')
        # or, if scan 12345 is a 2D grid scan
        x, y, z = generate_mesh(12345, axes=('sx', 'sy'), signal='roi2_sum')

    :param scan_files: multiple files or single 2D grid scan
    :param hdf_map: hdfmap.NeXus map, or None to generate
    :param axes: x-axis name, or for grid scans the names of ('xaxis', 'yaxis')
    :param signal: signal name
    :param values: name of the value that changes between multiple files
    :returns: X, Y, IMAGE rank 2 arrays
    """
    scans = self.scans(*scan_files, hdf_map=hdf_map)
    if len(scan_files) == 1:
        # Single 2D Grid scan
        scan = scans[0]
        try:
            x_axis, y_axis = axes
        except ValueError:
            raise Exception("axes should be specified as axes=('axes0', 'axes1')")
        with scan.load_hdf() as hdf:
            x_data = scan.map.eval(hdf, x_axis, default=0.0).squeeze()
            y_data = scan.map.eval(hdf, y_axis, default=0.0).squeeze()
            z_data = scan.map.eval(hdf, signal, default=0.0).squeeze()
        if x_data.size != y_data.size:
            raise Exception(f"arrays '{x_axis}'[{x_data.size}] and '{y_axis}'[{y_data.size}] have different sizes")
        if x_data.ndim == 1 and y_data.ndim == 1:
            y_data, x_data = np.meshgrid(y_data, x_data)
        if z_data.shape != x_data.shape or z_data.shape != y_data.shape:
            raise Exception(
                f"{repr(scan)} '{x_axis}', '{y_axis}' and '{signal}' shapes are not consistent: " +
                f"x: {x_data.shape}, y: {y_data.shape}, im: {z_data.shape}"
            )
        return x_data, y_data, z_data
    else:
        x_data, y_data, z_data = [], [], []
        for n, scan in enumerate(scans):
            with scan.load_hdf() as hdf:
                x = np.reshape(scan.map.eval(hdf, axes, default=0.0), -1)
                y = np.reshape(scan.map.eval(hdf, signal, default=0.0), -1)
                val = np.reshape(scan.map.eval(hdf, values, default=0.0), -1) if values is not None else np.array([n])
            if val.size == 1:
                val = np.tile(val, x.size)
            if y.size != x.size or val.size != x.size:
                raise Exception(
                    f"{repr(scan)} '{axes}', '{signal}' and '{values}' shapes are not consistent: " +
                    f"x, y, val = ({x.size}, {y.size}, {val.size})"
                )
            x_data.append(x)
            y_data.append(val)
            z_data.append(y)
        # create a regular sized array
        min_len = min(len(x) for x in x_data)
        # array size [len(scan_files), min_len]
        x_array = np.array([x[:min_len] for x in x_data])
        y_array = np.array([y[:min_len] for y in y_data])
        z_array = np.array([z[:min_len] for z in z_data])
        return x_array, y_array, z_array

generate_scans_title(*scan_files, metadata_str=None, hdf_map=None)

Generate title from multiple scan files

Source code in mmg_toolbox/utils/experiment.py 
295
296
297
298
299
def generate_scans_title(self, *scan_files: int | str, metadata_str: str | None = None,
                         hdf_map: hdfmap.NexusMap | None = None) -> str:
    """Generate title from multiple scan files"""
    scans = self.scans(*scan_files, hdf_map=hdf_map)
    return self._generate_scans_title(*scans, metadata_str=metadata_str)

get_all_data(*fields, default=np.array([0.0]))

Return dict of data for all files

Source code in mmg_toolbox/utils/experiment.py 
177
178
179
180
181
def get_all_data(self, *fields: str, default: np.ndarray = np.array([0.0])) -> dict[str, list]:
    """
    Return dict of data for all files
    """
    return self.join_scan_data(data_fields=list(fields), default=default)

get_scan_filename(scan_file=-1)

Return the full filename of a scan number

Source code in mmg_toolbox/utils/experiment.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def get_scan_filename(self, scan_file: int | str = -1) -> str:
    """Return the full filename of a scan number"""
    if isinstance(scan_file, int) or scan_file.isdigit():
        scan_file = int(scan_file)
        if scan_file < 1:
            scan_numbers = self._scan_numbers()
            return self.scan_list[scan_numbers[scan_file]]
        self._update_scan_list()
        if scan_file in self.scan_list:
            return self.scan_list[scan_file]
        scan_numbers = self._scan_numbers()
        return self.scan_list[scan_numbers[scan_file]]

    if os.path.isfile(scan_file):
        return os.path.abspath(scan_file)
    raise FileNotFoundError(f"scan file {scan_file} not found")

join_scan_data(*scan_files, hdf_map=None, data_fields=None, default=np.array([0.0]))

Join data from scans

Source code in mmg_toolbox/utils/experiment.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
def join_scan_data(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None,
                   data_fields: list[str] | None = None, default: np.ndarray = np.array([0.0])) -> dict[str, list]:
    """
    Join data from scans
    """
    scans = self.scans(*scan_files, hdf_map=hdf_map)
    data_fields = [self.config[C.scan_description]] if data_fields is None else data_fields
    data = {name: [] for name in data_fields}
    for scan in scans:
        with scan.load_hdf() as hdf:
            for name in data_fields:
                data[name].append(scan.map.eval(hdf, name, default=default))
    return data

load_xas(*scan_files, sample_name='', element_edge=None, mode='all', dls_loader=False, match_metadata=True, temp_tol=1.0, field_tol=0.1)

Read XAS spectra - see xas.SpectraContainer

spectra_list = exp.load_xas(12345, 12346, mode='TEY', match_metadata=True)

Parameters:

Name Type Description Default
scan_files int | str

List of filenames or scan numbers in folder

 ()
sample_name str | None

sample name, e.g. 'sample1' or None to load from NeXus file

 ''
element_edge str | None

element edge, e.g. 'FeL3' or None to determine from energy range

 None
mode str | list[str]

detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file

 'all'
dls_loader bool

bool, if True uses explicit loading of metadata from DLS MMG beamlines

 False
match_metadata bool

bool, if True uses metadata to determine from sample name

 True
temp_tol float

Tolerance for temperature comparison (default: 0.1 K)

 1.0
field_tol float

Tolerance for field comparison (default: 0.1 T)

 0.1

Returns:

Type Description
list[SpectraContainer]

List of SpectraContainer objects containing XAS spectra
Source code in mmg_toolbox/utils/experiment.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def load_xas(self, *scan_files: int | str, sample_name: str | None = '', element_edge: str | None = None,
             mode: str | list[str] = 'all', dls_loader: bool = False, match_metadata: bool = True,
             temp_tol: float = 1., field_tol: float = 0.1) -> list[SpectraContainer]:
    """
    Read XAS spectra - see xas.SpectraContainer

        spectra_list = exp.load_xas(12345, 12346, mode='TEY', match_metadata=True)

    :param scan_files: List of filenames or scan numbers in folder
    :param sample_name: sample name, e.g. 'sample1' or None to load from NeXus file
    :param element_edge: element edge, e.g. 'FeL3' or None to determine from energy range
    :param mode: detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file
    :param dls_loader: bool, if True uses explicit loading of metadata from DLS MMG beamlines
    :param match_metadata: bool, if True uses metadata to determine from sample name
    :param temp_tol: Tolerance for temperature comparison (default: 0.1 K)
    :param field_tol: Tolerance for field comparison (default: 0.1 T)
    :return: List of SpectraContainer objects containing XAS spectra
    """
    filenames = [self.get_scan_filename(file) for file in scan_files]
    kwargs = dict(sample_name=sample_name, element_edge=element_edge, mode=mode, dls_loader=dls_loader)
    if match_metadata and len(scan_files) > 1:
        return find_similar_measurements(*filenames, temp_tol=temp_tol, field_tol=field_tol, **kwargs)
    return load_xas_scans(*filenames, **kwargs)

scan(scan_file=-1)

read Nexus file as NexusDataHolder

Source code in mmg_toolbox/utils/experiment.py 
124
125
126
def scan(self, scan_file: int | str = -1) -> NexusDataHolder:
    """read Nexus file as NexusDataHolder"""
    return NexusDataHolder(self.get_scan_filename(scan_file), config=self.config)

scan_str(scan_file=0, metadata_str=None, hdf_map=None)

Read scan file and return metadata string

Source code in mmg_toolbox/utils/experiment.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def scan_str(self, scan_file: int | str = 0, metadata_str: str | None = None,
             hdf_map: hdfmap.NexusMap | None = None) -> str:
    """Read scan file and return metadata string"""
    if metadata_str is None:
        from ..tkguis.misc.beamline_metadata import META_STRING, BEAMLINE_META
        if self.instrument in BEAMLINE_META:
            metadata_str = BEAMLINE_META[self.instrument]
        else:
            metadata_str = META_STRING

    scan_file = self.get_scan_filename(scan_file)

    if hdf_map is None:
        hdf_map = hdfmap.create_nexus_map(scan_file)

    with hdfmap.load_hdf(self.get_scan_filename(scan_file)) as hdf:
        return hdf_map.format_hdf(hdf, metadata_str, raise_errors=True)

scans(*scan_files, hdf_map=None)

Read Nexus files as lazy NexusScan. All files use the same HdfMap, based on the first scan

Source code in mmg_toolbox/utils/experiment.py 
128
129
130
131
132
133
134
135
def scans(self, *scan_files: int | str, hdf_map: hdfmap.NexusMap | None = None) -> list[NexusScan]:
    """Read Nexus files as lazy NexusScan. All files use the same HdfMap, based on the first scan"""
    filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
    if not filenames:
        filenames = list(self.all_scans().values())
    if filenames and hdf_map is None:
        hdf_map = hdfmap.create_nexus_map(filenames[0])
    return [NexusScan(file, hdf_map, config=self.config) for file in filenames]

scans_str(*scan_files, metadata_str=None, hdf_map=None)

Return list of string descriptions for multiple files

Source code in mmg_toolbox/utils/experiment.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
def scans_str(self, *scan_files: int | str, metadata_str: str | None = None,
              hdf_map: hdfmap.NexusMap | None = None) -> list[str]:
    """Return list of string descriptions for multiple files"""
    if metadata_str is None:
        metadata_str = " : {str(start_time):30} : " + self.config[C.scan_description]
    filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
    if hdf_map is None:
        hdf_map = hdfmap.create_nexus_map(filenames[0])
    folder_file = ['/'.join(filename.split(os.sep)[-2:]) for filename in filenames]
    return [
        name + self.scan_str(file, metadata_str, hdf_map)
        for file, name in zip(filenames, folder_file)
    ]

xas_polarised_spectra(*scan_files, sample_name='', element_edge=None, mode='all', dls_loader=False, match_metadata=True, temp_tol=1.0, field_tol=0.1)

Read XAS spectra - see xas.SpectraContainer

pol1, pol2 = exp.xas_polarised_spectra(12345, 12346, mode='TEY', match_metadata=True)

Parameters:

Name Type Description Default
scan_files int | str

List of filenames or scan numbers in folder

 ()
sample_name str | None

sample name, e.g. 'sample1' or None to load from NeXus file

 ''
element_edge str | None

element edge, e.g. 'FeL3' or None to determine from energy range

 None
mode str | list[str]

detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file

 'all'
dls_loader bool

bool, if True uses explicit loading of metadata from DLS MMG beamlines

 False
match_metadata bool

bool, if True uses metadata to determine from sample name

 True
temp_tol float

Tolerance for temperature comparison (default: 0.1 K)

 1.0
field_tol float

Tolerance for field comparison (default: 0.1 T)

 0.1

Returns:

Type Description
tuple[SpectraContainer, SpectraContainer]

pol1, pol2 SpectraContainer objects containing averaged XAS spectra
Source code in mmg_toolbox/utils/experiment.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def xas_polarised_spectra(self, *scan_files: int | str, sample_name: str | None = '',
                          element_edge: str | None = None, mode: str | list[str] = 'all',
                          dls_loader: bool = False, match_metadata: bool = True,
                          temp_tol: float = 1., field_tol: float = 0.1) -> tuple[SpectraContainer, SpectraContainer]:
    """
    Read XAS spectra - see xas.SpectraContainer

        pol1, pol2 = exp.xas_polarised_spectra(12345, 12346, mode='TEY', match_metadata=True)

    :param scan_files: List of filenames or scan numbers in folder
    :param sample_name: sample name, e.g. 'sample1' or None to load from NeXus file
    :param element_edge: element edge, e.g. 'FeL3' or None to determine from energy range
    :param mode: detector values to load, 'all', 'default' or e.g. 'tey', 'tfy' as specified in file
    :param dls_loader: bool, if True uses explicit loading of metadata from DLS MMG beamlines
    :param match_metadata: bool, if True uses metadata to determine from sample name
    :param temp_tol: Tolerance for temperature comparison (default: 0.1 K)
    :param field_tol: Tolerance for field comparison (default: 0.1 T)
    :return: pol1, pol2 SpectraContainer objects containing averaged XAS spectra
    """
    spectra = self.load_xas(*scan_files, sample_name=sample_name, element_edge=element_edge,
                            mode=mode, dls_loader=dls_loader, match_metadata=match_metadata,
                            temp_tol=temp_tol, field_tol=field_tol)
    return average_polarised_scans(*spectra)