PATATO Bridge & Export

optari.patato_bridge

patato_bridge: functions to convert between PATATO and napari data structures

acquisition_start

acquisition_start(pa_data)

Wall-clock time of the first frame, or None if the scan cannot be dated.

PATATO's iThera and IPASC readers count timestamps in seconds since 0001-01-01, but an HDF5 file from another tool may use any epoch. So the timestamps are only read as wall-clock times when they match the date PATATO reads from the scan itself. A day of tolerance absorbs time zones: iThera records local time, IPASC UTC.

Source code in src/optari/patato_bridge.py

def acquisition_start(pa_data: "pat.PAData") -> datetime | None:
    """Wall-clock time of the first frame, or None if the scan cannot be dated.

    PATATO's iThera and IPASC readers count timestamps in seconds since 0001-01-01, but
    an HDF5 file from another tool may use any epoch. So the timestamps are only read as
    wall-clock times when they match the date PATATO reads from the scan itself. A day of
    tolerance absorbs time zones: iThera records local time, IPASC UTC.
    """
    scan_date = pa_data.get_scan_datetime()
    # NaN when the file records no date
    if not isinstance(scan_date, datetime):
        return None
    first_s = float(np.asarray(pa_data.get_timestamps())[0, 0])
    scan_date_s = (scan_date.replace(tzinfo=None) - YEAR_ONE).total_seconds()
    if not abs(first_s - scan_date_s) <= 24 * 3600:
        logger.warning(
            "scan timestamps do not match its date %s, frame times stay relative",
            scan_date,
        )
        return None
    return YEAR_ONE + timedelta(seconds=first_s)

build_napari_layers

build_napari_layers(pa_data)

Build napari image layers from an open pa_data handle.

Returns

layers : list of (data, kwargs) tuples Ready to pass to viewer.add_image. Everything a scan holds is an image. annotations and segmentations are layers OPTARI creates, not loads. patato_objects : dict Maps napari layer name → PATATO ImageSequence for later use (e.g. FOV queries, scale derivation).

Source code in src/optari/patato_bridge.py

