diff --git a/docs/validation/official30_qualification_330_20261001.rst b/docs/validation/official30_qualification_330_20261001.rst new file mode 100644 index 0000000000..203dc1eec4 --- /dev/null +++ b/docs/validation/official30_qualification_330_20261001.rst @@ -0,0 +1,112 @@ +Official30 qualification after main synchronization (#330) +========================================================= + +Production source ``4c40348354a0680850ac21cdc102d00e460558ee`` includes main +``2cbfc4a0771bf24d153b076bff4a5e321ac039f4`` and its exact dependency pins. +All 30 native-reference comparisons pass, with zero differences. The original +six compile failures and the subsequent eight execution/output failures are +resolved. This correctness gate enables representative performance work; +it is not a speedup claim. + +The determining owners +---------------------- + +``MeasurementSubject.row_identity_domain`` derives compatible row identities +without merging exact source-qualified image subjects. Common output admission +uses the derived domain; payload policies still enforce their own cardinality. +Ordinary/native single-subject payloads and incompatible scopes/object domains +retain their rejection controls. + +Carrier admission follows the authoritative compiled predecessor through +inherited bindings. ``CompiledStepPlan`` owns terminal-source projection, and +every producer transition and terminal physical header remain mandatory. + +``StepInputDependency`` determines which anchors originate at pipeline start; +inherited source bindings cannot discard predecessor/artifact-only lifecycle +anchors. ``CallableContract.group_scope_inputs`` owns execution-domain source +selection. The planner projects preliminary source bindings through those exact +refs, then uses its existing producer/lineage resolver. ExampleFly's auxiliary +OrigBlue filename source cannot restrict the composed RGBImage execution domain. +The typed output manifest remains authoritative for actual producer pixels. + +``OpenHCSMetadataWriter.OutputTarget`` derives production and reconciliation +targets. Runtime artifact targets enumerate actual declared writer destinations, +and durable typed source projections survive execution-local value-store cleanup. +The declared root remains included so unknown image addresses still fail closed. +Materialization's existing record factory owns the empty-output-plan case; +callers do not add absence guards around its execution-local store. + +Terminal and streaming materialization policies now live in the shared +materialization package. Internal diagnostic planes declare persistence without +external export intent. Source serialization preserves each nominal policy +subtype, including its options, backends and write mode. No function-name +exceptions, ignored extra science outputs, generated-name coordinate inference +or CSV normalization are used. + +Validation and limits +--------------------- + +* 586 local tests and 20 subtests pass; one existing optional Napari skip. +* Original CI-pinned R0 passes for OpenHCS, scripts and benchmark, without increases. +* Original NRA R1 passes its two policy-configured detectors with complete + dependency context and no increases; this is not an all-detector clean claim. +* Original ClassDef census: 5,122 declarations, 5,110 projected, all 12 OPEN retained. +* NRA admitted the two-class policy relocation and its import-consumer closure. + The initial binding-authority/annotation boundary rejection is retained. + Authored semantic changes have separate executed tests and native comparison. +* Native progression: 22/30, then 29/30, then 30/30 on the synced source above. + All final difference counts are zero. Existing 1e-6 absolute/relative numerical + tolerances, exact discrete/identity checks and declared-image checks remain. + +Tests include real image writes into two declared destinations, cleanup and +durable reopen with exact metadata/pixels, unknown-address rejection, implicit +main-flow controls, stored cross-axis and direct-source payload domains, and +three materialization-policy source roundtrips. Three stale fixtures were +reproduced on unchanged main and repaired from authoritative declarations/source +fields and the existing shared ACK fixture; scientific assertions remain strict. + +The final gate uses the production comparison suite with the retained native +outputs from ``official30-native-complete-fresh-20260928``. Those outputs are +scientific references. The native clocks and speedup columns in the raw reports +are retained observations, **not fresh comparative performance evidence**. This +gate does not regenerate native clocks or scaling figures and does not assert +complete parity for every possible ROI/export profile outside the suite. + +Ready-server startup and shutdown are excluded from pipeline clocks. The suite +uses one inline worker, one numerical thread, CPU5, and no memory observer. +Comparison and qualification work are separate from execution timing. Tests, +audits, builds and other replays do not overlap timed pipeline work. + +Receipts and reproduction +------------------------ + +``official30_qualification_330_evidence_20261001.tgz`` contains source and +dependency receipts, exact task recipes, original guards/census, local test XML, +unchanged-main controls, NRA move evidence, the saved ExampleFly anchor trace, +all three qualification summaries/observations, and the final 30 generated +pipeline sources and measured receipts. ``SHA256SUMS`` authenticates its files. +Large scientific inputs/outputs remain in the benchmark-run workspace. + +Extract the archive, check out the validated production revision, and use its +final qualification recipe with the configured benchmark Python and retained +native-reference root. The original controller environment was: + +.. code-block:: bash + + env PYTHONPATH="$PROJECT_ROOT" \ + OPENHCS_REFERENCE_EXPORT_PIPELINES_ROOT="$PROJECT_ROOT/benchmark/reference_exports/official30_value_completion_20260914" \ + OPENHCS_CPU_ONLY=true NUMBA_CACHE_DIR="$SHARED_NUMBA_CACHE" \ + OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1 \ + NUMEXPR_NUM_THREADS=1 VECLIB_MAXIMUM_THREADS=1 PYTHONHASHSEED=0 \ + taskset -c 5 "$BENCHMARK_PYTHON" "$QUALIFICATION_RECIPE" + +The recipe records its source assertion, manifest, retained native-reference +receipt and production ``run_comparison_suite`` arguments. Its local paths must +resolve to those retained inputs. If they are unavailable, produce fresh native +outputs using the public comparison route and configured CellProfiler environment. +The public throughput route remains ``scripts/benchmark_cppipe_well_throughput.py`` +with ``--manifest benchmark/manifests/official30_portable_axis1.json --mode 1w_1t``. + +This change fixes #330 and references #315, #214 and #324. It does not close +#315's separate parent/live acceptance. Generic runtime ownership remains PR326, +and execution/compilation performance work continues under #162. diff --git a/docs/validation/official30_qualification_330_evidence_20261001.tgz b/docs/validation/official30_qualification_330_evidence_20261001.tgz new file mode 100644 index 0000000000..35e7d883c0 Binary files /dev/null and b/docs/validation/official30_qualification_330_evidence_20261001.tgz differ diff --git a/openhcs/core/artifacts.py b/openhcs/core/artifacts.py index 7e56929a59..04755abaa8 100644 --- a/openhcs/core/artifacts.py +++ b/openhcs/core/artifacts.py @@ -708,9 +708,16 @@ def require_output_subject( @classmethod def validate_output_declaration(cls, spec: "ArtifactSpec") -> None: - """Explicit subject relations must agree under every recording owner.""" + """Reject incompatible row domains before payload-owner validation. - ArtifactSpecRelation.measurement_subject_for_output(spec) + Source-qualified image rows share one image-set domain. Their recording + owner determines whether multiple subjects can occupy the output; native + and ordinary recorded tables still require one exact subject. + """ + + subjects = ArtifactSpecRelation.measurement_subjects_for_output(spec) + if len({subject.row_identity_domain for subject in subjects}) > 1: + ArtifactSpecRelation.measurement_subject_for_output(spec) @classmethod def validate_native_output_declaration(cls, spec: "ArtifactSpec") -> None: @@ -1271,12 +1278,12 @@ def measurement_subject(self) -> "MeasurementSubject | None": return None @staticmethod - def measurement_subject_for_output( + def measurement_subjects_for_output( output: "ArtifactSpec | ArtifactOutputPlan", - ) -> "MeasurementSubject | None": - """Resolve the sole subject from an output's original relation owners.""" + ) -> "tuple[MeasurementSubject, ...]": + """Derive all exact subjects from the original relation owners.""" - subjects = tuple( + return tuple( dict.fromkeys( subject for relation in output.relations @@ -1284,6 +1291,14 @@ def measurement_subject_for_output( if subject is not None ) ) + + @staticmethod + def measurement_subject_for_output( + output: "ArtifactSpec | ArtifactOutputPlan", + ) -> "MeasurementSubject | None": + """Resolve one exact subject for a table-wide subject owner.""" + + subjects = ArtifactSpecRelation.measurement_subjects_for_output(output) if len(subjects) > 1: raise ValueError( f"Artifact output {output.ref()!r} declares multiple measurement " @@ -2008,8 +2023,7 @@ def merge_existing( and existing.sidecar_role is not incoming.sidecar_role ): raise ValueError( - f"Conflicting {self.role} artifact sidecar role for " - f"'{incoming.name}'." + f"Conflicting {self.role} artifact sidecar role for '{incoming.name}'." ) sidecar_role = ( existing.sidecar_role diff --git a/openhcs/core/compiled_step_plan.py b/openhcs/core/compiled_step_plan.py index 97af352073..0282c2d04e 100644 --- a/openhcs/core/compiled_step_plan.py +++ b/openhcs/core/compiled_step_plan.py @@ -234,7 +234,9 @@ def has_persistent_target(self) -> bool: def persists_to_backend(self, backend: str) -> bool: """Own admission of a storage target to persistent artifact publication.""" - return self.has_persistent_target and self.require_persistent_backend() == backend + return ( + self.has_persistent_target and self.require_persistent_backend() == backend + ) def require_persistent_backend(self) -> str: if self.persistent_backend is None: @@ -313,6 +315,14 @@ class CompiledStepPlan: chainbreaker: bool = False error: str | None = None + @property + def requires_terminal_source_projection(self) -> bool: + """Require an exact source anchor when declared bindings have no predecessor.""" + return ( + bool(self.source_binding_plan.binding_declarations) + and self.main_input_dependency.predecessor_step_index() is None + ) + def require_function_execution_ready(self) -> "CompiledStepPlan": """Validate the compiler-owned fields required by FunctionStep runtime.""" if not self.axis_id: diff --git a/openhcs/core/function_patterns.py b/openhcs/core/function_patterns.py index 7f5a771789..2c599eec76 100644 --- a/openhcs/core/function_patterns.py +++ b/openhcs/core/function_patterns.py @@ -807,6 +807,11 @@ class CompiledFunctionPattern: groups: tuple[CompiledFunctionGroup, ...] is_grouped: bool + @property + def runtime_domain(self) -> RuntimeInvocationDomain: + """Derive the shared anchor domain from the compiled invocations.""" + return RuntimeInvocationDomain.from_invocations(tuple(self.iter_invocations())) + @property def execution_scope(self) -> FunctionStepExecutionScope: """Return the uniform lifecycle scope derived from its invocations.""" diff --git a/openhcs/core/pipeline/artifact_planning.py b/openhcs/core/pipeline/artifact_planning.py index 8f77e5efc2..34695d84cc 100644 --- a/openhcs/core/pipeline/artifact_planning.py +++ b/openhcs/core/pipeline/artifact_planning.py @@ -33,28 +33,11 @@ CsvOptions, ImageFileOptions, MaterializedFilenameIdentity, - MaterializationSpec, ROIOptions, + StreamingOnlyMaterializationSpec, + TerminalMaterializationSpec, ) - -class TerminalMaterializationSpec(MaterializationSpec): - """Compiler-added persistence excluded from declared export comparison.""" - - def participates_in_runtime_export_observation(self) -> bool: - return False - - -class StreamingOnlyMaterializationSpec(MaterializationSpec): - """Compiler-added viewer materialization excluded from persistent exports.""" - - def participates_in_runtime_export_observation(self) -> bool: - return False - - def participates_in_persistent_materialization(self) -> bool: - return False - - class AutomaticArtifactOutputMaterializationStrategy( ArtifactTypeStrategyMatchMixin, MostDerivedContextStrategyMixin[type[ArtifactType]], diff --git a/openhcs/core/pipeline/compiler.py b/openhcs/core/pipeline/compiler.py index 1aae3ab073..8bbf601152 100644 --- a/openhcs/core/pipeline/compiler.py +++ b/openhcs/core/pipeline/compiler.py @@ -1451,7 +1451,7 @@ def validate_primary_image_carrier_requirements( source_binding_plan is None or not source_binding_plan.primary_plane_bindings ): - if plan.source_binding_plan.binding_declarations: + if plan.requires_terminal_source_projection: failures.append( f"{owner}: no exact primary source-binding " "projection represents this callable's main flow." @@ -1611,7 +1611,7 @@ def _carrier_source_plan( and current_source_binding_plan.primary_plane_bindings ): return producer, current_source_binding_plan - if producer.source_binding_plan.binding_declarations: + if producer.requires_terminal_source_projection: failures.append( f"{owner}: carrier requirement {requirement.value!r} has no " f"exact primary source-binding projection at producer step " diff --git a/openhcs/core/pipeline/path_planner.py b/openhcs/core/pipeline/path_planner.py index d561ab433d..8813ed637b 100644 --- a/openhcs/core/pipeline/path_planner.py +++ b/openhcs/core/pipeline/path_planner.py @@ -323,6 +323,19 @@ def get_execution_groups( if input_component_scopes is None else input_component_scopes ) + artifact_owned = bool(contracts) and all( + contract.group_scope_inputs for contract in contracts + ) + if artifact_owned: + source_bindings = ( + self.planner.source_bindings_for_snapshot(snapshot) + if source_bindings is None + else source_bindings + ).for_artifact_refs( + spec.ref() + for contract in contracts + for spec in contract.group_scope_inputs + ) scope = component_scopes.scope_for_group_by( group_by, ) @@ -337,7 +350,7 @@ def get_execution_groups( if source_scope.is_ungrouped else source_scope ) - if contracts and all(contract.group_scope_inputs for contract in contracts): + if artifact_owned: scope = self.artifact_owned_execution_scope( snapshot, contracts, diff --git a/openhcs/core/runtime_measurements.py b/openhcs/core/runtime_measurements.py index 0c2282fcbd..c42e5266e3 100644 --- a/openhcs/core/runtime_measurements.py +++ b/openhcs/core/runtime_measurements.py @@ -1605,6 +1605,17 @@ def source_image_name(self) -> str | None: return None return self.name + @property + def row_identity_domain(self) -> tuple[MeasurementScope, str | None, str | None]: + """Return the row domain independently of an image's source qualifier. + + Named image subjects share image-set row identity. Object and relationship + names identify different row domains and cannot be combined implicitly. + """ + + name = None if self.scope is MeasurementScope.IMAGE else self.name + return self.scope, name, self.id_field + @property def object_name(self) -> str | None: """Return the concrete object set represented by this subject, if any.""" diff --git a/openhcs/core/step_dependencies.py b/openhcs/core/step_dependencies.py index e0a752d6cc..fe5d188831 100644 --- a/openhcs/core/step_dependencies.py +++ b/openhcs/core/step_dependencies.py @@ -98,6 +98,10 @@ def predecessor_step_index(self) -> int | None: ) return self.source_step_index + def uses_pipeline_start_anchors(self) -> bool: + """Only a pipeline-start main input is filtered against raw source bindings.""" + return self.kind.is_pipeline_start + def require_pipeline_start(self) -> None: """Prove that this terminal dependency reaches the pipeline source.""" diff --git a/openhcs/core/steps/function_artifact_materialization.py b/openhcs/core/steps/function_artifact_materialization.py index 71e1ddf9b0..7d4a0da320 100644 --- a/openhcs/core/steps/function_artifact_materialization.py +++ b/openhcs/core/steps/function_artifact_materialization.py @@ -1201,12 +1201,11 @@ def runtime_artifact_materializations( """Derive actual materializations from compiled outputs and runtime values.""" materializations: list[RuntimeArtifactMaterialization] = [] - store = context.runtime_value_store for output_plan in plan.artifact_outputs.values(): if output_plan.materialization is None: continue records = actual_materialization_records( - store=store, + store=context.runtime_value_store, plan=plan, output_plan=output_plan, ) diff --git a/openhcs/core/steps/function_execution.py b/openhcs/core/steps/function_execution.py index e376923be2..1b075b9d82 100644 --- a/openhcs/core/steps/function_execution.py +++ b/openhcs/core/steps/function_execution.py @@ -265,8 +265,7 @@ def execution_group_anchor_patterns( pattern = self.plan.compiled_function_pattern source_owns_groups = ( - RuntimeInvocationDomain.from_invocations(tuple(pattern.iter_invocations())) - is RuntimeInvocationDomain.SOURCE_ANCHORED + pattern.runtime_domain is RuntimeInvocationDomain.SOURCE_ANCHORED ) if not grouped_patterns.groups: return grouped_patterns @@ -316,7 +315,7 @@ def source_bound_anchor_patterns( ) -> PatternGroups: """Restrict source-bound step anchors to compatible declared sources.""" - if self.plan.main_input_dependency.kind is StepInputDependencyKind.STEP_OUTPUT: + if not self.plan.main_input_dependency.uses_pipeline_start_anchors(): return grouped_patterns if not self.plan.source_binding_plan.has_primary_content: return grouped_patterns diff --git a/openhcs/core/steps/function_outputs.py b/openhcs/core/steps/function_outputs.py index 9fc7cb773d..919d4acaa5 100644 --- a/openhcs/core/steps/function_outputs.py +++ b/openhcs/core/steps/function_outputs.py @@ -6,7 +6,7 @@ import time from abc import ABC, abstractmethod from collections.abc import Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import ClassVar @@ -83,6 +83,8 @@ from openhcs.core.virtual_workspace_metadata import ( METADATA_CONFIG, AtomicMetadataWriter, + OpenHCSMetadataSubdirectories, + VirtualWorkspaceSourceProjectionEntries, ) from openhcs.microscopes.microscope_interfaces import FilenameParser @@ -722,9 +724,22 @@ def for_execution( return tuple( target for declaration in cls.__registry__.values() - if (target := declaration.from_execution(context, plan)) is not None + if (owner := declaration.from_execution(context, plan)) is not None + for target in owner.production_targets(context, plan) ) + def production_targets( + self, context: ProcessingContext, plan: CompiledStepPlan + ) -> tuple[OpenHCSMetadataWriter.OutputTarget, ...]: + """Project this declaration's exact storage destinations for the step.""" + return (self,) + + def reconciliation_targets( + self, context: ProcessingContext + ) -> tuple[OpenHCSMetadataWriter.OutputTarget, ...]: + """Resolve destinations after runtime values have been released.""" + return (self,) + def produced_records( self, context: ProcessingContext, plan: CompiledStepPlan ) -> tuple[ProducedOutputSemantics, ...]: @@ -805,7 +820,7 @@ def produced_projection_metadata( if context.filemanager is None: raise ValueError("OpenHCS metadata requires a file manager.") target = type(self).from_plan(plan) - if target != self: + if target is None or self not in target.production_targets(context, plan): raise ValueError( "Produced metadata plan does not own this output target." ) @@ -905,10 +920,9 @@ def project_runtime_artifacts( context, output_path_filter=ImageFileFormat.is_image_path, ): - if ( - not ImageFileFormat.is_image_path(output.path) - or Path(output.path).parent != Path(self.output_dir) - ): + if not ImageFileFormat.is_image_path(output.path) or Path( + output.path + ).parent != Path(self.output_dir): continue if output.metadata is None: raise ValueError( @@ -1022,9 +1036,10 @@ def finalize_completed_plate( for target in cls.OutputTarget.for_plan(plan): target_contexts.setdefault(target, context) - for target, context in target_contexts.items(): - if target.contains_images(context): - target.write(context) + for owner, context in target_contexts.items(): + for target in owner.reconciliation_targets(context): + if target.contains_images(context): + target.write(context) class ProducedImageMetadataCapability: @@ -1108,14 +1123,51 @@ def from_plan( class RuntimeArtifactMetadataTarget(OpenHCSMetadataWriter.OutputTarget): """Image artifacts persist independently of main-flow image/checkpoint storage.""" - @classmethod - def from_execution( - cls, context: ProcessingContext, plan: CompiledStepPlan - ) -> RuntimeArtifactMetadataTarget | None: - target = super().from_execution(context, plan) - if target is None or not target.contains_images(context): - return None - return target + def production_targets( + self, context: ProcessingContext, plan: CompiledStepPlan + ) -> tuple[RuntimeArtifactMetadataTarget, ...]: + """Derive directories from the same writer outputs used to save artifacts.""" + directories = dict.fromkeys( + Path(output.path).parent + for materialization in runtime_artifact_materializations(plan, context) + if materialization.spec.participates_in_persistent_materialization() + for output in materialization.outputs( + plan, context, output_path_filter=ImageFileFormat.is_image_path + ) + if ImageFileFormat.is_image_path(output.path) + ) + return tuple( + target + for directory in directories + for target in (self.for_directory(directory),) + if target.contains_images(context) + ) + + def reconciliation_targets( + self, context: ProcessingContext + ) -> tuple[RuntimeArtifactMetadataTarget, ...]: + """Use durable typed projections, without reloading cleaned artifact values.""" + subdirectories = OpenHCSMetadataSubdirectories.from_path( + METADATA_CONFIG.metadata_path(self.plate_root) + ) + directories = tuple( + Path(self.plate_root) / Path(path).parent + for _name, subdirectory in subdirectories.items() + for path, projection in VirtualWorkspaceSourceProjectionEntries.from_subdirectory( + subdirectory + ).entries.items() + if isinstance(projection, SourceArtifactProjection) + and projection.ref.backend == self.backend + ) + return tuple( + self.for_directory(directory) + for directory in dict.fromkeys((self.output_dir, *directories)) + ) + + def for_directory(self, directory: Path) -> RuntimeArtifactMetadataTarget: + """Retain the compiled plate/backend while projecting one declared directory.""" + sub_dir = directory.relative_to(self.plate_root) + return replace(self, output_dir=directory, sub_dir=str(sub_dir)) @classmethod def from_plan(cls, plan: CompiledStepPlan) -> RuntimeArtifactMetadataTarget | None: diff --git a/openhcs/core/virtual_workspace_metadata.py b/openhcs/core/virtual_workspace_metadata.py index 6e8cc71b6c..6ef9709f73 100644 --- a/openhcs/core/virtual_workspace_metadata.py +++ b/openhcs/core/virtual_workspace_metadata.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import logging import os from collections.abc import Mapping, Sequence @@ -354,6 +355,14 @@ class OpenHCSMetadataSubdirectories: metadata: OpenHCSMetadataPayload + @classmethod + def from_path(cls, path: Path) -> OpenHCSMetadataSubdirectories: + """Load the durable projection document for completed-plate reconciliation.""" + if not path.is_file(): + return cls({}) + with path.open(encoding="utf-8") as stream: + return cls(json.load(stream)) + def items(self) -> tuple[tuple[str, OpenHCSSubdirectoryPayload], ...]: subdirectories = self.metadata.get(FIELDS.SUBDIRECTORIES) if subdirectories is None: diff --git a/openhcs/processing/backends/cellprofiler/primary_object_diagnostics.py b/openhcs/processing/backends/cellprofiler/primary_object_diagnostics.py index 253aac98d3..2aa02996ae 100644 --- a/openhcs/processing/backends/cellprofiler/primary_object_diagnostics.py +++ b/openhcs/processing/backends/cellprofiler/primary_object_diagnostics.py @@ -30,11 +30,10 @@ from openhcs.core.runtime_object_labels import ObjectLabelPayload, ObjectLabelVariant from openhcs.processing.materialization import ( ImageFileOptions, - MaterializationSpec, + TerminalMaterializationSpec, MaterializedFilenameIdentity, ) - @dataclass(frozen=True, slots=True) class DiagnosticPlaneSource: """Decode source identity and validity once, sharing one mask across stages.""" @@ -175,7 +174,7 @@ def artifact_specs( ImageArtifactType, sidecar_role=ArtifactSidecarRole.QA_CHECKPOINT, viewer_streaming=ArtifactViewerStreaming.ON_DEMAND, - materialization=MaterializationSpec( + materialization=TerminalMaterializationSpec( ImageFileOptions( filename_suffix=".tif", filename_identity=MaterializedFilenameIdentity.ARTIFACT_NAME, diff --git a/openhcs/processing/materialization/__init__.py b/openhcs/processing/materialization/__init__.py index a8e58d6dde..4e920f0ee9 100644 --- a/openhcs/processing/materialization/__init__.py +++ b/openhcs/processing/materialization/__init__.py @@ -34,6 +34,10 @@ MaterializationRelativePathScope, SharedMaterializationRelativePathScope, ) +from openhcs.processing.materialization.persistence import ( + StreamingOnlyMaterializationSpec, + TerminalMaterializationSpec, +) from openhcs.processing.materialization.presets import ( csv_dataclass_materializer, csv_materializer, @@ -51,6 +55,8 @@ "MaterializationFormat", "WriteMode", "MaterializationSpec", + "TerminalMaterializationSpec", + "StreamingOnlyMaterializationSpec", "MaterializationContext", "Output", "PathHelper", diff --git a/openhcs/processing/materialization/persistence.py b/openhcs/processing/materialization/persistence.py new file mode 100644 index 0000000000..7d976d1d09 --- /dev/null +++ b/openhcs/processing/materialization/persistence.py @@ -0,0 +1,20 @@ +"""Shared persistence and export-observation policies for materialized artifacts.""" + +from .core import MaterializationSpec + + +class TerminalMaterializationSpec(MaterializationSpec): + """Retain internal artifacts without declaring an external pipeline export.""" + + def participates_in_runtime_export_observation(self) -> bool: + return False + + +class StreamingOnlyMaterializationSpec(MaterializationSpec): + """Prepare viewer outputs without declaring persistent or external exports.""" + + def participates_in_runtime_export_observation(self) -> bool: + return False + + def participates_in_persistent_materialization(self) -> bool: + return False diff --git a/openhcs/serialization/pycodify_formatters.py b/openhcs/serialization/pycodify_formatters.py index 7ce679c45b..02d0fc1a7d 100644 --- a/openhcs/serialization/pycodify_formatters.py +++ b/openhcs/serialization/pycodify_formatters.py @@ -201,12 +201,13 @@ def can_format(self, value) -> bool: return isinstance(value, MaterializationSpec) def format(self, value, context: FormatContext) -> SourceFragment: + materialization_type = type(value) import_pair = ( - "openhcs.processing.materialization.core", - "MaterializationSpec", + materialization_type.__module__, + materialization_type.__name__, ) class_name = NameMappingLookup.resolve( - context, import_pair, "MaterializationSpec" + context, import_pair, materialization_type.__name__ ) item_ctx = context.indented() output_frags = [to_source(output, item_ctx) for output in value.outputs] diff --git a/tests/conftest.py b/tests/conftest.py index 1333c85ee0..b166ac8ab1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,6 +7,7 @@ ensure_source_checkout_external_paths() import pytest +from zmqruntime.messages import AckReturnRoute, ProcessIdentity from polystore.imagej_distribution import ( FijiArchiveDistribution, @@ -27,6 +28,16 @@ pytest_plugins = [] +@pytest.fixture +def viewer_ack_return_route(): + """Nominal producer route for message construction without a live listener.""" + return AckReturnRoute( + url="tcp://127.0.0.1:8111", + incarnation="00000000-0000-0000-0000-000000000001", + owner=ProcessIdentity.current(), + ) + + @pytest.fixture(autouse=True) def cleanup_test_runtime_resources(): """Release connection and process resources after every test.""" diff --git a/tests/unit/test_artifact_output_materialization_planner.py b/tests/unit/test_artifact_output_materialization_planner.py index b317d95fd6..2f1878fe87 100644 --- a/tests/unit/test_artifact_output_materialization_planner.py +++ b/tests/unit/test_artifact_output_materialization_planner.py @@ -1,4 +1,5 @@ import numpy as np +from openhcs.processing.materialization.persistence import TerminalMaterializationSpec from polystore.filemanager import FileManager from polystore.memory import MemoryStorageBackend @@ -12,7 +13,6 @@ from openhcs.core.pipeline.artifact_planning import ( AutomaticArtifactOutputMaterializationStrategy, ArtifactOutputMaterializationPlanner, - TerminalMaterializationSpec, ) from openhcs.core.runtime_exports import RuntimeExportExpectation from openhcs.processing.materialization import ( diff --git a/tests/unit/test_artifact_output_ownership.py b/tests/unit/test_artifact_output_ownership.py index 48b69abc57..7d43b8eab8 100644 --- a/tests/unit/test_artifact_output_ownership.py +++ b/tests/unit/test_artifact_output_ownership.py @@ -146,7 +146,9 @@ def test_scalar_input_ambiguity_precedes_dependent_output_subject_validation( rows = ArtifactSpec.output( "Measurements", MeasurementsArtifactType, - relations=tuple(ObjectMeasurementSubjectRelation(spec.ref()) for spec in inputs), + relations=tuple( + ObjectMeasurementSubjectRelation(spec.ref()) for spec in inputs + ), ) @runtime_adapter( @@ -203,6 +205,67 @@ def measure(image, *, runtime): compile_function_pattern(measure, {}, {}) +@pytest.mark.parametrize( + "output_policy", + [NativeReturnArtifactOutputPolicy, AdapterRecordedArtifactOutputPolicy], +) +def test_table_wide_subject_owner_rejects_distinct_named_image_subjects(output_policy): + images = tuple( + ArtifactSpec.output(name, ImageArtifactType) for name in ("Stain1", "Stain2") + ) + measurements = ArtifactSpec.output( + "Measurements", + MeasurementsArtifactType, + relations=tuple( + ImageMeasurementSubjectRelation(image.ref()) for image in images + ), + ) + + @runtime_adapter( + "runtime", lambda request: object(), artifact_output_policy=output_policy + ) + @artifact_outputs(*images, measurements) + def measure(image, *, runtime): + raise AssertionError("Invalid declarations must not execute") + + with pytest.raises(ValueError, match="multiple measurement subjects"): + compile_function_pattern(measure, {}, {}) + + +def test_cellprofiler_row_owner_preserves_distinct_named_image_subjects(): + from openhcs.core.artifacts import ArtifactSpecRelation + from openhcs.processing.backends.cellprofiler.image_quality import ( + MeasureImageQualityModule, + ) + + images = tuple( + ArtifactSpec.output(name, ImageArtifactType) for name in ("Stain1", "Stain2") + ) + measurements = ArtifactSpec.output( + "Measurements", + MeasurementsArtifactType, + relations=tuple( + ImageMeasurementSubjectRelation(image.ref()) for image in images + ), + measurement_feature_owner=MeasureImageQualityModule, + ) + + @runtime_adapter( + "runtime", + lambda request: object(), + artifact_output_policy=CellProfilerRecordedArtifactOutputPolicy, + ) + @artifact_outputs(*images, measurements) + def measure(image, *, runtime): + raise AssertionError("Compile admission must not execute a callable") + + compiled = compile_function_pattern(measure, {}, {}) + subjects = ArtifactSpecRelation.measurement_subjects_for_output(measurements) + assert tuple(subject.name for subject in subjects) == ("Stain1", "Stain2") + assert len({subject.row_identity_domain for subject in subjects}) == 1 + assert compiled is not None + + def test_real_cellprofiler_declaration_compiles_without_a_table_wide_subject(): """The real module/provider declares the owner, not a compiler exemption.""" from openhcs.core.compiled_step_plan import CompiledStepPlan @@ -276,12 +339,17 @@ def test_real_cellprofiler_declaration_compiles_without_a_table_wide_subject(): authored = next(normalize_function_pattern(step.func).iter_items()) contract = provider.plans[(0, authored.key)].contract assert contract.artifact_output_policy is CellProfilerRecordedArtifactOutputPolicy - assert contract.metadata.artifact_output_policy is CellProfilerRecordedArtifactOutputPolicy + assert ( + contract.metadata.artifact_output_policy + is CellProfilerRecordedArtifactOutputPolicy + ) (measurement,) = contract.artifact_outputs.of_artifact_type( MeasurementsArtifactType ) assert measurement.measurement_feature_owner is IdentifyPrimaryObjectsModule - assert measurement.require_measurement_feature_owner() is IdentifyPrimaryObjectsModule + assert ( + measurement.require_measurement_feature_owner() is IdentifyPrimaryObjectsModule + ) assert ArtifactSpecRelation.measurement_subject_for_output(measurement) is None compiled = compile_function_pattern( diff --git a/tests/unit/test_cellprofiler_artifact_declarations.py b/tests/unit/test_cellprofiler_artifact_declarations.py index 6b6f2fb318..1515156c70 100644 --- a/tests/unit/test_cellprofiler_artifact_declarations.py +++ b/tests/unit/test_cellprofiler_artifact_declarations.py @@ -564,9 +564,24 @@ def test_declarations_carry_cross_step_object_and_measurement_flow() -> None: assert identify_contract.artifact_outputs.names_of_artifact_type( ObjectLabelsArtifactType ) == ("Nuclei",) + from openhcs.processing.backends.cellprofiler.primary_object_diagnostics import ( + PrimaryObjectDiagnosticPlanes, + ) + + identify_objects = ( + identify_contract.artifact_outputs.require_by_name_and_artifact_type( + "Nuclei", ObjectLabelsArtifactType + ) + ) assert identify_contract.artifact_outputs.names() == ( "IdentifyPrimaryObjects_1_measurements", "Nuclei", + *( + spec.name + for spec in PrimaryObjectDiagnosticPlanes.artifact_specs( + source_image=source_image, objects=identify_objects + ) + ), ) identify_measurement = ( identify_contract.artifact_outputs.require_by_name_and_artifact_type( @@ -574,12 +589,6 @@ def test_declarations_carry_cross_step_object_and_measurement_flow() -> None: MeasurementsArtifactType, ) ) - identify_objects = ( - identify_contract.artifact_outputs.require_by_name_and_artifact_type( - "Nuclei", - ObjectLabelsArtifactType, - ) - ) source_ref = ArtifactSpec.input("OrigBlue", ImageArtifactType).ref() assert identify_measurement.relations == ( ArtifactSpecRelation(source_ref), diff --git a/tests/unit/test_compiled_execution.py b/tests/unit/test_compiled_execution.py index 29f7bbd1c4..6ab25aa560 100644 --- a/tests/unit/test_compiled_execution.py +++ b/tests/unit/test_compiled_execution.py @@ -44,9 +44,8 @@ ) from openhcs.core.runtime_stores import RuntimeValueStore from openhcs.core.runtime_tabular_values import FieldSpec -from openhcs.core.pipeline.artifact_planning import TerminalMaterializationSpec from openhcs.processing.materialization import CsvOptions, MaterializationSpec - +from openhcs.processing.materialization.persistence import TerminalMaterializationSpec def _runtime_environment() -> CompiledRuntimeEnvironmentPlan: return CompiledRuntimeEnvironmentPlan( diff --git a/tests/unit/test_compiler_auto_materialization_policy.py b/tests/unit/test_compiler_auto_materialization_policy.py index 2d57a9f307..d41da9788c 100644 --- a/tests/unit/test_compiler_auto_materialization_policy.py +++ b/tests/unit/test_compiler_auto_materialization_policy.py @@ -5,7 +5,6 @@ import pytest from openhcs.core.compiled_step_plan import RuntimeArtifactMaterializationPlan -from openhcs.core.pipeline.artifact_planning import TerminalMaterializationSpec from openhcs.core.pipeline.compiler import PipelineCompiler from openhcs.core.pipeline.materialization_flag_planner import ( MaterializationFlagPlanner, @@ -15,7 +14,7 @@ MaterializationSpec, ROIOptions, ) - +from openhcs.processing.materialization.persistence import TerminalMaterializationSpec def test_disabling_automatic_materialization_keeps_only_declared_exports( monkeypatch, diff --git a/tests/unit/test_function_artifact_materialization.py b/tests/unit/test_function_artifact_materialization.py index 1559f16542..bca6321c5a 100644 --- a/tests/unit/test_function_artifact_materialization.py +++ b/tests/unit/test_function_artifact_materialization.py @@ -6,6 +6,7 @@ import numpy as np import pytest +from openhcs.processing.materialization.persistence import TerminalMaterializationSpec from polystore.base import DataSink, ensure_storage_registry, storage_registry from polystore.filemanager import FileManager from polystore.napari_stream import NapariStreamingBackend @@ -61,7 +62,6 @@ RuntimeContextObservation, RuntimeExecutionObservation, ) -from openhcs.core.pipeline.artifact_planning import TerminalMaterializationSpec from openhcs.core.pipeline.function_contracts import artifact_outputs from openhcs.core.runtime_artifact_values import ( RuntimeValue, @@ -619,7 +619,6 @@ def test_slice_aligned_object_label_arrays_preserve_source_slice_metadata(): assert dict(runtime_value.data.source_component_metadata or {}) == { "well": "A02", "channel": 1, - "extension": ".tif", } assert runtime_value.data.source_image_names == () assert runtime_value.data.source_image_provenance_planes == ( @@ -3668,7 +3667,9 @@ def fake_materialize( ) -def test_materialize_artifact_outputs_streams_singleton_roi_plane_from_output_plan(): +def test_materialize_artifact_outputs_streams_singleton_roi_plane_from_output_plan( + viewer_ack_return_route, +): output_plan = ArtifactOutputPlan( name="Nuclei", path="/memory/Nuclei.pkl", @@ -3721,6 +3722,7 @@ def test_materialize_artifact_outputs_streams_singleton_roi_plane_from_output_pl streamed_item = StreamingBatchMessageBuilder.build( napari_backend, StreamingBatchMessageRequest( + return_route=viewer_ack_return_route, data_list=[roi_content], file_paths=[roi_path], stream_request=stream_request, diff --git a/tests/unit/test_function_execution_lifecycle_anchors.py b/tests/unit/test_function_execution_lifecycle_anchors.py index 7396d972c6..288c3eecc3 100644 --- a/tests/unit/test_function_execution_lifecycle_anchors.py +++ b/tests/unit/test_function_execution_lifecycle_anchors.py @@ -4,6 +4,8 @@ from types import SimpleNamespace +import pytest + from openhcs.constants.constants import AllComponents from openhcs.core.artifacts import ( ArtifactInputPlan, @@ -19,11 +21,10 @@ from openhcs.core.pipeline.function_contracts import artifact_inputs from openhcs.core.pipeline.path_planner import PathPlanner, PathPlannerArtifactStage from openhcs.core.runtime_adapters import runtime_adapter -from openhcs.core.source_bindings import CompiledSourceBindingPlan +from openhcs.core.source_bindings import CompiledSourceBindingPlan, NamedSourceBinding from openhcs.core.step_dependencies import StepInputDependency from openhcs.core.steps.function_execution import PatternGroups, StepAnchorPatternFilter - def _anchor_filter(plan: object, output_manifest: object) -> StepAnchorPatternFilter: return StepAnchorPatternFilter( plan=plan, @@ -34,7 +35,17 @@ def _anchor_filter(plan: object, output_manifest: object) -> StepAnchorPatternFi ) -def test_storage_backed_cross_group_uses_producer_lifecycle_anchor() -> None: +@pytest.mark.parametrize( + "dependency", + ( + StepInputDependency.step_output( + source_step_index=4, + source_step_scope_id="non-adjacent-main-flow-producer", + ), + StepInputDependency.no_main_flow(), + ), +) +def test_storage_backed_cross_group_uses_producer_lifecycle_anchor(dependency) -> None: measurements = ArtifactSpec.input( "measurements", MeasurementsArtifactType, @@ -89,11 +100,10 @@ def consume_measurements(image, *, runtime): plan = SimpleNamespace( step_index=9, step_name="StorageBackedConsumer", - main_input_dependency=StepInputDependency.step_output( - source_step_index=4, - source_step_scope_id="non-adjacent-main-flow-producer", + main_input_dependency=dependency, + source_binding_plan=CompiledSourceBindingPlan( + bindings=(NamedSourceBinding(alias="OrigDNA"),) ), - source_binding_plan=CompiledSourceBindingPlan.empty(), execution_group_value="channel", execution_group_scope=ComponentGroupScope.from_raw( ("1",), diff --git a/tests/unit/test_function_outputs.py b/tests/unit/test_function_outputs.py index ebd66c519f..3e2a76a8fc 100644 --- a/tests/unit/test_function_outputs.py +++ b/tests/unit/test_function_outputs.py @@ -19,9 +19,12 @@ from openhcs.constants.constants import AllComponents, Backend, VariableComponents from openhcs.core.aligned_image_payload import AlignedImageSliceContext from openhcs.core.artifacts import ( + ArtifactOutputPlan, ImageArtifactType, ObjectLabelsArtifactType, ) +from openhcs.core.runtime_artifact_values import RuntimeValue +from openhcs.core.runtime_stores import RuntimeValueStore from openhcs.core.axis_filter import StepAxisFilterResolution, StepAxisFilterSet from openhcs.core.compiled_step_plan import ( CompiledStepPlan, @@ -88,7 +91,11 @@ from openhcs.microscopes.openhcs import OpenHCSMetadataHandler from openhcs.microscopes.source_schema import SourceSchemaFilenameParser from openhcs.processing.materialization.core import Output - +from openhcs.processing.materialization import ( + ImageFileOptions, + MaterializationSpec, + MaterializedFilenameIdentity, +) @pytest.mark.parametrize("backend", [Backend.ZARR.value, "custom-array-store"]) def test_memory_output_writer_projects_runtime_image_payload_for_array_storage( @@ -1753,6 +1760,13 @@ def test_runtime_multiplane_label_artifact_projects_persisted_source_binding( lambda _plan, _context: (materialization,), ) + output_plan = ArtifactOutputPlan( + name="neurite_outgrowth", + path="/memory/neurite_outgrowth.pkl", + artifact_type=ObjectLabelsArtifactType, + materialization=MaterializationSpec(ImageFileOptions()), + ) + plan.artifact_outputs = {output_plan.ref(): output_plan} target = RuntimeArtifactMetadataTarget.from_plan(plan) assert target is not None [(projection, virtual_path)] = target.runtime_artifact_projection_paths( @@ -2167,9 +2181,44 @@ def test_runtime_image_metadata_target_requires_persisted_images(tmp_path, conte plan = function_step_plan("SaveImages") plan.write_backend = Backend.MEMORY.value plan.output_plate_root = str(tmp_path) + plan.output_dir = directory plan.analysis_results_dir = str(directory) plan.runtime_artifact_materialization = RuntimeArtifactMaterializationPlan( - persistent_enabled=True, persistent_backend=Backend.DISK.value, + persistent_enabled=True, + persistent_backend=Backend.DISK.value, + ) + output_plan = ArtifactOutputPlan( + name="SavedImage", + path="/memory/SavedImage.pkl", + artifact_type=ImageArtifactType, + materialization=MaterializationSpec( + ImageFileOptions( + filename_suffix=".tif", + filename_identity=MaterializedFilenameIdentity.SOURCE_IDENTITY, + ) + ), + ) + plan.artifact_outputs = {output_plan.ref(): output_plan} + context.runtime_value_store = RuntimeValueStore() + context.runtime_value_store.record( + RuntimeValue.normalize( + output_plan, + ImageMetadataPayload( + np.ones((2, 2), dtype=np.uint8), + ImagePayloadMetadata( + source_component_metadata={ + "well": "A01", + "site": "1", + "channel": "2", + "z_index": "1", + "timepoint": "1", + } + ), + ), + axis_id="A01", + ), + path=output_plan.path, + backend=Backend.MEMORY.value, ) target = RuntimeArtifactMetadataTarget.from_plan(plan) assert target is not None @@ -2183,3 +2232,90 @@ def test_runtime_image_metadata_target_requires_persisted_images(tmp_path, conte OpenHCSMetadataWriter.write(context, plan) assert not (tmp_path / "openhcs_metadata.json").exists() assert directory.exists() == (contents != "absent") + + +@pytest.mark.parametrize("stray_image", [False, True]) +def test_declared_image_destinations_publish_and_reconcile_after_value_cleanup( + tmp_path, stray_image +): + filemanager = FileManager( + { + Backend.DISK.value: DiskStorageBackend(), + Backend.MEMORY.value: MemoryStorageBackend(), + } + ) + context = context_stub(filemanager, parser=SourceSchemaFilenameParser()) + context.runtime_value_store = RuntimeValueStore() + context.metadata_cache = {} + context.tiff_config = None + plan = function_step_plan("Declared image exports") + plan.streaming_configs = {} + plan.write_backend = Backend.MEMORY.value + plan.output_plate_root = str(tmp_path) + plan.output_dir = tmp_path / "images" + plan.sub_dir = "images" + plan.analysis_results_dir = str(tmp_path / "results") + plan.create_openhcs_metadata = True + plan.runtime_artifact_materialization = RuntimeArtifactMaterializationPlan( + persistent_enabled=True, + persistent_backend=Backend.DISK.value, + ) + context.step_plans = {plan.step_index: plan} + components = { + "well": "A01", + "site": "1", + "channel": "2", + "z_index": "1", + "timepoint": "1", + } + pixels = np.arange(20, dtype=np.uint16).reshape(4, 5) + payload = ImageMetadataPayload( + pixels, + ImagePayloadMetadata( + source_component_metadata=components, + source_dtype="uint16", + ), + ) + for name, directory in (("Copy", "nested/copies"), ("Review", "review")): + output_plan = ArtifactOutputPlan( + name=name, + path=f"/memory/{name}.pkl", + artifact_type=ImageArtifactType, + materialization=MaterializationSpec( + ImageFileOptions( + filename_suffix=".tif", + filename_identity=MaterializedFilenameIdentity.SOURCE_IDENTITY, + relative_path_template=f"{directory}/A01_s001_w2_z001_t001.tif", + ) + ), + ) + plan.artifact_outputs[output_plan.ref()] = output_plan + context.runtime_value_store.record( + RuntimeValue.normalize(output_plan, payload, axis_id="A01"), + path=output_plan.path, + backend=Backend.MEMORY.value, + ) + RuntimeArtifactMaterializationAuthority.materialize(context, plan) + OpenHCSMetadataWriter.write(context, plan) + context.runtime_value_store.clear() + if stray_image: + tifffile.imwrite(tmp_path / "images/review/unowned.tif", pixels) + with pytest.raises(MetadataWriteError, match="lack typed produced addresses"): + OpenHCSMetadataWriter.finalize_completed_plate({"A01": context}) + return + OpenHCSMetadataWriter.finalize_completed_plate({"A01": context}) + subdirectories = json.loads((tmp_path / "openhcs_metadata.json").read_text())[ + FIELDS.SUBDIRECTORIES + ] + assert set(subdirectories) == {"images/nested/copies", "images/review"} + for directory, subdirectory in subdirectories.items(): + path = f"{directory}/A01_s001_w2_z001_t001.tif" + assert subdirectory[FIELDS.IMAGE_FILES] == [path] + entries = VirtualWorkspaceSourceProjectionEntries.from_subdirectory( + subdirectory + ).entries + assert set(entries) == {path} + projection = entries[path] + assert projection.component_value(AllComponents.CHANNEL) == "2" + assert projection.source_metadata["site"] == "1" + np.testing.assert_array_equal(tifffile.imread(tmp_path / path), pixels) diff --git a/tests/unit/test_materialization_core.py b/tests/unit/test_materialization_core.py index f1669802a0..97d1cdab0b 100644 --- a/tests/unit/test_materialization_core.py +++ b/tests/unit/test_materialization_core.py @@ -34,7 +34,6 @@ ViewerStreamSourceIdentity, ) from zmqruntime.config import TransportMode -from zmqruntime.messages import AckReturnRoute, ProcessIdentity from zmqruntime.viewer_protocol import ViewerTransportEndpoint, ViewerWireField import openhcs # noqa: F401 @@ -374,16 +373,6 @@ class _TestViewerMicroscopeHandler(ViewerMicroscopeHandlerABC): metadata_handler = _TestViewerMetadataHandler() -@pytest.fixture -def viewer_ack_return_route(): - """Nominal producer route for message construction without a live listener.""" - return AckReturnRoute( - url="tcp://127.0.0.1:8111", - incarnation="00000000-0000-0000-0000-000000000001", - owner=ProcessIdentity.current(), - ) - - def _viewer_stream_backend_kwargs(): display_config = _TestViewerDisplayConfig() request = ViewerStreamRequest( diff --git a/tests/unit/test_path_planner_materialization.py b/tests/unit/test_path_planner_materialization.py index ba182ad70a..3e963ae484 100644 --- a/tests/unit/test_path_planner_materialization.py +++ b/tests/unit/test_path_planner_materialization.py @@ -3791,6 +3791,70 @@ def test_non_dict_group_by_uses_source_binding_identity_for_pipeline_start_scope ) +@pytest.mark.parametrize("stored_payload", (False, True)) +def test_auxiliary_source_does_not_restrict_declared_payload_execution(stored_payload): + payload = ArtifactSpec.input("ComposedImage", ImageArtifactType) + prefix = ArtifactSpec.input("FilenameSource", ImageArtifactType) + saved = ArtifactSpec.output( + "SavedImage", + ImageArtifactType, + relations=(GroupLineageSourceRelation(source=payload.ref()),), + ) + + @artifact_inputs(payload, prefix) + @artifact_outputs(saved) + def save_declared_payload(image): + return image + + planner = _artifact_planner_stub() + bindings = [ + NamedSourceBinding( + alias=prefix.name, + component_identity=(ComponentSelector(AllComponents.CHANNEL, "1"),), + ) + ] + if stored_payload: + _record_declared_output( + planner, + ArtifactOutputPlan( + name=payload.name, + path="/memory/composed.pkl", + artifact_type=payload.artifact_type, + group_keys=("1", "2", "3"), + group_component=AllComponents.SITE, + ), + ) + else: + bindings.append( + NamedSourceBinding( + alias=payload.name, + component_identity=(ComponentSelector(AllComponents.CHANNEL, "3"),), + ) + ) + snapshot = _snapshot( + is_function_step=True, + func=save_declared_payload, + group_by=GroupBy.CHANNEL, + variable_components=(VariableComponents.SITE,), + source_bindings=StepSourceBindingsConfig(enabled=True, bindings=tuple(bindings)), + input_source=InputSource.PREVIOUS_STEP, + ) + + scope = planner.execution_groups.get_execution_groups( + snapshot, + PathPlannerComponentScopes.empty(), + contracts=(CallableContract.from_callable(save_declared_payload),), + ) + + expected = ( + PathPlannerGroupScope.dynamic(AllComponents.CHANNEL) + if stored_payload + else PathPlannerGroupScope.from_raw(("3",), component=AllComponents.CHANNEL) + ) + assert scope == expected + assert not scope.contains_runtime_key("1") or scope.is_dynamic + + def test_main_flow_source_anchor_restricts_execution_to_its_exact_channel(): planner = _artifact_planner_stub() source_bindings = StepSourceBindingsConfig( diff --git a/tests/unit/test_primary_image_carrier_compile_gate.py b/tests/unit/test_primary_image_carrier_compile_gate.py index 517ae75b4b..5056b6dd0c 100644 --- a/tests/unit/test_primary_image_carrier_compile_gate.py +++ b/tests/unit/test_primary_image_carrier_compile_gate.py @@ -14,6 +14,7 @@ PrimaryImageCarrierTransition, preserves_primary_image_carrier, ) +from openhcs.core.artifacts import ArtifactSpec, ImageArtifactType from openhcs.core.compiled_step_plan import CompiledStepPlan from openhcs.core.component_group_scope import ComponentGroupScope from openhcs.core.context.processing_context import ProcessingContext @@ -22,6 +23,8 @@ CompiledFunctionInvocation, CompiledFunctionPattern, FunctionInvocationKey, + InvocationArtifactInputEdgePlan, + InvocationArtifactInputProjectionKey, ) from openhcs.core.pipeline.compilation_session import CompilationSession from openhcs.core.pipeline.compiler import PipelineCompiler @@ -51,11 +54,30 @@ def main_flow_plan_ancestry( return CompilationSession.main_flow_plan_ancestry(self, index) -def _compiled_pattern(callable_=color_to_gray) -> CompiledFunctionPattern: +def _compiled_pattern( + callable_=color_to_gray, *, input_image_name: str | None = None +) -> CompiledFunctionPattern: invocation = CompiledFunctionInvocation( key=FunctionInvocationKey(callable_.__name__, "default", 0), contract=CallableContract.from_callable(callable_), ) + if input_image_name is not None: + (edge_key,) = InvocationArtifactInputProjectionKey.for_input_count( + invocation.key, 1 + ) + invocation = invocation.with_artifact_input_edges( + ( + InvocationArtifactInputEdgePlan( + key=edge_key, + spec=ArtifactSpec.input( + input_image_name, ImageArtifactType, parameter_name="image" + ), + storage_plan=None, + projection=None, + consumes_main_flow=True, + ), + ) + ) return CompiledFunctionPattern( groups=(CompiledFunctionGroup("default", (invocation,)),), is_grouped=False, @@ -477,6 +499,63 @@ def unknown_transform(image): PipelineCompiler.validate_primary_image_carrier_requirements(session) +@pytest.mark.parametrize("source_kind", ("rgb", "gray", "missing-anchor")) +def test_inherited_bindings_trace_exact_named_carrier_edges( + tmp_path: Path, source_kind: str +) -> None: + path = tmp_path / "source.tif" + if source_kind == "gray": + tifffile.imwrite(path, np.zeros((8, 9), dtype=np.uint8)) + else: + tifffile.imwrite(path, np.zeros((8, 9, 3), dtype=np.uint8), photometric="rgb") + session = _session(tmp_path, (path,)) + bindings = session.plans[0].source_binding_plan + + @preserves_primary_image_carrier + def crop_like(image): + return image + + plans = {} + for index, (callable_, input_name) in enumerate( + ( + (crop_like, "missing" if source_kind == "missing-anchor" else "image"), + (crop_like, "FirstCrop"), + (color_to_gray, "SecondCrop"), + ) + ): + plans[index] = CompiledStepPlan( + step_index=index, + step_name=f"Carrier step {index}", + step_type="FunctionStep", + axis_id="A01", + step_scope_id=f"step-{index}", + main_input_dependency=( + StepInputDependency.pipeline_start() + if index == 0 + else StepInputDependency.step_output( + source_step_index=index - 1, + source_step_scope_id=f"step-{index - 1}", + ) + ), + source_binding_plan=bindings, + compiled_function_pattern=_compiled_pattern( + callable_, input_image_name=input_name + ), + ) + session.plans = plans + session.context.step_plans = plans + if source_kind == "rgb": + PipelineCompiler.validate_primary_image_carrier_requirements(session) + else: + expected = ( + "source_channel_axis" + if source_kind == "gray" + else "exact primary source-binding projection" + ) + with pytest.raises(ValueError, match=expected): + PipelineCompiler.validate_primary_image_carrier_requirements(session) + + def test_generic_main_flow_preservation_does_not_prove_carrier( tmp_path: Path, ) -> None: diff --git a/tests/unit/test_primary_object_diagnostics.py b/tests/unit/test_primary_object_diagnostics.py index 55534f1546..6c8f88cf27 100644 --- a/tests/unit/test_primary_object_diagnostics.py +++ b/tests/unit/test_primary_object_diagnostics.py @@ -129,6 +129,8 @@ def test_diagnostic_contract_has_source_producer_stage_identity_not_new_objects( assert spec.sidecar_role is ArtifactSidecarRole.QA_CHECKPOINT assert not spec.participates_in_main_flow assert spec.viewer_streaming is ArtifactViewerStreaming.ON_DEMAND + assert spec.materialization.participates_in_persistent_materialization() + assert not spec.materialization.participates_in_runtime_export_observation() assert spec.materialization.outputs == ( ImageFileOptions( filename_suffix=".tif", diff --git a/tests/unit/test_pycodify_formatters.py b/tests/unit/test_pycodify_formatters.py index 8a264d457e..b006698435 100644 --- a/tests/unit/test_pycodify_formatters.py +++ b/tests/unit/test_pycodify_formatters.py @@ -29,7 +29,13 @@ measure_colocalization_objects, ) from openhcs.processing.backends.cellprofiler.shape import measure_object_size_shape - +from openhcs.processing.materialization import ( + ImageFileOptions, + MaterializationSpec, + StreamingOnlyMaterializationSpec, + TerminalMaterializationSpec, + WriteMode, +) def configurable_test_function(image, threshold: int = 3, enabled: bool = True): return image @@ -42,6 +48,33 @@ def _source(value, *, clean_mode: bool = True) -> str: ) +@pytest.mark.parametrize( + "spec_type, exports, persists", + ( + (MaterializationSpec, True, True), + (TerminalMaterializationSpec, False, True), + (StreamingOnlyMaterializationSpec, False, False), + ), +) +def test_materialization_source_roundtrip_preserves_nominal_intent( + spec_type, exports, persists +): + original = spec_type( + ImageFileOptions(filename_suffix=".tif"), + allowed_backends=["disk"], + write_mode=WriteMode.ERROR, + ) + namespace = {} + exec(_source(original), namespace) + restored = namespace["config"] + assert type(restored) is spec_type + assert restored.outputs == original.outputs + assert restored.allowed_backends == original.allowed_backends + assert restored.write_mode is WriteMode.ERROR + assert restored.participates_in_runtime_export_observation() is exports + assert restored.participates_in_persistent_materialization() is persists + + def test_function_reference_formats_from_declared_identity_without_resolution( monkeypatch, ):