Source code for coseda.nexus.logs

"""Helpers for dense NXlog views over full-length metadata streams."""

from __future__ import annotations

from typing import Optional, Sequence, Tuple

import h5py
import numpy as np

from coseda.nexus.groups import ensure_nexus_parents

_GONIOMETER_LOGS: Sequence[Tuple[str, str]] = (
    ("stagepos_x", "stage_x_log"),
    ("stagepos_x_refined", "stage_x_refined_log"),
    ("stagepos_y", "stage_y_log"),
    ("stagepos_y_refined", "stage_y_refined_log"),
    ("stagepos_z", "stage_z_log"),
    ("alphatilt", "alpha_tilt_log"),
    ("betatilt", "beta_tilt_log"),
)

_DETECTOR_LOGS: Sequence[Tuple[str, str]] = (
    ("center_x", "beam_center_x_log"),
    ("center_y", "beam_center_y_log"),
    ("det_shift_x_mm", "detector_shift_x_log"),
    ("det_shift_y_mm", "detector_shift_y_log"),
)

_TIME_CANDIDATES = (
    "/entry/data/timestamp_image",
    "/entry/data/timestamp",
    "/entry/data/frame_id",
)


def _find_time_source(h5file: h5py.File, length: int) -> Optional[h5py.Dataset]:
    for path in _TIME_CANDIDATES:
        if path in h5file:
            ds = h5file[path]
            if ds.shape and ds.shape[0] == length:
                return ds
    return None


def _ensure_value_link(log_group: h5py.Group, value_ds: h5py.Dataset) -> None:
    if "value" in log_group:
        try:
            if log_group["value"].id == value_ds.id:
                return
        except Exception:
            pass
        del log_group["value"]
    log_group["value"] = value_ds


def _ensure_time_dataset(log_group: h5py.Group, length: int, source: Optional[h5py.Dataset]) -> None:
    if source is not None:
        if "time" in log_group:
            try:
                if log_group["time"].id == source.id:
                    return
            except Exception:
                pass
            del log_group["time"]
        log_group["time"] = source
        return

    if "time" in log_group:
        ds = log_group["time"]
        if ds.shape == (length,):
            return
        del log_group["time"]

    ds = log_group.create_dataset("time", data=np.arange(length, dtype=np.int64))
    ds.attrs["units"] = "index"
    ds.attrs["long_name"] = "Frame index"




def ensure_dense_logs(h5file: h5py.File) -> None:
    """
    Create NXlog views for dense metadata streams under /entry/instrument.
    """
    ensure_nexus_parents(h5file)

    goniometer = h5file.require_group("entry/instrument/goniometer")
    goniometer.attrs["NX_class"] = "NXgoniometer"

    for dataset_name, log_name in _GONIOMETER_LOGS:
        source_path = f"/entry/data/{dataset_name}"
        if source_path not in h5file:
            continue
        value_ds = h5file[source_path]
        if not value_ds.shape:
            continue
        log_group = goniometer.require_group(log_name)
        log_group.attrs["NX_class"] = "NXlog"
        _ensure_value_link(log_group, value_ds)
        time_source = _find_time_source(h5file, value_ds.shape[0])
        _ensure_time_dataset(log_group, value_ds.shape[0], time_source)

    detector = h5file.require_group("entry/instrument/detector")
    detector.attrs["NX_class"] = "NXdetector"

    for dataset_name, log_name in _DETECTOR_LOGS:
        source_path = f"/entry/data/{dataset_name}"
        if source_path not in h5file:
            continue
        value_ds = h5file[source_path]
        if not value_ds.shape:
            continue
        log_group = detector.require_group(log_name)
        log_group.attrs["NX_class"] = "NXlog"
        _ensure_value_link(log_group, value_ds)
        time_source = _find_time_source(h5file, value_ds.shape[0])
        _ensure_time_dataset(log_group, value_ds.shape[0], time_source)