def build_napari_layers(pa_data: "pat.PAData") -> tuple[list[tuple], dict]:
    """Build napari image layers from an open *pa_data* handle.

    Returns
    -------
    layers : list of ``(data, kwargs)`` tuples
        Ready to pass to ``viewer.add_image``. Everything a scan holds is an image.
        annotations and segmentations are layers OPTARI creates, not loads.
    patato_objects : dict
        Maps napari layer name → PATATO ``ImageSequence`` for later use
        (e.g. FOV queries, scale derivation).
    """
    _us_fallback = settings.general.US_FALLBACK_SCALE
    _pa_fallback = settings.general.PA_FALLBACK_SCALE

    patato_objects: dict = {}
    layers: list = []

    timestamps = np.array(pa_data.get_timestamps())
    start = acquisition_start(pa_data)
    wavelengths = [int(w) for w in pa_data.get_wavelengths()]

    # --- ultrasound ---
    # IPASC scans have no ultrasound. The acquisition frame count then comes from the time series instead.
    us_obj = pa_data.get_ultrasound()
    if hasattr(us_obj, "da"):
        us_img = display_data_from_patato_obj(us_obj)
        n_acq_frames = us_img.shape[0]
        patato_objects["US"] = us_obj

        layers.append(
            (
                us_img,
                {
                    "colormap": layer_colormap(HDF5Tags.ULTRASOUND),
                    "name": "US",
                    "scale": scale_from_patato_obj(us_obj, _us_fallback),
                    "opacity": 1.0,
                    "metadata": {
                        "type": "us",
                        "timestamps": timestamps,
                    },
                },
            )
        )
    else:
        n_acq_frames = pa_data.shape[0]

    def _translate(image_sequence, ndim: int) -> tuple[float, ...]:
        return (0.0,) * (ndim - 2) + tuple(
            image_sequence.da.attrs.get(TRANSLATE_ATTR, (0.0, 0.0))
        )

    def _settings(image_sequence) -> dict:
        # what OPTARI made the layer with, else the file's own attributes (vendor recons)
        attrs = image_sequence.da.attrs
        if SETTINGS_ATTR in attrs:
            return json.loads(attrs[SETTINGS_ATTR])
        return dict(attrs)

    def _frame_list(image_sequence, n_frames: int) -> list[int]:
        frames_info = image_sequence.da.attrs.get(
            "frames", image_sequence.da.attrs.get("frame")
        )
        if frames_info is None:
            return list(range(n_frames))
        # atleast_1d covers a scalar of any numpy/Python int type as well as lists and
        # arrays. anything int() rejects is a corrupt attribute and should fail
        return [int(f) for f in np.atleast_1d(frames_info)]

    # --- reconstructions ---
    for (recon_name, idx), recon in pa_data.get_scan_reconstructions().items():
        recon_raw = display_data_from_patato_obj(recon)
        recon_frame_list = _frame_list(recon, recon_raw.shape[0])
        recon_img = expand_to_acquisition_frames(
            recon_raw, recon_frame_list, n_acq_frames
        )

        layer_name = recon.da.attrs.get(
            LAYER_NAME_ATTR, f"Recon: {recon_name}_{idx}"
        )
        patato_objects[layer_name] = recon
        layers.append(
            (
                recon_img,
                {
                    "colormap": layer_colormap(HDF5Tags.RECONSTRUCTION),
                    "name": layer_name,
                    "scale": scale_from_patato_obj(recon, _pa_fallback),
                    "translate": _translate(recon, recon_img.ndim),
                    "opacity": 1.0,
                    "blending": "multiplicative",
                    "auto_contrast": True,
                    "metadata": {
                        "type": "pa",
                        "pa_kind": "recon",
                        "wavelengths": wavelengths,
                        "axis1_name": "Channel",
                        "axis1_labels": wavelengths,
                        "timestamps": timestamps,
                        "acquisition_start": start,
                        "frames": recon_frame_list,
                        "settings": _settings(recon),
                    },
                },
            )
        )

    # --- derived PA image groups that may already exist in HDF5 ---
    derived_specs = [
        (HDF5Tags.UNMIXED, "Unmixed", "unmixed", None),
        (HDF5Tags.THB, "THb", "unmixed_param", "thb"),
        (HDF5Tags.SO2, "sO2", "unmixed_param", "so2"),
    ]
    for group_name, prefix, pa_kind, parameter in derived_specs:
        for (dataset_name, idx), image in pa_data.get_scan_images(
            group_name, ignore_default=True
        ).items():
            raw = display_data_from_patato_obj(image)
            frame_list = _frame_list(image, raw.shape[0])
            data = expand_to_acquisition_frames(raw, frame_list, n_acq_frames)

            axis1_labels = list(
                map(str, np.asarray(image.ax_1_labels).tolist())
            )
            source_layer = image.da.attrs.get("source_layer")
            if source_layer is None:
                source_layer = f"Recon: {dataset_name}_{idx}"
            metadata = {
                "type": "pa",
                "pa_kind": pa_kind,
                "source_layer": str(source_layer),
                "frames": frame_list,
                "axis1_name": "Channel",
                "axis1_labels": axis1_labels,
                "timestamps": timestamps,
                "acquisition_start": start,
                "settings": _settings(image),
            }
            if pa_kind == "unmixed":
                metadata["chromophores"] = axis1_labels
            if parameter is not None:
                metadata["parameter"] = parameter

            layers.append(
                (
                    data,
                    {
                        "colormap": layer_colormap(group_name),
                        "name": image.da.attrs.get(
                            LAYER_NAME_ATTR, f"{prefix}: {dataset_name}_{idx}"
                        ),
                        "scale": scale_from_patato_obj(image, _pa_fallback),
                        "translate": _translate(image, data.ndim),
                        "opacity": 1.0,
                        "blending": "multiplicative",
                        "auto_contrast": True,
                        "metadata": metadata,
                    },
                )
            )

    return layers, patato_objects

