From 3802f168aa82bba8fc58182d22f41650497986b7 Mon Sep 17 00:00:00 2001 From: Peter Grotz <199680633+peter-grotz@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:55:53 -0700 Subject: [PATCH 1/3] Record the real cell count and finish time for packaging build_packaging_process was called before package_cells with cell_count=0, and the record was embedded verbatim in every cell, so every processing.json claimed zero cells packaged. Its end_date_time was also stamped before any packaging work ran. package_cells now resolves every cell before writing any, then builds the record from the resulting count and the actual finish time. It takes a factory rather than a record because neither value is known at the call site. --- src/exaspim_swc_processing/packaging.py | 25 +++++++---- tests/test_packaging.py | 59 ++++++++++++++++++++++++- 2 files changed, 75 insertions(+), 9 deletions(-) diff --git a/src/exaspim_swc_processing/packaging.py b/src/exaspim_swc_processing/packaging.py index 2b1f5e4..790b42a 100644 --- a/src/exaspim_swc_processing/packaging.py +++ b/src/exaspim_swc_processing/packaging.py @@ -20,9 +20,9 @@ from __future__ import annotations import logging -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from aind_data_schema.components.identifiers import Code @@ -161,7 +161,7 @@ def package_cells( stage_processes: Sequence[DataProcess], pipeline: Code, creation_time: datetime, - packaging_process: DataProcess, + describe_packaging: Callable[[int, datetime], DataProcess], specs: Sequence[ArtifactSpec] = ARTIFACT_SPECS, overrides: Mapping[str, object] | None = None, ) -> PackagingResult: @@ -181,8 +181,10 @@ def package_cells( The pipeline that produced the run. creation_time : datetime Creation time shared by every cell in the run. - packaging_process : DataProcess - Record of this packaging step, appended to each cell's processing record. + describe_packaging : Callable[[int, datetime], DataProcess] + Builds this packaging step's record, given the number of cells written and the + time the work finished. A callable rather than a record because both values are + only known once discovery has run, yet the record is embedded in every cell. specs : Sequence[ArtifactSpec], optional Artifacts to collect, by default :data:`~exaspim_swc_processing.layout.ARTIFACT_SPECS`. overrides : Mapping[str, object] | None, optional @@ -195,10 +197,11 @@ def package_cells( skipped rather than written without metadata. """ specs = tuple(specs) - processes = [*stage_processes, packaging_process] - packaged: list[PackagedCell] = [] skipped: list[SkippedCell] = [] + # Resolve every cell before writing any: the packaging record embedded in each cell + # states how many the run produced, which is not known until the last is resolved. + planned = [] for reconstruction, cell in discover_cells(stage_root, specs).items(): missing = cell.missing_roles(specs) if missing: @@ -219,10 +222,16 @@ def package_cells( logger.warning("Skipping %s: %s", reconstruction.stem, error) skipped.append(SkippedCell(reconstruction, str(error))) continue + planned.append((reconstruction, cell, description)) + + finished = datetime.now(timezone.utc) + processes = [*stage_processes, describe_packaging(len(planned), finished)] + record = build_cell_processing(processes, pipeline) + packaged: list[PackagedCell] = [] + for reconstruction, cell, description in planned: cell_dir = output_root / description.name _write_cell_directory(cell, cell_dir, specs) - record = build_cell_processing(processes, pipeline) (cell_dir / DATA_DESCRIPTION_FILENAME).write_text( description.model_dump_json(indent=2), encoding="utf-8" ) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 2277b93..6dddec2 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -1,6 +1,7 @@ """Tests for :mod:`exaspim_swc_processing.packaging`.""" import json +from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path @@ -124,6 +125,33 @@ def _packaging_process(parent: ParentMetadata, cells: int = 2) -> DataProcess: ) +def _describe(parent: ParentMetadata) -> Callable[[int, datetime], DataProcess]: + """Return the factory :func:`package_cells` calls once the count is known. + + Parameters + ---------- + parent : ParentMetadata + Resolved parent, supplying provenance. + + Returns + ------- + Callable[[int, datetime], DataProcess] + Builds the step record from the cell count and finish time. + """ + + def build(cells: int, finished: datetime) -> DataProcess: + return build_packaging_process( + parent, + PACKAGER, + start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), + end_time=finished, + output_path="cells", + cell_count=cells, + ) + + return build + + def _package(stage_root: Path, tmp_path: Path, **kwargs: object) -> object: """Package the fixture tree with default arguments. @@ -149,7 +177,7 @@ def _package(stage_root: Path, tmp_path: Path, **kwargs: object) -> object: _stage_processes(), PIPELINE, CREATION_TIME, - _packaging_process(parent), + _describe(parent), **kwargs, ) @@ -296,3 +324,32 @@ def test_experimenters_can_be_overridden() -> None: experimenters=["Cameron Arshadi"], ) assert step.experimenters == ["Cameron Arshadi"] + + +def test_the_recorded_cell_count_is_the_number_actually_written( + stage_root: Path, tmp_path: Path +) -> None: + """The count was previously fixed at call time, so every cell recorded zero.""" + result = _package(stage_root, tmp_path) + record = json.loads( + (result.packaged[0].directory / "processing.json").read_text(encoding="utf-8") + ) + packaging = [ + p for p in record["data_processes"] if p["output_parameters"].get("cells_packaged") + ] + assert packaging[0]["output_parameters"]["cells_packaged"] == len(result.packaged) + + +def test_packaging_ends_after_the_cells_are_resolved(stage_root: Path, tmp_path: Path) -> None: + """The end time was previously stamped before any packaging work had run.""" + before = datetime.now(timezone.utc) + result = _package(stage_root, tmp_path) + record = json.loads( + (result.packaged[0].directory / "processing.json").read_text(encoding="utf-8") + ) + ends = [ + datetime.fromisoformat(p["end_date_time"]) + for p in record["data_processes"] + if p["output_parameters"].get("cells_packaged") + ] + assert ends[0] >= before From 3082b5c421f475c94faba671f7272888cee68406 Mon Sep 17 00:00:00 2001 From: Peter Grotz <199680633+peter-grotz@users.noreply.github.com> Date: Wed, 23 Sep 2026 10:01:24 -0700 Subject: [PATCH 2/3] Drop the run-level cell count and read acquisitions from the registry Every asset is a single cell, so recording how many cells the run produced says nothing about the asset it sits in. build_packaging_process no longer takes cell_count and package_cells' factory needs only the finish time, which remains necessary so the record does not claim to have finished before the work ran. The acquisition now resolves DocDB v2 -> v1 -> S3, matching default_sources, with a staged copy only as a last resort. The registry holds what the indexer published; a mounted bundle may carry a stale copy or none. check_orientation reads the axes from it, so a wrong one mis-registers every cell without failing. Both fetchers are generalised to any core field. --- src/exaspim_swc_processing/acquisition.py | 118 +++++++++++++++++++ src/exaspim_swc_processing/packaging.py | 22 ++-- src/exaspim_swc_processing/sources.py | 31 +++-- tests/test_acquisition.py | 134 ++++++++++++++++++++++ tests/test_packaging.py | 42 +++---- 5 files changed, 297 insertions(+), 50 deletions(-) create mode 100644 src/exaspim_swc_processing/acquisition.py create mode 100644 tests/test_acquisition.py diff --git a/src/exaspim_swc_processing/acquisition.py b/src/exaspim_swc_processing/acquisition.py new file mode 100644 index 0000000..51d7c1f --- /dev/null +++ b/src/exaspim_swc_processing/acquisition.py @@ -0,0 +1,118 @@ +"""Obtain a dataset's ``acquisition`` from the registry, falling back to S3. + +The transform needs the acquisition's axis orientation, and ``RegistrationPipeline`` +takes it as a file path. Where that file comes from matters: a dataset's mounted bundle +may carry a stale copy, or none, while the registry holds the record the indexer +published. + +The order matches :func:`~exaspim_swc_processing.sources.default_sources` -- DocDB v2, +then v1, then the object in S3 -- so the registry is authoritative and S3 answers only +for assets it has not indexed. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Sequence +from pathlib import Path + +from exaspim_swc_processing.parent_metadata import MetadataSource +from exaspim_swc_processing.sources import ( + DOCDB_HOST, + OPEN_DATA_BUCKET, + Fetcher, + docdb_fetcher, + s3_fetcher, +) + +logger = logging.getLogger(__name__) + +ACQUISITION_FIELD = "acquisition" +"""Top-level field of a DocDB record holding the acquisition.""" + +ACQUISITION_KEY = "acquisition.json" +"""Object at the asset root holding the acquisition.""" + + +class AcquisitionNotFoundError(LookupError): + """Raised when no source holds an acquisition for the dataset.""" + + +def acquisition_sources( + host: str = DOCDB_HOST, bucket: str = OPEN_DATA_BUCKET +) -> list[tuple[MetadataSource, Fetcher]]: + """Build the resolution order for an acquisition. + + Parameters + ---------- + host : str, optional + DocDB host, by default :data:`~exaspim_swc_processing.sources.DOCDB_HOST`. + bucket : str, optional + S3 bucket, by default :data:`~exaspim_swc_processing.sources.OPEN_DATA_BUCKET`. + + Returns + ------- + list[tuple[MetadataSource, Fetcher]] + Sources in priority order: DocDB v2, DocDB v1, then S3. + """ + return [ + (MetadataSource.DOCDB_V2, docdb_fetcher("v2", host, field=ACQUISITION_FIELD)), + (MetadataSource.DOCDB_V1, docdb_fetcher("v1", host, field=ACQUISITION_FIELD)), + (MetadataSource.S3, s3_fetcher(bucket, filename=ACQUISITION_KEY)), + ] + + +def resolve_acquisition( + asset_name: str, + destination: Path, + sources: Sequence[tuple[MetadataSource, Fetcher]] | None = None, + local_fallback: str = "", +) -> tuple[Path, MetadataSource | None]: + """Write the dataset's acquisition to a file and say where it came from. + + Parameters + ---------- + asset_name : str + Processed dataset name. + destination : Path + File to write the acquisition to when it comes from a registry or S3. + sources : Sequence[tuple[MetadataSource, Fetcher]] | None, optional + Sources in priority order. Defaults to :func:`acquisition_sources`. + local_fallback : str, optional + An acquisition already on disk, used only when no source answers. + + Returns + ------- + tuple[Path, MetadataSource | None] + The file to read, and the source it came from -- ``None`` when the local + fallback was used. + + Raises + ------ + AcquisitionNotFoundError + If no source answers and there is no usable local fallback. + """ + for source, fetch in sources if sources is not None else acquisition_sources(): + try: + document = fetch(asset_name) + except Exception as error: # noqa: BLE001 - an unreachable source is not fatal + logger.info("Acquisition source %s failed: %s", source.value, error) + continue + if not document: + continue + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(json.dumps(document, indent=2), encoding="utf-8") + logger.info("Acquisition for %s read from %s", asset_name, source.value) + return destination, source + + if local_fallback and Path(local_fallback).is_file(): + logger.warning( + "No acquisition for %s in DocDB or S3; using the copy staged at %s", + asset_name, + local_fallback, + ) + return Path(local_fallback), None + raise AcquisitionNotFoundError( + f"No acquisition for {asset_name} in DocDB v2, DocDB v1 or S3, and no local copy" + ) diff --git a/src/exaspim_swc_processing/packaging.py b/src/exaspim_swc_processing/packaging.py index 790b42a..1169afb 100644 --- a/src/exaspim_swc_processing/packaging.py +++ b/src/exaspim_swc_processing/packaging.py @@ -109,7 +109,6 @@ def build_packaging_process( start_time: datetime, end_time: datetime, output_path: str, - cell_count: int, experimenters: Sequence[str] = DEFAULT_EXPERIMENTERS, ) -> DataProcess: """Describe the packaging step itself. @@ -126,8 +125,6 @@ def build_packaging_process( When packaging finished. output_path : str Where the cell directories were written, relative to ``/results``. - cell_count : int - Number of cells packaged in the run. experimenters : Sequence[str], optional Who is responsible for the run, by default :data:`DEFAULT_EXPERIMENTERS`. @@ -149,7 +146,7 @@ def build_packaging_process( start_date_time=start_time, end_date_time=end_time, output_path=output_path, - output_parameters={**parent.provenance(), "cells_packaged": cell_count}, + output_parameters=parent.provenance(), notes="Regrouped pipeline outputs into one derived asset per reconstruction.", ) @@ -161,7 +158,7 @@ def package_cells( stage_processes: Sequence[DataProcess], pipeline: Code, creation_time: datetime, - describe_packaging: Callable[[int, datetime], DataProcess], + describe_packaging: Callable[[datetime], DataProcess], specs: Sequence[ArtifactSpec] = ARTIFACT_SPECS, overrides: Mapping[str, object] | None = None, ) -> PackagingResult: @@ -181,10 +178,11 @@ def package_cells( The pipeline that produced the run. creation_time : datetime Creation time shared by every cell in the run. - describe_packaging : Callable[[int, datetime], DataProcess] - Builds this packaging step's record, given the number of cells written and the - time the work finished. A callable rather than a record because both values are - only known once discovery has run, yet the record is embedded in every cell. + describe_packaging : Callable[[datetime], DataProcess] + Builds this packaging step's record, given the time the work finished. A + callable rather than a record because the record is embedded in every cell and + so must be built before any cell is written, yet must not claim to have finished + before the work ran. specs : Sequence[ArtifactSpec], optional Artifacts to collect, by default :data:`~exaspim_swc_processing.layout.ARTIFACT_SPECS`. overrides : Mapping[str, object] | None, optional @@ -199,8 +197,8 @@ def package_cells( specs = tuple(specs) skipped: list[SkippedCell] = [] - # Resolve every cell before writing any: the packaging record embedded in each cell - # states how many the run produced, which is not known until the last is resolved. + # Resolve every cell before writing any, so the shared packaging record can carry a + # finish time that follows the work rather than preceding it. planned = [] for reconstruction, cell in discover_cells(stage_root, specs).items(): missing = cell.missing_roles(specs) @@ -225,7 +223,7 @@ def package_cells( planned.append((reconstruction, cell, description)) finished = datetime.now(timezone.utc) - processes = [*stage_processes, describe_packaging(len(planned), finished)] + processes = [*stage_processes, describe_packaging(finished)] record = build_cell_processing(processes, pipeline) packaged: list[PackagedCell] = [] diff --git a/src/exaspim_swc_processing/sources.py b/src/exaspim_swc_processing/sources.py index 51203ca..8ec3f8f 100644 --- a/src/exaspim_swc_processing/sources.py +++ b/src/exaspim_swc_processing/sources.py @@ -96,7 +96,10 @@ def _docdb_client(version: str, host: str) -> DocDbClient: def docdb_fetcher( - version: str, host: str = DOCDB_HOST, client: DocDbClient | None = None + version: str, + host: str = DOCDB_HOST, + client: DocDbClient | None = None, + field: str = CORE_FIELD, ) -> Fetcher: """Build a fetcher reading one DocDB endpoint. @@ -112,12 +115,15 @@ def docdb_fetcher( Host serving the index, by default :data:`DOCDB_HOST`. client : DocDbClient | None, optional A pre-built client. Constructed on first use when omitted. + field : str, optional + Core field to read, by default :data:`CORE_FIELD`. Any top-level field of the + record works, e.g. ``"acquisition"``. Returns ------- Fetcher - Returns the asset's raw ``data_description``, or ``None`` if the endpoint has no - record for it. + Returns the asset's raw document for ``field``, or ``None`` if the endpoint has + no record for it. """ def _fetch(asset_name: str) -> dict | None: @@ -136,16 +142,14 @@ def _fetch(asset_name: str) -> dict | None: resolved = client if client is not None else _docdb_client(version, host) records = resolved.retrieve_docdb_records( filter_query={"name": asset_name}, - projection={CORE_FIELD: 1}, + projection={field: 1}, limit=1, ) if not records: return None - document = records[0].get(CORE_FIELD) + document = records[0].get(field) if not document: - logger.warning( - "DocDB %s has a record for %r with no %s", version, asset_name, CORE_FIELD - ) + logger.warning("DocDB %s has a record for %r with no %s", version, asset_name, field) return None return document @@ -173,6 +177,7 @@ def _s3_client() -> S3Client: def s3_fetcher( bucket: str = OPEN_DATA_BUCKET, client: S3Client | None = None, + filename: str = DATA_DESCRIPTION_KEY, ) -> Fetcher: """Build a fetcher reading the asset's ``data_description.json`` from S3. @@ -185,12 +190,14 @@ def s3_fetcher( Bucket holding the asset, by default :data:`OPEN_DATA_BUCKET`. client : S3Client | None, optional A pre-built S3 client. Constructed on first use when omitted. + filename : str, optional + Object to read from the asset root, by default :data:`DATA_DESCRIPTION_KEY`. Returns ------- Fetcher - Returns the raw ``data_description``, or ``None`` if the object is absent or is - not readable as JSON. + Returns the raw document, or ``None`` if the object is absent or is not readable + as JSON. """ def _fetch(asset_name: str) -> dict | None: @@ -207,12 +214,12 @@ def _fetch(asset_name: str) -> dict | None: The raw document, or ``None`` if absent or unreadable. """ resolved = client if client is not None else _s3_client() - key = f"{asset_name}/{DATA_DESCRIPTION_KEY}" + key = f"{asset_name}/{filename}" try: response = resolved.get_object(Bucket=bucket, Key=key) return json.loads(response["Body"].read()) except Exception as error: # noqa: BLE001 - any read failure means "not here" - logger.info("No %s in s3://%s/%s (%s)", DATA_DESCRIPTION_KEY, bucket, key, error) + logger.info("No %s in s3://%s/%s (%s)", filename, bucket, key, error) return None return _fetch diff --git a/tests/test_acquisition.py b/tests/test_acquisition.py new file mode 100644 index 0000000..952139b --- /dev/null +++ b/tests/test_acquisition.py @@ -0,0 +1,134 @@ +"""Tests for :mod:`exaspim_swc_processing.acquisition`.""" + +import json +from pathlib import Path + +import pytest + +from exaspim_swc_processing.acquisition import ( + ACQUISITION_FIELD, + ACQUISITION_KEY, + AcquisitionNotFoundError, + acquisition_sources, + resolve_acquisition, +) +from exaspim_swc_processing.parent_metadata import MetadataSource + +DATASET = "exaSPIM_826509_2026-05-12_18-03-50_processed_2026-05-22_14-04-22" +DOCUMENT = {"axes": [{"dimension": 0, "direction": "Superior_to_inferior"}]} + + +def _source(value: object) -> object: + """Return a fetcher yielding a fixed value. + + Parameters + ---------- + value : object + Document to return, or ``None`` for a source with no record. + + Returns + ------- + object + A fetcher. + """ + return lambda _name: value + + +def _raising(_name: str) -> None: + """Stand in for a source that cannot be reached. + + Parameters + ---------- + _name : str + Asset name, ignored. + + Raises + ------ + RuntimeError + Always. + """ + raise RuntimeError("endpoint down") + + +def test_the_registry_is_preferred_over_s3(tmp_path: Path) -> None: + """DocDB holds what the indexer published; a staged copy may be stale.""" + sources = [ + (MetadataSource.DOCDB_V2, _source(DOCUMENT)), + (MetadataSource.S3, _source({"axes": "wrong"})), + ] + path, source = resolve_acquisition(DATASET, tmp_path / "acq.json", sources) + assert source is MetadataSource.DOCDB_V2 + assert json.loads(path.read_text(encoding="utf-8")) == DOCUMENT + + +def test_v1_answers_when_v2_has_no_record(tmp_path: Path) -> None: + """Most exaSPIM assets are not in the v2 index.""" + sources = [ + (MetadataSource.DOCDB_V2, _source(None)), + (MetadataSource.DOCDB_V1, _source(DOCUMENT)), + ] + _, source = resolve_acquisition(DATASET, tmp_path / "acq.json", sources) + assert source is MetadataSource.DOCDB_V1 + + +def test_s3_answers_for_an_unindexed_asset(tmp_path: Path) -> None: + """The object at the dataset root is the last registry-side source.""" + sources = [ + (MetadataSource.DOCDB_V2, _source(None)), + (MetadataSource.S3, _source(DOCUMENT)), + ] + _, source = resolve_acquisition(DATASET, tmp_path / "acq.json", sources) + assert source is MetadataSource.S3 + + +def test_an_unreachable_source_is_skipped(tmp_path: Path) -> None: + """A DocDB outage must not stop a transform that S3 can serve.""" + sources = [ + (MetadataSource.DOCDB_V2, _raising), + (MetadataSource.S3, _source(DOCUMENT)), + ] + _, source = resolve_acquisition(DATASET, tmp_path / "acq.json", sources) + assert source is MetadataSource.S3 + + +def test_the_staged_copy_is_the_last_resort(tmp_path: Path) -> None: + """A mounted bundle still works when nothing is published.""" + staged = tmp_path / "staged.json" + staged.write_text(json.dumps(DOCUMENT), encoding="utf-8") + path, source = resolve_acquisition( + DATASET, tmp_path / "acq.json", [(MetadataSource.S3, _source(None))], str(staged) + ) + assert path == staged + assert source is None + + +def test_no_acquisition_anywhere_is_an_error(tmp_path: Path) -> None: + """Transforming without the axis orientation would mis-register silently.""" + with pytest.raises(AcquisitionNotFoundError, match="No acquisition"): + resolve_acquisition(DATASET, tmp_path / "acq.json", [(MetadataSource.S3, _source(None))]) + + +def test_a_missing_staged_copy_is_not_used(tmp_path: Path) -> None: + """A path that does not exist is not a fallback.""" + with pytest.raises(AcquisitionNotFoundError): + resolve_acquisition( + DATASET, + tmp_path / "acq.json", + [(MetadataSource.S3, _source(None))], + str(tmp_path / "absent.json"), + ) + + +def test_the_default_order_is_v2_then_v1_then_s3() -> None: + """The registry is authoritative; S3 answers only for unindexed assets.""" + assert [source for source, _ in acquisition_sources()] == [ + MetadataSource.DOCDB_V2, + MetadataSource.DOCDB_V1, + MetadataSource.S3, + ] + + +def test_the_field_and_object_names_match_the_schema() -> None: + """DocDB nests the acquisition under a field; S3 publishes it as a file.""" + assert ACQUISITION_FIELD == "acquisition" + assert ACQUISITION_KEY == "acquisition.json" diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 6dddec2..f3ba54a 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -100,15 +100,13 @@ def _stage_processes() -> list[DataProcess]: return [DataProcess.model_validate(record) for record in payload] -def _packaging_process(parent: ParentMetadata, cells: int = 2) -> DataProcess: +def _packaging_process(parent: ParentMetadata) -> DataProcess: """Build the packaging step record. Parameters ---------- parent : ParentMetadata Resolved parent, supplying provenance. - cells : int, optional - Number of cells packaged, by default 2. Returns ------- @@ -121,11 +119,10 @@ def _packaging_process(parent: ParentMetadata, cells: int = 2) -> DataProcess: start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), end_time=datetime(2026, 8, 20, 4, 5, tzinfo=timezone.utc), output_path="cells", - cell_count=cells, ) -def _describe(parent: ParentMetadata) -> Callable[[int, datetime], DataProcess]: +def _describe(parent: ParentMetadata) -> Callable[[datetime], DataProcess]: """Return the factory :func:`package_cells` calls once the count is known. Parameters @@ -135,18 +132,17 @@ def _describe(parent: ParentMetadata) -> Callable[[int, datetime], DataProcess]: Returns ------- - Callable[[int, datetime], DataProcess] - Builds the step record from the cell count and finish time. + Callable[[datetime], DataProcess] + Builds the step record from the finish time. """ - def build(cells: int, finished: datetime) -> DataProcess: + def build(finished: datetime) -> DataProcess: return build_packaging_process( parent, PACKAGER, start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), end_time=finished, output_path="cells", - cell_count=cells, ) return build @@ -245,7 +241,6 @@ def test_metadata_source_is_recorded_in_the_packaging_step( parameters = step.output_parameters.model_dump() assert parameters["metadata_source"] == "s3" assert parameters["upgraded_to"] == "2.4.1" - assert parameters["cells_packaged"] == 2 def test_fallback_source_is_tagged_for_later_rederivation(stage_root: Path, tmp_path: Path) -> None: @@ -320,26 +315,11 @@ def test_experimenters_can_be_overridden() -> None: start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), end_time=datetime(2026, 8, 20, 4, 5, tzinfo=timezone.utc), output_path="cells", - cell_count=2, experimenters=["Cameron Arshadi"], ) assert step.experimenters == ["Cameron Arshadi"] -def test_the_recorded_cell_count_is_the_number_actually_written( - stage_root: Path, tmp_path: Path -) -> None: - """The count was previously fixed at call time, so every cell recorded zero.""" - result = _package(stage_root, tmp_path) - record = json.loads( - (result.packaged[0].directory / "processing.json").read_text(encoding="utf-8") - ) - packaging = [ - p for p in record["data_processes"] if p["output_parameters"].get("cells_packaged") - ] - assert packaging[0]["output_parameters"]["cells_packaged"] == len(result.packaged) - - def test_packaging_ends_after_the_cells_are_resolved(stage_root: Path, tmp_path: Path) -> None: """The end time was previously stamped before any packaging work had run.""" before = datetime.now(timezone.utc) @@ -350,6 +330,16 @@ def test_packaging_ends_after_the_cells_are_resolved(stage_root: Path, tmp_path: ends = [ datetime.fromisoformat(p["end_date_time"]) for p in record["data_processes"] - if p["output_parameters"].get("cells_packaged") + if p["name"] == PACKAGING_STEP_NAME ] assert ends[0] >= before + + +def test_no_run_level_count_is_recorded(stage_root: Path, tmp_path: Path) -> None: + """Every asset is a single cell, so a count of the run's cells means nothing there.""" + result = _package(stage_root, tmp_path) + record = json.loads( + (result.packaged[0].directory / PROCESSING_FILENAME).read_text(encoding="utf-8") + ) + for process in record["data_processes"]: + assert "cells_packaged" not in (process.get("output_parameters") or {}) From 49bae4804fbe7524abc498f49738762860df447a Mon Sep 17 00:00:00 2001 From: Peter Grotz <199680633+peter-grotz@users.noreply.github.com> Date: Wed, 23 Sep 2026 10:23:21 -0700 Subject: [PATCH 3/3] Stop recording packaging as a processing step Packaging hardlinks reconstructions and writes metadata; the SWCs that come out are byte-identical to the ones that go in. A DataProcess for it describes assembling the asset rather than producing the data, in a document meant for the latter. Where the parent metadata came from stays on DataDescription.tags, which is indexed and so searchable, unlike output_parameters. Removing the step also removes the need for the finish-time factory, so package_cells returns to a single pass. --- src/exaspim_swc_processing/packaging.py | 88 ++-------------- tests/test_packaging.py | 129 +++--------------------- 2 files changed, 24 insertions(+), 193 deletions(-) diff --git a/src/exaspim_swc_processing/packaging.py b/src/exaspim_swc_processing/packaging.py index 1169afb..f1dc398 100644 --- a/src/exaspim_swc_processing/packaging.py +++ b/src/exaspim_swc_processing/packaging.py @@ -12,22 +12,24 @@ requirement that the upstream assets cannot currently satisfy, so the asset would advertise itself as incomplete. Quality control belongs to a later stage and is not written here. -The packaging step records itself as a :class:`~aind_data_schema.core.processing.DataProcess` -appended to each cell's record. It is a real processing step — it is what produced the -directory — and it carries the provenance of where the parent metadata was read from. +Packaging does not record itself as a +:class:`~aind_data_schema.core.processing.DataProcess`. It changes no data -- the +reconstructions are hardlinked, byte for byte -- so a record of it would describe +assembling the asset rather than producing the data, in a document meant for the latter. +Where the parent metadata was read from is carried on ``DataDescription.tags``, which is +indexed and therefore searchable, unlike ``output_parameters``. """ from __future__ import annotations import logging -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass -from datetime import datetime, timezone +from datetime import datetime from pathlib import Path from aind_data_schema.components.identifiers import Code -from aind_data_schema.core.processing import DataProcess, ProcessStage -from aind_data_schema_models.process_names import ProcessName +from aind_data_schema.core.processing import DataProcess from exaspim_swc_processing.data_description import ( DataDescriptionDerivationError, @@ -41,12 +43,6 @@ logger = logging.getLogger(__name__) -PACKAGING_STEP_NAME = "exaspim_swc_packaging" -"""``DataProcess.name`` of the step this module performs.""" - -DEFAULT_EXPERIMENTERS = ("Peter Grotz",) -"""Default ``DataProcess.experimenters`` for the packaging step.""" - DATA_DESCRIPTION_FILENAME = "data_description.json" PROCESSING_FILENAME = "processing.json" @@ -103,54 +99,6 @@ class PackagingResult: skipped: tuple[SkippedCell, ...] -def build_packaging_process( - parent: ParentMetadata, - code: Code, - start_time: datetime, - end_time: datetime, - output_path: str, - experimenters: Sequence[str] = DEFAULT_EXPERIMENTERS, -) -> DataProcess: - """Describe the packaging step itself. - - Parameters - ---------- - parent : ParentMetadata - Resolved parent metadata, supplying the provenance of where it was read from. - code : Code - The packaging code: repository URL, name and version. - start_time : datetime - When packaging started. - end_time : datetime - When packaging finished. - output_path : str - Where the cell directories were written, relative to ``/results``. - experimenters : Sequence[str], optional - Who is responsible for the run, by default :data:`DEFAULT_EXPERIMENTERS`. - - Returns - ------- - DataProcess - The step record, carrying :meth:`ParentMetadata.provenance` in - ``output_parameters``. Typed ``OTHER`` because regrouping files into per-cell - assets is not one of the vocabulary's defined operations; the docs direct - unmatched operations to ``ANALYSIS`` or ``OTHER``, and ``OTHER`` requires the - ``name`` and ``notes`` set here. - """ - return DataProcess( - process_type=ProcessName.OTHER, - name=PACKAGING_STEP_NAME, - stage=ProcessStage.PROCESSING, - code=code, - experimenters=list(experimenters), - start_date_time=start_time, - end_date_time=end_time, - output_path=output_path, - output_parameters=parent.provenance(), - notes="Regrouped pipeline outputs into one derived asset per reconstruction.", - ) - - def package_cells( stage_root: Path, output_root: Path, @@ -158,7 +106,6 @@ def package_cells( stage_processes: Sequence[DataProcess], pipeline: Code, creation_time: datetime, - describe_packaging: Callable[[datetime], DataProcess], specs: Sequence[ArtifactSpec] = ARTIFACT_SPECS, overrides: Mapping[str, object] | None = None, ) -> PackagingResult: @@ -178,11 +125,6 @@ def package_cells( The pipeline that produced the run. creation_time : datetime Creation time shared by every cell in the run. - describe_packaging : Callable[[datetime], DataProcess] - Builds this packaging step's record, given the time the work finished. A - callable rather than a record because the record is embedded in every cell and - so must be built before any cell is written, yet must not claim to have finished - before the work ran. specs : Sequence[ArtifactSpec], optional Artifacts to collect, by default :data:`~exaspim_swc_processing.layout.ARTIFACT_SPECS`. overrides : Mapping[str, object] | None, optional @@ -195,11 +137,10 @@ def package_cells( skipped rather than written without metadata. """ specs = tuple(specs) + record = build_cell_processing(list(stage_processes), pipeline) + packaged: list[PackagedCell] = [] skipped: list[SkippedCell] = [] - # Resolve every cell before writing any, so the shared packaging record can carry a - # finish time that follows the work rather than preceding it. - planned = [] for reconstruction, cell in discover_cells(stage_root, specs).items(): missing = cell.missing_roles(specs) if missing: @@ -220,14 +161,7 @@ def package_cells( logger.warning("Skipping %s: %s", reconstruction.stem, error) skipped.append(SkippedCell(reconstruction, str(error))) continue - planned.append((reconstruction, cell, description)) - finished = datetime.now(timezone.utc) - processes = [*stage_processes, describe_packaging(finished)] - record = build_cell_processing(processes, pipeline) - - packaged: list[PackagedCell] = [] - for reconstruction, cell, description in planned: cell_dir = output_root / description.name _write_cell_directory(cell, cell_dir, specs) (cell_dir / DATA_DESCRIPTION_FILENAME).write_text( diff --git a/tests/test_packaging.py b/tests/test_packaging.py index f3ba54a..6847beb 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -1,7 +1,6 @@ """Tests for :mod:`exaspim_swc_processing.packaging`.""" import json -from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path @@ -13,9 +12,7 @@ from exaspim_swc_processing.packaging import ( DATA_DESCRIPTION_FILENAME, - PACKAGING_STEP_NAME, PROCESSING_FILENAME, - build_packaging_process, package_cells, ) from exaspim_swc_processing.parent_metadata import MetadataSource, ParentMetadata @@ -100,54 +97,6 @@ def _stage_processes() -> list[DataProcess]: return [DataProcess.model_validate(record) for record in payload] -def _packaging_process(parent: ParentMetadata) -> DataProcess: - """Build the packaging step record. - - Parameters - ---------- - parent : ParentMetadata - Resolved parent, supplying provenance. - - Returns - ------- - DataProcess - The step record. - """ - return build_packaging_process( - parent, - PACKAGER, - start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), - end_time=datetime(2026, 8, 20, 4, 5, tzinfo=timezone.utc), - output_path="cells", - ) - - -def _describe(parent: ParentMetadata) -> Callable[[datetime], DataProcess]: - """Return the factory :func:`package_cells` calls once the count is known. - - Parameters - ---------- - parent : ParentMetadata - Resolved parent, supplying provenance. - - Returns - ------- - Callable[[datetime], DataProcess] - Builds the step record from the finish time. - """ - - def build(finished: datetime) -> DataProcess: - return build_packaging_process( - parent, - PACKAGER, - start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), - end_time=finished, - output_path="cells", - ) - - return build - - def _package(stage_root: Path, tmp_path: Path, **kwargs: object) -> object: """Package the fixture tree with default arguments. @@ -173,7 +122,6 @@ def _package(stage_root: Path, tmp_path: Path, **kwargs: object) -> object: _stage_processes(), PIPELINE, CREATION_TIME, - _describe(parent), **kwargs, ) @@ -218,31 +166,6 @@ def test_written_asset_validates_as_metadata(stage_root: Path, tmp_path: Path) - assert record.processing is not None -def test_packaging_step_is_recorded_in_each_cell(stage_root: Path, tmp_path: Path) -> None: - """The step that produced the directory appears in its own provenance.""" - result = _package(stage_root, tmp_path) - processing = Processing.model_validate_json( - (result.packaged[0].directory / PROCESSING_FILENAME).read_text(encoding="utf-8") - ) - names = [process.name for process in processing.data_processes] - assert PACKAGING_STEP_NAME in names - assert processing.dependency_graph[PACKAGING_STEP_NAME] == ["aligned_swc_processing"] - - -def test_metadata_source_is_recorded_in_the_packaging_step( - stage_root: Path, tmp_path: Path -) -> None: - """Where the parent metadata came from is auditable from the published asset.""" - result = _package(stage_root, tmp_path) - processing = Processing.model_validate_json( - (result.packaged[0].directory / PROCESSING_FILENAME).read_text(encoding="utf-8") - ) - step = next(p for p in processing.data_processes if p.name == PACKAGING_STEP_NAME) - parameters = step.output_parameters.model_dump() - assert parameters["metadata_source"] == "s3" - assert parameters["upgraded_to"] == "2.4.1" - - def test_fallback_source_is_tagged_for_later_rederivation(stage_root: Path, tmp_path: Path) -> None: """Tags are indexed, so fallback-derived assets can be found and rebuilt.""" result = _package(stage_root, tmp_path) @@ -298,48 +221,22 @@ def test_overrides_apply_to_every_cell(stage_root: Path, tmp_path: Path) -> None assert description.project_name == "Neuron Reconstruction" -def test_packaging_process_records_the_output_path() -> None: - """The step points at where it wrote.""" - step = _packaging_process(_parent()) - assert str(step.output_path) == "cells" - assert step.process_type == "Other" - assert step.notes - assert step.experimenters == ["Peter Grotz"] - - -def test_experimenters_can_be_overridden() -> None: - """A run attributable to a person records that person.""" - step = build_packaging_process( - _parent(), - PACKAGER, - start_time=datetime(2026, 8, 20, 4, tzinfo=timezone.utc), - end_time=datetime(2026, 8, 20, 4, 5, tzinfo=timezone.utc), - output_path="cells", - experimenters=["Cameron Arshadi"], - ) - assert step.experimenters == ["Cameron Arshadi"] - - -def test_packaging_ends_after_the_cells_are_resolved(stage_root: Path, tmp_path: Path) -> None: - """The end time was previously stamped before any packaging work had run.""" - before = datetime.now(timezone.utc) +def test_packaging_does_not_record_itself_as_a_process(stage_root: Path, tmp_path: Path) -> None: + """Hardlinking files into a directory produces no data, so it is not a process.""" result = _package(stage_root, tmp_path) - record = json.loads( - (result.packaged[0].directory / "processing.json").read_text(encoding="utf-8") + processing = Processing.model_validate_json( + (result.packaged[0].directory / PROCESSING_FILENAME).read_text(encoding="utf-8") ) - ends = [ - datetime.fromisoformat(p["end_date_time"]) - for p in record["data_processes"] - if p["name"] == PACKAGING_STEP_NAME - ] - assert ends[0] >= before + names = [process.name for process in processing.data_processes] + assert names == [process.name for process in _stage_processes()] + assert "exaspim_swc_packaging" not in names + assert set(processing.dependency_graph) == set(names) -def test_no_run_level_count_is_recorded(stage_root: Path, tmp_path: Path) -> None: - """Every asset is a single cell, so a count of the run's cells means nothing there.""" +def test_the_metadata_source_stays_discoverable_on_tags(stage_root: Path, tmp_path: Path) -> None: + """Dropping the process loses output_parameters; tags carry the source and index.""" result = _package(stage_root, tmp_path) - record = json.loads( - (result.packaged[0].directory / PROCESSING_FILENAME).read_text(encoding="utf-8") + description = json.loads( + (result.packaged[0].directory / DATA_DESCRIPTION_FILENAME).read_text(encoding="utf-8") ) - for process in record["data_processes"]: - assert "cells_packaged" not in (process.get("output_parameters") or {}) + assert any(tag.startswith("metadata-source:") for tag in description["tags"])