Keep image composition and masks in the declared native domain

openhcs/core/aligned_image_payload.py

@@ -37,6 +37,7 @@
     image_payload_geometry,
     image_payload_mask,
     image_payload_metadata,
+    project_image_mask_to_data_domain,
     preserved_image_plane_projection,
     with_image_payload_data,
 )
@@ -581,8 +582,7 @@ def composition_memory_domain(
     def compose(
         self, *, memory_type: str | None = None, device_id: int | None = None,
     ) -> Any:
-        # Intensity reconciliation projects pixels to the numerical host domain.
-        # Resolve the destination from the original carrier before that projection.
+        # Resolve the destination from the original carrier before reconciliation.
         memory_type, device_id = self.composition_memory_domain(
             tuple(image_payload_data(payload) for payload in self.composition_payloads),
             memory_type=memory_type, device_id=device_id,
@@ -713,8 +713,13 @@ def _complete_image_payload_mask(
                 f"domain; got mask {tuple(np.shape(mask))!r} for slice "
                 f"{tuple(np.shape(payload_data))!r}."
             )
-        return mask
-    return np.ones(mask_domain.default_mask_shape(), dtype=bool)
+        return project_image_mask_to_data_domain(
+            mask, payload_data, metadata=image_payload_metadata(payload),
+        )
+    data = image_payload_data(payload_data)
+    return MemoryType(detect_memory_type(data)).ones_like(
+        data, shape=mask_domain.default_mask_shape(), dtype=bool,
+    )
 
 
 def unstack_image_payload_context(
@@ -838,7 +843,7 @@ def compose_mask(
         if shared_spatial_shape is not None and all(
             tuple(np.shape(mask)) == shared_spatial_shape for mask in masks
         ):
-            return self.combined_mask()
+            return self.combined_mask(composed)
         resolved_masks = tuple(
             _complete_image_payload_mask(payload, data, mask)
             for payload, data, mask in zip(
@@ -848,16 +853,24 @@ def compose_mask(
                 strict=True,
             )
         )
-        memory_type = detect_memory_type(resolved_masks[0])
+        shapes = tuple(image_payload_geometry(mask).shape for mask in resolved_masks)
+        if any(shape != shapes[0] for shape in shapes[1:]):
+            resolved_masks = tuple(
+                metadata.for_leading_source_plane(index)
+                .mask_domain(composed[index])
+                .broadcast_to_data(mask)
+                for index, mask in enumerate(resolved_masks)
+            )
+        memory_type = detect_memory_type(image_payload_data(composed))
         memory_type_owner = MemoryType(memory_type)
         stacked = stack_runtime_slices(
             resolved_masks,
             memory_type,
-            memory_type_owner.device_id_of(resolved_masks[0]),
+            memory_type_owner.device_id_of(image_payload_data(composed)),
         )
         return memory_type_owner.astype(stacked, bool)
 
-    def combined_mask(self) -> RuntimeArrayData | None:
+    def combined_mask(self, composed: Any) -> RuntimeArrayData | None:
         masks = self.present_masks
         if not masks:
             return None
@@ -867,12 +880,21 @@ def combined_mask(self) -> RuntimeArrayData | None:
                 "Image bundle mask intersection requires one exact declared "
                 f"spatial mask shape; got {mask_shapes!r}."
             )
-        memory_type = MemoryType(detect_memory_type(masks[0]))
-        combined = memory_type.astype(masks[0], bool)
-        for mask in masks[1:]:
+        data = image_payload_data(composed)
+        memory_type = MemoryType(detect_memory_type(data))
+        device_id = memory_type.device_id_of(data)
+        prepared = tuple(
+            memory_type.astype(
+                MemoryType(detect_memory_type(mask)).convert_to(mask, memory_type, device_id),
+                bool,
+            )
+            for mask in masks
+        )
+        combined = prepared[0]
+        for mask in prepared[1:]:
             combined = memory_type.logical_and(
                 combined,
-                memory_type.astype(mask, bool),
+                mask,
             )
         return combined
 
@@ -914,14 +936,13 @@ def compose_mixed_channel_payloads(
         device_id: int | None,
     ) -> RuntimeArrayData:
         """Promote channel-free payloads using declared channel-axis semantics."""
-        numpy_payloads = tuple(
-            np.asarray(
-                convert_memory(
-                    data=payload,
-                    source_type=detect_memory_type(payload),
-                    target_type=MEMORY_TYPE_NUMPY,
-                    gpu_id=device_id,
-                )
+        target = MemoryType(memory_type)
+        prepared_payloads = tuple(
+            convert_memory(
+                data=payload,
+                source_type=detect_memory_type(payload),
+                target_type=memory_type,
+                gpu_id=device_id,
             )
             for payload in payloads
         )
@@ -934,7 +955,7 @@ def compose_mixed_channel_payloads(
             )
         channel_counts = tuple(
             int(payload.shape[channel_axis])
-            for payload, axis in zip(numpy_payloads, channel_axes, strict=True)
+            for payload, axis in zip(prepared_payloads, channel_axes, strict=True)
             if axis is not None
         )
         channel_count = channel_counts[0]
@@ -953,7 +974,7 @@ def compose_mixed_channel_payloads(
                     if index != axis
                 )
             )
-            for payload, axis in zip(numpy_payloads, channel_axes, strict=True)
+            for payload, axis in zip(prepared_payloads, channel_axes, strict=True)
         )
         if any(shape != source_shapes[0] for shape in source_shapes[1:]):
             raise ValueError(
@@ -964,23 +985,17 @@ def compose_mixed_channel_payloads(
             (
                 payload
                 if axis is not None
-                else np.repeat(
-                    np.expand_dims(payload, axis=channel_axis),
-                    channel_count,
-                    axis=channel_axis,
+                else target.broadcast_to(
+                    target.reshape(
+                        payload,
+                        (*payload.shape[:channel_axis], 1, *payload.shape[channel_axis:]),
+                    ),
+                    (*payload.shape[:channel_axis], channel_count, *payload.shape[channel_axis:]),
                 )
             )
-            for payload, axis in zip(numpy_payloads, channel_axes, strict=True)
-        )
-        stacked = np.stack(promoted, axis=0)
-        if memory_type == MEMORY_TYPE_NUMPY:
-            return stacked
-        return convert_memory(
-            data=stacked,
-            source_type=MEMORY_TYPE_NUMPY,
-            target_type=memory_type,
-            gpu_id=device_id,
+            for payload, axis in zip(prepared_payloads, channel_axes, strict=True)
         )
+        return target.stack_arrays(list(promoted), device_id)
 
 
 @dataclass(frozen=True, slots=True)
@@ -1472,7 +1487,11 @@ def composition_metadata_mode(self) -> ImagePayloadMetadataCompositionMode:
     @property
     def dtype(self) -> Any:
         return np.result_type(
-            *(image_payload_data(payload).dtype for payload in self.slices)
+            *(
+                MemoryType(detect_memory_type(image_payload_data(payload)))
+                .canonical_dtype_name(image_payload_data(payload).dtype)
+                for payload in self.slices
+            )
         )
 
     def plane_axis_for_output_context(

openhcs/core/runtime_image_values.py

@@ -28,6 +28,7 @@
 from openhcs.core.runtime_array_values import (
     DataBackedRuntimeArrayPayload,
     RuntimeArrayData,
+    is_array_payload,
 )
 from openhcs.core.runtime_plane_projection import (
     RuntimePlaneAxis,
@@ -231,13 +232,16 @@ def normalize_intensity_payload(
     ) -> Any:
         """Normalize the declared current domain, independently of storage dtype."""
         data = image_payload_data(payload)
-        array = np.asarray(MemoryType(detect_memory_type(data)).to_numpy(data))
-        target_dtype = self.normalization_dtype(array.dtype, dtype)
+        array = data if is_array_payload(data) else np.asarray(data)
+        memory_type = MemoryType(detect_memory_type(array))
+        target_dtype = self.normalization_dtype(
+            memory_type.canonical_dtype_name(array.dtype), dtype,
+        )
         if target_dtype is None:
             return payload
         if self.has_normalized_intensity:
             return self.payload_with(
-                array.astype(target_dtype, copy=False), image_payload_mask(payload),
+                memory_type.astype(array, target_dtype), image_payload_mask(payload),
             )
         if self.has_leading_intensity_axis and self.source_plane_intensity_scales:
             if len(self.source_plane_intensity_scales) != len(array):
@@ -258,7 +262,10 @@ def normalize_intensity_payload(
                 ),
             )
             return metadata.payload_with(
-                np.stack(tuple(plane for plane, _ in normalized_planes)),
+                memory_type.stack_arrays(
+                    [plane for plane, _ in normalized_planes],
+                    memory_type.device_id_of(array),
+                ),
                 image_payload_mask(payload),
             )
         normalized, proof_scale = self.normalized_intensity_array(
@@ -271,20 +278,22 @@ def normalize_intensity_payload(
 
     @staticmethod
     def normalized_intensity_array(
-        array: np.ndarray, *, target_dtype: np.dtype, scale: float | None,
-    ) -> tuple[np.ndarray, int | None]:
+        array: Any, *, target_dtype: np.dtype, scale: float | None,
+    ) -> tuple[Any, int | None]:
         """Apply one numerical recipe without projecting image source identity."""
+        memory_type = MemoryType(detect_memory_type(array))
+        source_dtype = np.dtype(memory_type.canonical_dtype_name(array.dtype))
         if scale is None:
             # Bare arrays are admitted at this original numerical boundary. A
             # promoted float uses its declared source scale, never a range guess.
-            scale = image_intensity_scale_for_dtype(array.dtype)
-        normalized = array.astype(target_dtype, copy=False)
+            scale = image_intensity_scale_for_dtype(source_dtype)
+        normalized = memory_type.astype(array, target_dtype)
         proof_scale = None
         if scale is not None:
             if not np.isfinite(scale) or scale <= 0:
                 raise ValueError("Source intensity scale must be finite and positive.")
             normalized = normalized / float(scale)
-            if np.issubdtype(array.dtype, np.integer) and float(scale).is_integer():
+            if np.issubdtype(source_dtype, np.integer) and float(scale).is_integer():
                 proof_scale = int(scale)
         return normalized, proof_scale
 
@@ -1614,7 +1623,13 @@ def project_image_mask_to_data_domain(
     """Validate a mask against explicit image-domain metadata."""
     if mask is None:
         return None
-    mask_array = np.asarray(mask, dtype=bool)
+    data_array = image_payload_data(data)
+    mask_array = image_payload_data(mask) if is_array_payload(mask) else np.asarray(mask)
+    target = MemoryType(detect_memory_type(data_array))
+    mask_array = MemoryType(detect_memory_type(mask_array)).convert_to(
+        mask_array, target, target.device_id_of(data_array),
+    )
+    mask_array = target.astype(mask_array, bool)
     mask_shape = tuple(mask_array.shape)
     resolved_metadata = image_payload_metadata(data) if metadata is None else metadata
     mask_domain = resolved_metadata.mask_domain(data)
@@ -2218,7 +2233,8 @@ def projected_channel_mask(
         channel_index: int,
         channel_axis: int,
     ) -> Any:
-        mask_array = np.asarray(mask, dtype=bool)
+        mask_array = image_payload_data(mask) if is_array_payload(mask) else np.asarray(mask)
+        mask_array = MemoryType(detect_memory_type(mask_array)).astype(mask_array, bool)
         if mask_array.shape != image_payload_geometry(source_data).shape:
             return mask_array
         channel_mask = cls.channel_axis_slice(
@@ -2231,9 +2247,12 @@ def projected_channel_mask(
             == image_payload_geometry(channel_data).shape
         ):
             return channel_mask
-        squeezed_mask = np.squeeze(
+        squeezed_mask = MemoryType(detect_memory_type(channel_mask)).reshape(
             channel_mask,
-            axis=channel_axis % channel_mask.ndim,
+            tuple(
+                size for axis, size in enumerate(channel_mask.shape)
+                if axis != channel_axis % len(channel_mask.shape)
+            ),
         )
         if (
             image_payload_geometry(squeezed_mask).shape
@@ -2281,9 +2300,11 @@ def default_mask_shape(self) -> tuple[int, ...]:
             if axis != self.channel_axis
         )
 
-    def broadcast_to_data(self, mask: Any) -> np.ndarray:
+    def broadcast_to_data(self, mask: Any) -> Any:
         """Broadcast a valid mask across its declared non-spatial axes."""
-        mask_array = np.asarray(mask, dtype=bool)
+        mask_array = image_payload_data(mask) if is_array_payload(mask) else np.asarray(mask)
+        memory_type = MemoryType(detect_memory_type(mask_array))
+        mask_array = memory_type.astype(mask_array, bool)
         mask_shape = tuple(mask_array.shape)
         if mask_shape == self.data_shape:
             return mask_array
@@ -2293,8 +2314,8 @@ def broadcast_to_data(self, mask: Any) -> np.ndarray:
             broadcast_shape = [1] * len(self.data_shape)
             for mask_axis, data_axis in enumerate(self.spatial_axes_yx):
                 broadcast_shape[data_axis] = mask_shape[mask_axis]
-            return np.broadcast_to(
-                mask_array.reshape(tuple(broadcast_shape)),
+            return memory_type.broadcast_to(
+                memory_type.reshape(mask_array, tuple(broadcast_shape)),
                 self.data_shape,
             )
         channel_free_shape = (
@@ -2311,7 +2332,9 @@ def broadcast_to_data(self, mask: Any) -> np.ndarray:
                 f"Mask shape {mask_shape!r} is not valid for image "
                 f"shape {self.data_shape!r}."
             )
-        return np.broadcast_to(
-            np.expand_dims(mask_array, axis=self.channel_axis),
+        broadcast_shape = list(mask_shape)
+        broadcast_shape.insert(self.channel_axis, 1)
+        return memory_type.broadcast_to(
+            memory_type.reshape(mask_array, tuple(broadcast_shape)),
             self.data_shape,
         )

plans/runtime_plumbing/IMPLEMENTATION.md

@@ -11,7 +11,7 @@ The active performance objective remains execution and total pipeline latency, w
 | P2 | RuntimePatternDiscoveryCache; declared source bindings | Implemented in c1650d8. Production consumer probe confirms one physical filename parse across discovery, matching and indexed group selection, and re-admission when parser semantics change. Produced filename aliases remain live. |
 | P3 | StepOutputManifestStore; CompiledFunctionPattern | Implemented in 88522fb. Ordered address maps update only new records; producer-scoped invalidation eliminates revision counters. Existing immutable DurableSourceMetadata fixes publication slots; source metadata and filename aliases remain live. Publication policy explicitly approved by user. Derived path order is retained without changing insertion order. |
 | P4 | ProcessingContext; source projection authority; manifest | Implemented in c1650d8. Held authority rebinds strong owners; optional file eligibility and document changes remain live. Lazy logging and retained P3 order remove per-group formatting/sorting. Ambiguity admission still precedes cached payload return. |
-| P5 | Compiled invocation; ArrayBridge MemoryType; ImageFileFormat | Core implemented in c1d8728 and 09b353e. Target/device are captured; conversions use the real active source, including skipped calls. Native stacking removes mandatory GPU-host-GPU traffic. Disk format parent owns host projection; memory checkpoints keep their declared framework. Actual GPU receiving covers shape/dtype/device, independent buffers, mixed NumPy/CuPy stacks/conversion and direct disk writes. Mixed intrinsic-channel promotion and mask broadcasting still use host operations and need generic ArrayBridge capabilities before P5 is fully complete. |
+| P5 | Compiled invocation; ArrayBridge MemoryType; ImageFileFormat | Core implemented in c1d8728 and 09b353e. Target/device are captured; conversions use the real active source, including skipped calls. Native stacking removes mandatory GPU-host-GPU traffic. Disk format parent owns host projection; memory checkpoints keep their declared framework. #1080 completes native mixed-channel promotion, intensity reconciliation and mask geometry through ArrayBridge's existing operation owner. GTX 1080/CuPy composition receiving forbids downloads and checks exact values, masks, destination devices and independent stacks. Pyclesperanto lacks native changed-shape reshape/broadcast and boolean allocation; those operations report unsupported capability. P5 remains incomplete until a real mixed-device production pipeline has a complete before/after conversion profile, including skipped calls and disk boundaries. No CPU benchmark speedup is inferred from these GPU checks. |
 
 #641 is merged and removes post-compilation kernel warming; normal registry READY is the only child-cache consumer. Full catalog READY qualification passed. #630 (runtime ownership refactor), #635 (intensity maxima), #597 (saved compiler resolution) and #608 (morphology graph intermediates) are merged following the user's explicit request to merge locally qualified work. Their earlier negative whole-pipeline observations remain documented; merge is not evidence of an end-to-end speedup.
 

pyproject.toml

@@ -104,7 +104,7 @@ dependencies = [
     "objectstate>=1.3.1,<2",
     "python-introspect>=0.2.1,<0.3",
     "metaclass-registry>=0.2.2,<0.3",
-    "arraybridge>=0.3.6,<0.4",
+    "arraybridge>=0.3.7,<0.4",
     "polystore>=0.3.3,<0.4",
     "pyqt-reactive>=0.3.27,<0.4",
 ]