display_data_from_patato_obj

display_data_from_patato_obj(image_sequence)

Convert PATATO image data to napari's display orientation.

Source code in src/optari/patato_bridge.py

def display_data_from_patato_obj(image_sequence) -> np.ndarray:
    """Convert PATATO image data to napari's display orientation."""
    return np.flip(np.array(image_sequence.da[:, :, :, 0, :]), axis=-2)

expand_to_acquisition_frames

expand_to_acquisition_frames(data, frames, n_acq_frames)

Place per-output-frame data at its frames on the acquisition frame axis.

Frames that were not processed stay zero, so every PA layer shares the frame slider with the ultrasound. Data that already spans the acquisition is returned unchanged.

Source code in src/optari/patato_bridge.py

def expand_to_acquisition_frames(
    data: np.ndarray, frames: list[int], n_acq_frames: int
) -> np.ndarray:
    """Place per-output-frame *data* at its *frames* on the acquisition frame axis.

    Frames that were not processed stay zero, so every PA layer shares the frame slider
    with the ultrasound. Data that already spans the acquisition is returned unchanged.
    """
    if data.shape[0] == n_acq_frames:
        return data
    # A frame list that does not match the data would put images on the wrong frames.
    if len(frames) != data.shape[0] or not all(
        0 <= f < n_acq_frames for f in frames
    ):
        raise ValueError(
            f"frame indices {frames} do not match {data.shape[0]} image frame(s) "
            f"within {n_acq_frames} acquisition frame(s)"
        )
    expanded = np.zeros((n_acq_frames, *data.shape[1:]), dtype=data.dtype)
    expanded[frames] = data
    return expanded

fov_from_objects

fov_from_objects(patato_objects)

Return (fov_x_m, fov_y_m) from the first object with valid FOV.

Returns None if no object exposes usable FOV metadata.

Source code in src/optari/patato_bridge.py

def fov_from_objects(patato_objects: dict) -> "tuple[float, float] | None":
    """Return ``(fov_x_m, fov_y_m)`` from the first object with valid FOV.

    Returns ``None`` if no object exposes usable FOV metadata.
    """
    for obj in patato_objects.values():
        fov = _fov_m(obj)
        if fov is not None and min(fov) > 0:
            return fov
    return None

layer_colormap

layer_colormap(group)

Colormap for an HDF5 image group, whether loaded from the scan or computed.

Source code in src/optari/patato_bridge.py

def layer_colormap(group: str) -> str:
    """Colormap for an HDF5 image *group*, whether loaded from the scan or computed."""
    return settings.general.LAYER_COLOR_MAPS.get(
        group, DEFAULT_COLORMAPS[group]
    )

patato_roi_from_geometry

patato_roi_from_geometry(geometry, fov_x_m, fov_y_m, *, z=0.0, run=0.0, rep=0.0, frame_idx=0, roi_id=None, track_id=None, roi_group_uid=None)

Convert an ROI geometry into a PATATO ROI object for the native writer.

The geometry's source becomes PATATO's roi_class, so provenance survives export.

Source code in src/optari/patato_bridge.py

