Receivers take the components that split windows and layers (0.4.0)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "polystore"
-version = "0.3.4"
+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",
 ]
 

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."""

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."""

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",
 ]

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",
 ]

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,
     )

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"]

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]