Tomo export: skip unfinished runs automatically; export recorded projections by hand

src/hextools/exporters/auto_exporter.py

@@ -40,7 +40,7 @@
 CSV_COLUMNS = ("scan_id", "plan_name", "uid", "status", "export_status")
 
 
-def _record_run(csv_path: Path, run: BlueskyRunV3, succeeded: bool) -> None:
+def _record_run(csv_path: Path, run: BlueskyRunV3, export_status: str) -> None:
     uid = run.start["uid"]
     try:
         write_header = not csv_path.exists() or csv_path.stat().st_size == 0
@@ -54,7 +54,7 @@ def _record_run(csv_path: Path, run: BlueskyRunV3, succeeded: bool) -> None:
                     run.start.get("plan_name"),
                     uid,
                     run.stop.get("exit_status"),
-                    "success" if succeeded else "failure",
+                    export_status,
                 ]
             )
     except OSError:
@@ -66,10 +66,17 @@ def auto_export_run(
 ) -> None:
     """Run every exporter registered for ``run``'s specs, raising if any of them failed.
 
+    A run that did not end in success is skipped; export it by hand if needed.
     If ``csv_path`` is given, append the run's scan_id, plan_name, uid and export
     status to it.
     """
     uid = run.start["uid"]
+    exit_status = run.stop.get("exit_status")
+    if exit_status != "success":
+        logger.info(f"Run {uid} ended with exit status {exit_status}; not exporting")
+        if csv_path is not None:
+            _record_run(Path(csv_path), run, "skipped")
+        return
     specs = {(spec.name, spec.version) for spec in run.specs}
     exporters = [EXPORTERS[spec] for spec in specs if spec in EXPORTERS]
     if not exporters:
@@ -85,7 +92,7 @@ def auto_export_run(
             failed.append(exporter.__name__)  # type: ignore
 
     if csv_path is not None:
-        _record_run(Path(csv_path), run, succeeded=not failed)
+        _record_run(Path(csv_path), run, "failure" if failed else "success")
     if failed:
         raise RuntimeError(f"Exporters failed for run {uid}: {failed}")
 

src/hextools/exporters/tomography.py

@@ -14,6 +14,7 @@
     confirm_required_start_fields,
     create_baseline_dataset,
     find_hdf_datasets,
+    read_recorded_captures,
     run_metadata_to_json,
 )
 from hextools.utils import ensure_available