def patato_roi_from_geometry(
    geometry: RoiGeometry,
    fov_x_m: float,
    fov_y_m: float,
    *,
    z: float = 0.0,
    run: float = 0.0,
    rep: float = 0.0,
    frame_idx: int = 0,
    roi_id: int | None = None,
    track_id: int | None = None,
    roi_group_uid: str | None = None,
) -> object:
    """Convert an ROI geometry into a PATATO ROI object for the native writer.

    The geometry's ``source`` becomes PATATO's ``roi_class``, so provenance survives export.
    """
    from patato.utils.rois.roi_type import ROI as PatatoROI  # type: ignore[import]

    roi = PatatoROI.from_polygon_mm(
        verts_yx_mm=geometry.verts_mm(fov_x_m, fov_y_m),
        fov=(fov_x_m, fov_y_m),
        z_position=z,
        run=run,
        repetition=rep,
        ax0_index=np.array([frame_idx]),
        roi_class=geometry.source,
        position=geometry.tissue_class,  # PATATO's own kwarg name
        generated=True,
        shape_type=geometry.kind,
    )
    if roi_id is not None:
        roi.roi_id = int(roi_id)
    if track_id is not None:
        roi.roi_group_id = int(track_id)  # PATATO's own attribute name
    if roi_group_uid is not None:
        roi.roi_group_uid = str(roi_group_uid)
    return roi

roi_records_from_scan_rois

roi_records_from_scan_rois(pa_data, fov_x_m, fov_y_m)

Load PATATO ROIs as one frame-owned record per valid frame.

Source code in src/optari/patato_bridge.py

def roi_records_from_scan_rois(
    pa_data: "pat.PAData", fov_x_m: float, fov_y_m: float
) -> list[ROIRecord]:
    """Load PATATO ROIs as one frame-owned record per valid frame."""
    try:
        rois = pa_data.get_rois()
        n_frames = int(pa_data.shape[0])
    except Exception as exc:
        raise ValueError("could not read stored ROI annotations") from exc

    records: list[ROIRecord] = []
    used_ids: set[int] = set()
    next_id = 0
    next_track = 0
    frameless = []
    for key, roi in rois.items():
        try:
            verts_m = np.asarray(roi.points, dtype=float)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"stored ROI {key!r} has invalid geometry"
            ) from exc
        geometry = RoiGeometry(
            verts_m=verts_m,
            kind=roi.shape_type,
            source=roi.roi_class,
            tissue_class=roi.position,
        )
        frames = np.unique(roi.ax0_index)
        # A frame this scan does not have means the ROI was drawn on another acquisition.
        if np.any((frames < 0) | (frames >= n_frames)):
            raise ValueError(
                f"stored ROI {key!r} references frame(s) {frames.tolist()}, "
                f"but the scan has {n_frames}"
            )
        if not frames.size:
            # No frame recorded at all (e.g. iLabs .iROI files): the ROI covers the whole acquisition.
            frames = np.arange(n_frames, dtype=int)
            frameless.append(key)

        # PATATO calls this "roi_group_id", in OPTARI it is `track_id`.
        track_id = (
            next_track if roi.roi_group_id is None else int(roi.roi_group_id)
        )
        next_track = max(next_track, track_id + 1)
        persisted_id = None if roi.roi_id is None else int(roi.roi_id)

        # Every frame copy of one ROI shares the group's identity. a scan written
        # before uids existed gets a fresh one for the whole group, not per frame.
        group_uid = str(roi.roi_group_uid or new_roi_group_uid())
        for frame_id in frames:
            roi_id = persisted_id if len(frames) == 1 else None
            if roi_id is None or roi_id in used_ids:
                while next_id in used_ids:
                    next_id += 1
                roi_id = next_id
            used_ids.add(roi_id)
            next_id = max(next_id, roi_id + 1)
            records.append(
                geometry.to_record(
                    roi_id=roi_id,
                    track_id=track_id,
                    frame_id=int(frame_id),
                    roi_group_uid=group_uid,
                    fov_x_m=fov_x_m,
                    fov_y_m=fov_y_m,
                )
            )
    if frameless:
        logger.warning(
            "%d stored ROI(s) have no frame assigned and are shown on all %d frames: %s",
            len(frameless),
            n_frames,
            ", ".join(f"{name} #{number}" for name, number in frameless),
        )
    return records

scale_from_patato_obj

scale_from_patato_obj(obj, fallback)

