From 816a96d8031c28eb5620ef16dda6e5b2491bc704 Mon Sep 17 00:00:00 2001 From: sligara7 Date: Mon, 26 Oct 2026 16:17:07 -0400 Subject: [PATCH 1/2] Tomo export: skip unfinished runs in the auto-exporter; export recorded projections by hand An aborted tomo flyscan records fewer events than the PandA captured before it was disarmed, so Tiled refuses to serve the capture (422 'Can not reshape (2335,) array data to (570, 4)', HEX scan 14, 2026-10-26) and the auto-export failed. The auto-exporter now skips runs that did not end in success and records them as skipped. export_tomo_flyscan_to_h5, run by hand on such a run, reads each PandA capture from its file up to the recorded events and averages over the recorded projections. Assisted-by: copilot:claude-opus-5-5 --- src/hextools/exporters/auto_exporter.py | 13 ++- src/hextools/exporters/tomography.py | 24 +++- src/hextools/exporters/utils.py | 20 ++++ tests/test_export_unfinished_run.py | 147 ++++++++++++++++++++++++ 4 files changed, 198 insertions(+), 6 deletions(-) create mode 100644 tests/test_export_unfinished_run.py diff --git a/src/hextools/exporters/auto_exporter.py b/src/hextools/exporters/auto_exporter.py index 6858f7c..9e1dedf 100644 --- a/src/hextools/exporters/auto_exporter.py +++ b/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}") diff --git a/src/hextools/exporters/tomography.py b/src/hextools/exporters/tomography.py index bd76998..8957e52 100644 --- a/src/hextools/exporters/tomography.py +++ b/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,31 @@ 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}") + recorded = (run.stop or {}).get("num_events", {}).get("primary", num_projections) + # 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 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") diff --git a/src/hextools/exporters/utils.py b/src/hextools/exporters/utils.py index ef36203..12e0498 100644 --- a/src/hextools/exporters/utils.py +++ b/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) diff --git a/tests/test_export_unfinished_run.py b/tests/test_export_unfinished_run.py new file mode 100644 index 0000000..26f8811 --- /dev/null +++ b/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)) From 54924a6acd6657227f21fdfb61405da06ccd665f Mon Sep 17 00:00:00 2001 From: sligara7 Date: Tue, 27 Oct 2026 10:25:43 -0400 Subject: [PATCH 2/2] Narrow the stop document's num_events before reading it (ty) Assisted-by: copilot:claude-opus-5-5 --- src/hextools/exporters/tomography.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/hextools/exporters/tomography.py b/src/hextools/exporters/tomography.py index 8957e52..07b998e 100644 --- a/src/hextools/exporters/tomography.py +++ b/src/hextools/exporters/tomography.py @@ -112,10 +112,15 @@ 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}") - recorded = (run.stop or {}).get("num_events", {}).get("primary", num_projections) + 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 recorded < num_projections + 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 "