Source code for cosedaUI.h5_metadata_loader

"""Background worker for collecting essential HDF5 metadata."""

from __future__ import annotations

import h5py
import numpy as np
from PyQt6.QtCore import QObject, pyqtSignal

from coseda.nexus.paths import get_mask_dataset
from h5py import SoftLink, ExternalLink



class H5MetadataLoader(QObject):
    """Load lightweight metadata from an HDF5 file without blocking the UI."""

    finished = pyqtSignal(dict)
    failed = pyqtSignal(str)

    def __init__(self, h5_path: str):
        super().__init__()
        self.h5_path = h5_path



    def run(self):
        info = {
            'total_frames': 0,
            'stagepos_exists': False,
            'intensity_exists': False,
            'has_peaks': False,
            'has_center': False,
            'mask': None,
            'has_npeaks': False,
        }
        try:
            with h5py.File(self.h5_path, 'r') as file:
                data_group = file.get('entry/data')
                if data_group is None:
                    raise KeyError("entry/data group missing in HDF5 file.")
                # Resolve images dataset robustly (follow internal links)
                link_obj = data_group.get('images', getlink=True)
                if isinstance(link_obj, SoftLink):
                    target = link_obj.path
                    if target in file:
                        images_ds = file[target]
                    else:
                        raise KeyError(f"entry/data/images soft-links to '{target}', which is missing.")
                elif isinstance(link_obj, ExternalLink):
                    # Internal-only expected; surface a clearer message if external
                    raise KeyError(f"entry/data/images is an external link to '{link_obj.filename}:{link_obj.path}', which is not accessible from this environment.")
                elif link_obj is None:
                    images_ds = None
                else:
                    # Hard link or dataset
                    images_ds = data_group.get('images')

                if images_ds is None:
                    raise KeyError("entry/data/images dataset missing in HDF5 file.")

                try:
                    info['total_frames'] = images_ds.shape[0]
                except Exception as exc:
                    raise KeyError(f"entry/data/images exists but shape could not be read: {exc}")

                info['stagepos_exists'] = all(
                    key in data_group for key in ['stagepos_x_refined', 'stagepos_y']
                )
                info['intensity_exists'] = any(
                    key in data_group for key in ['frame_mean_intensities', 'frame_total_intensities']
                )
                info['has_peaks'] = (
                    'peakXPosRaw' in data_group and 'peakYPosRaw' in data_group
                )
                info['has_center'] = (
                    'center_x' in data_group and 'center_y' in data_group
                )
                info['has_npeaks'] = 'nPeaks' in data_group
                info['bit_depth'] = images_ds.dtype.itemsize * 8
                mask_ds = get_mask_dataset(file)
                if mask_ds is not None:
                    if mask_ds.ndim == 2:
                        info['mask'] = np.asarray(mask_ds[()])
            self.finished.emit(info)
        except Exception as exc:
            self.failed.emit(str(exc))