Derive a (t, y, x) napari scale in mm/pixel from a PATATO object.

Reads the object's fov ((fov_x_m, fov_y_m) in metres) and shape_2d ((ny, nx) in pixels). Returns fallback if either value is absent or zero.

Source code in src/optari/patato_bridge.py

def scale_from_patato_obj(obj, fallback: tuple) -> tuple:
    """Derive a ``(t, y, x)`` napari scale in mm/pixel from a PATATO object.

    Reads the object's ``fov`` (``(fov_x_m, fov_y_m)`` in metres) and
    ``shape_2d`` (``(ny, nx)`` in pixels).  Returns *fallback* if either
    value is absent or zero.
    """
    fov = _fov_m(obj)
    if fov is None:
        return fallback
    fov_x_m, fov_y_m = fov
    ny, nx = obj.shape_2d[-2:]  # pixels
    if not (ny and nx and fov_x_m and fov_y_m):
        logger.warning(
            "no usable FOV on %s, using fallback scale %s", obj, fallback
        )
        return fallback
    return (fallback[0], fov_y_m / ny * 1000, fov_x_m / nx * 1000)

segmentation_from_scan

segmentation_from_scan(pa_data)

Load a previously exported segmentation mask and its class metadata.

Returns None when the scan has no segmentation, or one that has not optari compatible metadata

Source code in src/optari/patato_bridge.py

def segmentation_from_scan(pa_data: "pat.PAData") -> dict | None:
    """Load a previously exported segmentation mask and its class metadata.

    Returns ``None`` when the scan has no segmentation, or one that has not optari compatible metadata
    """
    seg = pa_data.get_segmentation()
    if seg is None:
        return None
    dataset = pa_data.scan_reader.file[HDF5Tags.SEGMENTATION]
    if "optari_meta" not in dataset.attrs:
        logger.info(
            "scan has a segmentation dataset OPTARI did not write, ignoring it"
        )
        return None
    meta = json.loads(dataset.attrs["optari_meta"])
    meta["mask"] = np.asarray(seg, dtype=MASK_DTYPE)  # (n_frames, H, W)
    meta["class_names"] = {int(k): v for k, v in meta["class_names"].items()}
    return meta

optari.io.export_pipeline

HDF5 and IPASC scan export, and the ROI table XLSX round-trip.

export_scan_to_hdf5() writes the scan through PATATO's own save_hdf5(), then reopens the file to rewrite the ROI and derived-data datasets (segmentation mask, derived PA images, clinical metadata) so in-session OPTARI edits are captured, not just what was already in the source file. No controller reads a PATATO array or writes an HDF5 attribute directly outside this module and patato_bridge.py.

default_roi_table_filename

default_roi_table_filename()

roi_data_<operator>_<analysis id>_<timestamp>.xlsx.

The timestamp is UTC so it matches creation_time in the provenance sheet.

Source code in src/optari/io/export_pipeline.py

def default_roi_table_filename() -> str:
    """``roi_data_<operator>_<analysis id>_<timestamp>.xlsx``.

    The timestamp is UTC so it matches ``creation_time`` in the provenance sheet.
    """
    timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return (
        f"roi_data_{filename_token(settings.general.OPERATOR)}"
        f"_{filename_token(settings.general.ANALYSIS_ID)}_{timestamp}.xlsx"
    )

export_roi_table_to_xlsx

export_roi_table_to_xlsx(df_saved, path=None)

Write the ROI table as an XLSX file, prompting for a path if path is None.

Source code in src/optari/io/export_pipeline.py

