Code Description

MMG_Toolbox

Magnetic Materials Group Toolbox

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

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: dict[int, str] = {}
        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.all_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 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: ScanFile = -1) -> str:
        """Return the full filename of a scan number"""
        if isinstance(scan_file, (NexusScan, NexusDataHolder)):
            scan_file = scan_file.filename
        elif isinstance(scan_file, int) or scan_file.isdigit():
            scan_file = int(scan_file)
            self._update_scan_list()
            if scan_file in self.scan_list:
                # scan_file is scan number
                return self.scan_list[scan_file]
            elif -len(self.scan_list) < scan_file < len(self.scan_list):
                # scan_file is index
                scan_numbers = list(self.scan_list)
                return self.scan_list[scan_numbers[scan_file]]
            else:
                FileNotFoundError(f"scan number {scan_file} not found")
        if os.path.isfile(scan_file):
            # scan file is filename
            return os.path.abspath(scan_file)
        raise FileNotFoundError(f"scan file {scan_file} not found")

    def get_scan_number(self, scan_file: ScanFile = -1) -> int:
        """Return scan number of file or number"""
        filename = self.get_scan_filename(scan_file)
        files = list(self.scan_list.values())
        index = files.index(filename)
        return list(self.scan_list)[index]

    def get_nearby_scan_numbers(self, scan_file: ScanFile = -1, before=10, after=10) -> list[int]:
        """Return scan numbers near the given scan, based on position rather than scan numbers"""
        scan_number = self.get_scan_number(scan_file)
        all_scan_numbers = list(self.scan_list)
        scan_index = all_scan_numbers.index(scan_number)
        scan_indexes = range(
            scan_index - before if scan_index > before else 0,
            scan_index + after if scan_index + after < len(all_scan_numbers) else len(all_scan_numbers) - 1
        )
        return [all_scan_numbers[idx] for idx in scan_indexes]

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

    def scans(self, *scan_files: ScanFile, 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"""
        if scan_files and isinstance(scan_files[0], (NexusScan, NexusDataHolder)):
            return list(scan_files)
        filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
        if not filenames:
            filenames = self.all_scan_files()
        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: ScanFile,  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': 'energy', # matches if 'scan' in scan('name1')
                'name2': value, # matches if scan('name2') ~= value
                'name3': (value, tol), # matches if abs(scan('name3') - value) < tol
            }
            match_scans = exp.find_scans(*scan_numbers, **matches)

        :param scan_files: set of scan numbers or filenames, if not given will search all scans in folder. If single value, searches nearby scans.
        :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 objects that match all requirements
        """
        if len(scan_files) == 0:
            scan_files = self.all_scan_numbers()
        if len(scan_files) == 1:
            scan_files = self.get_nearby_scan_numbers(scan_files[0])
        scans = self.scans(*scan_files, hdf_map=hdf_map)

        def check(scn: NexusScan):
            for name, match_value in matches.items():
                file_value = scn.eval(name)
                if isinstance(match_value, str):
                    chk = match_value in file_value
                elif isinstance(match_value, (float, int)):
                    chk = abs(match_value - file_value) < 0.01
                else:
                    chk = abs(match_value[0] - file_value) < match_value[1]
                if not chk:
                    return False
            return True
        search = (scan for scan in scans if check(scan))

        if first_only:
            return [next(search)]
        return list(search)

    def join_scan_data(self, *scan_files: ScanFile, 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 - return list of values from each scan, in dict for each data field

            data = exp.join_scan_data(*scan_files, data_fields=['scan_command', 'Ta'])
            data['scan_command'] -> ['scan ...', 'scan ...', ...]
            data['Ta'] -> [10, 20, ...]

        :param scan_files: set of scan numbers or filenames
        :param hdf_map: if given, uses this hdfmap rather than generating one.
        :param data_fields: list of data fields
        :param default: default value
        :return: dict of data for all files
        """
        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}
        expression = ','.join(data_fields)
        for scan in scans:
            values = scan.eval(expression, default=default)  # use eval to ensure local_data is available
            for name, value in zip(data_fields, values):
                data[name].append(value)
        return data

    def join_scan_arrays(self, *scan_files: ScanFile, hdf_map: hdfmap.NexusMap | None = None,
                         data_fields: list[str] | None = None, default: np.ndarray = np.array([0.0])) -> tuple[np.ndarray, ...]:
        """
        Join data from scans - return array of values for each scan, for each data field

            cmd, temp = exp.join_scan_arrays(*scan_files, data_fields=['scan_command', 'Ta'])

        :param scan_files: set of scan numbers or filenames
        :param hdf_map: if given, uses this hdfmap rather than generating one.
        :param data_fields: list of data fields
        :param default: default value
        :return: Numpy arrays
        """
        data = self.join_scan_data(*scan_files, hdf_map=hdf_map, data_fields=data_fields, default=default)
        return tuple(np.array(array) for array in data.values())

    def get_value_changes(self, *scan_files: ScanFile, hdf_map: hdfmap.NexusMap | None = None,
                          omit_change: bool = True, sort: bool = True) -> dict[str, np.ndarray]:
        """
        Returns dict of all values that changed between scan files
        Only numeric values are checked.

            data = exp.find_value_changes(*scan_numbers)
            print('\n'.join(f"{name} : {np.std(array):.2f}" for name, array in data.items()))

        :param scan_files: set of scan numbers or filenames
        :param hdf_map: if given, uses this hdfmap rather than generating one.
        :param omit_change: if true, returns only the changed values
        :param sort: if true, returns only the changed values
        :return: dict of all values that changed between scan files
        """
        scans = self.scans(*scan_files, hdf_map=hdf_map)
        m = hdf_map or scans[0].map
        with scans[0].load_hdf() as hdf:
            parameters = {
                name: path for name, path in (m.metadata or m.values).items()
                if (dataset := hdf.get(path)) and np.issubdtype(dataset.dtype, np.number)
            }
        # Pre-allocate
        data = {
            parameter: np.zeros(len(scans))
            for parameter in parameters
        }
        # Load values
        for n, scan in enumerate(scans):
            values = scan.values(*parameters)
            for name, value in zip(parameters, values):
                data[name][n] = value
        # Omit values that don't change
        if omit_change:
            data = {
                parameter: array
                for parameter, array in data.items()
                if not all(np.isnan(array)) and (np.nanstd(array) / np.nanmean(array)) > 0.01
            }
        # Sort by std
        if sort:
            data = dict(
                sorted(data.items(), key=lambda item: np.nanstd(item[1]), reverse=True)
            )
        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: ScanFile, hdf_map: hdfmap.NexusMap | None = None,
                      axes: str | tuple[str, str] = 'axes', signal: str = 'signal',
                      values: str | tuple[str, str] | None = None, interpolate: bool = False) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
        """
        Generate 2D mesh from scan or scans

            # to plot grid of eta vs Tsample vs roi2_sum (each scan over eta)
            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')
            # or, if scans are varying by sx and sy in a grid
            x, y, z = generate_mesh(*range(0, 25), values=('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
        :param interpolate: if true, interpolates between points
        :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:
            try:
                # 2D grid scan with 2 dimensions changing
                x_axis, y_axis = values
                z_axis = signal
                x_data, y_data, z_data = self.join_scan_arrays(*scans, data_fields=[x_axis, y_axis, z_axis])
                if interpolate:
                    return interpolate_to_grid(x_data, y_data, z_data)
                return coordinates_to_grid(x_data, y_data, z_data)
            except (ValueError, TypeError):
                pass
            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 x.size == 1:
                    x = np.tile(x, val.size)
                elif 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: ScanFile = -1, 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: ScanFile, 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: ScanFile, 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: ScanFile, 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
        """
        kwargs = dict(sample_name=sample_name, element_edge=element_edge, mode=mode, dls_loader=dls_loader)
        spectra = [scan.xas_spectra(**kwargs) for scan in self.scans(*scan_files)]
        if match_metadata:
            if len(scan_files) == 1:
                scan_numbers = self.get_nearby_scan_numbers(scan_files[0])
                return self.load_xas(*scan_numbers, match_metadata=True, temp_tol=temp_tol, field_tol=field_tol, **kwargs)
            else:
                return find_similar_measurements(*spectra, temp_tol=temp_tol, field_tol=field_tol)
        return spectra

    def xas_polarised_spectra(self, *scan_files: ScanFile, 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

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

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

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': 'energy', # matches if 'scan' in scan('name1')
    'name2': value, # matches if scan('name2') ~= value
    'name3': (value, tol), # matches if abs(scan('name3') - value) < tol
}
match_scans = exp.find_scans(*scan_numbers, **matches)

Parameters:

Name	Type	Description	Default
scan_files	ScanFile	set of scan numbers or filenames, if not given will search all scans in folder. If single value, searches nearby scans.	()
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 objects that match all requirements

Source code in mmg_toolbox/utils/experiment.py

def find_scans(self, *scan_files: ScanFile,  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': 'energy', # matches if 'scan' in scan('name1')
            'name2': value, # matches if scan('name2') ~= value
            'name3': (value, tol), # matches if abs(scan('name3') - value) < tol
        }
        match_scans = exp.find_scans(*scan_numbers, **matches)

    :param scan_files: set of scan numbers or filenames, if not given will search all scans in folder. If single value, searches nearby scans.
    :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 objects that match all requirements
    """
    if len(scan_files) == 0:
        scan_files = self.all_scan_numbers()
    if len(scan_files) == 1:
        scan_files = self.get_nearby_scan_numbers(scan_files[0])
    scans = self.scans(*scan_files, hdf_map=hdf_map)

    def check(scn: NexusScan):
        for name, match_value in matches.items():
            file_value = scn.eval(name)
            if isinstance(match_value, str):
                chk = match_value in file_value
            elif isinstance(match_value, (float, int)):
                chk = abs(match_value - file_value) < 0.01
            else:
                chk = abs(match_value[0] - file_value) < match_value[1]
            if not chk:
                return False
        return True
    search = (scan for scan in scans if check(scan))

    if first_only:
        return [next(search)]
    return list(search)

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

Generate 2D mesh from scan or scans

# to plot grid of eta vs Tsample vs roi2_sum (each scan over eta)
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')
# or, if scans are varying by sx and sy in a grid
x, y, z = generate_mesh(*range(0, 25), values=('sx', 'sy'), signal='roi2_sum')

Parameters:

Name	Type	Description	Default
scan_files	ScanFile	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	'signal'
values	str | tuple[str, str] | None	name of the value that changes between multiple files	None
interpolate	bool	if true, interpolates between points	False

Returns:

Type	Description
tuple[ndarray, ndarray, ndarray]	X, Y, IMAGE rank 2 arrays

Source code in mmg_toolbox/utils/experiment.py

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

        # to plot grid of eta vs Tsample vs roi2_sum (each scan over eta)
        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')
        # or, if scans are varying by sx and sy in a grid
        x, y, z = generate_mesh(*range(0, 25), values=('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
    :param interpolate: if true, interpolates between points
    :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:
        try:
            # 2D grid scan with 2 dimensions changing
            x_axis, y_axis = values
            z_axis = signal
            x_data, y_data, z_data = self.join_scan_arrays(*scans, data_fields=[x_axis, y_axis, z_axis])
            if interpolate:
                return interpolate_to_grid(x_data, y_data, z_data)
            return coordinates_to_grid(x_data, y_data, z_data)
        except (ValueError, TypeError):
            pass
        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 x.size == 1:
                x = np.tile(x, val.size)
            elif 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

def generate_scans_title(self, *scan_files: ScanFile, 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

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_nearby_scan_numbers(scan_file=-1, before=10, after=10)

Return scan numbers near the given scan, based on position rather than scan numbers

Source code in mmg_toolbox/utils/experiment.py

def get_nearby_scan_numbers(self, scan_file: ScanFile = -1, before=10, after=10) -> list[int]:
    """Return scan numbers near the given scan, based on position rather than scan numbers"""
    scan_number = self.get_scan_number(scan_file)
    all_scan_numbers = list(self.scan_list)
    scan_index = all_scan_numbers.index(scan_number)
    scan_indexes = range(
        scan_index - before if scan_index > before else 0,
        scan_index + after if scan_index + after < len(all_scan_numbers) else len(all_scan_numbers) - 1
    )
    return [all_scan_numbers[idx] for idx in scan_indexes]

get_scan_filename(scan_file=-1)

Return the full filename of a scan number

Source code in mmg_toolbox/utils/experiment.py

def get_scan_filename(self, scan_file: ScanFile = -1) -> str:
    """Return the full filename of a scan number"""
    if isinstance(scan_file, (NexusScan, NexusDataHolder)):
        scan_file = scan_file.filename
    elif isinstance(scan_file, int) or scan_file.isdigit():
        scan_file = int(scan_file)
        self._update_scan_list()
        if scan_file in self.scan_list:
            # scan_file is scan number
            return self.scan_list[scan_file]
        elif -len(self.scan_list) < scan_file < len(self.scan_list):
            # scan_file is index
            scan_numbers = list(self.scan_list)
            return self.scan_list[scan_numbers[scan_file]]
        else:
            FileNotFoundError(f"scan number {scan_file} not found")
    if os.path.isfile(scan_file):
        # scan file is filename
        return os.path.abspath(scan_file)
    raise FileNotFoundError(f"scan file {scan_file} not found")

get_scan_number(scan_file=-1)

Return scan number of file or number

Source code in mmg_toolbox/utils/experiment.py

def get_scan_number(self, scan_file: ScanFile = -1) -> int:
    """Return scan number of file or number"""
    filename = self.get_scan_filename(scan_file)
    files = list(self.scan_list.values())
    index = files.index(filename)
    return list(self.scan_list)[index]

get_value_changes(*scan_files, hdf_map=None, omit_change=True, sort=True)

    Returns dict of all values that changed between scan files
    Only numeric values are checked.

        data = exp.find_value_changes(*scan_numbers)
        print('

'.join(f"{name} : {np.std(array):.2f}" for name, array in data.items()))

    :param scan_files: set of scan numbers or filenames
    :param hdf_map: if given, uses this hdfmap rather than generating one.
    :param omit_change: if true, returns only the changed values
    :param sort: if true, returns only the changed values
    :return: dict of all values that changed between scan files

Source code in mmg_toolbox/utils/experiment.py

def get_value_changes(self, *scan_files: ScanFile, hdf_map: hdfmap.NexusMap | None = None,
                      omit_change: bool = True, sort: bool = True) -> dict[str, np.ndarray]:
    """
    Returns dict of all values that changed between scan files
    Only numeric values are checked.

        data = exp.find_value_changes(*scan_numbers)
        print('\n'.join(f"{name} : {np.std(array):.2f}" for name, array in data.items()))

    :param scan_files: set of scan numbers or filenames
    :param hdf_map: if given, uses this hdfmap rather than generating one.
    :param omit_change: if true, returns only the changed values
    :param sort: if true, returns only the changed values
    :return: dict of all values that changed between scan files
    """
    scans = self.scans(*scan_files, hdf_map=hdf_map)
    m = hdf_map or scans[0].map
    with scans[0].load_hdf() as hdf:
        parameters = {
            name: path for name, path in (m.metadata or m.values).items()
            if (dataset := hdf.get(path)) and np.issubdtype(dataset.dtype, np.number)
        }
    # Pre-allocate
    data = {
        parameter: np.zeros(len(scans))
        for parameter in parameters
    }
    # Load values
    for n, scan in enumerate(scans):
        values = scan.values(*parameters)
        for name, value in zip(parameters, values):
            data[name][n] = value
    # Omit values that don't change
    if omit_change:
        data = {
            parameter: array
            for parameter, array in data.items()
            if not all(np.isnan(array)) and (np.nanstd(array) / np.nanmean(array)) > 0.01
        }
    # Sort by std
    if sort:
        data = dict(
            sorted(data.items(), key=lambda item: np.nanstd(item[1]), reverse=True)
        )
    return data

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

Join data from scans - return array of values for each scan, for each data field

cmd, temp = exp.join_scan_arrays(*scan_files, data_fields=['scan_command', 'Ta'])

Parameters:

Name	Type	Description	Default
scan_files	ScanFile	set of scan numbers or filenames	()
hdf_map	NexusMap | None	if given, uses this hdfmap rather than generating one.	None
data_fields	list[str] | None	list of data fields	None
default	ndarray	default value	array([0.0])

Returns:

Type	Description
tuple[ndarray, ...]	Numpy arrays

Source code in mmg_toolbox/utils/experiment.py

def join_scan_arrays(self, *scan_files: ScanFile, hdf_map: hdfmap.NexusMap | None = None,
                     data_fields: list[str] | None = None, default: np.ndarray = np.array([0.0])) -> tuple[np.ndarray, ...]:
    """
    Join data from scans - return array of values for each scan, for each data field

        cmd, temp = exp.join_scan_arrays(*scan_files, data_fields=['scan_command', 'Ta'])

    :param scan_files: set of scan numbers or filenames
    :param hdf_map: if given, uses this hdfmap rather than generating one.
    :param data_fields: list of data fields
    :param default: default value
    :return: Numpy arrays
    """
    data = self.join_scan_data(*scan_files, hdf_map=hdf_map, data_fields=data_fields, default=default)
    return tuple(np.array(array) for array in data.values())

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

Join data from scans - return list of values from each scan, in dict for each data field

data = exp.join_scan_data(*scan_files, data_fields=['scan_command', 'Ta'])
data['scan_command'] -> ['scan ...', 'scan ...', ...]
data['Ta'] -> [10, 20, ...]

Parameters:

Name	Type	Description	Default
scan_files	ScanFile	set of scan numbers or filenames	()
hdf_map	NexusMap | None	if given, uses this hdfmap rather than generating one.	None
data_fields	list[str] | None	list of data fields	None
default	ndarray	default value	array([0.0])

Returns:

Type	Description
dict[str, list]	dict of data for all files

Source code in mmg_toolbox/utils/experiment.py

def join_scan_data(self, *scan_files: ScanFile, 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 - return list of values from each scan, in dict for each data field

        data = exp.join_scan_data(*scan_files, data_fields=['scan_command', 'Ta'])
        data['scan_command'] -> ['scan ...', 'scan ...', ...]
        data['Ta'] -> [10, 20, ...]

    :param scan_files: set of scan numbers or filenames
    :param hdf_map: if given, uses this hdfmap rather than generating one.
    :param data_fields: list of data fields
    :param default: default value
    :return: dict of data for all files
    """
    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}
    expression = ','.join(data_fields)
    for scan in scans:
        values = scan.eval(expression, default=default)  # use eval to ensure local_data is available
        for name, value in zip(data_fields, values):
            data[name].append(value)
    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	ScanFile	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

def load_xas(self, *scan_files: ScanFile, 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
    """
    kwargs = dict(sample_name=sample_name, element_edge=element_edge, mode=mode, dls_loader=dls_loader)
    spectra = [scan.xas_spectra(**kwargs) for scan in self.scans(*scan_files)]
    if match_metadata:
        if len(scan_files) == 1:
            scan_numbers = self.get_nearby_scan_numbers(scan_files[0])
            return self.load_xas(*scan_numbers, match_metadata=True, temp_tol=temp_tol, field_tol=field_tol, **kwargs)
        else:
            return find_similar_measurements(*spectra, temp_tol=temp_tol, field_tol=field_tol)
    return spectra

scan(scan_file=-1)

read Nexus file as NexusDataHolder

Source code in mmg_toolbox/utils/experiment.py

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

scan_str(scan_file=-1, metadata_str=None, hdf_map=None)

Read scan file and return metadata string

Source code in mmg_toolbox/utils/experiment.py

def scan_str(self, scan_file: ScanFile = -1, 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

def scans(self, *scan_files: ScanFile, 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"""
    if scan_files and isinstance(scan_files[0], (NexusScan, NexusDataHolder)):
        return list(scan_files)
    filenames = [self.get_scan_filename(scan_file) for scan_file in scan_files]
    if not filenames:
        filenames = self.all_scan_files()
    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

def scans_str(self, *scan_files: ScanFile, 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	ScanFile	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

def xas_polarised_spectra(self, *scan_files: ScanFile, 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)

NexusDataHolder

Bases: DataHolder, NexusScan

Nexus data holder class - Automatically reads scannable and metadata from file - acts like the old .dat DataHolder class - has additional functions to read data from NeXus file

Example: scan = NexusDataHolder('12345.nxs') scan.eta -> returns array scan.metadata.metadata -> returns value scan('signal') -> evaluate expression

Parameters:

Name	Type	Description	Default
filename	str | None	path to Nexus file	required
hdf_map	NexusMap | None	NexusMap object or None to generate	None
flatten_scannables	bool	if True, flattens all scannable arrays to 1D	True

Source code in mmg_toolbox/nexus/nexus_scan.py

class NexusDataHolder(DataHolder, NexusScan):
    """
    Nexus data holder class
     - Automatically reads scannable and metadata from file
     - acts like the old .dat DataHolder class
     - has additional functions to read data from NeXus file

    Example:
        scan = NexusDataHolder('12345.nxs')
        scan.eta -> returns array
        scan.metadata.metadata -> returns value
        scan('signal') -> evaluate expression

    :param filename: path to Nexus file
    :param hdf_map: NexusMap object or None to generate
    :param flatten_scannables: if True, flattens all scannable arrays to 1D
    """
    filename: str
    map: NexusMap
    metadata: DataHolder

    def __init__(self, filename: str | None, hdf_map: NexusMap | None = None, flatten_scannables: bool = True,
                 config: dict | None = None):
        NexusScan.__init__(self, filename, hdf_map, config)

        with load_hdf(filename) as hdf:
            metadata = self.map.get_metadata(hdf)
            scannables = self.map.get_scannables(hdf, flatten=flatten_scannables)
        DataHolder.__init__(self, **scannables)
        self.metadata = DataHolder(**metadata)

    def __repr__(self):
        return f"NexusDataHolder('{self.filename}')"

NexusScan

Bases: NexusLoader

Light-weight NeXus file reader

>>> scan = NexusScan('scan.nxs')
>>> scan('scan_command')
 'scan x ...'
>>> x, y = scan('axes, signal / monitor')
>>> scan.plot()  # default plot
>>> scan.plot.image()  # other plot options
>>> result = scan.fit.multi_peak_fit()
>>> print(scan)  # print scan default metadata
>>> print(scan.info())  # print scan namespace
>>> scan.map.add_roi('name', ...)  # add ROI to namespace
>>> scan.image(0)  # return first detector image as array
>>> scan.volume()  # return image stack
>>> data = scan.get_plot_data()  # return dict of plot data

Parameters:

Name	Type	Description	Default
nxs_filename	str	path to nexus file	required
hdf_map	NexusMap | None	NexusMap object or None	None
config	dict | None	configuration dict	None

Source code in mmg_toolbox/nexus/nexus_scan.py

class NexusScan(NexusLoader):
    """
    Light-weight NeXus file reader

        >>> scan = NexusScan('scan.nxs')
        >>> scan('scan_command')
         'scan x ...'
        >>> x, y = scan('axes, signal / monitor')
        >>> scan.plot()  # default plot
        >>> scan.plot.image()  # other plot options
        >>> result = scan.fit.multi_peak_fit()
        >>> print(scan)  # print scan default metadata
        >>> print(scan.info())  # print scan namespace
        >>> scan.map.add_roi('name', ...)  # add ROI to namespace
        >>> scan.image(0)  # return first detector image as array
        >>> scan.volume()  # return image stack
        >>> data = scan.get_plot_data()  # return dict of plot data

    :param nxs_filename: path to nexus file
    :param hdf_map: NexusMap object or None
    :param config: configuration dict
    """
    MAX_STR_LEN: int = 100

    def __init__(self, nxs_filename: str, hdf_map: NexusMap | None = None, config: dict | None = None):
        super().__init__(nxs_filename, hdf_map)
        self.config: dict = config or beamline_config()
        self.beamline = self.config.get(C.beamline, None)

        # add scan number to eval namespace
        self.add_local(scan_number=self.scan_number(), beamline=self.beamline)
        # add alternate names
        self.map.add_named_expression(**self.config.get(C.replace_names, {}))
        # add ROIs
        for (name, cen_i, cen_j, wid_i, wid_j, det_name) in self.config.get(C.roi, []):
            self.map.add_roi(name, cen_i, cen_j, wid_i, wid_j, det_name)

        from mmg_toolbox.fitting import ScanFitManager, poisson_errors
        self.fit = ScanFitManager(self)
        self._error_function = poisson_errors
        from mmg_toolbox.plotting.scan_plot_manager import ScanPlotManager
        self.plot = ScanPlotManager(self)

    def __repr__(self):
        if self.beamline:
            return f"NexusScan<{self.beamline}>({self.scan_number()}: '{self.filename}')"
        return f"NexusScan('{self.filename}')"

    def __str__(self):
        try:
            return self.metadata_str()
        except Exception as ex:
            return f"{repr(self)}\n  Metadata failed with: \n{ex}\n"

    def metadata_str(self, expression: str | None = None):
        """Generate metadata string from beamline config"""
        if expression is None:
            expression = self.config.get(C.metadata_string, '')
        return self.format(expression)

    def info(self, arrays=False, values=False, combined=False,
             metadata=False, scannables=True, image_data=True,
             local=False, alternate=True) -> str:
        """Return string of namespace information"""
        local_data = (
                "Local Data:\n" +
                "\n".join(
                    f"  {name}: {value}" for name, value in self._local_data.items()
                ) +
                "\n"
        ) if local else ""
        alternate_data = (
            "Alternate Names:\n  Name: Expression" +
            "\n".join(
                f"  {name}: '{expr}'" for name, expr in self.map._alternate_names.items()
            ) +
            "\n"
        ) if alternate else ""
        map_data = self.map.info_names(arrays=arrays, values=values, combined=combined,
                                       metadata=metadata, scannables=scannables, image_data=image_data)
        return local_data + alternate_data + map_data

    def rois(self, append: str = '_total') -> list[str]:
        """
        Return ROI expressions available in scan namespace

            rois = [scan.eval(roi) for roi in scan.rois('_total')]

        :param append: str to append to each ROI name, e.g. '_total', '_max', '_min', '_mean'
        :return: list of ROI names that can be used in eval.
        """
        # search for ROIs in HdfMap expressions
        alternate_name_rois = {
            next((name.removesuffix(sfx) for sfx in ROI_SUFFIXES if name.endswith(sfx)), name)
            for name, expression in self.map._alternate_names.items()
            if expression.startswith('d_')
        }
        return [roi + append for roi in alternate_name_rois]

    def scan_number(self) -> int:
        return get_scan_number(self.filename)

    def title(self) -> str:
        return f"#{self.scan_number()}"

    def label(self) -> str:
        return f"#{self.scan_number()}"

    def start_end_duration(self) -> tuple[datetime.datetime, datetime.datetime, datetime.timedelta]:
        """Return start and end times of scan, plus duration as datetime and timedelta objects"""
        scan_start, scan_end = self.times(NX_START, NX_END, default=None)
        if scan_start is None:
            raise KeyError(f"scan start time not found in {self.filename}")
        if scan_end is None:
            scan_end = get_file_time(self.filename)
        return scan_start, scan_end, scan_end - scan_start

    def load_hdf(self) -> h5py.File:
        """Load the Hdf file"""
        return load_hdf(self.filename)

    def hdf_tree_string(self, group: str = '/', all_links: bool = True, attributes: bool = True) -> str:
        """
        Generate string of the hdf file structure, similar to h5ls. Uses h5py.visititems

        :param all_links: bool, if True, also show links
        :param group: only display tree structure of this group (default root)
        :param attributes: if True, display the attributes of groups and datasets
        :return: str
        """
        return hdfmap.hdf_tree_string(self.filename, all_links=all_links, group=group, attributes=attributes)

    def hdf_find(self, *field_or_class: str | list[str], find_all: bool = False) -> (h5py.Dataset | h5py.Group) | list[h5py.Dataset | h5py.Group]:
        """
        Find datasets and groups within hdf file

            HdfDataset = scan.hdf_find('NXentry', ['NXdata', 'measurement'], 'signal')

        Warning: HDF file stays in open state while Dataset or Group objects exist.

        :param field_or_class: names to search for, in hierarchical order. Lists treated as OR
        :param find_all: whether to return all datasets or only the first match
        :returns: matching Dataset or Group, or list of matching Datasets or Groups
        """
        with self.load_hdf() as hdf:
            if find_all:
                return nx_find_all(hdf, *field_or_class)
            return nx_find(hdf, *field_or_class)

    def datasets(self, *args) -> list[h5py.Dataset]:
        """Return HDF5 datasets from NeXus file (leaves file in open state)"""
        with self.load_hdf() as hdf:
            return [hdf[self.map.combined[name]] for name in args]

    def arrays(self, *args, units: str = '', default: np.ndarray = np.array([np.nan])) -> list[np.ndarray]:
        """Return Numpy arrays"""
        with self.load_hdf() as hdf:
            return [
                get_dataset_value(self.map.combined[name], hdf, units=units, default=default)
                for name in args
            ]

    def values(self, *args, value_func=np.mean,
               units: str = '', default: np.ndarray = np.array(np.nan)) -> list[np.floating]:
        """Return float values"""
        with self.load_hdf() as hdf:
            return [
                value_func(get_dataset_value(self.map.combined[name], hdf, units=units, default=default))
                for name in args
            ]

    def times(self, *args, default: datetime.datetime | None = None) -> list[datetime.datetime | None]:
        """Return datetime object"""
        with self.load_hdf() as hdf:
            data = [
                dataset2data(hdf[self.map.combined[name]])
                if name in self.map.combined else default
                for name in args
            ]
            dt = [
                obj if obj is None or isinstance(obj, datetime.datetime) else
                datetime.datetime.fromisoformat(obj) if isinstance(obj, str) else
                datetime.datetime.fromtimestamp(float(obj))
                for obj in data
            ]
        return dt

    def strings(self, *args, units=False) -> list[str]:
        """Return string value"""
        with self.load_hdf() as hdf:
            return [dataset2str(hdf[self.map.combined[name]], units=units) for name in args]

    def image(self, index: int | str | None = None) -> np.ndarray:
        """
        Return image or selection from default detector

            im = scan.image(0)  # First image
            im = scan.image(None)  # middle image
            im = scan.image('sum')  # sum of all images

        :param index: point in the scan to show the image, 'sum' gives sum over all images, None gives the middle image
        :return: (n,m) array
        """
        if not self.map.image_data:
            raise ValueError(f'{repr(self)} contains no image data')
        with self.load_hdf() as hdf:
            if index == 'sum':
                volume = self.map.get_image(hdf, ())
                image = volume.sum(axis=tuple(range(volume.ndim-2)))  # this will fail for filenames
            else:
                image = self.map.get_image(hdf, index)

            if issubclass(type(image), str):
                # TIFF image, NXdetector/image_data -> array('file.tif')
                file_directory = os.path.dirname(self.filename)
                image_filename = os.path.join(file_directory, image)
                if not os.path.isfile(image_filename):
                    raise FileNotFoundError(f"File not found: {image_filename}")
                image = read_tiff(image_filename)
            elif image.ndim == 0:
                # image is file path number, NXdetector/path -> arange(n_points)
                scan_number = get_scan_number(self.filename)
                file_directory = os.path.dirname(self.filename)
                detector_names = list(self.map.image_data.keys())
                for detector_name in detector_names:
                    image_filename = os.path.join(file_directory, f"{scan_number}-{detector_name}-files/{image:05.0f}.tif")
                    if os.path.isfile(image_filename):
                        break
                if not os.path.isfile(image_filename):
                    raise FileNotFoundError(f"File not found: {image_filename}")
                image = read_tiff(image_filename)
            elif image.ndim != 2:
                raise Exception(f"detector image[{index}] is the wrong shape: {image.shape}")
            return image

    def volume(self) -> np.ndarray:
        """Return complete stack of images"""
        return self.map.get_image(self.load_hdf(), ())

    def image_background(self, index: int | tuple | slice | str | None = (), n_bins: int  = 100) -> np.ndarray:
        """
        Return the modal value of the detector image,
        which usually gives the background value.

        The modal value is determined by histograming the image (or image stack) and taking
        the value of the largest bin.

        :param index: index of image to return, use () for full image stack.
        :param n_bins: number of histogram bins
        :return: modal value or values per image
        """
        image = self.map.get_image(self.load_hdf(), index)  # hdf data only
        n, bins = np.histogram(np.log10(image[image>0].flatten()), bins=n_bins)
        return 10 ** bins[np.argmax(n)]

    def get_max_index(self, expression: str = 'IMAGE') -> tuple[float, tuple[int, ...]]:
        """Return the index and value of the maximum pixel in a detector image or volume"""
        data = self.eval(expression)
        max_idx = tuple(int(v) for v in np.unravel_index(np.argmax(data), data.shape))
        return float(data[max_idx]), max_idx

    def table(self, delimiter=', ', string_spec='', format_spec='f', default_decimals=8) -> str:
        """Return data table"""
        with self.load_hdf() as hdf:
            return self.map.create_scannables_table(hdf, delimiter, string_spec, format_spec, default_decimals)

    def get_scan_time(self) -> np.ndarray:
        """Return array of datetimes for each scan point"""
        start, stop, duration = self.start_end_duration()
        count_time = self.eval('_t', default=None)  # '_t' is an alternate for count_time and others
        if count_time is None:
            # assume regular spacing in time
            shape = self.map.scannables_shape()
            time_per_point = duration.total_seconds() / np.prod(*shape)
            count_time = np.full(shape, time_per_point)
        return np.array([start + datetime.timedelta(seconds=x) for x in np.cumsum(count_time)])

    def replace_default_names(self, expression: str) -> str:
        """Replace 'axes', 'signal' and 'IMAGE' in expression with correct values"""
        axes_names, signal_names = self.map.nexus_default_names()
        detector_names = list(self.map.image_data)
        # Replace axes instances
        expression = re.sub(
            r'axes(\d*)',
            lambda m: list(axes_names)[int(m.group(1) or 0)],
            expression
        )
        # Replace signal instances
        expression = re.sub(
            r'signal(\d*)',
            lambda m: list(signal_names)[int(m.group(1) or 0)],
            expression
        )
        # Replace IMAGE instances
        expression = re.sub(
            r'IMAGE(\d*)',
            lambda m: list(detector_names)[int(m.group(1) or 0)],
            expression
        )
        # Replace other HdfMap names
        replace_names = {name: self.map.generate_ids(name)[0] for name in self.map if name in expression}
        expression = multiple_replace(replace_names, expression)
        return expression

    def _get_plot_axis(self, hdf: h5py.File | h5py.Group, axis_name: str,
                       reduce_shape: bool = True, flatten: bool = False) -> tuple[np.ndarray, str]:
        """
        Return plot axis data and label for given axis name

            data, label = scan.get_plot_axis('axes', flatten=True)

        :param hdf: h5py.File or h5py.Group
        :param axis_name: axis name as given in self.map
        :param reduce_shape: reduces shape (summing additional axes) of >2D arrays to self.map.scannables_shape
        :param flatten: flatten data array if True
        :return: (data, label) tuple
        """
        # Default scannables if not generated by hdfmap
        label = axis_name = self.replace_default_names(axis_name)
        data = self.map.eval(hdf, axis_name, local_data=self._local_data)
        if np.ndim(data) > 1 and reduce_shape:
            # reduce high dimensional arrays to the default scannable shape
            shape = self.map.scannables_shape()
            if np.ndim(data) == len(shape) + 2:
                # Image data
                data = np.sum(data, axis=(-1, -2))
            if np.shape(data) != shape:
                raise ValueError(f"2+D Arrays must have same shape: {axis_name}{np.shape(data)} != {shape}")
        if flatten:
            data = np.reshape(data, -1)
        return data, label

    def get_plot_axis(self, axis_name: str, reduce_shape: bool = True, flatten: bool = False) -> tuple[np.ndarray, str]:
        """
        Return plot axis data and label for given axis name

            data, label = scan.get_plot_axis('axes', flatten=True)

        :param axis_name: axis name as given in self.map
        :param reduce_shape: reduces shape (summing additional axes) of >2D arrays to self.map.scannables_shape
        :param flatten: flattens output if True
        :return: (data, label) tuple
        """
        with self.load_hdf() as hdf:
            return self._get_plot_axis(hdf, axis_name, reduce_shape=reduce_shape, flatten=flatten)

    def get_plot_data(self, x_axis: str | None = None, *y_axis: str | None, z_axis: str | None = None) -> dict:
        """
        Return dict of plottable data

            data = scan.get_plot_data('axes', 'signal')
            plt.plot(data['x'], data['y'])
            plt.xlabel(data['xlabel'])
            plt.ylabel(data['ylabel'])
            plt.title(data['title'])
            plt.legend(data['legend'])

        :param x_axis: axis name or expression as given in self.map
        :param y_axis: axis name or expression as given in self.map
        :param z_axis: axis name or expression as given in self.map
        :returns: {
            'xlabel': str label of first axes
            'ylabel': str label of first signal
            'xdata': flattened array of first axes
            'ydata': flattened array of first signal
            'axes_names': list of axes names,
            'signal_names': list of signal + auxiliary signal names,
            'axes_data': list of ND arrays of data for axes,
            'signal_data': list of ND array of data for signal + auxiliary signals,
            'axes_labels': list of axes labels as 'name [units]',
            'signal_labels': list of signal labels,
            'data': dict of all scannables axes,
            'title': str title as 'filename\nNXtitle'
        if dataset is a 2D grid scan, additional rows:
            'grid_xlabel': str label of grid x-axis
            'grid_ylabel': str label of grid y-axis
            'grid_label': str label of height or colour
            'grid_xdata': 2D array of x-coordinates
            'grid_ydata': 2D array of y-coordinates
            'grid_data': 2D array of height or colour
        }
        """
        with self.load_hdf() as hdf:
            data = self.map.get_plot_data(hdf)
            cmd = self.map.eval(hdf, Md.cmd)
            if len(cmd) > self.MAX_STR_LEN:
                cmd = shorten_string(cmd)
            x_data, x_lab = self._get_plot_axis(hdf, x_axis or 'axes', reduce_shape=True, flatten=True)

            y_data, y_labs = [], []
            y_axis = y_axis or [None]
            for n, _y_axis in enumerate(y_axis):
                _y_data, _y_lab = self._get_plot_axis(hdf, _y_axis or f"signal{n}", reduce_shape=True, flatten=True)
                y_data.append(_y_data)
                y_labs.append(_y_lab)
            y_data = np.array(y_data)
            y_error = self._error_function(y_data)

            additional = {
                'x': x_data,
                'y': y_data[0],
                'xdata': x_data,
                'ydata': y_data.T,
                'yerror': y_error.T,
                'xlabel': x_lab,
                'ylabel': y_labs[0],
                'title': f"#{self.scan_number()}\n{cmd}",
                'legend': y_labs
            }
            if z_axis is not None:
                z_data, z_lab = self._get_plot_axis(hdf, z_axis, reduce_shape=True, flatten=True)
                additional['zdata'] = z_data
                additional['zlabel'] = z_lab
            # 2+D data
            shape = self.map.scannables_shape()
            if len(shape) >= 2:
                x_data = x_data.reshape(shape)
                y_data, y_lab = self._get_plot_axis(hdf, y_axis[0] or 'axes1')
                z_data, z_lab = self._get_plot_axis(hdf,
                    y_axis[1] if len(y_axis) > 1 else z_axis or 'signal',
                )
                # reduce dimensions to 2
                #TODO: taking the first dimension might not always be right
                x_data = x_data[(..., ) + (0, ) * (x_data.ndim - 2)]
                y_data = y_data[(..., ) + (0, ) * (y_data.ndim - 2)]
                z_data = z_data.sum(axis=tuple(range(2, z_data.ndim)))

                if y_data.shape != x_data.shape:
                    raise ValueError(f"Shape of '{y_lab}' {y_data.shape} != '{x_lab}' {x_data.shape}")
                if z_data.shape != x_data.shape:
                    raise ValueError(f"Shape of '{z_lab}' {z_data.shape} != '{x_lab}' {x_data.shape}")

                additional['grid_xlabel'] = x_lab
                additional['grid_ylabel'] = y_lab
                additional['grid_label'] = z_lab
                additional['grid_xdata'] = x_data
                additional['grid_ydata'] = y_data
                additional['grid_data'] = z_data
            data.update(additional)
            return data

    def xas_spectra(self, sample_name: str | None = None, element_edge: str | None = None, mode: str | list[str] = 'all',
                    dls_loader: bool = False) -> SpectraContainer:
        """
        Load XAS Spectra from the scan file

            >>> spectra = scan.xas_spectra()
            >>> spectra = spectra.remove_background('slope')
            >>> spectra.plot()

        :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
        :return: SpectraContainer
        """
        return load_xas_scans(self.filename, sample_name=sample_name, element_edge=element_edge,
                              mode=mode, dls_loader=dls_loader, hdfmap=self.map)[0]

    def instrument_model(self) -> NXInstrumentModel:
        """Build and instrument model from NeXus file"""
        with self.load_hdf() as hdf:
            return NXInstrumentModel(hdf)

    def save(self, filename: str):
        """Save object as HDF5 file"""
        from mmg_toolbox import __version__
        date = str(datetime.datetime.now())

        # Write new file
        with h5py.File(filename, 'w') as f:
            nw.add_entry_links(f, self.filename)
            entry = nw.add_nxentry(f, 'NexusScan', default=False)
            # Filename
            nw.add_nxfield(entry, 'filename', self.filename)
            # LocalData
            nw.add_nxprocess(
                root=entry,
                name='local_data',
                program='mmg_toolbox.nexus.nexus_scan',
                version=__version__,
                date=date,
                **self._local_data  # save local_data as NXparameters
            )
            nw.add_nxnote(
                root=entry,
                name='config',
                description='NexusScan configuration',
                data=self.config,
            )

    def load_local_data(self):
        """If the scan file contains previously saved NexusScan information, the local data and config will be loaded."""

        with self.load_hdf() as hdf:
            orig_filename, local_data, config = _load_local_data(hdf)

        self.__init__(orig_filename, None, config=config)
        self.add_local(**local_data)

    def save_nxdata(self, root: h5py.Group, name: str = 'data', x_axis: str | None = None, *y_axis: str | None,
                    z_axis: str | None = None, default: bool = False):
        """Save scan data as NXdata group"""
        plot_data = self.get_plot_data(x_axis, *y_axis, z_axis=z_axis)
        nx_data = nw.add_nxdata(
            root=root,
            name=name,
            axes=plot_data['axes_names'],
            signal=plot_data['signal_names'][0],
            default=default
        )
        for name, data in zip(plot_data['axes_names'], plot_data['axes_data']):
            nw.add_nxfield(nx_data, name, data)
        for name, data in zip(plot_data['signal_names'], plot_data['signal_data']):
            nw.add_nxfield(nx_data, name, data)

    def save_csv(self, filename: str, *scannables: str):
        """Save axes and signal as CSV file"""
        # Load scannables data
        names = (self.replace_default_names(scannable) for scannable in scannables)
        header = '# mmg_toolbox.nexus.nexus_scan\n'
        header += '# ' + ', '.join(names)
        scannable_arrays = self.eval(','.join(scannables))
        array_len = max(len(array) for array in scannable_arrays)
        scannable_arrays = [
            np.tile(array, array_len) if array.size == 1 else array
            for array in scannable_arrays
        ]
        with open(filename, 'w') as f:
            f.write(header)
            f.write('\n')
            for values in zip(*scannable_arrays):
                f.write(', '.join(map(str, values)))
                f.write('\n')

arrays(*args, units='', default=np.array([np.nan]))

Return Numpy arrays

Source code in mmg_toolbox/nexus/nexus_scan.py

def arrays(self, *args, units: str = '', default: np.ndarray = np.array([np.nan])) -> list[np.ndarray]:
    """Return Numpy arrays"""
    with self.load_hdf() as hdf:
        return [
            get_dataset_value(self.map.combined[name], hdf, units=units, default=default)
            for name in args
        ]

datasets(*args)

Return HDF5 datasets from NeXus file (leaves file in open state)

Source code in mmg_toolbox/nexus/nexus_scan.py

def datasets(self, *args) -> list[h5py.Dataset]:
    """Return HDF5 datasets from NeXus file (leaves file in open state)"""
    with self.load_hdf() as hdf:
        return [hdf[self.map.combined[name]] for name in args]

get_max_index(expression='IMAGE')

Return the index and value of the maximum pixel in a detector image or volume

Source code in mmg_toolbox/nexus/nexus_scan.py

def get_max_index(self, expression: str = 'IMAGE') -> tuple[float, tuple[int, ...]]:
    """Return the index and value of the maximum pixel in a detector image or volume"""
    data = self.eval(expression)
    max_idx = tuple(int(v) for v in np.unravel_index(np.argmax(data), data.shape))
    return float(data[max_idx]), max_idx

get_plot_axis(axis_name, reduce_shape=True, flatten=False)

Return plot axis data and label for given axis name

data, label = scan.get_plot_axis('axes', flatten=True)

Parameters:

Name	Type	Description	Default
axis_name	str	axis name as given in self.map	required
reduce_shape	bool	reduces shape (summing additional axes) of >2D arrays to self.map.scannables_shape	True
flatten	bool	flattens output if True	False

Returns:

Type	Description
tuple[ndarray, str]	(data, label) tuple

Source code in mmg_toolbox/nexus/nexus_scan.py

def get_plot_axis(self, axis_name: str, reduce_shape: bool = True, flatten: bool = False) -> tuple[np.ndarray, str]:
    """
    Return plot axis data and label for given axis name

        data, label = scan.get_plot_axis('axes', flatten=True)

    :param axis_name: axis name as given in self.map
    :param reduce_shape: reduces shape (summing additional axes) of >2D arrays to self.map.scannables_shape
    :param flatten: flattens output if True
    :return: (data, label) tuple
    """
    with self.load_hdf() as hdf:
        return self._get_plot_axis(hdf, axis_name, reduce_shape=reduce_shape, flatten=flatten)

get_plot_data(x_axis=None, *y_axis, z_axis=None)

    Return dict of plottable data

        data = scan.get_plot_data('axes', 'signal')
        plt.plot(data['x'], data['y'])
        plt.xlabel(data['xlabel'])
        plt.ylabel(data['ylabel'])
        plt.title(data['title'])
        plt.legend(data['legend'])

    :param x_axis: axis name or expression as given in self.map
    :param y_axis: axis name or expression as given in self.map
    :param z_axis: axis name or expression as given in self.map
    :returns: {
        'xlabel': str label of first axes
        'ylabel': str label of first signal
        'xdata': flattened array of first axes
        'ydata': flattened array of first signal
        'axes_names': list of axes names,
        'signal_names': list of signal + auxiliary signal names,
        'axes_data': list of ND arrays of data for axes,
        'signal_data': list of ND array of data for signal + auxiliary signals,
        'axes_labels': list of axes labels as 'name [units]',
        'signal_labels': list of signal labels,
        'data': dict of all scannables axes,
        'title': str title as 'filename

NXtitle' if dataset is a 2D grid scan, additional rows: 'grid_xlabel': str label of grid x-axis 'grid_ylabel': str label of grid y-axis 'grid_label': str label of height or colour 'grid_xdata': 2D array of x-coordinates 'grid_ydata': 2D array of y-coordinates 'grid_data': 2D array of height or colour }

Source code in mmg_toolbox/nexus/nexus_scan.py

def get_plot_data(self, x_axis: str | None = None, *y_axis: str | None, z_axis: str | None = None) -> dict:
    """
    Return dict of plottable data

        data = scan.get_plot_data('axes', 'signal')
        plt.plot(data['x'], data['y'])
        plt.xlabel(data['xlabel'])
        plt.ylabel(data['ylabel'])
        plt.title(data['title'])
        plt.legend(data['legend'])

    :param x_axis: axis name or expression as given in self.map
    :param y_axis: axis name or expression as given in self.map
    :param z_axis: axis name or expression as given in self.map
    :returns: {
        'xlabel': str label of first axes
        'ylabel': str label of first signal
        'xdata': flattened array of first axes
        'ydata': flattened array of first signal
        'axes_names': list of axes names,
        'signal_names': list of signal + auxiliary signal names,
        'axes_data': list of ND arrays of data for axes,
        'signal_data': list of ND array of data for signal + auxiliary signals,
        'axes_labels': list of axes labels as 'name [units]',
        'signal_labels': list of signal labels,
        'data': dict of all scannables axes,
        'title': str title as 'filename\nNXtitle'
    if dataset is a 2D grid scan, additional rows:
        'grid_xlabel': str label of grid x-axis
        'grid_ylabel': str label of grid y-axis
        'grid_label': str label of height or colour
        'grid_xdata': 2D array of x-coordinates
        'grid_ydata': 2D array of y-coordinates
        'grid_data': 2D array of height or colour
    }
    """
    with self.load_hdf() as hdf:
        data = self.map.get_plot_data(hdf)
        cmd = self.map.eval(hdf, Md.cmd)
        if len(cmd) > self.MAX_STR_LEN:
            cmd = shorten_string(cmd)
        x_data, x_lab = self._get_plot_axis(hdf, x_axis or 'axes', reduce_shape=True, flatten=True)

        y_data, y_labs = [], []
        y_axis = y_axis or [None]
        for n, _y_axis in enumerate(y_axis):
            _y_data, _y_lab = self._get_plot_axis(hdf, _y_axis or f"signal{n}", reduce_shape=True, flatten=True)
            y_data.append(_y_data)
            y_labs.append(_y_lab)
        y_data = np.array(y_data)
        y_error = self._error_function(y_data)

        additional = {
            'x': x_data,
            'y': y_data[0],
            'xdata': x_data,
            'ydata': y_data.T,
            'yerror': y_error.T,
            'xlabel': x_lab,
            'ylabel': y_labs[0],
            'title': f"#{self.scan_number()}\n{cmd}",
            'legend': y_labs
        }
        if z_axis is not None:
            z_data, z_lab = self._get_plot_axis(hdf, z_axis, reduce_shape=True, flatten=True)
            additional['zdata'] = z_data
            additional['zlabel'] = z_lab
        # 2+D data
        shape = self.map.scannables_shape()
        if len(shape) >= 2:
            x_data = x_data.reshape(shape)
            y_data, y_lab = self._get_plot_axis(hdf, y_axis[0] or 'axes1')
            z_data, z_lab = self._get_plot_axis(hdf,
                y_axis[1] if len(y_axis) > 1 else z_axis or 'signal',
            )
            # reduce dimensions to 2
            #TODO: taking the first dimension might not always be right
            x_data = x_data[(..., ) + (0, ) * (x_data.ndim - 2)]
            y_data = y_data[(..., ) + (0, ) * (y_data.ndim - 2)]
            z_data = z_data.sum(axis=tuple(range(2, z_data.ndim)))

            if y_data.shape != x_data.shape:
                raise ValueError(f"Shape of '{y_lab}' {y_data.shape} != '{x_lab}' {x_data.shape}")
            if z_data.shape != x_data.shape:
                raise ValueError(f"Shape of '{z_lab}' {z_data.shape} != '{x_lab}' {x_data.shape}")

            additional['grid_xlabel'] = x_lab
            additional['grid_ylabel'] = y_lab
            additional['grid_label'] = z_lab
            additional['grid_xdata'] = x_data
            additional['grid_ydata'] = y_data
            additional['grid_data'] = z_data
        data.update(additional)
        return data

get_scan_time()

Return array of datetimes for each scan point

Source code in mmg_toolbox/nexus/nexus_scan.py

def get_scan_time(self) -> np.ndarray:
    """Return array of datetimes for each scan point"""
    start, stop, duration = self.start_end_duration()
    count_time = self.eval('_t', default=None)  # '_t' is an alternate for count_time and others
    if count_time is None:
        # assume regular spacing in time
        shape = self.map.scannables_shape()
        time_per_point = duration.total_seconds() / np.prod(*shape)
        count_time = np.full(shape, time_per_point)
    return np.array([start + datetime.timedelta(seconds=x) for x in np.cumsum(count_time)])

hdf_find(*field_or_class, find_all=False)

Find datasets and groups within hdf file

HdfDataset = scan.hdf_find('NXentry', ['NXdata', 'measurement'], 'signal')

Warning: HDF file stays in open state while Dataset or Group objects exist.

Parameters:

Name	Type	Description	Default
field_or_class	str | list[str]	names to search for, in hierarchical order. Lists treated as OR	()
find_all	bool	whether to return all datasets or only the first match	False

Returns:

Type	Description
Dataset | Group | list[Dataset | Group]	matching Dataset or Group, or list of matching Datasets or Groups

Source code in mmg_toolbox/nexus/nexus_scan.py

def hdf_find(self, *field_or_class: str | list[str], find_all: bool = False) -> (h5py.Dataset | h5py.Group) | list[h5py.Dataset | h5py.Group]:
    """
    Find datasets and groups within hdf file

        HdfDataset = scan.hdf_find('NXentry', ['NXdata', 'measurement'], 'signal')

    Warning: HDF file stays in open state while Dataset or Group objects exist.

    :param field_or_class: names to search for, in hierarchical order. Lists treated as OR
    :param find_all: whether to return all datasets or only the first match
    :returns: matching Dataset or Group, or list of matching Datasets or Groups
    """
    with self.load_hdf() as hdf:
        if find_all:
            return nx_find_all(hdf, *field_or_class)
        return nx_find(hdf, *field_or_class)

hdf_tree_string(group='/', all_links=True, attributes=True)

Generate string of the hdf file structure, similar to h5ls. Uses h5py.visititems

Parameters:

Name	Type	Description	Default
all_links	bool	bool, if True, also show links	True
group	str	only display tree structure of this group (default root)	'/'
attributes	bool	if True, display the attributes of groups and datasets	True

Returns:

Type	Description
str	str

Source code in mmg_toolbox/nexus/nexus_scan.py

def hdf_tree_string(self, group: str = '/', all_links: bool = True, attributes: bool = True) -> str:
    """
    Generate string of the hdf file structure, similar to h5ls. Uses h5py.visititems

    :param all_links: bool, if True, also show links
    :param group: only display tree structure of this group (default root)
    :param attributes: if True, display the attributes of groups and datasets
    :return: str
    """
    return hdfmap.hdf_tree_string(self.filename, all_links=all_links, group=group, attributes=attributes)

image(index=None)

Return image or selection from default detector

im = scan.image(0)  # First image
im = scan.image(None)  # middle image
im = scan.image('sum')  # sum of all images

Parameters:

Name	Type	Description	Default
index	int | str | None	point in the scan to show the image, 'sum' gives sum over all images, None gives the middle image	None

Returns:

Type	Description
ndarray	(n,m) array

Source code in mmg_toolbox/nexus/nexus_scan.py

def image(self, index: int | str | None = None) -> np.ndarray:
    """
    Return image or selection from default detector

        im = scan.image(0)  # First image
        im = scan.image(None)  # middle image
        im = scan.image('sum')  # sum of all images

    :param index: point in the scan to show the image, 'sum' gives sum over all images, None gives the middle image
    :return: (n,m) array
    """
    if not self.map.image_data:
        raise ValueError(f'{repr(self)} contains no image data')
    with self.load_hdf() as hdf:
        if index == 'sum':
            volume = self.map.get_image(hdf, ())
            image = volume.sum(axis=tuple(range(volume.ndim-2)))  # this will fail for filenames
        else:
            image = self.map.get_image(hdf, index)

        if issubclass(type(image), str):
            # TIFF image, NXdetector/image_data -> array('file.tif')
            file_directory = os.path.dirname(self.filename)
            image_filename = os.path.join(file_directory, image)
            if not os.path.isfile(image_filename):
                raise FileNotFoundError(f"File not found: {image_filename}")
            image = read_tiff(image_filename)
        elif image.ndim == 0:
            # image is file path number, NXdetector/path -> arange(n_points)
            scan_number = get_scan_number(self.filename)
            file_directory = os.path.dirname(self.filename)
            detector_names = list(self.map.image_data.keys())
            for detector_name in detector_names:
                image_filename = os.path.join(file_directory, f"{scan_number}-{detector_name}-files/{image:05.0f}.tif")
                if os.path.isfile(image_filename):
                    break
            if not os.path.isfile(image_filename):
                raise FileNotFoundError(f"File not found: {image_filename}")
            image = read_tiff(image_filename)
        elif image.ndim != 2:
            raise Exception(f"detector image[{index}] is the wrong shape: {image.shape}")
        return image

image_background(index=(), n_bins=100)

Return the modal value of the detector image, which usually gives the background value.

The modal value is determined by histograming the image (or image stack) and taking the value of the largest bin.

Parameters:

Name	Type	Description	Default
index	int | tuple | slice | str | None	index of image to return, use () for full image stack.	()
n_bins	int	number of histogram bins	100

Returns:

Type	Description
ndarray	modal value or values per image

Source code in mmg_toolbox/nexus/nexus_scan.py

def image_background(self, index: int | tuple | slice | str | None = (), n_bins: int  = 100) -> np.ndarray:
    """
    Return the modal value of the detector image,
    which usually gives the background value.

    The modal value is determined by histograming the image (or image stack) and taking
    the value of the largest bin.

    :param index: index of image to return, use () for full image stack.
    :param n_bins: number of histogram bins
    :return: modal value or values per image
    """
    image = self.map.get_image(self.load_hdf(), index)  # hdf data only
    n, bins = np.histogram(np.log10(image[image>0].flatten()), bins=n_bins)
    return 10 ** bins[np.argmax(n)]

info(arrays=False, values=False, combined=False, metadata=False, scannables=True, image_data=True, local=False, alternate=True)

Return string of namespace information

Source code in mmg_toolbox/nexus/nexus_scan.py

def info(self, arrays=False, values=False, combined=False,
         metadata=False, scannables=True, image_data=True,
         local=False, alternate=True) -> str:
    """Return string of namespace information"""
    local_data = (
            "Local Data:\n" +
            "\n".join(
                f"  {name}: {value}" for name, value in self._local_data.items()
            ) +
            "\n"
    ) if local else ""
    alternate_data = (
        "Alternate Names:\n  Name: Expression" +
        "\n".join(
            f"  {name}: '{expr}'" for name, expr in self.map._alternate_names.items()
        ) +
        "\n"
    ) if alternate else ""
    map_data = self.map.info_names(arrays=arrays, values=values, combined=combined,
                                   metadata=metadata, scannables=scannables, image_data=image_data)
    return local_data + alternate_data + map_data

instrument_model()

Build and instrument model from NeXus file

Source code in mmg_toolbox/nexus/nexus_scan.py

def instrument_model(self) -> NXInstrumentModel:
    """Build and instrument model from NeXus file"""
    with self.load_hdf() as hdf:
        return NXInstrumentModel(hdf)

load_hdf()

Load the Hdf file

Source code in mmg_toolbox/nexus/nexus_scan.py

def load_hdf(self) -> h5py.File:
    """Load the Hdf file"""
    return load_hdf(self.filename)

load_local_data()

If the scan file contains previously saved NexusScan information, the local data and config will be loaded.

Source code in mmg_toolbox/nexus/nexus_scan.py

def load_local_data(self):
    """If the scan file contains previously saved NexusScan information, the local data and config will be loaded."""

    with self.load_hdf() as hdf:
        orig_filename, local_data, config = _load_local_data(hdf)

    self.__init__(orig_filename, None, config=config)
    self.add_local(**local_data)

metadata_str(expression=None)

Generate metadata string from beamline config

Source code in mmg_toolbox/nexus/nexus_scan.py

def metadata_str(self, expression: str | None = None):
    """Generate metadata string from beamline config"""
    if expression is None:
        expression = self.config.get(C.metadata_string, '')
    return self.format(expression)

replace_default_names(expression)

Replace 'axes', 'signal' and 'IMAGE' in expression with correct values

Source code in mmg_toolbox/nexus/nexus_scan.py

def replace_default_names(self, expression: str) -> str:
    """Replace 'axes', 'signal' and 'IMAGE' in expression with correct values"""
    axes_names, signal_names = self.map.nexus_default_names()
    detector_names = list(self.map.image_data)
    # Replace axes instances
    expression = re.sub(
        r'axes(\d*)',
        lambda m: list(axes_names)[int(m.group(1) or 0)],
        expression
    )
    # Replace signal instances
    expression = re.sub(
        r'signal(\d*)',
        lambda m: list(signal_names)[int(m.group(1) or 0)],
        expression
    )
    # Replace IMAGE instances
    expression = re.sub(
        r'IMAGE(\d*)',
        lambda m: list(detector_names)[int(m.group(1) or 0)],
        expression
    )
    # Replace other HdfMap names
    replace_names = {name: self.map.generate_ids(name)[0] for name in self.map if name in expression}
    expression = multiple_replace(replace_names, expression)
    return expression

rois(append='_total')

Return ROI expressions available in scan namespace

rois = [scan.eval(roi) for roi in scan.rois('_total')]

Parameters:

Name	Type	Description	Default
append	str	str to append to each ROI name, e.g. '_total', '_max', '_min', '_mean'	'_total'

Returns:

Type	Description
list[str]	list of ROI names that can be used in eval.

Source code in mmg_toolbox/nexus/nexus_scan.py

def rois(self, append: str = '_total') -> list[str]:
    """
    Return ROI expressions available in scan namespace

        rois = [scan.eval(roi) for roi in scan.rois('_total')]

    :param append: str to append to each ROI name, e.g. '_total', '_max', '_min', '_mean'
    :return: list of ROI names that can be used in eval.
    """
    # search for ROIs in HdfMap expressions
    alternate_name_rois = {
        next((name.removesuffix(sfx) for sfx in ROI_SUFFIXES if name.endswith(sfx)), name)
        for name, expression in self.map._alternate_names.items()
        if expression.startswith('d_')
    }
    return [roi + append for roi in alternate_name_rois]

save(filename)

Save object as HDF5 file

Source code in mmg_toolbox/nexus/nexus_scan.py

def save(self, filename: str):
    """Save object as HDF5 file"""
    from mmg_toolbox import __version__
    date = str(datetime.datetime.now())

    # Write new file
    with h5py.File(filename, 'w') as f:
        nw.add_entry_links(f, self.filename)
        entry = nw.add_nxentry(f, 'NexusScan', default=False)
        # Filename
        nw.add_nxfield(entry, 'filename', self.filename)
        # LocalData
        nw.add_nxprocess(
            root=entry,
            name='local_data',
            program='mmg_toolbox.nexus.nexus_scan',
            version=__version__,
            date=date,
            **self._local_data  # save local_data as NXparameters
        )
        nw.add_nxnote(
            root=entry,
            name='config',
            description='NexusScan configuration',
            data=self.config,
        )

save_csv(filename, *scannables)

Save axes and signal as CSV file

Source code in mmg_toolbox/nexus/nexus_scan.py

def save_csv(self, filename: str, *scannables: str):
    """Save axes and signal as CSV file"""
    # Load scannables data
    names = (self.replace_default_names(scannable) for scannable in scannables)
    header = '# mmg_toolbox.nexus.nexus_scan\n'
    header += '# ' + ', '.join(names)
    scannable_arrays = self.eval(','.join(scannables))
    array_len = max(len(array) for array in scannable_arrays)
    scannable_arrays = [
        np.tile(array, array_len) if array.size == 1 else array
        for array in scannable_arrays
    ]
    with open(filename, 'w') as f:
        f.write(header)
        f.write('\n')
        for values in zip(*scannable_arrays):
            f.write(', '.join(map(str, values)))
            f.write('\n')

save_nxdata(root, name='data', x_axis=None, *y_axis, z_axis=None, default=False)

Save scan data as NXdata group

Source code in mmg_toolbox/nexus/nexus_scan.py

def save_nxdata(self, root: h5py.Group, name: str = 'data', x_axis: str | None = None, *y_axis: str | None,
                z_axis: str | None = None, default: bool = False):
    """Save scan data as NXdata group"""
    plot_data = self.get_plot_data(x_axis, *y_axis, z_axis=z_axis)
    nx_data = nw.add_nxdata(
        root=root,
        name=name,
        axes=plot_data['axes_names'],
        signal=plot_data['signal_names'][0],
        default=default
    )
    for name, data in zip(plot_data['axes_names'], plot_data['axes_data']):
        nw.add_nxfield(nx_data, name, data)
    for name, data in zip(plot_data['signal_names'], plot_data['signal_data']):
        nw.add_nxfield(nx_data, name, data)

start_end_duration()

Return start and end times of scan, plus duration as datetime and timedelta objects

Source code in mmg_toolbox/nexus/nexus_scan.py

def start_end_duration(self) -> tuple[datetime.datetime, datetime.datetime, datetime.timedelta]:
    """Return start and end times of scan, plus duration as datetime and timedelta objects"""
    scan_start, scan_end = self.times(NX_START, NX_END, default=None)
    if scan_start is None:
        raise KeyError(f"scan start time not found in {self.filename}")
    if scan_end is None:
        scan_end = get_file_time(self.filename)
    return scan_start, scan_end, scan_end - scan_start

strings(*args, units=False)

Return string value

Source code in mmg_toolbox/nexus/nexus_scan.py

def strings(self, *args, units=False) -> list[str]:
    """Return string value"""
    with self.load_hdf() as hdf:
        return [dataset2str(hdf[self.map.combined[name]], units=units) for name in args]

table(delimiter=', ', string_spec='', format_spec='f', default_decimals=8)

Return data table

Source code in mmg_toolbox/nexus/nexus_scan.py

def table(self, delimiter=', ', string_spec='', format_spec='f', default_decimals=8) -> str:
    """Return data table"""
    with self.load_hdf() as hdf:
        return self.map.create_scannables_table(hdf, delimiter, string_spec, format_spec, default_decimals)

times(*args, default=None)

Return datetime object

Source code in mmg_toolbox/nexus/nexus_scan.py

def times(self, *args, default: datetime.datetime | None = None) -> list[datetime.datetime | None]:
    """Return datetime object"""
    with self.load_hdf() as hdf:
        data = [
            dataset2data(hdf[self.map.combined[name]])
            if name in self.map.combined else default
            for name in args
        ]
        dt = [
            obj if obj is None or isinstance(obj, datetime.datetime) else
            datetime.datetime.fromisoformat(obj) if isinstance(obj, str) else
            datetime.datetime.fromtimestamp(float(obj))
            for obj in data
        ]
    return dt

values(*args, value_func=np.mean, units='', default=np.array(np.nan))

Return float values

Source code in mmg_toolbox/nexus/nexus_scan.py

def values(self, *args, value_func=np.mean,
           units: str = '', default: np.ndarray = np.array(np.nan)) -> list[np.floating]:
    """Return float values"""
    with self.load_hdf() as hdf:
        return [
            value_func(get_dataset_value(self.map.combined[name], hdf, units=units, default=default))
            for name in args
        ]

volume()

Return complete stack of images

Source code in mmg_toolbox/nexus/nexus_scan.py

def volume(self) -> np.ndarray:
    """Return complete stack of images"""
    return self.map.get_image(self.load_hdf(), ())

xas_spectra(sample_name=None, element_edge=None, mode='all', dls_loader=False)

Load XAS Spectra from the scan file

>>> spectra = scan.xas_spectra()
>>> spectra = spectra.remove_background('slope')
>>> spectra.plot()

Parameters:

Name	Type	Description	Default
sample_name	str | None	sample name, e.g. 'sample1' or None to load from NeXus file	None
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

Returns:

Type	Description
SpectraContainer	SpectraContainer

Source code in mmg_toolbox/nexus/nexus_scan.py

def xas_spectra(self, sample_name: str | None = None, element_edge: str | None = None, mode: str | list[str] = 'all',
                dls_loader: bool = False) -> SpectraContainer:
    """
    Load XAS Spectra from the scan file

        >>> spectra = scan.xas_spectra()
        >>> spectra = spectra.remove_background('slope')
        >>> spectra.plot()

    :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
    :return: SpectraContainer
    """
    return load_xas_scans(self.filename, sample_name=sample_name, element_edge=element_edge,
                          mode=mode, dls_loader=dls_loader, hdfmap=self.map)[0]

create_notebooks(directory=None, beamline=None, start_jupyter=None, prefix='', quiet=False)

Create notebooks at start of experiment

Parameters:

Name	Type	Description	Default
directory	str | None	directory where to create notebooks, or visit ID for current beamline	None
beamline	str | None	Beamline name, e.g. i06-1	None
start_jupyter	bool | None	start jupyter lab	None
prefix	str	prefix for notebook/ script filenames	''
quiet	bool	quiet mode (no user inputs)	False

Source code in mmg_toolbox/scripts/experiment_startup.py

def create_notebooks(directory: str | None = None, beamline: str | None = None,
                     start_jupyter: bool | None = None, prefix: str = '', quiet: bool = False):
    """
    Create notebooks at start of experiment

    :param directory: directory where to create notebooks, or visit ID for current beamline
    :param beamline: Beamline name, e.g. i06-1
    :param start_jupyter: start jupyter lab
    :param prefix: prefix for notebook/ script filenames
    :param quiet: quiet mode (no user inputs)
    """
    print("Creating experiment notebooks for mmg_toolbox")
    print(f"Directory: {directory}, Beamline: {beamline}")
    # Beamline
    if beamline is None:
        if directory is None and not quiet:
            beamline = input("Which beamline? e.g. i06-1:")
        else:
            beamline = get_beamline_from_directory(directory or '', default='')

    # Visits (from beamline)
    visits = get_dls_visits(beamline)
    current_visit = next(iter(visits.keys()), os.getcwd())
    if directory is None and not quiet:
        if visits:
            print(f"Visits for beamline {beamline}:\n  " + "\n  ".join(visits))
        directory = input(f"Which visit or directory? [default: {current_visit}]:")
    if not directory:
        directory = current_visit
    if directory in visits:
        directory = visits[directory]

    # Directory (from visits)
    directory = os.path.abspath(directory)
    if not os.path.isdir(directory):
        raise FileNotFoundError(f"Directory {directory} does not exist")
    processing_directory = get_processing_directory(directory)
    if not quiet and os.path.isdir(get_processing_directory(directory)):
        use_processing = input("Use the processing directory? [Y/n]:")
        if use_processing.lower() == 'y':
            directory = processing_directory

    # Beamline (again, from directory)
    if not beamline:
        beamline = get_beamline_from_directory(directory, default='')

    print(f"Creating notebooks for {beamline} in {directory}")
    title = f"{beamline} {os.path.basename(directory)} mmg_toolbox Example Script"
    create_beamline_scripts(beamline, directory, prefix=prefix, title=title)

    # Start Jupyter
    if not quiet and start_jupyter is None:
        start_jupyter = input("Would you like to start Jupyter? [Y/n]:")
        start_jupyter = start_jupyter.lower() == 'y'
    if start_jupyter:
        open_jupyter_lab(directory)

data_file_reader(filename, beamline=None)

Read Nexus or dat file as DataHolder

Source code in mmg_toolbox/utils/file_reader.py

def data_file_reader(filename: str, beamline: str | None = None) -> NexusDataHolder | DataHolder:
    """
    Read Nexus or dat file as DataHolder
    """
    beamline = beamline or get_beamline(None, filename=filename)
    if filename.endswith('.dat'):
        return read_dat_file(filename)
    return read_nexus_file(filename, beamline=beamline)

find_scan_files(scan_file, directory=None, instrument=None, extension='.nxs')

Recursively search a directory for scan files, returns a generator

fn = find_scan_files(12345, '/dls/i16/data')
filepath = next(fn, None)  # returns first filepath with '12345' in the title
all_paths = list(fn)

Parameters:

Name	Type	Description	Default
scan_file	str | int	scan number or part of filename	required
directory	str | None	directory to search for scan files, or None to use instrument	None
instrument	str | None	if directory is None, use instrument to use DLS data directory (None to use current beamline)	None
extension	str	extension to search for scan files	'.nxs'

Returns:

Type	Description
Iterator[str]	generator object for search

Source code in mmg_toolbox/utils/env_functions.py

def find_scan_files(scan_file: str | int, directory: str | None = None,
                    instrument: str | None = None, extension: str = '.nxs') -> Iterator[str]:
    """
    Recursively search a directory for scan files, returns a generator

        fn = find_scan_files(12345, '/dls/i16/data')
        filepath = next(fn, None)  # returns first filepath with '12345' in the title
        all_paths = list(fn)

    :param scan_file: scan number or part of filename
    :param directory: directory to search for scan files, or None to use instrument
    :param instrument: if directory is None, use instrument to use DLS data directory (None to use current beamline)
    :param extension: extension to search for scan files
    :return: generator object for search
    """
    scan_file = str(scan_file)
    if directory is None:
        if instrument is None:
            instrument = get_beamline()
        directory = os.path.join(DLS, instrument.lower(), 'data')

    def scantree(path: str):
        for entry in os.scandir(path):
            if os.access(entry.path, os.R_OK) and not entry.name.startswith('.'):
                if entry.is_file() and entry.name.endswith(extension) and scan_file in entry.name:
                    yield entry.path
                elif entry.is_dir():
                    yield from scantree(entry.path)
    return scantree(directory)

get_dls_visits(instrument=None, year=None, max_visits=None, omit_empty=False)

Return dict of {visit: path} for each visible visit in the beamline directory

visits = get_dls_visits('i16')

Parameters:

Name	Type	Description	Default
instrument	str | None	displays visits for this instrument, or None for current beamline.	None
year	str | int | None	displays visits in this year, or None for current year.	None
max_visits	int | None	maximum number of visits to display, nor None for all visits	None
omit_empty	bool	if True, omits visits containing no scans	False

Source code in mmg_toolbox/utils/env_functions.py

def get_dls_visits(instrument: str | None = None, year: str | int | None = None,
                   max_visits: int | None = None, omit_empty: bool = False) -> dict[str, str]:
    """
    Return dict of {visit: path} for each visible visit in the beamline directory

        visits = get_dls_visits('i16')

    :param instrument: displays visits for this instrument, or None for current beamline.
    :param year: displays visits in this year, or None for current year.
    :param max_visits: maximum number of visits to display, nor None for all visits
    :param omit_empty: if True, omits visits containing no scans
    """
    if instrument is None:
        instrument = get_beamline()
    if year is None:
        year = datetime.now().year

    if max_visits:
        visits = get_dls_visits(instrument, year, max_visits=None, omit_empty=omit_empty)
        return {visit: visits[visit] for visit in list(visits)[:max_visits]}

    dls_dir = os.path.join(DLS, instrument.lower(), 'data', str(year))
    if os.path.isdir(dls_dir):
        return {
            os.path.basename(path): path
            for path in sorted(
                (
                    file.path for file in os.scandir(dls_dir)
                    if file.is_dir() and
                       os.access(file.path, os.R_OK) and
                       (contains_filetype(file.path, '.nxs') if omit_empty else True)
                 ),
                key=lambda x: os.path.getmtime(x), reverse=True
            )
        }
    return {}

get_dls_visits_str(instrument=None, year=None, max_visits=3, omit_empty=True)

Return str of visits for each visible visit int he beamline directory

print(get_dls_visits_str('i16', 2025))

Parameters:

Name	Type	Description	Default
instrument	str | None	displays visits for this instrument, or None for current beamline.	None
year	str | int | None	displays visits in this year, or None for current year.	None
max_visits	int | None	maximum number of visits to display, nor None for all visits	3
omit_empty	bool	if True, omits visits containing no scans	True

Returns:

Type	Description
str	multiline string

Source code in mmg_toolbox/utils/env_functions.py

def get_dls_visits_str(instrument: str | None = None, year: str | int | None = None,
                       max_visits: int | None = 3, omit_empty: bool = True) -> str:
    """
    Return str of visits for each visible visit int he beamline directory

        print(get_dls_visits_str('i16', 2025))

    :param instrument: displays visits for this instrument, or None for current beamline.
    :param year: displays visits in this year, or None for current year.
    :param max_visits: maximum number of visits to display, nor None for all visits
    :param omit_empty: if True, omits visits containing no scans
    :return: multiline string
    """
    instrument = instrument or get_beamline()
    year = year or datetime.now().year
    visits = get_dls_visits(instrument, year)
    if not visits:
        return f"No visits for instrument '{instrument}'"

    strings = []
    for visit, path in visits.items():
        scans = scan_number_mapping(path)
        if len(scans) > 0:
            last_scan_number = list(scans)[-1]
            last_scan_path = list(scans.values())[-1]
            last_scan_time = datetime.fromtimestamp(os.path.getmtime(last_scan_path), tz=timezone.utc)
            last_scan = f", #{last_scan_number:<8} ({last_scan_time:%Y-%m-%d %H:%M})"
        else:
            last_scan = ""
            if omit_empty:
                continue
        strings.append(f"{visit:10} Scans: {len(scans):4}{last_scan}")
        if len(strings) == max_visits:
            break
    return f"{instrument} visits in {year}\n" + '\n'.join(strings)

scan_number_mapping(*folders, extension='.nxs')

Build mapping of scan number to scan file

Source code in mmg_toolbox/utils/env_functions.py

def scan_number_mapping(*folders: str, extension: str = '.nxs') -> dict[int, str]:
    """Build mapping of scan number to scan file"""
    mapping = {
        number: filename
        for folder in folders
        for filename in list_files(folder, extension=extension)
        if (number := get_scan_number(filename)) > 0
    }
    return dict(sorted(mapping.items()))