From 8d59883761f1ecc7b21abb81bbff68f364ec77d8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 03:54:02 -0400 Subject: [PATCH 1/3] Receivers take the components that split windows and layers; bump to 0.3.5 The Fiji window grouping and the napari route key re-read zmqruntime's flat ViewerComponentMode enum. The calling viewer now passes the components its own slot family places in separate windows or layers, so these helpers name no viewer vocabulary. normalize_component_layout (test-only) and the NapariDisplayWireField/FijiDisplayWireField rosters (the OpenHCS display settings dataclasses are the wire keys now) are deleted. Co-Authored-By: Claude Opus 5.5 --- pyproject.toml | 2 +- src/polystore/fiji_stream.py | 7 -- src/polystore/napari_stream.py | 7 -- src/polystore/streaming/receivers/__init__.py | 6 +- .../streaming/receivers/core/__init__.py | 4 +- .../receivers/core/window_projection.py | 39 ++-------- .../streaming/receivers/napari/__init__.py | 3 +- .../streaming/receivers/napari/layer_key.py | 70 +----------------- tests/test_streaming_receiver_core.py | 73 +++++-------------- 9 files changed, 36 insertions(+), 175 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d06a3c5..bd9151c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "polystore" -version = "0.3.4" +version = "0.3.5" description = "Framework-agnostic multi-backend storage abstraction for ML and scientific computing" readme = "README.md" requires-python = ">=3.11" diff --git a/src/polystore/fiji_stream.py b/src/polystore/fiji_stream.py index 95da974..a988b13 100644 --- a/src/polystore/fiji_stream.py +++ b/src/polystore/fiji_stream.py @@ -12,7 +12,6 @@ """ import logging -from enum import Enum from zmqruntime.viewer_protocol import ( ViewerBatchItemWireField, @@ -37,12 +36,6 @@ logger = logging.getLogger(__name__) -class FijiDisplayWireField(str, Enum): - """Fiji-specific display fields inside the shared viewer display payload.""" - - LUT = "lut" - AUTO_CONTRAST = "auto_contrast" - class FijiMessageMetadata: """Typed access to optional Fiji message metadata.""" diff --git a/src/polystore/napari_stream.py b/src/polystore/napari_stream.py index 551a54a..6e3b1a3 100644 --- a/src/polystore/napari_stream.py +++ b/src/polystore/napari_stream.py @@ -13,7 +13,6 @@ """ import logging -from enum import Enum from zmqruntime.viewer_protocol import ( ViewerBatchItemWireField, @@ -33,12 +32,6 @@ logger = logging.getLogger(__name__) -class NapariDisplayWireField(str, Enum): - """Napari-specific display fields inside the shared viewer display payload.""" - - COLORMAP = "colormap" - VARIABLE_SIZE_HANDLING = "variable_size_handling" - class NapariStreamingBackend(StreamingBackend): """Napari streaming backend with automatic registration.""" diff --git a/src/polystore/streaming/receivers/__init__.py b/src/polystore/streaming/receivers/__init__.py index c6d58db..c625ccc 100644 --- a/src/polystore/streaming/receivers/__init__.py +++ b/src/polystore/streaming/receivers/__init__.py @@ -11,12 +11,11 @@ GroupedWindowItems, WindowProjectionPayloadProvider, WindowProjectionSource, - group_items_by_component_modes, + group_items_into_windows, ) from polystore.streaming.receivers.fiji.fiji_batch_processor import FijiBatchProcessor from polystore.streaming.receivers.napari import ( NapariBatchProcessor, - normalize_component_layout, build_route_key, ) @@ -27,9 +26,8 @@ "GroupedWindowItems", "WindowProjectionPayloadProvider", "WindowProjectionSource", - "group_items_by_component_modes", + "group_items_into_windows", "FijiBatchProcessor", "NapariBatchProcessor", - "normalize_component_layout", "build_route_key", ] diff --git a/src/polystore/streaming/receivers/core/__init__.py b/src/polystore/streaming/receivers/core/__init__.py index 786f526..16fd06c 100644 --- a/src/polystore/streaming/receivers/core/__init__.py +++ b/src/polystore/streaming/receivers/core/__init__.py @@ -9,7 +9,7 @@ GroupedWindowItems, WindowProjectionPayloadProvider, WindowProjectionSource, - group_items_by_component_modes, + group_items_into_windows, ) __all__ = [ @@ -19,5 +19,5 @@ "GroupedWindowItems", "WindowProjectionPayloadProvider", "WindowProjectionSource", - "group_items_by_component_modes", + "group_items_into_windows", ] diff --git a/src/polystore/streaming/receivers/core/window_projection.py b/src/polystore/streaming/receivers/core/window_projection.py index e25627f..032e3f9 100644 --- a/src/polystore/streaming/receivers/core/window_projection.py +++ b/src/polystore/streaming/receivers/core/window_projection.py @@ -15,18 +15,10 @@ from zmqruntime.viewer_protocol import ( ViewerBatchDisplayPayload, ViewerBatchItemWireField, - ViewerComponentMode, ViewerWireMapping, ViewerWireValue, ) - -WINDOW_COMPONENT_MODES = ( - ViewerComponentMode.WINDOW, - ViewerComponentMode.CHANNEL, - ViewerComponentMode.SLICE, - ViewerComponentMode.FRAME, -) WindowLabel = tuple[str, ViewerWireValue] WindowProjectionItemT = TypeVar("WindowProjectionItemT") WindowProjectionProviderT = TypeVar( @@ -56,12 +48,9 @@ def from_mapping( @dataclass(frozen=True, slots=True) class GroupedWindowItems(Generic[WindowProjectionItemT]): - """Projection result for a single batch.""" + """Items of one batch grouped into windows.""" - window_components: list[str] - channel_components: list[str] - slice_components: list[str] - frame_components: list[str] + window_components: tuple[str, ...] windows: dict[str, list[WindowProjectionItemT]] fixed_window_labels: dict[str, tuple[WindowLabel, ...]] @@ -143,26 +132,13 @@ def _required_mapping( return dict(value) -def group_items_by_component_modes( +def group_items_into_windows( items: Sequence[WindowProjectionSource[WindowProjectionItemT]], display_layout: ViewerBatchDisplayPayload, + window_components: Sequence[str], ) -> GroupedWindowItems[WindowProjectionItemT]: - """Project items into window groups using declared component modes.""" - mode_groups = display_layout.component_mode_groups(WINDOW_COMPONENT_MODES) - mode_groups.require_all_supported("window projection") - - window_components = list( - mode_groups.components_for_mode(ViewerComponentMode.WINDOW) - ) - channel_components = list( - mode_groups.components_for_mode(ViewerComponentMode.CHANNEL) - ) - slice_components = list( - mode_groups.components_for_mode(ViewerComponentMode.SLICE) - ) - frame_components = list( - mode_groups.components_for_mode(ViewerComponentMode.FRAME) - ) + """Group items into one window per value combination of ``window_components``.""" + window_components = tuple(window_components) windows: dict[str, list[WindowProjectionItemT]] = {} fixed_window_labels: dict[str, tuple[WindowLabel, ...]] = {} @@ -226,9 +202,6 @@ def group_items_by_component_modes( return GroupedWindowItems( window_components=window_components, - channel_components=channel_components, - slice_components=slice_components, - frame_components=frame_components, windows=windows, fixed_window_labels=fixed_window_labels, ) diff --git a/src/polystore/streaming/receivers/napari/__init__.py b/src/polystore/streaming/receivers/napari/__init__.py index 6472c4c..ce5e578 100644 --- a/src/polystore/streaming/receivers/napari/__init__.py +++ b/src/polystore/streaming/receivers/napari/__init__.py @@ -2,8 +2,7 @@ from polystore.streaming.receivers.napari.napari_batch_processor import NapariBatchProcessor from polystore.streaming.receivers.napari.layer_key import ( - normalize_component_layout, build_route_key, ) -__all__ = ["NapariBatchProcessor", "normalize_component_layout", "build_route_key"] +__all__ = ["NapariBatchProcessor", "build_route_key"] diff --git a/src/polystore/streaming/receivers/napari/layer_key.py b/src/polystore/streaming/receivers/napari/layer_key.py index 857b8cd..f17063c 100644 --- a/src/polystore/streaming/receivers/napari/layer_key.py +++ b/src/polystore/streaming/receivers/napari/layer_key.py @@ -2,53 +2,23 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from polystore.streaming.identity import StreamProducerIdentity, StreamRouteKeyAuthority from polystore.streaming_constants import StreamingDataType -from zmqruntime.viewer_protocol import ( - ViewerBatchDisplayPayload, - ViewerComponentMode, - ViewerDisplayConfigWireField, - ViewerWireMapping, - ViewerWireValue, -) - - -def normalize_component_layout( - display_config: ViewerBatchDisplayPayload | ViewerWireMapping, -) -> ViewerBatchDisplayPayload: - """Return canonical display layout from a viewer display-config payload.""" - if isinstance(display_config, ViewerBatchDisplayPayload): - return display_config - if isinstance(display_config, dict): - return ViewerBatchDisplayPayload( - component_modes=_required_mapping( - display_config, - ViewerDisplayConfigWireField.COMPONENT_MODES.value, - ), - component_order=_required_sequence( - display_config, - ViewerDisplayConfigWireField.COMPONENT_ORDER.value, - ), - ) - - raise TypeError( - "Napari component layout requires ViewerBatchDisplayPayload or mapping, " - f"got {type(display_config).__name__}." - ) +from zmqruntime.viewer_protocol import ViewerWireValue def build_route_key( producer_identity: StreamProducerIdentity | Mapping[str, ViewerWireValue], component_info: Mapping[str, ViewerWireValue], - display_layout: ViewerBatchDisplayPayload, + layer_components: Sequence[str], data_type: StreamingDataType, ) -> str: """Build a hidden route key from producer identity, layer components, and type.""" producer = StreamProducerIdentity.from_payload(producer_identity) route_parts: list[str] = list(producer.route_parts()) - for component in display_layout.components_for_mode(ViewerComponentMode.LAYER): + for component in layer_components: if component not in component_info: raise ValueError( f"Napari route key missing layer component {component!r}." @@ -59,35 +29,3 @@ def build_route_key( return f"{route_key}{data_type.napari_layer_suffix}" - -def _required_mapping( - payload: Mapping[str, ViewerWireValue], - field_name: str, -) -> dict[str, str]: - if field_name not in payload: - raise ValueError(f"Display config missing required field {field_name!r}.") - value = payload[field_name] - if not isinstance(value, Mapping): - raise TypeError( - f"Display config field {field_name!r} must be a mapping, " - f"got {type(value).__name__}." - ) - return { - str(component): str(mode) - for component, mode in value.items() - } - - -def _required_sequence( - payload: Mapping[str, ViewerWireValue], - field_name: str, -) -> list[str]: - if field_name not in payload: - raise ValueError(f"Display config missing required field {field_name!r}.") - value = payload[field_name] - if isinstance(value, str) or not isinstance(value, list | tuple): - raise TypeError( - f"Display config field {field_name!r} must be a sequence, " - f"got {type(value).__name__}." - ) - return [str(component) for component in value] diff --git a/tests/test_streaming_receiver_core.py b/tests/test_streaming_receiver_core.py index f3ce531..dea0827 100644 --- a/tests/test_streaming_receiver_core.py +++ b/tests/test_streaming_receiver_core.py @@ -13,12 +13,9 @@ from polystore.streaming.receivers.core import ( DebouncedBatchEngine, WindowProjectionSource, - group_items_by_component_modes, -) -from polystore.streaming.receivers.napari import ( - build_route_key, - normalize_component_layout, + group_items_into_windows, ) +from polystore.streaming.receivers.napari import build_route_key from polystore.streaming_constants import StreamingDataType @@ -64,7 +61,7 @@ def artifact_output( ) -def test_group_items_by_component_modes_keys_windows_by_producer_identity() -> None: +def test_group_items_into_windows_keys_windows_by_producer_identity() -> None: image_identity = PipelineProducerFixture.main_output( step_name="RawLoad", pipeline_position=1, @@ -90,18 +87,16 @@ def test_group_items_by_component_modes_keys_windows_by_producer_identity() -> N component_modes = {"well": "frame", "channel": "channel"} component_order = ["well", "channel"] - grouped = group_items_by_component_modes( + grouped = group_items_into_windows( WindowProjectionSource.from_wire_payloads(items), display_layout=ViewerBatchDisplayPayload( component_modes=component_modes, component_order=component_order, ), + window_components=(), ) - assert grouped.window_components == [] - assert grouped.channel_components == ["channel"] - assert grouped.frame_components == ["well"] - assert grouped.slice_components == [] + assert grouped.window_components == () assert grouped.fixed_window_labels[ "origin_pipeline_kind_artifact_projection_Nuclei_step_2_name_Segment" ] == (("producer", "3. Segment Nuclei"),) @@ -123,7 +118,7 @@ def test_named_main_outputs_share_projection_and_keep_exact_provenance() -> None pipeline_position=2, ) - grouped = group_items_by_component_modes( + grouped = group_items_into_windows( WindowProjectionSource.from_wire_payloads( [ { @@ -142,6 +137,7 @@ def test_named_main_outputs_share_projection_and_keep_exact_provenance() -> None component_modes={"channel": "channel"}, component_order=("channel",), ), + window_components=(), ) assert tuple(grouped.windows) == ( @@ -158,33 +154,23 @@ def test_named_main_outputs_share_projection_and_keep_exact_provenance() -> None assert build_route_key( producer_identity=first, component_info={"channel": 1}, - display_layout=ViewerBatchDisplayPayload( - component_modes={"channel": "stack"}, - component_order=("channel",), - ), + layer_components=(), data_type=StreamingDataType.IMAGE, ) == build_route_key( producer_identity=second, component_info={"channel": 2}, - display_layout=ViewerBatchDisplayPayload( - component_modes={"channel": "stack"}, - component_order=("channel",), - ), + layer_components=(), data_type=StreamingDataType.IMAGE, ) - layer_layout = ViewerBatchDisplayPayload( - component_modes={"channel": "layer"}, - component_order=("channel",), - ) assert build_route_key( producer_identity=first, component_info={"channel": 1}, - display_layout=layer_layout, + layer_components=("channel",), data_type=StreamingDataType.IMAGE, ) != build_route_key( producer_identity=second, component_info={"channel": 2}, - display_layout=layer_layout, + layer_components=("channel",), data_type=StreamingDataType.IMAGE, ) @@ -202,7 +188,7 @@ def test_named_main_outputs_cannot_claim_the_same_projection_slot() -> None: ) try: - group_items_by_component_modes( + group_items_into_windows( WindowProjectionSource.from_wire_payloads( [ { @@ -221,6 +207,7 @@ def test_named_main_outputs_cannot_claim_the_same_projection_slot() -> None: component_modes={"channel": "channel"}, component_order=("channel",), ), + window_components=(), ) except ValueError as error: assert "same component coordinate" in str(error) @@ -228,14 +215,14 @@ def test_named_main_outputs_cannot_claim_the_same_projection_slot() -> None: raise AssertionError("distinct producer slot collision must fail loudly") -def test_group_items_by_component_modes_rejects_missing_metadata() -> None: +def test_group_items_into_windows_rejects_missing_metadata() -> None: producer = PipelineProducerFixture.main_output( step_name="RawLoad", pipeline_position=1, ) try: - group_items_by_component_modes( + group_items_into_windows( WindowProjectionSource.from_wire_payloads( [{"producer_identity": producer.to_payload()}] ), @@ -243,6 +230,7 @@ def test_group_items_by_component_modes_rejects_missing_metadata() -> None: component_modes={"well": "window"}, component_order=["well"], ), + window_components=('well',), ) except ValueError as error: assert "metadata" in str(error) @@ -291,26 +279,18 @@ def test_napari_route_key_builder_uses_producer_layer_components_and_payload_typ step_name="Segment", pipeline_position=2, ) - component_modes = {"well": "layer", "channel": "stack", "site": "layer"} - component_order = ["well", "channel", "site"] component_info = {"well": "A01", "channel": 2, "site": 3} key_image = build_route_key( producer_identity=producer, component_info=component_info, - display_layout=ViewerBatchDisplayPayload( - component_modes=component_modes, - component_order=component_order, - ), + layer_components=("well", "site"), data_type=StreamingDataType.IMAGE, ) key_shapes = build_route_key( producer_identity=producer, component_info=component_info, - display_layout=ViewerBatchDisplayPayload( - component_modes=component_modes, - component_order=component_order, - ), + layer_components=("well", "site"), data_type=StreamingDataType.SHAPES, ) @@ -329,10 +309,7 @@ def test_napari_route_key_builder_rejects_missing_layer_component() -> None: build_route_key( producer_identity=producer, component_info={"well": "A01"}, - display_layout=ViewerBatchDisplayPayload( - component_modes={"well": "layer", "site": "layer"}, - component_order=["well", "site"], - ), + layer_components=("well", "site"), data_type=StreamingDataType.IMAGE, ) except ValueError as error: @@ -341,16 +318,6 @@ def test_napari_route_key_builder_rejects_missing_layer_component() -> None: raise AssertionError("missing layer component must fail loudly") -def test_normalize_component_layout_dict_config() -> None: - display_layout = normalize_component_layout( - { - "component_modes": {"well": "layer", "channel": "stack"}, - "component_order": ["well", "channel"], - } - ) - assert list(display_layout.component_order) == ["well", "channel"] - assert display_layout.component_modes["well"] == "layer" - def test_debounced_batch_engine_flush_processes_pending_once() -> None: processed: list[tuple[list[dict], dict]] = [] From 32366a0908717cebca0b9968333e31e53b479c0d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 04:02:45 -0400 Subject: [PATCH 2/3] Bump to 0.4.0 and require zmqruntime 0.5 The receiver API change is breaking, and zmqruntime 0.5 deletes the mode enum these receivers used to read. Co-Authored-By: Claude Opus 5.5 --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index bd9151c..9da0f32 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "polystore" -version = "0.3.5" +version = "0.4.0" description = "Framework-agnostic multi-backend storage abstraction for ML and scientific computing" readme = "README.md" requires-python = ">=3.11" @@ -45,7 +45,7 @@ dependencies = [ "zarr>=3.0.10", # Reads Zarr 2 and 3 stores; output format is backend-owned "numcodecs>=0.14.0", "ome-zarr>=0.12.2", # Required for OME-ZARR HCS compliance - "zmqruntime>=0.3.0,<0.5", + "zmqruntime>=0.5.0,<0.6", "platformdirs>=4.0.0", ] From 7f83c4650b9c705570ccb79cdb0e6d6acfd3b886 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 12:01:20 -0400 Subject: [PATCH 3/3] Lock zmqruntime 0.5.0 for the 0.4.0 release Co-Authored-By: Claude Opus 5.5 --- uv.lock | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/uv.lock b/uv.lock index 6d8b68e..b91334b 100644 --- a/uv.lock +++ b/uv.lock @@ -2905,7 +2905,7 @@ wheels = [ [[package]] name = "polystore" -version = "0.3.4" +version = "0.4.0" source = { editable = "." } dependencies = [ { name = "arraybridge" }, @@ -3046,7 +3046,7 @@ requires-dist = [ { name = "torch", marker = "extra == 'torch'", specifier = ">=2.0.0" }, { name = "torchvision", marker = "extra == 'gpu'", specifier = ">=0.21.0,<0.23.0" }, { name = "zarr", specifier = ">=3.0.10" }, - { name = "zmqruntime", specifier = ">=0.3.0,<0.5" }, + { name = "zmqruntime", specifier = ">=0.5.0,<0.6" }, ] provides-extras = ["zarr", "torch", "jax", "tensorflow", "cupy", "streaming", "omero", "bioformats", "all", "dev", "docs", "gpu"] @@ -5028,7 +5028,7 @@ wheels = [ [[package]] name = "zmqruntime" -version = "0.4.0" +version = "0.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "annotated-types" }, @@ -5040,7 +5040,7 @@ dependencies = [ { name = "pyzmq" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/25/7c/3d6ebbe5b83462876988bf1ca417542269fd3c19f6402c72089ac554c950/zmqruntime-0.4.0.tar.gz", hash = "sha256:cfc3799fc5bf98aa7a57cee0d759b6a6fa7108136e8f937db122b54ddf165279", size = 718700, upload-time = "2026-10-04T23:28:12.247Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a8/5f/0e4bb73ef7762ac9bbb16dd398f4488e2c69edbc3d552fa91f4f87585245/zmqruntime-0.5.0.tar.gz", hash = "sha256:f78efa887a16c4b454211647adf901c3d4ea6f816a7449a8fcc0da2b8b4d02d3", size = 720752, upload-time = "2026-10-10T15:20:09.309Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/7a/26/b4e6f1d70b44410acd6eb4a9d88000dcf1e8ab83a3cc008c213a39c0790c/zmqruntime-0.4.0-py3-none-any.whl", hash = "sha256:6c3ea4a8a8d2e516faeae670f3e9961ecfc1237cc1a8145118f04a27ed903fa0", size = 95908, upload-time = "2026-10-04T23:28:10.621Z" }, + { url = "https://files.pythonhosted.org/packages/00/a1/af68de3633f5a17a4029481f518ce95867ce500d76d6d327a9c0ef91b658/zmqruntime-0.5.0-py3-none-any.whl", hash = "sha256:9aa136114501abe81e551fae58f517e26265a6eaf5555e26829d8c3ff3750523", size = 96024, upload-time = "2026-10-10T15:20:07.657Z" }, ]