def export_roi_table_to_xlsx(df_saved, path=None) -> str | None:
    """Write the ROI table as an XLSX file, prompting for a path if *path* is None."""

    if path is not None:
        filename = str(path)
    else:
        filename, _ = QFileDialog.getSaveFileName(
            None,
            "Save ROIs as Excel",
            default_roi_table_filename(),
            "Excel Files (*.xlsx)",
        )
        if not filename:
            return None
        if not filename.endswith(".xlsx"):
            filename += ".xlsx"

    meta = file_origin(roi_table_schema_version=ROI_TABLE_SCHEMA_VERSION)
    meta_df = pd.DataFrame(
        {"key": list(meta.keys()), "value": [str(v) for v in meta.values()]}
    )
    with pd.ExcelWriter(Path(filename)) as writer:
        df_saved.to_excel(writer, sheet_name=ROI_TABLE_SHEET, index=False)
        meta_df.to_excel(writer, sheet_name=ROI_TABLE_META_SHEET, index=False)

    return filename

export_scan_to_hdf5

export_scan_to_hdf5(controller, destination)

Export scan data and OPTARI additions (ROIs and derived images).

Source code in src/optari/io/export_pipeline.py

def export_scan_to_hdf5(controller, destination: Path) -> bool:
    """Export scan data and OPTARI additions (ROIs and derived images)."""
    if controller.pa_data is None:
        logger.warning("No scan loaded, nothing to export")
        return False

    destination = _hdf5_destination(destination)
    if destination is None:
        return False

    try:
        with atomic_destination(destination) as temporary:
            controller.pa_data.save_hdf5(str(temporary))
            _write_file_origin(temporary)
            exported = pat.PAData.from_hdf5(str(temporary), mode="r+")
            try:
                _write_rois(controller, exported)
                _write_derived_data(controller, exported)
            finally:
                # Closed before the move, which Windows refuses for an open file.
                exported.close()
    except Exception:
        logger.exception("Failed to export scan to %s", destination.name)
        return False
    logger.info("exported scan to %s", destination)
    return True

export_scan_to_ipasc

export_scan_to_ipasc(controller, destination)

Export the raw time series of the loaded scan as a native IPASC file.

Source code in src/optari/io/export_pipeline.py

def export_scan_to_ipasc(controller, destination: Path) -> bool:
    """Export the raw time series of the loaded scan as a native IPASC file."""
    if controller.pa_data is None:
        logger.warning("No scan loaded, nothing to export")
        return False

    destination = _hdf5_destination(destination)
    if destination is None:
        return False

    try:
        pat.write_ipasc(controller.pa_data.scan_reader, str(destination))
    except Exception:
        logger.exception(
            "Failed to export raw time series to %s", destination.name
        )
        return False

    logger.info("exported raw time series to %s", destination)
    return True

file_origin

file_origin(**extra)

origin recorded in every file OPTARI writes.

Source code in src/optari/io/export_pipeline.py

def file_origin(**extra) -> dict:
    """origin recorded in every file OPTARI writes."""
    return {
        "tool": "OPTARI",
        "tool_version": __version__,
        "operator": settings.general.OPERATOR,
        "analysis_id": settings.general.ANALYSIS_ID,
        "creation_time": dt.datetime.now(dt.timezone.utc).isoformat(),
        "format_version": OPTARI_FILE_FORMAT_VERSION,
        "config_schema_version": CONFIG_SCHEMA_VERSION,
        "source_url": OPTARI_SOURCE_URL,
        "documentation_url": OPTARI_DOCS_URL,
        "publication_doi": OPTARI_PUBLICATION_DOI,
        **extra,
    }

import_roi_table_from_xlsx

import_roi_table_from_xlsx(path=None)

Read a previously exported ROI table, prompting for a file if path is None.

Source code in src/optari/io/export_pipeline.py

