Source code for ewoksbm29.io.read_mesh

import h5py
import numpy
from silx.io import h5py_utils
from silx.utils.retry import RetryError

from ..models.dahu import MeshScanParameters




@h5py_utils.retry(retry_timeout=10)
def read_scan_parameters(
    scan_file_path: str,
    scan_number: int,
) -> MeshScanParameters:
    positioners = f"{scan_number}.1/instrument/positioners"
    with h5py_utils.File(scan_file_path) as fh:
        positioners = fh[positioners]
        return _infer_mesh_scan_parameters(positioners)



def _infer_mesh_scan_parameters(positioners: h5py.Group) -> MeshScanParameters:
    motors = {}

    for name, ds in positioners.items():
        if len(ds.shape) == 1:
            motors[name] = numpy.asarray(ds[()], dtype=float)

    if len(motors) != 2:
        raise RetryError("A mesh scan must involve exactly two moving positioners.")

    # The slow motor = fewest changes
    changes = {
        name: numpy.count_nonzero(numpy.diff(values)) for name, values in motors.items()
    }
    slow_motor_name = min(changes, key=changes.get)
    fast_motor_name = max(changes, key=changes.get)
    slow = motors[slow_motor_name]
    fast = motors[fast_motor_name]

    # Determine slow motor start/stop/nsteps
    unique_slow = numpy.unique(numpy.round(slow, 6))
    slow_motor_start = float(unique_slow[0])
    slow_motor_stop = float(unique_slow[-1])
    slow_motor_step = len(unique_slow) - 1

    # Determine fast motor start/stop/nsteps
    # Split into segments using changes in slow motor
    slow_breaks = numpy.where(numpy.diff(slow) != 0)[0] + 1
    fast_segments = numpy.split(fast, slow_breaks)

    # Fast motor start/stop
    starts = numpy.array([fast_segment[0] for fast_segment in fast_segments])
    stops = numpy.array([fast_segment[-1] for fast_segment in fast_segments])
    fast_motor_start = float(numpy.median(starts))
    fast_motor_stop = float(numpy.median(stops))
    fast_motor_step = len(fast_segments[0]) - 1

    # Determine zigzag (back'n'forth)
    directions = numpy.sign(stops - starts)
    backnforth = len(set(directions)) > 1

    return MeshScanParameters(
        fast_motor_name=fast_motor_name,
        fast_motor_start=fast_motor_start,
        fast_motor_stop=fast_motor_stop,
        fast_motor_step=fast_motor_step,
        slow_motor_name=slow_motor_name,
        slow_motor_start=slow_motor_start,
        slow_motor_stop=slow_motor_stop,
        slow_motor_step=slow_motor_step,
        backnforth=backnforth,
    )