@@ -111,14 +112,36 @@ def export_tomo_flyscan_to_h5(
     num_projections = run_start.get("num_points")
     if not isinstance(images_to_average, int):
         raise ValueError(f"images_to_average is not an int: {images_to_average!r}")
+    num_events = (run.stop or {}).get("num_events")
+    recorded = num_events.get("primary") if isinstance(num_events, Mapping) else None
+    # A run that did not finish leaves files longer than its events, which Tiled
+    # will not serve, so its captures are read from the files up to what it recorded.
+    unfinished = (
+        isinstance(num_projections, int)
+        and isinstance(recorded, int)
+        and recorded < num_projections
+    )
+    if unfinished:
+        logger.warning(
+            f"Run recorded {recorded} of {num_projections} projections; exporting "
+            "those. Linked detector data may hold frames past the last projection."
+        )
     averaged = {}
-    for name, (_, _, ndim) in primary.items():
-        if name in det_names or (images_to_average == 1 and ndim == 1):
+    for name, (relpath, dataset, ndim) in primary.items():
+        if name in det_names or (
+            images_to_average == 1 and ndim == 1 and not unfinished
+        ):
             continue
         if not isinstance(num_projections, int):
             raise ValueError(f"num_points is not an int: {num_projections!r}")
+        if unfinished:
+            values = read_recorded_captures(
+                scan_id_dir_path / relpath, dataset, recorded, images_to_average
+            )
+        else:
+            values = run["primary"][name].read()
         averaged[name] = average_per_projection(
-            run["primary"][name].read(), num_projections, images_to_average
+            values, recorded if unfinished else num_projections, images_to_average
         )
         links[f"{name}_raw"] = links.pop(name)
         logger.info(f"Averaging {name} over {images_to_average} images per projection")

src/hextools/exporters/utils.py

@@ -122,3 +122,23 @@ def find_hdf_datasets(
             len(item["attributes"]["structure"]["shape"]),
         )
     return found
+
+
+def read_recorded_captures(
+    path: Path, dataset: str, num_events: int, images_to_average: int
+) -> np.ndarray:
+    """Read the PandA captures a run recorded, straight from the HDF5 file.
+
+    A run that did not finish records fewer events than the PandA captured before
+    it was disarmed, and Tiled refuses to serve a file longer than the run's
+    events, so read only the first ``num_events * images_to_average`` captures.
+    """
+    wanted = num_events * images_to_average
+    with h5py.File(path, "r") as f:
+        values = f[dataset][:wanted]
+    if len(values) < wanted:
+        raise ValueError(
+            f"{path}:{dataset} holds {len(values)} captures, fewer than the "
+            f"{wanted} the run recorded"
+        )
+    return values.reshape(num_events, images_to_average)

tests/test_export_unfinished_run.py

@@ -0,0 +1,147 @@
+"""Exporting a tomography flyscan that did not finish.
+
+At HEX on 2026-10-26, scan 14 (3601 projections x 4 images) was aborted after 570
+projections. The run recorded 570 x 4 PandA captures, but the PandA kept capturing
+until it was disarmed and its file held 2335, so Tiled refused to serve the capture
+(422 "Can not reshape (2335,) array data to (570, 4)") and the auto-export failed.
+"""
+
+import csv
+from pathlib import Path
+from unittest.mock import MagicMock, Mock
+
+import h5py
+import numpy as np
+import pytest
+from bluesky_tiled_plugins import CatalogOfBlueskyRuns
+from bluesky_tiled_plugins.clients.bluesky_run import BlueskyRunV3
+
+from hextools.exporters import auto_exporter, tomography
+from hextools.exporters.utils import read_recorded_captures
+from hextools.specs import TOMO_FLYSCAN_V1
+
+RECORDED, IMAGES, IN_FILE = 570, 4, 2335
+
+
+def _run(
+    exit_status: str, rotation_angle: Mock | None = None, recorded: int = RECORDED
+) -> MagicMock:
+    run = MagicMock(spec=BlueskyRunV3)
+    run.start = {
+        "uid": "uid-14",
+        "scan_id": 14,
+        "plan_name": "tomo_flyscan",
+        "data_session": "pass-319162",
+        "cycle": "2026-3",
+        "num_points": 3601,
+        "images_to_average": IMAGES,
+    }
+    run.stop = {"exit_status": exit_status, "num_events": {"primary": recorded}}
+    run.specs = [TOMO_FLYSCAN_V1]
+    run.__getitem__.side_effect = {"primary": {"rotation_angle": rotation_angle}}.get
+    return run
+
+
+def _csv_rows(csv_path: Path) -> list[tuple[str, str]]:
+    rows = csv.DictReader(csv_path.read_text().splitlines())
+    return [(r["status"], r["export_status"]) for r in rows]
+
+
+@pytest.mark.parametrize("exit_status", ["abort", "stop", "fail"])
+def test_auto_export_skips_a_run_that_did_not_finish(
+    monkeypatch, tmp_path: Path, exit_status: str
+):
+    exporter = Mock(__name__="export_tomo_flyscan_to_h5")
+    key = (TOMO_FLYSCAN_V1.name, TOMO_FLYSCAN_V1.version)
+    monkeypatch.setitem(auto_exporter.EXPORTERS, key, exporter)
+    csv_path = tmp_path / "exports.csv"
+
+    auto_exporter.auto_export_run(_run(exit_status), Mock(), csv_path=csv_path)
+
+    exporter.assert_not_called()
+    assert _csv_rows(csv_path) == [(exit_status, "skipped")]
+
+
+def test_auto_export_runs_exporters_for_a_finished_run(monkeypatch, tmp_path: Path):
+    exporter = Mock(__name__="export_tomo_flyscan_to_h5")
+    key = (TOMO_FLYSCAN_V1.name, TOMO_FLYSCAN_V1.version)
+    monkeypatch.setitem(auto_exporter.EXPORTERS, key, exporter)
+    csv_path = tmp_path / "exports.csv"
+
+    auto_exporter.auto_export_run(
+        _run("success", recorded=3601), Mock(), csv_path=csv_path
+    )
+
+    exporter.assert_called_once()
+    assert _csv_rows(csv_path) == [("success", "success")]
+
+
+def _panda_file(path: Path) -> np.ndarray:
+    angles = np.arange(IN_FILE, dtype=float)
+    with h5py.File(path, "w") as f:
+        f["/Angle"] = angles
+    return angles
+
+
+def test_reads_only_the_captures_the_run_recorded(tmp_path: Path):
+    angles = _panda_file(tmp_path / "panda.h5")
+    values = read_recorded_captures(tmp_path / "panda.h5", "/Angle", RECORDED, IMAGES)
+    np.testing.assert_array_equal(
+        values, angles[: RECORDED * IMAGES].reshape(RECORDED, IMAGES)
+    )
+
+
+def test_refuses_a_file_shorter_than_the_run(tmp_path: Path):
+    _panda_file(tmp_path / "panda.h5")
+    with pytest.raises(ValueError, match="fewer"):
+        read_recorded_captures(tmp_path / "panda.h5", "/Angle", 600, IMAGES)
+
+
+@pytest.fixture
+def export_dir(monkeypatch, tmp_path: Path) -> Path:
+    """Export under tmp_path, with the dark/flat, baseline and json steps stubbed."""
+    monkeypatch.setattr(tomography, "Path", lambda p: tmp_path / str(p).lstrip("/"))
+    monkeypatch.setattr(tomography, "resolve_dark_flat_run", lambda *a: None)
+    monkeypatch.setattr(tomography, "create_baseline_dataset", lambda *a: None)
+    monkeypatch.setattr(tomography, "run_metadata_to_json", lambda *a: None)
+    scan_dir = tmp_path / "nsls2/data/hex/proposals/2026-3/pass-319162"
+    path = scan_dir / "tomography/metadata/scan_00014"
+    path.mkdir(parents=True)
+    return path
+
+
+def _export(monkeypatch, run: MagicMock, export_dir: Path) -> np.ndarray:
+    datasets = {
+        "kinetix-det1": ("det.h5", "/entry/data/data", 4),
+        "rotation_angle": ("../panda.h5", "/Angle", 2),
+    }
+    monkeypatch.setattr(tomography, "find_hdf_datasets", lambda *a: datasets)
+    tomography.export_tomo_flyscan_to_h5(run, MagicMock(spec=CatalogOfBlueskyRuns))
+    with h5py.File(export_dir / "scan_00014.nxs") as f:
+        return f["/entry/data/rotation_angle"][()]
+
+
+def test_exports_the_recorded_projections_of_an_aborted_run(
+    monkeypatch, export_dir: Path
+):
+    angles = _panda_file(export_dir.parent / "panda.h5")
+    rotation_angle = Mock(shape=(RECORDED, IMAGES))
+    rotation_angle.read.side_effect = RuntimeError(
+        "422: Can not reshape (2335,) array data to (570, 4)"
+    )
+
+    exported = _export(monkeypatch, _run("abort", rotation_angle), export_dir)
+
+    expected = angles[: RECORDED * IMAGES].reshape(RECORDED, IMAGES).mean(axis=1)
+    np.testing.assert_allclose(exported, expected)
+
+
+def test_a_finished_run_is_still_read_through_tiled(monkeypatch, export_dir: Path):
+    captures = np.arange(3601 * IMAGES, dtype=float).reshape(3601, IMAGES)
+    rotation_angle = Mock(shape=captures.shape)
+    rotation_angle.read.return_value = captures
+
+    run = _run("success", rotation_angle, recorded=3601)
+    exported = _export(monkeypatch, run, export_dir)
+
+    np.testing.assert_allclose(exported, captures.mean(axis=1))