def import_roi_table_from_xlsx(path=None) -> tuple[pd.DataFrame, str] | None:
    """Read a previously exported ROI table, prompting for a file if *path* is None."""
    if path is not None:
        filename = str(path)
    else:
        filename, _ = QFileDialog.getOpenFileName(
            None,
            "Import ROI Analysis Table",
            "",
            "Excel Files (*.xlsx)",
        )
        if not filename:
            return None

    path = Path(filename)

    try:
        meta_df = pd.read_excel(path, sheet_name=ROI_TABLE_META_SHEET)
        meta = dict(zip(meta_df["key"], meta_df["value"]))
    except (ValueError, KeyError) as exc:
        raise ValueError(
            f"'{Path(path).name}' is not a OPTARI ROI table."
        ) from exc

    file_version = str(meta.get("roi_table_schema_version", ""))
    if file_version != str(ROI_TABLE_SCHEMA_VERSION):
        raise ValueError(
            f"ROI table schema version {file_version or 'missing'} cannot be read by "
            f"this OPTARI version, which expects version {ROI_TABLE_SCHEMA_VERSION}."
        )

    df = pd.read_excel(path, sheet_name=ROI_TABLE_SHEET)

    missing_required = [
        c for c in SAVED_FIXED_SOURCE_COLUMNS if c not in df.columns
    ]
    if missing_required:
        raise ValueError(
            f"ROI table is missing required column(s): {', '.join(missing_required)}."
        )

    # Without a group identity a row cannot be grouped or restored, and a blank one
    # would drop out of every groupby silently rather than failing.
    if df["roi_group_uid"].fillna("").astype(str).str.strip().eq("").any():
        raise ValueError("ROI table has row(s) with no roi_group_uid.")

    # Reconcile by name, not position, the export column set follows FEATURE_REGISTRY
    # and can grow between releases without a schema version bump.
    expected = saved_export_columns()
    dropped = [c for c in df.columns if c not in expected]
    added = [c for c in expected if c not in df.columns]
    if dropped:
        logger.warning(
            "ignoring unknown column(s) in %s: %s", path, ", ".join(dropped)
        )
    if added:
        logger.warning(
            "filling absent column(s) in %s: %s", path, ", ".join(added)
        )

    return df.reindex(columns=expected).reset_index(drop=True), filename

ipasc_export_report

ipasc_export_report(destination)

Summarise how complete the IPASC metadata of an exported file are.

Reports which of IPASC's minimal fields are present. Fields IPASC marks "report if present" are absent when the source format never recorded them, so their absence is not a failure and is not counted here.

Source code in src/optari/io/export_pipeline.py

def ipasc_export_report(destination: Path) -> str:
    """Summarise how complete the IPASC metadata of an exported file are.

    Reports which of IPASC's minimal fields are present. Fields IPASC marks "report if
    present" are absent when the source format never recorded them, so their
    absence is not a failure and is not counted here.
    """
    import pacfish as pf
    from pacfish import MetadataAcquisitionTags

    data = pf.load_data(str(destination))
    missing = [
        datum.tag
        for datum in MetadataAcquisitionTags.TAGS
        if datum.mandatory and datum.tag not in data.meta_data_acquisition
    ]
    detectors = data.meta_data_device.get("detectors", {})
    if not detectors:
        missing.append("detector positions")

    if missing:
        return f"IPASC minimal metadata incomplete: {', '.join(missing)} not recorded."
    return f"IPASC minimal metadata complete, {len(detectors)} detection elements."

optari.io.discovery

Finding studies and scans on disk. Headless: no Qt, no napari, no viewer.

ScanInfo dataclass

ScanInfo(kind, internal_name)

Metadata for a discovered scan.

discover_scans

discover_scans(folder)

Find every scan directly under folder, keyed by path and deduplicated by scan_key.

When both an HDF5 file and an iThera folder resolve to the same key, the HDF5 one wins.

Source code in src/optari/io/discovery.py

