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..07b998e 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,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") 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))