diff --git a/docs/validation/automatic-point-domain-152-20261005.rst b/docs/validation/automatic-point-domain-152-20261005.rst new file mode 100644 index 0000000000..7feae93997 --- /dev/null +++ b/docs/validation/automatic-point-domain-152-20261005.rst @@ -0,0 +1,168 @@ +Automatic ROI publication: one rendered geometric source domain +============================================================== + +Source checkpoint only; installed/public acceptance is pending. Normal merges +now include qualified732 head2f4c68ff05876efb48e14743ac8027064b1b09ab and actual +mainf8f39e136098ce528f9847d07116dcb790610c43. The PR is stacked on the original +732 owner branch while that dependency remains unmerged; it must not bulk-deliver +732 or the historical494 Root stack. Existing receiving targets and Dewey's +sole public client remain unchanged. + +Determining source and ownership +------------------------------- + +The point writer already validates the exact source-plane domain through +ROIFractionalZ.source_component_domain. A MeasurementTable has no image +plane_axis. RuntimeArtifactMaterialization previously tried to derive stream +domains before rendering; PointROIOptions does not emit image-plane slices, +so that path fell back to the common scalar identity and discarded Z. Native +Points then correctly rejected a fractional coordinate without projected Z. +Saved archive reopening already consumes the validated geometric domain. + +PointROIOutput now owns that same validated geometry domain and its first-plane +route anchor. Output's existing publication adapter consumes those behaviors. +The target prepares one MaterializationBatch, derives stream metadata from the +actual backend-accepted outputs, then saves that same batch. No writer runs a +second time. The earlier spec/data reconstruction of stream domains is removed. +No pixel-plane axis is fabricated for a point table. No missing-axis guard, +source codec, stored provenance, compiler, renderer or scientific callable is +changed. True other-well/site/channel/time, disordered/duplicate-Z, calibration +and out-of-domain coordinates remain rejected by the original domain owner. + +Source census used the existing refactor-audit measure_source owner on28 relevant +production/test files selected by declarations, zero parse omissions. It covers +RuntimeArtifactMaterialization, MaterializationBatch, Output, PointROIOptions, +ROIFractionalZ, stream metadata authority and their imports/calls. Dynamic +receiver identity is semantic review, not an AST proof. Actual dependency +ViewerStreamRequest/BackendKwargs were read from receiving12's installed +PolyStore; no dependency source or installed files are edited. + +Singer explicitly released these materialization/projection hunks. Her743 +invocation/module/object-row-policy changes are disjoint; receiving12 is not +overlaid. Borrowed operations remain byte-identical to the recorded source. + +Remaining qualification +----------------------- + +Original POINT-DOMAIN-CONTROLS02 completed86PASS in7.13 seconds, whole helper +terminal0 in13.81 seconds, peak467052KiB RSS, no swaps. It exercises actual +materialization/save/request construction with source-Z origins0 and10, +unchanged fractional2.375/features, full four-plane domains without an invented +image axis, native Points model reopening and original malformed-domain +negatives. Source tier only: no installed MCP/native/public endpoint was started. +Original controls01 remains82PASS/4FAIL: missing replace import in this patch +and an older receiver fixture without the current source-element identity +method. Both corrections use existing owners; the real NapariStreamLayerItem +replaces the incomplete SimpleNamespace, no production identity fallback. + +Exact logs/times are retained in the existing issue-batch engineering494 root: +POINT-DOMAIN-CONTROLS01.log/.time and POINT-DOMAIN-CONTROLS02.log/.time. + +The28 legacy prepare spies were migrated to observe the original backend-kwargs +owner after actual rendering, without replacing its batch or saver. Empty-label +metadata-only fixtures now contain tiny real labels and use their declared +min_area0; metadata/domain assertions remain. The RGB fixture now explicitly +declares its output-image source address rather than relying on a fake save to +invent a scalar channel from three source inputs. No production RGB fallback +was added. + +Original FAMILY03 retained201PASS/10FAIL; nine cases observed no ROI because +their old fixtures were empty, and one had the undeclared RGB scalar source. +FAMILY04 remains210PASS/1FAIL: the original payload-scope volume-label control +expected a represented Z domain. That assertion is unchanged and now passes. +The completed label-family repair is described below; the old negative is not +relabelled or erased. + +Completed volume-label geometry owner +------------------------------------- + +ROIOutput extends Output's existing source-address, domain and item-field hooks. +PointROIOutput retains its stricter fractional-Z declaration. Both live output +publication and StreamingService's saved archive reload consume +ROIArchiveSourceMetadata.source_component_domain. Label geometry comes from +the original PolyStore writer's plane_indices/plane_shape and the complete +source-plane provenance, including empty label planes. The route address uses +that domain's first plane, not a common scalar address with Z omitted. + +The former NapariShapePlaneMetadata reader is deleted. ROIPlaneMetadata owns +the same external geometry fields for materialization, archive reload and +native Shapes, delegating index/rank/bounds validation to the existing +RuntimeProjectionPlaneMetadata. Mixed indexed/unindexed shapes, incomplete +source domains and incompatible component extents remain errors. This closes +IDEN-1 (pixel-axis versus geometric-domain identity), IMPL-12 (duplicate reader) +and BOUND-2 (bypassed existing plane validation); it adds no registry, codec, +stored axis mirror or compiler reconstruction. + +The existing image metadata projector still owns image fields and singleton +compiled component projection. ROI geometry only supplements those fields. +ROIMaterializationPlaneMetadataAuthority's existing declared stack composition +is unchanged. Disk/native-model controls cover both a payload with no image +plane_axis and the original compiled Z projection, at source-Z origins0 and10. +They preserve four source planes even when only planes0 and2 contain contours, +label7, calibration2/.65/.65 and actual saved-archive geometry. The native +Shapes model preserves the local Z positions; this is not a Qt or MCP claim. + +Final POINT-DOMAIN-FAMILY10 completed386PASS across nine related test files in +16.71 seconds. Original helper terminal0, elapsed22.04 seconds, peak724056KiB +RSS, swaps0. The original211 family, singleton projections, fractional points, +strict malformed-domain controls and the full native streaming-handler source +family pass. Two older native fixtures were migrated to the existing visible +layer and source-item-derived feature contracts; no identity fallback or +production guard was added. New control fixture constructor/path/wire mistakes +and the singleton regression are retained in FAMILY05 through FAMILY09, not +claimed as production failures or overwritten. + +Final named AST evidence uses the existing refactor-audit measure_source owner: +43 related production/test files plus three actual installed dependency files, +254 named sites, zero parse omissions. POINT-DOMAIN-AST05 retains the prior +owner/consumer snapshot; AST06 pins final file bytes. This is focused static +evidence with semantic MRO/call review, not a global NRA/R1 completeness claim. +External ROI extractor/converter and viewer transport declarations were read +from receiving09's retained dependency backer; no dependency install was edited. + +Original evidence under the same engineering494 root:: + + POINT-DOMAIN-FAMILY10.log 6c3db41eaea3b29c118fd05d1629b51cbe8c30eeed56462b58e888d639a616bd + POINT-DOMAIN-FAMILY10.time c12f049178962dc688297dc3915a690241b02dc3d10f82d8e99f250a38deb607 + POINT-DOMAIN-AST06.json 32cc000ad83501334903318a35f6aab08a60f68f24acd49ca2af3ccdc26c4e6d + +The batch uses the retained paired interpreter and engineering722/source_controls.py, +which reports the source owner and receiving09 dependency backer explicitly. +No client, native, viewer, environment or scientific execution was launched. + +The remaining live boundary is public automatic producer settlement and +same-viewer saved point/volume-label reload with exact fractional-Z, complete +domain, calibration, features and XY/XZ/YZ placement. It needs one ordinary +qualified package and an explicitly handed-over existing engineering route. +Dewey's original95/743 and subsequent732 receiving remain separate and untouched. + +The separately retained saved two-channel label witness is not claimed repaired: +its two-dimensional geometry is mislabeled with a SOURCE_BINDING pixel axis. +That needs the original label producer's plane-versus-contributor declaration, +not rounding points, inventing scalar channel or weakening native route guards. + +Normal current-main integration +------------------------------- + +The qualified732 merge is58a745d1a063d90691ea0584f06d18f157d40f74; +the subsequent main merge is1435f215c0b02531b888bb4f54a989a9a5498115. +Neither changed borrowed scripts/blind_analysis operations or external gitlinks. +Foreign untracked diagnostics and submodule dirt remain untouched. + +Original POINT-DOMAIN-INTEGRATION11 completed390PASS in17.76 seconds; the +same source runner terminated0, elapsed23.25 seconds, peak729892KiB RSS, +swaps0. It used the retained receiving09 dependency backer, not an overlay on +receiving13. No test was rerun to relabel an earlier failure. + +Original evidence under engineering494:: + + POINT-DOMAIN-INTEGRATION11.log 2883cef3e402477f2e53e494288bb65f171a9fedb03de664989ea857c677a522 + POINT-DOMAIN-INTEGRATION11.time f8972f6ce1bef7ee76b0c96560ddab0c81e063bde5d490fbaf6cdddc724628eb + +Singer's remaining522 work is confined to viewer_controls.py and +napari_viewer_server.py: declared XYZ point-coordinate navigation and its +handler hook. Existing main source-member selection and retirement are retained. +This integration does not implement or claim acceptance of those remaining hooks. +One future ordinary package and exact released incarnation after732 will exercise +the own tiny four-plane synthetic producer, automatic settlement and saved +point/label reopening. That public boundary remains unverified. diff --git a/openhcs/core/image_file_serialization.py b/openhcs/core/image_file_serialization.py index 2a93f8125e..1d4f0ef2f5 100644 --- a/openhcs/core/image_file_serialization.py +++ b/openhcs/core/image_file_serialization.py @@ -478,11 +478,16 @@ def storage_config( self, payload: Any, configured: TiffConfig | None ) -> TiffConfig | None: metadata = image_payload_metadata(payload) - if not metadata.persists_whole_image(): + if metadata.persists_whole_image(): + axes = list("ZYX") + elif metadata.plane_axis is not None: + # Q is a container frame, not a guessed physical Z/channel axis. + # Its exact OpenHCS component domain remains in source projection. + axes = list("QYX") + else: return configured data = image_payload_data(payload) channel_axis = metadata.normalized_source_channel_axis(data) - axes = list("ZYX") planarconfig = None photometric = TiffPhotometric.MINISBLACK if channel_axis is not None: @@ -501,7 +506,7 @@ def storage_config( photometric = TiffPhotometric.RGB if len(axes) != data.ndim: raise ValueError( - "Intrinsic TIFF pixels must retain exactly their declared Z/Y/X and channel axes." + "TIFF pixels must retain exactly their declared plane, Y/X and channel axes." ) return replace( configured if configured is not None else TiffConfig(), diff --git a/openhcs/core/roi_source_metadata.py b/openhcs/core/roi_source_metadata.py index f6d90351fc..adfb3275d4 100644 --- a/openhcs/core/roi_source_metadata.py +++ b/openhcs/core/roi_source_metadata.py @@ -3,20 +3,127 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from dataclasses import replace +from dataclasses import dataclass, replace +from math import prod from typing import ClassVar from polystore.roi import ROI +from zmqruntime.viewer_protocol import ViewerWireValue from openhcs.core.runtime_image_values import ImagePayloadMetadata +from openhcs.core.runtime_slice_projection import RuntimeProjectionPlaneMetadata +from openhcs.core.source_image_provenance import SourceComponentMetadata from openhcs.serialization.json import to_jsonable +@dataclass(frozen=True, slots=True) +class ROIPlaneMetadata: + """The original ROI-local plane indices, independent of image pixel axes.""" + + metadata: Mapping[str, object] + + @classmethod + def from_shape_payload(cls, shape: Mapping[str, object]) -> ROIPlaneMetadata: + metadata = shape.get("metadata") + return cls(metadata if isinstance(metadata, Mapping) else {}) + + def has_plane_metadata(self) -> bool: + present = ("plane_indices" in self.metadata, "plane_shape" in self.metadata) + if any(present) and not all(present): + raise ValueError("ROI plane metadata requires both indices and shape.") + return all(present) + + def indices(self) -> tuple[int, ...]: + return self.projection().plane_indices + + def shape(self) -> tuple[int, ...]: + return self.projection().plane_shape + + def projection(self) -> RuntimeProjectionPlaneMetadata: + return RuntimeProjectionPlaneMetadata( + plane_indices=self._tuple_field("plane_indices"), + plane_shape=self._tuple_field("plane_shape"), + ) + + def _tuple_field(self, field: str) -> tuple[int, ...]: + value = self.metadata.get(field) + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise ValueError(f"ROI plane metadata field {field!r} must be a sequence.") + return tuple(int(item) for item in value) + + @classmethod + def common_shape(cls, metadata_items: Sequence[Mapping[str, object]]) -> tuple[int, ...]: + planes = tuple(cls(metadata) for metadata in metadata_items) + indexed = tuple(plane for plane in planes if plane.has_plane_metadata()) + if not indexed: + return () + if len(indexed) != len(planes): + raise ValueError("ROI payload mixes plane-indexed and unindexed shapes.") + shapes = tuple(plane.shape() for plane in indexed) + if any(shape != shapes[0] for shape in shapes[1:]): + raise ValueError(f"ROI payload has inconsistent plane_shape metadata: {shapes!r}.") + return shapes[0] + + @classmethod + def source_component_domain( + cls, rois: Sequence[ROI], metadata: ImagePayloadMetadata, + ) -> tuple[SourceComponentMetadata, ...] | None: + shape = cls.common_shape(tuple(roi.metadata for roi in rois)) + if not shape: + return None + domain = metadata.source_provenance.source_image_provenance_planes.component_metadata + if len(domain) != prod(shape) or any(plane is None for plane in domain): + raise ValueError("Plane-indexed ROI requires its complete source-plane domain.") + values = cls.component_values(metadata) + if tuple(len(value) for value in values.values()) != shape and not ( + shape == (1,) and not values + ): + raise ValueError("ROI plane shape conflicts with its represented component domain.") + return tuple(domain) + + @staticmethod + def component_values(metadata: ImagePayloadMetadata) -> dict[str, tuple[object, ...]]: + from openhcs.constants.constants import AllComponents + + return { + component: tuple(dict.fromkeys(values)) + for component, values in metadata.source_provenance.varying_plane_component_values( + tuple(AllComponents) + ).items() + } + + class ROIArchiveSourceMetadata: """Native ROI sidecar projection, not a second source-metadata schema.""" FIELD: ClassVar[str] = "openhcs_image_payload_metadata" + @classmethod + def source_component_domain( + cls, rois: Sequence[ROI], metadata: ImagePayloadMetadata, + ) -> tuple[SourceComponentMetadata, ...] | None: + from openhcs.core.roi_point_metadata import ROIFractionalZ + + point_domain = ROIFractionalZ.source_component_domain(rois, metadata) + return ( + point_domain if point_domain is not None + else ROIPlaneMetadata.source_component_domain(rois, metadata) + ) + + @classmethod + def stream_item_fields( + cls, rois: Sequence[ROI], metadata: ImagePayloadMetadata | None, + image_fields: Mapping[str, ViewerWireValue], + ) -> dict[str, ViewerWireValue]: + from zmqruntime.viewer_protocol import ViewerWireField + + fields = dict(image_fields) + if metadata is not None and ROIPlaneMetadata.source_component_domain(rois, metadata) is not None: + values = ROIPlaneMetadata.component_values(metadata) + if values: + fields[ViewerWireField.PLANE_COMPONENT_VALUES.value] = values + return fields + @classmethod def bind(cls, rois: Sequence[ROI], metadata: ImagePayloadMetadata) -> list[ROI]: encoded = to_jsonable(metadata) diff --git a/openhcs/core/steps/function_artifact_materialization.py b/openhcs/core/steps/function_artifact_materialization.py index 6240c0c2a3..01ac57e1cb 100644 --- a/openhcs/core/steps/function_artifact_materialization.py +++ b/openhcs/core/steps/function_artifact_materialization.py @@ -4,7 +4,7 @@ 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 types import MappingProxyType from typing import TYPE_CHECKING, ClassVar @@ -57,6 +57,7 @@ from openhcs.core.streaming_config_factory import StreamingViewerSurface from openhcs.processing.materialization.core import ( BackendKwargs, + BackendKwargsAuthority, MaterializationSpec, MaterializationValue, Output, @@ -169,8 +170,19 @@ def materialize_outputs( str(materialization.base_path), filemanager, backends, - self.backend_kwargs( + {}, + context=context, + artifact_source_identity=materialization.source_identity, + artifact_filename_identity=materialization.filename_source_identity, + variable_components=materialization.output_plan.variable_components, + pipeline_position=plan.pipeline_position, + output_plan=materialization.output_plan, + ) + batch = replace(batch, context=replace( + batch.context, + backend_kwargs=BackendKwargsAuthority.normalize(self.backend_kwargs( materialization=materialization, + outputs=batch.outputs, persistent_backend_kwargs=persistent_backend_kwargs, streaming_viewer_surfaces=streaming_viewer_surfaces, fallback_source_identity=( @@ -184,15 +196,9 @@ def materialize_outputs( context=context, filemanager=filemanager, images_dir=images_dir, - stream_output_paths=stream_output_paths, - ), - context=context, - artifact_source_identity=materialization.source_identity, - artifact_filename_identity=materialization.filename_source_identity, - variable_components=materialization.output_plan.variable_components, - pipeline_position=plan.pipeline_position, - output_plan=materialization.output_plan, - ) + stream_output_paths=tuple(output.path for output in batch.outputs), + )), + )) saved_materializations.append( MaterializedRuntimeArtifact( outputs_by_backend=MappingProxyType( @@ -251,6 +257,7 @@ def backend_kwargs( self, *, materialization: "RuntimeArtifactMaterialization", + outputs: tuple[Output, ...], persistent_backend_kwargs: BackendKwargs, streaming_viewer_surfaces: Mapping[str, StreamingViewerSurface], fallback_source_identity: SourceImageIdentity | None, @@ -297,11 +304,14 @@ def backend_kwargs( if not streamable_viewer_surfaces: return result - source_metadata_items = materialization.stream_source_metadata_items( - fallback_source_identity - ) producer = ViewerStreamProducer.from_identity(producer_identity) for backend, viewer_surface in streamable_viewer_surfaces.items(): + backend_instance = filemanager._get_backend(backend) + source_metadata_items = materialization.stream_source_metadata_items( + tuple(output for output in outputs + if backend_instance.accepts_payload(output.content, output.path)), + fallback_source_identity, + ) stream_backend_kwargs = StreamComponentMessageExtraAuthority.from_context( viewer_surface, context=context, @@ -510,32 +520,19 @@ class RuntimeArtifactMaterialization: def stream_source_metadata_items( self, + outputs: tuple[Output, ...], fallback_source_identity: SourceImageIdentity | None, ) -> StreamSourceComponentMetadataItems: """Project this occurrence's sources at the streaming request boundary.""" fallback_source_identity = ( fallback_source_identity or self.payload_source_identity(self.data) ) - metadata = image_payload_metadata(self.data) - if metadata.plane_axis is not None: - return StreamSourceComponentMetadataItems.from_image_metadata( - metadata, - fallback_source_identity=fallback_source_identity, - ) - emitted_identities = self.spec.emitted_source_identities(self.data) - if emitted_identities: - return StreamSourceComponentMetadataItems.from_source_identities( - emitted_identities, - fallback_source_identity=fallback_source_identity, - ) return StreamSourceComponentMetadataItems.from_values( - ( - ( - fallback_source_identity.component_metadata - if fallback_source_identity is not None - else None - ), - ) + metadata + for output in outputs + for metadata in output.stream_source_metadata_items( + fallback_source_identity, + ).values ) @staticmethod diff --git a/openhcs/core/steps/stream_component_semantics.py b/openhcs/core/steps/stream_component_semantics.py index 5b8bd2b168..6f29fad64f 100644 --- a/openhcs/core/steps/stream_component_semantics.py +++ b/openhcs/core/steps/stream_component_semantics.py @@ -119,9 +119,17 @@ def partition_indices( component_order: tuple[str, ...], ) -> tuple[tuple[int, ...], ...]: """Group ordered items by the metadata common to one wire batch.""" + return cls.partition_item_fields( + cls.item_fields(metadata, component_order) for metadata in metadata_items + ) + + @staticmethod + def partition_item_fields( + fields: Iterable[dict[str, ViewerWireValue]], + ) -> tuple[tuple[int, ...], ...]: + """Partition the actual rendered image or geometry fields once.""" partitions: list[tuple[dict[str, ViewerWireValue], list[int]]] = [] - for index, metadata in enumerate(metadata_items): - item_fields = cls.item_fields(metadata, component_order) + for index, item_fields in enumerate(fields): for partition_fields, indices in partitions: if partition_fields == item_fields: indices.append(index) diff --git a/openhcs/core/viewer_streaming_service.py b/openhcs/core/viewer_streaming_service.py index 3cd83d7b79..c95d5c4ac0 100644 --- a/openhcs/core/viewer_streaming_service.py +++ b/openhcs/core/viewer_streaming_service.py @@ -33,7 +33,6 @@ from openhcs.core.runtime_plane_projection import RuntimePlaneAxisValueProjection from openhcs.core.runtime_slice_projection import RuntimeSliceProjection from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata -from openhcs.core.roi_point_metadata import ROIFractionalZ from openhcs.core.runtime_image_values import ( ImagePayloadMetadata, image_payload_data, @@ -828,7 +827,7 @@ def stream_rois( paths: list[str] = [] loaded_indices: list[int] = [] archive_metadata: list[ImagePayloadMetadata | None] = [] - point_domains: dict[str, tuple[SourceComponentMetadata, ...]] = {} + geometric_domains: dict[str, tuple[SourceComponentMetadata, ...]] = {} for i, filename in enumerate(request.roi_filenames, 1): file_path = Path(self.source.plate_path) / filename @@ -848,9 +847,9 @@ def stream_rois( ) archive_metadata.append(metadata) if metadata is not None: - point_domain = ROIFractionalZ.source_component_domain(rois, metadata) - if point_domain is not None: - point_domains[filename] = point_domain + geometric_domain = ROIArchiveSourceMetadata.source_component_domain(rois, metadata) + if geometric_domain is not None: + geometric_domains[filename] = geometric_domain data_list.append(rois) paths.append(filename) loaded_indices.append(i - 1) @@ -922,12 +921,12 @@ def stream_rois( if metadata is not None ) metadata_by_path.update( - (path, dict(domain[0])) for path, domain in point_domains.items() + (path, dict(domain[0])) for path, domain in geometric_domains.items() ) source_metadata_items = StreamSourceComponentMetadataItems.from_values( component_metadata for path in paths - for component_metadata in point_domains.get(path, (metadata_by_path[path],)) + for component_metadata in geometric_domains.get(path, (metadata_by_path[path],)) ) message_authority = StreamComponentMessageExtraAuthority.from_viewer_surface( viewer_surface, @@ -947,12 +946,15 @@ def stream_rois( self.source.calibrated_metadata(metadata) for metadata in archive_metadata ] component_order = message_authority.layout.component_order - for indices in StreamImagePayloadMetadataProjector.partition_indices( - payload_metadata, component_order - ): - item_fields = StreamImagePayloadMetadataProjector.item_fields( - payload_metadata[indices[0]], component_order + item_fields_by_archive = tuple( + ROIArchiveSourceMetadata.stream_item_fields( + rois, metadata, + StreamImagePayloadMetadataProjector.item_fields(metadata, component_order), ) + for rois, metadata in zip(data_list, payload_metadata, strict=True) + ) + for indices in StreamImagePayloadMetadataProjector.partition_item_fields(item_fields_by_archive): + item_fields = item_fields_by_archive[indices[0]] stream_backend_kwargs = message_authority.viewer_backend_kwargs( producer=producer.for_indices( tuple(loaded_indices[index] for index in indices), total diff --git a/openhcs/processing/materialization/core.py b/openhcs/processing/materialization/core.py index f01b753568..035c0cc4e9 100644 --- a/openhcs/processing/materialization/core.py +++ b/openhcs/processing/materialization/core.py @@ -81,6 +81,7 @@ ) from openhcs.core.steps.stream_component_semantics import ( StreamImagePayloadMetadataProjector, + StreamSourceComponentMetadataItems, StreamViewerComponentMetadataProjector, StreamScopedDisplayConfig, ) @@ -317,6 +318,29 @@ def source_component_metadata(self) -> SourceComponentMetadata | None: return None return self.source_identity.component_metadata + @property + def viewer_source_identity(self) -> SourceImageIdentity | None: + """Return the address of this rendered stream item, not its domain.""" + return self.source_identity + + def stream_source_metadata_items( + self, fallback_source_identity: SourceImageIdentity | None, + ) -> StreamSourceComponentMetadataItems: + """Observe sources from the actual rendered output's declared planes.""" + if self.metadata is not None: + return StreamSourceComponentMetadataItems.from_image_metadata( + self.metadata, fallback_source_identity=fallback_source_identity, + ) + return StreamSourceComponentMetadataItems.from_values( + (fallback_source_identity.component_metadata + if fallback_source_identity is not None else None,) + ) + + def viewer_item_fields(self) -> dict[str, ViewerWireValue]: + return StreamImagePayloadMetadataProjector.item_fields_for_plane_components( + self.metadata, self.variable_components, + ) + @property def viewer_stream_requires_source_metadata(self) -> bool: return True @@ -371,6 +395,60 @@ def with_variable_components( ) +@dataclass(frozen=True) +class ROIOutput(Output): + """A rendered ROI carries a geometric domain, not an image pixel axis.""" + + @property + def source_component_domain(self) -> tuple[SourceComponentMetadata, ...] | None: + from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata + + if self.metadata is None: + return None + return ROIArchiveSourceMetadata.source_component_domain(self.content, self.metadata) + + @property + def viewer_source_identity(self) -> SourceImageIdentity | None: + domain = self.source_component_domain + if domain is None: + return super().viewer_source_identity + identity = self.source_identity + return SourceImageIdentity( + path=identity.path if identity is not None else None, + component_metadata=domain[0], + ) + + def stream_source_metadata_items( + self, fallback_source_identity: SourceImageIdentity | None, + ) -> StreamSourceComponentMetadataItems: + domain = self.source_component_domain + if domain is None: + return super().stream_source_metadata_items(fallback_source_identity) + return StreamSourceComponentMetadataItems.from_values(domain) + + def viewer_item_fields(self) -> dict[str, ViewerWireValue]: + from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata + + return ROIArchiveSourceMetadata.stream_item_fields( + self.content, self.metadata, super().viewer_item_fields(), + ) + + +@dataclass(frozen=True) +class PointROIOutput(ROIOutput): + """A point archive's represented source domain is geometric, not pixel planes.""" + + @property + def source_component_domain(self) -> tuple[SourceComponentMetadata, ...]: + from openhcs.core.roi_point_metadata import ROIFractionalZ + + if self.metadata is None: + raise ValueError("Point ROI output requires exact source metadata.") + domain = ROIFractionalZ.source_component_domain(self.content, self.metadata) + if domain is None: + raise ValueError("Point ROI output requires fractional-Z metadata.") + return domain + @dataclass(frozen=True) class TextOutput(Output): """Materialization output whose canonical payload is text.""" @@ -1726,7 +1804,7 @@ def _output_fields( dict[str, ViewerWireValue], ViewerDisplayConfigABC, ] | None: - source_identity = output.source_identity + source_identity = output.viewer_source_identity if source_identity is not None: source_identity = source_identity.with_parsed_path_components( self.values.stream_request.source.identity.microscope_handler.parser @@ -1741,12 +1819,7 @@ def _output_fields( "Viewer stream materialization requires output metadata with " "source_component_metadata." ) - item_fields = ( - StreamImagePayloadMetadataProjector.item_fields_for_plane_components( - output.metadata, - output.variable_components, - ) - ) + item_fields = output.viewer_item_fields() display_config = output.viewer_display_config( self.values.stream_request.display_config ) @@ -3167,7 +3240,7 @@ def _write_roi_zip( source_spatial_domain=source_domain_authority.domain_for_target(target) ) outs.append( - Output( + ROIOutput( path=target.archive.path, content=ROIArchiveSourceMetadata.bind(target_rois, item_metadata), metadata=item_metadata, @@ -3402,14 +3475,14 @@ def _write_point_roi_zip( if not rois: raise ValueError("Point ROI ZIP requires at least one measured object.") metadata = ImagePayloadMetadata(source_provenance=provenance) - ROIFractionalZ.source_component_domain(rois, metadata) - return [ - Output( - path=ctx.paths(options).primary_output_path(options), - content=ROIArchiveSourceMetadata.bind(rois, metadata), - metadata=metadata, - ) - ] + output = PointROIOutput( + path=ctx.paths(options).primary_output_path(options), + content=ROIArchiveSourceMetadata.bind(rois, metadata), + metadata=metadata, + ) + # Admit the domain before any persistence or publication uses this output. + output.source_component_domain + return [output] @writer_for( diff --git a/openhcs/runtime/napari_streaming_handlers.py b/openhcs/runtime/napari_streaming_handlers.py index d32ebbee0d..876f19cd06 100644 --- a/openhcs/runtime/napari_streaming_handlers.py +++ b/openhcs/runtime/napari_streaming_handlers.py @@ -22,7 +22,7 @@ from openhcs.constants import AllComponents from openhcs.core.artifacts import ObjectArtifactSubjectBinding from openhcs.core.config import NapariDisplayConfig -from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata +from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata, ROIPlaneMetadata from openhcs.core.runtime_image_values import ( ImagePayloadMetadata, ) @@ -632,7 +632,7 @@ def shape_component_values( ) -> ComponentMap: if not self.bindings: return dict(item.address.components) - plane_indices = NapariShapePlaneMetadata(shape_dict).indices() + plane_indices = ROIPlaneMetadata.from_shape_payload(shape_dict).indices() return self.item_component_values(item, plane_indices) @@ -791,61 +791,10 @@ def _image_aggregate_extents( def _shape_aggregate_extents(data: LayerData) -> tuple[int, ...]: if not isinstance(data, Sequence) or isinstance(data, (str, bytes)): return () - shapes: list[tuple[int, ...]] = [] - missing_plane_metadata = 0 - for shape_dict in data: - if not isinstance(shape_dict, Mapping): - continue - plane_metadata = NapariShapePlaneMetadata(shape_dict) - if plane_metadata.has_plane_metadata(): - shapes.append(plane_metadata.shape()) - else: - missing_plane_metadata += 1 - if not shapes: - return () - if missing_plane_metadata: - raise ValueError( - "Napari shape payload mixes plane-indexed and unindexed shapes; " - "all shapes in an aggregate stack route must carry plane metadata." - ) - first = shapes[0] - if any(shape != first for shape in shapes): - raise ValueError( - "Napari shape payload has inconsistent plane_shape metadata: " - f"{shapes!r}." - ) - return first - - -@dataclass(frozen=True, slots=True) -class NapariShapePlaneMetadata: - """Plane-index metadata carried by one serialized ROI shape.""" - - shape_dict: ShapePayloadMap - - @property - def metadata(self) -> Mapping[str, ShapePayloadValue]: - metadata = self.shape_dict.get("metadata") - if not isinstance(metadata, Mapping): - return {} - return metadata - - def has_plane_metadata(self) -> bool: - return "plane_indices" in self.metadata and "plane_shape" in self.metadata - - def indices(self) -> tuple[int, ...]: - return self._tuple_field("plane_indices") - - def shape(self) -> tuple[int, ...]: - return self._tuple_field("plane_shape") - - def _tuple_field(self, field: str) -> tuple[int, ...]: - value = self.metadata.get(field) - if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): - raise ValueError( - f"Napari shape plane metadata field {field!r} must be a sequence." - ) - return tuple(int(item) for item in value) + return ROIPlaneMetadata.common_shape(tuple( + ROIPlaneMetadata.from_shape_payload(shape_dict).metadata + for shape_dict in data if isinstance(shape_dict, Mapping) + )) NapariLayerCreator: TypeAlias = Callable[ diff --git a/tests/unit/test_function_artifact_materialization.py b/tests/unit/test_function_artifact_materialization.py index 7d22590462..05d4c326d3 100644 --- a/tests/unit/test_function_artifact_materialization.py +++ b/tests/unit/test_function_artifact_materialization.py @@ -121,6 +121,7 @@ from openhcs.core.source_spatial_domain import SourceSpatialDomain from openhcs.core.steps.abstract import StepExecutionObservation from openhcs.core.steps.function_artifact_materialization import ( + ArtifactMaterializationTargetPlan, PersistentArtifactMaterializationTargetPlan, StreamingOnlyArtifactMaterializationTargetPlan, actual_materialization_records, @@ -223,6 +224,31 @@ def stream_request_from_backend_kwargs(backend_kwargs): ] +def _observe_materialization(monkeypatch, observe): + """Observe completed owner inputs without replacing rendered batches.""" + original = ArtifactMaterializationTargetPlan.backend_kwargs + + def recording(self, **kwargs): + values = original(self, **kwargs) + materialization = kwargs["materialization"] + observe( + materialization.spec, + materialization.data, + str(materialization.base_path), + kwargs["filemanager"], + list(values), + values, + context=kwargs["context"], + artifact_source_identity=materialization.source_identity, + artifact_filename_identity=materialization.filename_source_identity, + variable_components=materialization.output_plan.variable_components, + rendered_outputs=kwargs["outputs"], + ) + return values + + monkeypatch.setattr(ArtifactMaterializationTargetPlan, "backend_kwargs", recording) + + class MetadataHandlerStub: def __init__(self, values=None): self.values = values or {} @@ -818,12 +844,8 @@ def test_materialize_artifact_outputs_uses_runtime_store_payload( def fake_materialize(_spec, data, path, *_args, **_kwargs): materialized.append((data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -863,12 +885,8 @@ def test_materialize_artifact_outputs_attaches_image_schema_provenance(monkeypat def fake_materialize(_spec, data, path, *_args, **_kwargs): materialized.append((data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan( output_plan, @@ -941,12 +959,8 @@ def fake_materialize(spec, data, path, manager, *_args, **kwargs): variable_components=kwargs["variable_components"], ) ) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) plan = _plan( output_plan, @@ -1006,12 +1020,8 @@ def test_materialize_artifact_outputs_does_not_require_vfs_payload_for_store_rec def fake_materialize(_spec, data, path, *_args, **_kwargs): materialized.append((data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -1056,12 +1066,8 @@ def test_materialize_artifact_outputs_uses_runtime_record_identity_not_final_pat def fake_materialize(_spec, data, path, *_args, **_kwargs): materialized.append((data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -1105,12 +1111,8 @@ def test_materialize_artifact_outputs_uses_declared_measurement_csv_spec( def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -1122,7 +1124,8 @@ def fake_materialize(spec, data, path, *_args, **_kwargs): assert path == "/analysis/A01_measurements_step7.roi.zip" -def test_runtime_artifact_plan_materializes_3d_point_measurements(): +@pytest.mark.parametrize("source_z_origin", (0, 10)) +def test_runtime_artifact_plan_materializes_3d_point_measurements(source_z_origin): features = ObjectCoreMeasurementFeature output_plan = ArtifactOutputPlan( name="nuclei_centres", @@ -1162,7 +1165,7 @@ def test_runtime_artifact_plan_materializes_3d_point_measurements(): paths=(source_path,) * 4, component_metadata=tuple( {"well": "A01", "site": 1, "channel": 1, "z_index": z, "timepoint": 1} - for z in range(4) + for z in range(source_z_origin, source_z_origin + 4) ), ), subject=MeasurementSubject(MeasurementScope.OBJECT, "nuclei", "object_label"), @@ -1185,6 +1188,31 @@ def test_runtime_artifact_plan_materializes_3d_point_measurements(): assert ROIFractionalZ.decode(result.content[0].metadata) == ROIFractionalZ(2.375) assert ROIArchiveSourceMetadata.decode(result.content).source_path == source_path + # The measurement table has no pixel-plane axis. Its rendered point archive + # nevertheless owns every represented Z plane, not only its scalar address. + assert result.metadata.plane_axis is None + domain = materialization.stream_source_metadata_items((result,), None) + assert [item["z_index"] for item in domain.values] == list( + range(source_z_origin, source_z_origin + 4) + ) + assert result.viewer_source_identity.component_metadata["z_index"] == source_z_origin + config = streaming_config_stub() + plan = _plan( + output_plan, + variable_components=(VariableComponents.Z_INDEX,), + streaming_configs={"napari_stream": config}, + ) + [saved] = StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs( + context.filemanager, plan, context, + ) + [streamed] = saved.outputs_for_backend("napari_stream") + assert streamed.content[0].metadata[ROIFractionalZ.FIELD] == 2.375 + request = context.filemanager.saved[-1][3]["stream_request"] + assert "z_index" in request.display_config.COMPONENT_ORDER + assert request.source.metadata.component_metadata_for_item(streamed.path, 0)["z_index"] == source_z_origin + z_domain = request.message_extra["component_value_domain"] + assert z_domain["z_index"] == list(range(source_z_origin, source_z_origin + 4)) + def test_multi_plane_measurement_materialization_uses_aggregate_artifact_name(): output_plan = ArtifactOutputPlan( @@ -2086,12 +2114,8 @@ def test_materialize_artifact_outputs_unions_measurement_subject_records( def fake_materialize(_spec, data, path, *_args, **_kwargs): assert not data.source_provenance.has_values materialized.append((tuple(data.iter_row_mappings()), path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -2384,12 +2408,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan( output_plan, streaming_configs={"napari_stream": streaming_config_stub()} @@ -2445,12 +2465,8 @@ def test_materialize_artifact_outputs_uses_actual_group_records(monkeypatch): def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -2720,12 +2736,8 @@ def test_materialize_artifact_outputs_uses_group_measurement_artifact_identity( def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -2779,12 +2791,8 @@ def test_materialize_artifact_outputs_uses_grouped_scope_not_record_location( def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan, group_by_value="channel"), context) @@ -2806,11 +2814,11 @@ def test_materialize_artifact_outputs_uses_null_component_group_identity_for_str paths_by_group={ "2": "/memory/channel2_Nuclei.pkl", }, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) group_plan = output_plan.for_group("2") labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_component_metadata={ "well": "A01", "site": "1", @@ -2840,12 +2848,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, streaming_configs={"napari_stream": streaming_config_stub()} @@ -2872,10 +2876,10 @@ def test_materialize_artifact_outputs_streams_aggregate_artifact_with_incomplete name="Nuclei", path="/memory/Nuclei.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_component_metadata={ "well": "A01", "site": "1", @@ -2905,12 +2909,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) streaming_config = streaming_config_stub() @@ -3050,12 +3050,8 @@ def test_materialize_artifact_outputs_uses_declared_metadata_json_spec( def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -3079,12 +3075,8 @@ def test_materialize_artifact_outputs_skips_special_without_explicit_spec( def fake_materialize(*args, **kwargs): materialized.append((args, kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -3167,12 +3159,8 @@ def test_materialize_artifact_outputs_skips_explicitly_disabled_artifact_without def fake_materialize(*args, **kwargs): materialized.append((args, kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -3203,12 +3191,8 @@ def test_materialize_artifact_outputs_uses_declared_object_labels_roi_spec(monke def fake_materialize(spec, data, path, *_args, **_kwargs): materialized.append((spec, data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context) @@ -3227,11 +3211,11 @@ def test_materialize_artifact_outputs_can_target_streaming_without_persistent_ba name="labels", path="/memory/labels.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_path="/input/A01_s001_w1.TIF", source_component_metadata={"well": "A01", "channel": 1}, source_spatial_domain=SourceSpatialDomain(source_shape_yx=(100, 200)), @@ -3255,12 +3239,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan(output_plan, streaming_configs={"napari_stream": streaming_config}), context) @@ -3339,12 +3319,8 @@ def fake_materialize( **_kwargs, ): materialized.append((backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan( output_plan, @@ -3365,11 +3341,11 @@ def test_materialize_artifact_outputs_uses_artifact_source_metadata_for_streamin name="labels", path="/memory/labels.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_path="/input/A01_s002_w3_z001_t001.TIF", source_component_metadata={"channel": 3}, source_spatial_domain=SourceSpatialDomain(source_shape_yx=(100, 200)), @@ -3399,12 +3375,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, @@ -3448,11 +3420,11 @@ def test_materialize_artifact_outputs_streams_payload_component_metadata( name="Nuclei", path="/memory/Nuclei.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_path="/input/01_POS002_D.TIF", source_component_metadata={"well": "01", "site": "POS002", "channel": "D"}, source_spatial_domain=SourceSpatialDomain(source_shape_yx=(100, 200)), @@ -3480,12 +3452,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, plan, context) @@ -3537,12 +3505,8 @@ def test_materialize_artifact_outputs_uses_runtime_plane_group_identity( def fake_materialize(_spec, data, path, *_args, **_kwargs): materialized.append((data, path)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan( output_plan, @@ -3569,11 +3533,11 @@ def test_materialize_artifact_outputs_merges_parser_axes_into_source_metadata( name="Nuclei", path="/memory/Nuclei.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2), dtype=np.int32)), source_path="/input/A01_s002_w3_z001_t001.TIF", source_component_metadata={"instrument": "test"}, source_spatial_domain=SourceSpatialDomain(source_shape_yx=(100, 200)), @@ -3603,12 +3567,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan(output_plan, streaming_configs={"napari_stream": streaming_config}), context) @@ -3638,11 +3598,11 @@ def test_materialize_artifact_outputs_uses_variable_components_for_streaming_ide name="Nuclei", path="/memory/Nuclei.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2, 2), dtype=np.int32)), plane_axis=( RuntimePlaneAxis.RUNTIME_SLICE if domain_scope is ObjectLabelDomainScope.PLANE @@ -3650,7 +3610,7 @@ def test_materialize_artifact_outputs_uses_variable_components_for_streaming_ide ), domain=( ObjectLabelDomain( - declared_object_id_domains=((), ()), + declared_object_id_domains=((1,), (1,)), scope=ObjectLabelDomainScope.PLANE, ) if domain_scope is ObjectLabelDomainScope.PLANE @@ -3699,12 +3659,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, @@ -3808,14 +3764,14 @@ def test_materialize_artifact_outputs_streams_source_binding_roi_plane_metadata( name="segmentation_masks", path="/memory/segmentation_masks.pkl", artifact_type=ObjectLabelsArtifactType, - materialization=roi_zip(), + materialization=roi_zip(min_area=0), ) streaming_config = streaming_config_stub() labels = ObjectLabelPayload( - variant_data=ObjectLabelVariantData(labels=np.zeros((2, 2, 2), dtype=np.int32)), + variant_data=ObjectLabelVariantData(labels=np.ones((2, 2, 2), dtype=np.int32)), plane_axis=RuntimePlaneAxis.SOURCE_BINDING, domain=ObjectLabelDomain( - declared_object_id_domains=((), ()), + declared_object_id_domains=((1,), (1,)), scope=ObjectLabelDomainScope.PLANE, ), source_image_provenance_planes=SourceImageProvenancePlanes.from_components( @@ -3861,12 +3817,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, @@ -3908,6 +3860,13 @@ def test_materialize_rgb_artifact_streams_filename_channel_identity( rgb_payload = ImageMetadataPayload( data=np.ones((5, 7, 3), dtype=np.float32), metadata=ImagePayloadMetadata( + # RGB pixels need their explicit output-image address; three source + # contributors alone do not establish a scalar channel on the result. + source_path="/input/A01_s001_w3_z001_t001.TIF", + source_component_metadata={ + "well": "A01", "site": "1", "channel": "3", + "z_index": "1", "timepoint": "1", + }, source_image_provenance_planes=SourceImageProvenancePlanes.from_components( paths=( "/input/A01_s001_w3_z001_t001.TIF", @@ -3959,12 +3918,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, @@ -4040,12 +3995,8 @@ def test_materialize_image_uses_declared_filename_source_identity(monkeypatch): def fake_materialize(_spec, _data, path, *_args, **_kwargs): materialized.append(path) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan, variable_components=(VariableComponents.CHANNEL,)), context) @@ -4267,12 +4218,8 @@ def fake_materialize( **_kwargs, ): materialized.append((spec, data, path, backends, backend_kwargs)) - return SimpleNamespace(save=lambda: SavedMaterializationOutputs({})) - monkeypatch.setattr( - "openhcs.core.steps.function_artifact_materialization.prepare_materialization", - fake_materialize, - ) + _observe_materialization(monkeypatch, fake_materialize) StreamingOnlyArtifactMaterializationTargetPlan().materialize_outputs(filemanager, _plan( output_plan, diff --git a/tests/unit/test_function_artifact_materialization_io_hotpath.py b/tests/unit/test_function_artifact_materialization_io_hotpath.py index 4ca638da96..3bbd894f75 100644 --- a/tests/unit/test_function_artifact_materialization_io_hotpath.py +++ b/tests/unit/test_function_artifact_materialization_io_hotpath.py @@ -120,6 +120,7 @@ def unexpected_stream_metadata(*_args: object, **_kwargs: object) -> None: target = PersistentArtifactMaterializationTargetPlan("disk") kwargs = target.backend_kwargs( materialization=materialization, + outputs=(Output(path="/images/SavedImage.tif", content=payload),), persistent_backend_kwargs={"disk": RawBackendKwargs()}, streaming_viewer_surfaces=streaming_viewer_surfaces, fallback_source_identity=None, diff --git a/tests/unit/test_image_file_serialization.py b/tests/unit/test_image_file_serialization.py index fc8ac4fd43..b718ce197f 100644 --- a/tests/unit/test_image_file_serialization.py +++ b/tests/unit/test_image_file_serialization.py @@ -585,3 +585,52 @@ def test_intrinsic_tiff_write_uses_declared_axes_not_rgb_shaped_dimensions( assert header.pixel_semantics.channel_axis == channel_axis assert header.pixel_semantics.channel_count == (None if channel_axis is None else 3) assert header.image_shape_yx == (4, 4) + + +@pytest.mark.parametrize("plane_count", (1, 3, 4, 5)) +@pytest.mark.parametrize("dtype", (np.uint16, np.int32)) +@pytest.mark.parametrize("channel_axis", (None, -1, 1)) +def test_runtime_plane_tiff_write_retains_scalar_frames_and_declared_channels( + tmp_path, plane_count, dtype, channel_axis +): + from polystore.config import TiffCompression, TiffConfig, TiffPhotometric + from openhcs.processing.materialization.core import Output, RawBackendKwargs + + shape = ( + (plane_count, 5, 7) if channel_axis is None + else ((plane_count, 5, 7, 3) if channel_axis == -1 else (plane_count, 3, 5, 7)) + ) + pixels = np.arange(np.prod(shape), dtype=dtype).reshape(shape) + metadata = ImagePayloadMetadata( + plane_axis=RuntimePlaneAxis.RUNTIME_SLICE, + source_channel_axis=channel_axis, + ) + path = tmp_path / "artifact.labels.tif" + configured = TiffConfig(compression=TiffCompression.DEFLATE, compression_level=1) + (batch, kwargs), = RawBackendKwargs(tiff_config=configured).filemanager_batches( + (Output(str(path), pixels, metadata),) + ) + assert len(batch) == 1 and batch[0].content is pixels + config = kwargs['tiff_config'] + assert config.compression is TiffCompression.DEFLATE + assert config.compression_level == 1 + assert config.photometric is (TiffPhotometric.MINISBLACK if channel_axis is None else TiffPhotometric.RGB) + assert config.axes == ('QYX' if channel_axis is None else ('QYXS' if channel_axis == -1 else 'QSYX')) + tifffile.imwrite(path, pixels, **config.tifffile_write_kwargs()) + np.testing.assert_array_equal(tifffile.imread(path), pixels) + with tifffile.TiffFile(path) as tif: + assert tif.series[0].axes == config.axes + assert len(tif.pages) == plane_count + assert tif.pages[0].samplesperpixel == (1 if channel_axis is None else 3) + header = TiffImageFileFormat().require_source_metadata(path) + assert header.pixel_semantics.channel_axis == channel_axis + assert header.pixel_semantics.channel_count == (None if channel_axis is None else 3) + assert header.image_shape_yx == (5, 7) + + +def test_runtime_plane_tiff_rejects_undeclared_extra_payload_axis(tmp_path): + metadata = ImagePayloadMetadata(plane_axis=RuntimePlaneAxis.RUNTIME_SLICE) + with pytest.raises(ValueError, match='declared plane'): + TiffImageFileFormat().write( + tmp_path / 'invalid.tif', metadata.payload_with(np.zeros((4, 5, 7, 2))) + ) diff --git a/tests/unit/test_materialization_core.py b/tests/unit/test_materialization_core.py index 3be3e5a87c..462fb6c592 100644 --- a/tests/unit/test_materialization_core.py +++ b/tests/unit/test_materialization_core.py @@ -55,7 +55,7 @@ ObjectCoreMeasurementFeature, ) from openhcs.core.roi_point_metadata import ROIFractionalZ -from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata +from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata, ROIPlaneMetadata from openhcs.runtime.viewer_component_system import ViewerLayerAxisProjection from openhcs.core.runtime_object_label_building import ( SourceImageObjectLabelBuildRequest, @@ -833,11 +833,24 @@ def test_point_roi_materialization_native_reopen_preserves_fractional_z( routed_component_values={"z_index": z_values}, axis_offsets=(0,), ) + from openhcs.runtime.napari_streaming_handlers import ( + NapariStreamLayerAddress, NapariStreamLayerItem, + ) + from openhcs.runtime.viewer_component_system import ViewerComponentValueDomainPayload + from polystore.streaming_constants import StreamingDataType + points, properties = viewer_server._build_nd_points( [ - SimpleNamespace( + NapariStreamLayerItem( data=NapariROIConverter.rois_to_shapes(rois), - address=SimpleNamespace(components=source_domain[0]), + producer=StreamProducerIdentity.fixed_output( + FixedStreamProducerIdentityKind.MANUAL, "point_reopen", + ), + address=NapariStreamLayerAddress( + source_domain[0], archive, StreamingDataType.POINTS, + ), + image_metadata=metadata, + plane_component_domain=ViewerComponentValueDomainPayload(()), ) ], projection, @@ -852,6 +865,102 @@ def test_point_roi_materialization_native_reopen_preserves_fractional_z( assert native_layer.features.loc[0, "response"] == 4.75 +@pytest.mark.unit +@pytest.mark.parametrize("source_z_origin", (0, 10)) +@pytest.mark.parametrize("variable_components", ((), (VariableComponents.Z_INDEX,))) +def test_payload_label_roi_reopen_preserves_geometric_plane_domain( + tmp_path, source_z_origin, variable_components, +): + labels = np.zeros((4, 5, 7), dtype=np.int32) + labels[0, 1:4, 2:5] = 7 + labels[2, 1:4, 2:5] = 7 + payload = ObjectLabelPayload( + variant_data=ObjectLabelVariantData(labels=labels), + domain=ObjectLabelDomain(scope=ObjectLabelDomainScope.PAYLOAD), + plane_axis=None, + source_image_provenance_planes=SourceImageProvenancePlanes.from_components( + paths=("/source/volume.ome.tif",) * 4, + component_metadata=tuple( + {"well": "A01", "site": 1, "channel": 1, "z_index": z, "timepoint": 1} + for z in range(source_z_origin, source_z_origin + 4) + ), + ), + source_spatial_domain=SourceSpatialDomain(source_shape_yx=(5, 7)), + parent_image_source_voxel_spacing=SourceVoxelSpacing((2.0, 0.65, 0.65)), + ) + archive = materialize( + MaterializationSpec(ROIOptions(min_area=0)), + data=payload, path=str(tmp_path / "volume"), + filemanager=FileManager({"disk": DiskStorageBackend()}), + backends=["disk"], variable_components=variable_components, + ) + rois = load_rois_from_zip(Path(archive)) + metadata = ROIArchiveSourceMetadata.decode(rois) + assert payload.plane_axis is None + assert metadata.plane_axis is ( + RuntimePlaneAxis.RUNTIME_SLICE if variable_components else None + ) + assert metadata.source_voxel_spacing.values_zyx == (2.0, 0.65, 0.65) + domain = ROIArchiveSourceMetadata.source_component_domain(rois, metadata) + z_values = tuple(range(source_z_origin, source_z_origin + 4)) + assert tuple(plane["z_index"] for plane in domain) == z_values + assert [ROIPlaneMetadata(roi.metadata).indices() for roi in rois] == [(0,), (2,)] + fields = ROIArchiveSourceMetadata.stream_item_fields(rois, metadata, {}) + assert fields[ViewerWireField.PLANE_COMPONENT_VALUES.value] == { + "z_index": tuple(str(value) for value in z_values), + } + # This geometric-domain projection never invents an image-plane field. + assert ViewerWireField.PLANE_AXIS.value not in fields + from openhcs.runtime.napari_streaming_handlers import ( + NapariAggregateAxisBinding, NapariAggregateAxisBindingSet, + NapariShapeLayerPayload, NapariStreamLayerAddress, NapariStreamLayerItem, + ) + from openhcs.runtime.viewer_component_system import ViewerComponentValueDomainPayload + from polystore.streaming_constants import StreamingDataType + plane_domain = ViewerComponentValueDomainPayload.from_wire_mapping( + fields[ViewerWireField.PLANE_COMPONENT_VALUES.value], context="saved label ROI", + ) + assert plane_domain.to_wire_mapping() == {"z_index": list(z_values)} + + native = NapariShapeLayerPayload.build( + layer_items=[NapariStreamLayerItem( + data=NapariROIConverter.rois_to_shapes(rois), + producer=StreamProducerIdentity.fixed_output( + FixedStreamProducerIdentityKind.MANUAL, "volume_reopen", + ), + address=NapariStreamLayerAddress(domain[0], archive, StreamingDataType.SHAPES), + image_metadata=metadata, + plane_component_domain=plane_domain, + )], + axis_projection=ViewerLayerAxisProjection( + projected_axis_components=("z_index",), + component_values={"z_index": list(z_values)}, + routed_component_values={"z_index": list(z_values)}, axis_offsets=(0,), + ), + aggregate_axis_bindings=NapariAggregateAxisBindingSet(( + NapariAggregateAxisBinding("z_index", 0, z_values), + )), + ) + assert native.ndim == 3 + assert [np.unique(shape[:, 0]).tolist() for shape in native.data] == [[0], [2]] + assert native.features["label"] == [7, 7] + from napari.layers import Shapes + layer = Shapes(native.data, shape_type=native.shape_types, features=native.features) + assert layer.ndim == 3 + assert layer.features["label"].tolist() == [7, 7] + + +@pytest.mark.unit +@pytest.mark.parametrize("metadata", ( + {"plane_indices": (4,), "plane_shape": (4,)}, + {"plane_indices": (0,)}, + {"plane_indices": (0, 1), "plane_shape": (4,)}, +)) +def test_roi_geometric_plane_metadata_rejects_invalid_projection(metadata): + with pytest.raises(ValueError): + ROIPlaneMetadata.common_shape((metadata,)) + + @pytest.mark.unit @pytest.mark.parametrize( ("source_path", "rows", "error"), diff --git a/tests/unit/test_napari_streaming_handlers.py b/tests/unit/test_napari_streaming_handlers.py index d14374b03d..7d577e1efe 100644 --- a/tests/unit/test_napari_streaming_handlers.py +++ b/tests/unit/test_napari_streaming_handlers.py @@ -3166,7 +3166,7 @@ class FakeViewer: dims = FakeDims() text_overlay = FakeTextOverlay() - layer = type("Layer", (), {"data": np.zeros((2, 2, 8, 16))})() + layer = type("Layer", (), {"data": np.zeros((2, 2, 8, 16)), "visible": True})() server = _FakeNapariServer() server.layer_route_state = NapariLayerRouteStateStore.empty() server.viewer = FakeViewer() @@ -5985,24 +5985,23 @@ def test_napari_shapes_layer_display_applies_route_global_axis_translate(): component_values={"channel": [4], "site": [1, 2]}, axis_offsets=(3, 0), ) + item = _layer_item( + {"channel": 4, "site": 1}, + [ + { + "type": "polygon", + "coordinates": [[0, 0], [0, 2], [2, 2], [2, 0]], + "metadata": {"source_spatial_shape_yx": (3, 3)}, + } + ], + stream_layer_data_type=StreamingDataType.SHAPES, + ) napari_viewer_server.NapariShapesLayerDisplayHandler().handle( napari_viewer_server.NapariLayerDisplayRequest( pipeline=pipeline, presentation=presentation, - items=[ - _layer_item( - {"channel": 4, "site": 1}, - [ - { - "type": "polygon", - "coordinates": [[0, 0], [0, 2], [2, 2], [2, 0]], - "metadata": {"source_spatial_shape_yx": (3, 3)}, - } - ], - stream_layer_data_type=StreamingDataType.SHAPES, - ) - ], + items=[item], display_config=NapariDisplayConfig(), ) ) @@ -6019,6 +6018,7 @@ def test_napari_shapes_layer_display_applies_route_global_axis_translate(): "source_spatial_shape_yx": ["(3, 3)"], "label": [1], "path": ["test"], + NapariStreamLayerItem.ELEMENT_IDENTITY_FEATURE: [item.element_identity(0)], } assert layer_kwargs["ndim"] == 4 assert layer_kwargs["edge_color"] == "label"