def discover_scans(folder: Path) -> dict[Path, ScanInfo]:
    """Find every scan directly under *folder*, keyed by path and deduplicated by ``scan_key``.

    When both an HDF5 file and an iThera folder resolve to the same key, the HDF5
    one wins.
    """
    by_key: dict[str, tuple[Path, ScanInfo]] = {}
    for path in sorted(folder.glob("*.hdf5")):
        kind = scan_type(path)
        if kind is not None:
            info = ScanInfo(kind, _read_internal_scan_name(path, kind))
            by_key[scan_key(path)] = (path, info)

    # An iThera folder only counts if no HDF5 file holds the same scan.
    for path in folder.glob("Scan_*"):
        key = scan_key(path)
        kind = scan_type(path)
        if kind is not None and key not in by_key:
            info = ScanInfo(kind, _read_internal_scan_name(path, kind))
            by_key[key] = (path, info)

    return dict(
        sorted(by_key.values(), key=lambda item: scan_sort_key(item[0]))
    )

discover_studies

discover_studies(root, max_depth=3)

Map each study folder under root to its scans, in folder-name order.

A study is simply any folder that holds at least one scan, so a flat folder of scans comes back as a single study and no naming convention is imposed beyond the Scan_* one discover_scans already relies on. A folder that is itself a study is not descended into.

Source code in src/optari/io/discovery.py

def discover_studies(
    root: Path, max_depth: int = 3
) -> dict[Path, dict[Path, ScanInfo]]:
    """Map each study folder under *root* to its scans, in folder-name order.

    A study is simply any folder that holds at least one scan, so a flat folder
    of scans comes back as a single study and no naming convention is imposed
    beyond the ``Scan_*`` one ``discover_scans`` already relies on. A folder that
    is itself a study is not descended into.
    """
    studies: dict[Path, dict[Path, ScanInfo]] = {}

    def walk(folder: Path, depth: int) -> None:
        """Recurse into *folder* up to *max_depth*, stopping at the first folder that holds a scan."""
        scans = discover_scans(folder)
        if scans:
            studies[folder] = scans
            return
        if depth >= max_depth:
            return
        try:
            children = sorted(
                child
                for child in folder.iterdir()
                # Following symlinks here risks walking a cycle or wandering
                # outside the dataset the user picked.
                if child.is_dir() and not child.is_symlink()
            )
        except OSError:
            logger.warning("could not list '%s', skipping", folder)
            return
        for child in children:
            walk(child, depth + 1)

    walk(Path(root), 0)
    return studies

scan_key

scan_key(scan_path)

Return the Scan_<n> prefix of scan_path's name, or the full name if it doesn't match.

Source code in src/optari/io/discovery.py

def scan_key(scan_path: Path) -> str:
    """Return the ``Scan_<n>`` prefix of *scan_path*'s name, or the full name if it doesn't match."""
    name = scan_path.stem if scan_path.is_file() else scan_path.name
    m = re.match(r"^(Scan_\d+)", name)
    return m.group(1) if m else name

scan_sort_key

scan_sort_key(scan_path)

Sort key that orders Scan_<n> names numerically, ahead of any other name sorted alphabetically.

Source code in src/optari/io/discovery.py

def scan_sort_key(scan_path: Path):
    """Sort key that orders ``Scan_<n>`` names numerically, ahead of any other name sorted alphabetically."""
    key = scan_key(scan_path)
    m = re.match(r"^Scan_(\d+)$", key)
    if m:
        return (0, int(m.group(1)), key)
    return (1, 0, key)

scan_type

scan_type(path)

Identify the format of a scan on disk, or None if the path is not a scan.

Source code in src/optari/io/discovery.py

def scan_type(path: Path) -> str | None:
    """
    Identify the format of a scan on disk, or ``None`` if the path is not a scan.
    """
    if path.is_dir():
        return "ithera" if any(path.glob("*.msot")) else None
    if path.suffix.lower() != ".hdf5":
        return None
    try:
        with h5py.File(path, "r") as file:
            if IPASCTags.BINARY_DATA in file:
                return "ipasc"
            if HDF5Tags.RAW_DATA in file:
                return "hdf5"
    except OSError:
        logger.debug("could not open '%s' as HDF5", path, exc_info=True)
    return None