diff --git a/external/arraybridge b/external/arraybridge index 1e53d03d9f..ee4c0e93d5 160000 --- a/external/arraybridge +++ b/external/arraybridge @@ -1 +1 @@ -Subproject commit 1e53d03d9f468322a1c085c8be29485fb139caf2 +Subproject commit ee4c0e93d58b6d88bafbcf5168c8db67ca34fade diff --git a/openhcs/core/aligned_image_payload.py b/openhcs/core/aligned_image_payload.py index 9b58b9abc4..98fc567cdf 100644 --- a/openhcs/core/aligned_image_payload.py +++ b/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( diff --git a/openhcs/core/runtime_image_values.py b/openhcs/core/runtime_image_values.py index 4f2b20eee9..50c51fcc18 100644 --- a/openhcs/core/runtime_image_values.py +++ b/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, ) diff --git a/plans/runtime_plumbing/IMPLEMENTATION.md b/plans/runtime_plumbing/IMPLEMENTATION.md index 69e34220f9..cddcfb8a2b 100644 --- a/plans/runtime_plumbing/IMPLEMENTATION.md +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 848375294b..eef8e6b0ad 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/tests/unit/test_aligned_image_payload_device.py b/tests/unit/test_aligned_image_payload_device.py index b4335dd7c9..3a5b6b88a6 100644 --- a/tests/unit/test_aligned_image_payload_device.py +++ b/tests/unit/test_aligned_image_payload_device.py @@ -65,6 +65,12 @@ def __getitem__(self, key): def astype(self, dtype, copy=False): return type(self)(self.values.astype(dtype, copy=copy), self.device.id) + def __truediv__(self, scalar): + return type(self)(self.values / scalar, self.device.id) + + def reshape(self, shape): + return type(self)(self.values.reshape(shape), self.device.id) + def upload(values): state['uploads'].append(state['device']) return DeviceArray(values, state['device']) @@ -74,6 +80,12 @@ def upload(values): runtime=SimpleNamespace(getDeviceCount=lambda: 2), Device=DeviceScope, ), array=upload, + broadcast_to=lambda value, shape: DeviceArray( + np.broadcast_to(value.values, shape), value.device.id, + ), + ones=lambda shape, dtype=bool: DeviceArray( + np.ones(shape, dtype=dtype), state['device'], + ), stack=lambda values, axis=0: DeviceArray( np.stack([value.values for value in values], axis=axis), state['device'], ), @@ -116,6 +128,7 @@ def test_mixed_intensity_composition_uses_declared_memory_conversion( 'memory_type': destination, 'device_id': 1 if destination == 'cupy' else None, } result = context.compose(**kwargs) + assert bool(state['downloads']) == (destination == 'numpy') output = image_payload_data(result) requested = 'cupy' if destination == 'implicit' else destination owner = MemoryType(requested) @@ -128,10 +141,7 @@ def test_mixed_intensity_composition_uses_declared_memory_conversion( expected = raw.mask.values if composition is ImagePayloadStackContext or not declared_spatial_extent: expected = np.stack(tuple(value.mask.values for value in inputs)) - # Bundle's non-shared mask preserves its mask owner's device; dense stack - # masks are returned on the declared output device through the ancestor. - if composition is ImagePayloadStackContext: - assert mask_owner.device_id_of(mask) == owner.device_id_of(output) + assert mask_owner.device_id_of(mask) == owner.device_id_of(output) np.testing.assert_array_equal(masks, expected) assert state['downloads'] assert all(device == 1 for device in state['downloads']) @@ -173,6 +183,42 @@ def test_image_bundle_stacks_on_the_payload_framework_device(monkeypatch) -> Non assert observed == [(payloads, MemoryType.CUPY.value, 7)] +@pytest.mark.parametrize('mask_kind', ('full', 'channel-free', 'absent')) +@pytest.mark.parametrize('destination', ('numpy', 'cupy')) +def test_mixed_channel_bundle_promotes_masks_in_the_output_domain( + declared_cupy_leaf, mask_kind, destination, +): + DeviceArray, state = declared_cupy_leaf + gray = np.arange(6, dtype=np.uint8).reshape(2, 3) + color = np.repeat(gray[..., None], 3, axis=2) + spatial = gray % 2 == 0 + color_mask = ( + None if mask_kind == 'absent' else DeviceArray( + np.repeat(spatial[..., None], 3, axis=2) + if mask_kind == 'full' else spatial, + ) + ) + inputs = ( + ImagePayloadMetadata(source_channel_axis=2).payload_with( + DeviceArray(color), color_mask, + ), + ImagePayloadMetadata().payload_with(DeviceArray(gray), DeviceArray(spatial)), + ) + result = ImagePayloadBundleContext.from_payloads(inputs).compose( + memory_type=destination, device_id=1 if destination == 'cupy' else None, + ) + if destination == 'cupy': + assert not state['downloads'] + owner = MemoryType(destination) + data, mask = image_payload_data(result), image_payload_mask(result) + assert owner.device_id_of(mask) == owner.device_id_of(data) + np.testing.assert_array_equal(owner.to_numpy(data), np.stack((color, color))) + expected = np.stack((np.ones_like(spatial) if mask_kind == 'absent' else spatial, spatial)) + if mask_kind == 'full': + expected = np.repeat(expected[..., None], 3, axis=3) + np.testing.assert_array_equal(owner.to_numpy(mask), expected) + + def test_independent_cohort_copy_retains_nested_named_bundle_domains(): from openhcs.core.aligned_image_payload import ( AlignedImageSliceContext, AlignedImageStack, ImageOutputBundle,