Source code for cosedaUI.workflow_wizzard

"""Workflow guards to decide which steps are available based on HDF5 contents."""

import h5py
import os
from coseda.nexus.paths import get_mask_dataset



def has_peak_dataset(h5file_path):
    """Return True if the HDF5 file has a non-empty peak dataset."""
    if not h5file_path or not os.path.exists(h5file_path):
        return False
    try:
        with h5py.File(h5file_path, 'r') as f:
            # Accept any of these as a sign of peak data
            possible = [
                "entry/data/peakXPosRaw",
                "entry/data/peakpos_x",
                "entry/data/peak_x",
            ]
            for dset in possible:
                if dset in f and f[dset].shape[0] > 0:
                    return True
    except Exception:
        pass
    return False




def has_detector_shift(h5file_path):
    """Return True if the HDF5 file has detector shift/center datasets."""
    if not h5file_path or not os.path.exists(h5file_path):
        return False
    try:
        with h5py.File(h5file_path, 'r') as f:
            # Detector shift is typically in these
            if "entry/data/center_x" in f and "entry/data/center_y" in f:
                if f["entry/data/center_x"].shape[0] > 0 and f["entry/data/center_y"].shape[0] > 0:
                    return True
            # Or possibly as a group
            if "entry/data/detector_shift" in f:
                return True
    except Exception:
        pass
    return False





def has_mask_dataset(h5file_path):
    """Return True if a 2D mask dataset exists in HDF5."""
    if not h5file_path or not os.path.exists(h5file_path):
        return False
    try:
        with h5py.File(h5file_path, 'r') as f:
            ds = get_mask_dataset(f)
            if ds is None:
                return False
            return ds.ndim == 2 and ds.shape[0] > 0 and ds.shape[1] > 0
    except Exception:
        return False




def check_peakfinding_allowed(h5file_path):
    """Peakfinding is allowed only when a mask dataset exists."""
    return has_mask_dataset(h5file_path)




def check_findcenters_allowed(h5file_path):
    """Find centers only allowed if there are peaks."""
    return has_peak_dataset(h5file_path)




def check_refinecenters_allowed(h5file_path):
    """Refine centers only allowed if detector shift present."""
    return has_detector_shift(h5file_path)




def check_index_integrate_allowed(h5file_path):
    """Index & Integrate requires a mask dataset."""
    return has_mask_dataset(h5file_path)