Unify typed viewer display projection

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "polystore"
-version = "0.1.24"
+version = "0.1.25"
 description = "Framework-agnostic multi-backend storage abstraction for ML and scientific computing"
 readme = "README.md"
 requires-python = ">=3.11"

src/polystore/__init__.py

@@ -2,7 +2,7 @@
 Polystore package exports.
 """
 
-__version__ = "0.1.24"
+__version__ = "0.1.25"
 
 from .atomic import (
     FileLockError,

src/polystore/fiji_stream.py

@@ -14,25 +14,25 @@
 import logging
 from enum import Enum
 
+from zmqruntime.viewer_protocol import (
+    ViewerBatchItemWireField,
+    ViewerBatchWireField,
+    ViewerWireMapping,
+    ViewerWireValue,
+)
+
 from .constants import Backend
-from .streaming_constants import StreamingDataType
+from .roi_converters import FijiROIConverter
 from .streaming import (
     FilePath,
     RoiStreamPayload,
-    StreamingBuiltBatch,
     StreamingBackend,
+    StreamingBuiltBatch,
     StreamingComponentNamesRequest,
     StreamingItemPreparationRequest,
-    ViewerDisplayPayloadExtra,
 )
 from .streaming.viewer_transport import ViewerStreamItemPayload, ViewerStreamRequest
-from .roi_converters import FijiROIConverter
-from zmqruntime.viewer_protocol import (
-    ViewerBatchItemWireField,
-    ViewerBatchWireField,
-    ViewerWireMapping,
-    ViewerWireValue,
-)
+from .streaming_constants import StreamingDataType
 
 logger = logging.getLogger(__name__)
 
@@ -44,23 +44,6 @@ class FijiDisplayWireField(str, Enum):
     AUTO_CONTRAST = "auto_contrast"
 
 
-class FijiDisplayPayload:
-    """Display payload projection for Fiji stream messages."""
-
-    @staticmethod
-    def auto_contrast_value(display_config) -> bool:
-        return display_config.auto_contrast
-
-    @classmethod
-    def from_display_config(cls, display_config) -> dict[str, ViewerWireValue]:
-        return {
-            FijiDisplayWireField.LUT.value: display_config.get_lut_name(),
-            FijiDisplayWireField.AUTO_CONTRAST.value: cls.auto_contrast_value(
-                display_config
-            ),
-        }
-
-
 class FijiMessageMetadata:
     """Typed access to optional Fiji message metadata."""
 
@@ -86,14 +69,6 @@ class FijiStreamingBackend(StreamingBackend):
     VIEWER_TYPE = 'fiji'
     SHM_PREFIX = 'fiji_'
 
-    def display_payload_extra(
-        self,
-        stream_request: ViewerStreamRequest,
-    ) -> ViewerDisplayPayloadExtra:
-        return ViewerDisplayPayloadExtra.from_mapping(
-            FijiDisplayPayload.from_display_config(stream_request.display_config)
-        )
-
     def message_extra(
         self,
         stream_request: ViewerStreamRequest,

src/polystore/napari_stream.py

@@ -15,21 +15,20 @@
 import logging
 from enum import Enum
 
+from zmqruntime.viewer_protocol import (
+    ViewerBatchItemWireField,
+    ViewerWireValue,
+)
+
 from .constants import Backend
+from .roi_converters import NapariROIConverter
 from .streaming import (
     FilePath,
     RoiStreamPayload,
     StreamingBackend,
     StreamingItemPreparationRequest,
-    ViewerDisplayPayloadExtra,
-)
-from .streaming.viewer_transport import ViewerStreamItemPayload, ViewerStreamRequest
-from .roi_converters import NapariROIConverter
-from zmqruntime.viewer_protocol import (
-    ViewerBatchItemWireField,
-    ViewerWireMapping,
-    ViewerWireValue,
 )
+from .streaming.viewer_transport import ViewerStreamItemPayload
 
 logger = logging.getLogger(__name__)
 
@@ -41,41 +40,13 @@ class NapariDisplayWireField(str, Enum):
     VARIABLE_SIZE_HANDLING = "variable_size_handling"
 
 
-class NapariDisplayPayload:
-    """Display payload projection for Napari stream messages."""
-
-    @staticmethod
-    def variable_size_handling_value(display_config):
-        variable_size_handling = display_config.variable_size_handling
-        if variable_size_handling is None:
-            return None
-        return variable_size_handling.value
-
-    @classmethod
-    def from_display_config(cls, display_config) -> dict[str, ViewerWireValue]:
-        return {
-            NapariDisplayWireField.COLORMAP.value: display_config.get_colormap_name(),
-            NapariDisplayWireField.VARIABLE_SIZE_HANDLING.value: (
-                cls.variable_size_handling_value(display_config)
-            ),
-        }
-
-
 class NapariStreamingBackend(StreamingBackend):
     """Napari streaming backend with automatic registration."""
     _backend_type = Backend.NAPARI_STREAM.value
 
     VIEWER_TYPE = 'napari'
     SHM_PREFIX = 'napari_'
 
-    def display_payload_extra(
-        self,
-        stream_request: ViewerStreamRequest,
-    ) -> ViewerDisplayPayloadExtra:
-        return ViewerDisplayPayloadExtra.from_mapping(
-            NapariDisplayPayload.from_display_config(stream_request.display_config)
-        )
-
     def _prepare_shapes_data(
         self,
         data: RoiStreamPayload,

src/polystore/streaming/_streaming_backend.py

@@ -17,38 +17,39 @@
 from pathlib import Path
 from types import MappingProxyType
 from typing import TypeAlias
+
 import numpy as np
 import zmq
 from arraybridge import convert_memory, detect_memory_type
 from arraybridge.types import MemoryType as ArrayBridgeMemoryType
+from zmqruntime.ack_listener import GlobalAckListener
+from zmqruntime.config import ZMQConfig
+from zmqruntime.viewer_protocol import (
+    ViewerBatchItemWireField,
+    ViewerBatchMessagePayload,
+    ViewerComponentMetadataPayload,
+    ViewerDisplayConfigWireField,
+    ViewerTransportEndpoint,
+    ViewerWireMapping,
+    ViewerWirePayload,
+    ViewerWireValue,
+)
 
 from ..base import DataSink
 from ..formats import PIXEL_PAYLOAD_EXTENSIONS
-from ..streaming_constants import StreamingDataType
 from ..roi import ROI, ROI_ZIP_EXTENSION
 from ..roi_converters import ROIShapeNapariPayloadConverter
+from ..streaming_constants import StreamingDataType
 from ..zmq_config import POLYSTORE_ZMQ_CONFIG
 from .viewer_transport import (
     ViewerMicroscopeHandlerABC,
+    ViewerStreamBackendKwargs,
     ViewerStreamBatchItemInput,
     ViewerStreamBatchItemSource,
-    ViewerStreamBackendKwargs,
     ViewerStreamItemPayload,
     ViewerStreamRequest,
     ViewerTransportDefaults,
 )
-from zmqruntime.ack_listener import GlobalAckListener
-from zmqruntime.config import ZMQConfig
-from zmqruntime.viewer_protocol import (
-    ViewerBatchItemWireField,
-    ViewerBatchMessagePayload,
-    ViewerComponentMetadataPayload,
-    ViewerDisplayConfigWireField,
-    ViewerTransportEndpoint,
-    ViewerWirePayload,
-    ViewerWireMapping,
-    ViewerWireValue,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -86,9 +87,6 @@ def to_wire_mapping(self) -> dict[str, ViewerWireValue]:
         )
 
 
-EMPTY_DISPLAY_PAYLOAD_EXTRA = ViewerDisplayPayloadExtra()
-
-
 @dataclass(frozen=True)
 class StreamingComponentDomainValue:
     """Viewer component value normalized for a batch-level domain."""
@@ -692,7 +690,9 @@ def display_payload_extra(
         self,
         stream_request: ViewerStreamRequest,
     ) -> ViewerDisplayPayloadExtra:
-        return EMPTY_DISPLAY_PAYLOAD_EXTRA
+        return ViewerDisplayPayloadExtra.from_mapping(
+            stream_request.display_config.display_payload_extra()
+        )
 
     def message_extra(
         self,

src/polystore/streaming/receivers/fiji/fiji_batch_processor.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from polystore.streaming.receivers.core import DebouncedBatchEngine
 
@@ -9,59 +9,52 @@
 class FijiBatchProcessor:
     """
     Batch processor for Fiji viewer with configurable batching strategies.
-
-    Accumulates items and builds hyperstacks based on batch_size configuration:
-    - None: Wait for all items in operation, then build hyperstack once
-    - N: Rebuild hyperstack every N items incrementally
-
+
     Uses debouncing to collect items arriving in rapid succession.
     """
-    
+
     def __init__(
         self,
         fiji_server,
-        batch_size: Optional[int] = None,
         debounce_delay_ms: int = 500,
         max_debounce_wait_ms: int = 2000,
     ):
         """
         Initialize batch processor.
-        
+
         Args:
             fiji_server: Reference to FijiViewerServer for display operations
-            batch_size: Number of items to batch before displaying
-                       None = wait for all (default), N = display every N items
             debounce_delay_ms: Wait time after last item before processing (ms)
             max_debounce_wait_ms: Maximum total wait time before forcing display (ms)
         """
         self.fiji_server = fiji_server
-        self.batch_size = batch_size
         self.debounce_delay_ms = debounce_delay_ms
         self.max_debounce_wait_ms = max_debounce_wait_ms
-        
+
         self._engine = DebouncedBatchEngine(
             process_fn=self._process_batch,
             debounce_delay_ms=debounce_delay_ms,
             max_debounce_wait_ms=max_debounce_wait_ms,
         )
-        
+
         logger.info(
-            f"FijiBatchProcessor: Created with batch_size={batch_size}, "
-            f"debounce={debounce_delay_ms}ms, max_wait={max_debounce_wait_ms}ms"
+            "FijiBatchProcessor: Created with debounce=%sms, max_wait=%sms",
+            debounce_delay_ms,
+            max_debounce_wait_ms,
         )
-    
+
     def add_items(
         self,
         window_key: str,
-        items: List[Dict[str, Any]],
-        display_config: Dict[str, Any],
+        items: list[dict[str, Any]],
+        display_config: dict[str, Any],
         images_dir: str,
-        component_names_metadata: Dict[str, Any],
-        component_value_domain: Dict[str, Any],
+        component_names_metadata: dict[str, Any],
+        component_value_domain: dict[str, Any],
     ):
         """
         Add items to the batch for processing.
-        
+
         Args:
             window_key: Unique identifier for the Fiji window
             items: List of items to add (images)
@@ -88,7 +81,7 @@ def flush(self) -> None:
         """Force immediate processing of the pending batch."""
         self._engine.flush()
 
-    def _process_batch(self, items: List[Dict[str, Any]], context: Dict[str, Any]) -> None:
+    def _process_batch(self, items: list[dict[str, Any]], context: dict[str, Any]) -> None:
         """Process callback used by shared debounced batch engine."""
         display_config = context["display_config"]
         images_dir = context["images_dir"]

src/polystore/streaming/receivers/napari/napari_batch_processor.py

@@ -42,31 +42,15 @@ class NapariBatchProcessor:
     adapts batch payloads into the server display operation.
     """
 
-    def __init__(
-        self,
-        napari_server,
-        batch_size: int | None = None,
-        debounce_delay_ms: int = 1000,
-        max_debounce_wait_ms: int = 5000,
-    ):
+    def __init__(self, napari_server):
         """
         Initialize batch processor.
 
         Args:
             napari_server: Reference to NapariViewerServer for display operations
-            batch_size: Reserved for compatibility with viewer configuration
-            debounce_delay_ms: Qt-thread debounce delay owned by the caller
-            max_debounce_wait_ms: Reserved for compatibility with viewer configuration
         """
         self.napari_server = napari_server
-        self.batch_size = batch_size
-        self.debounce_delay_ms = debounce_delay_ms
-        self.max_debounce_wait_ms = max_debounce_wait_ms
-
-        logger.info(
-            f"NapariBatchProcessor: Created with batch_size={batch_size}, "
-            f"debounce={debounce_delay_ms}ms, max_wait={max_debounce_wait_ms}ms"
-        )
+        logger.info("NapariBatchProcessor: Created")
 
     def add_items(
         self,

src/polystore/streaming/viewer_transport.py

@@ -43,6 +43,10 @@ class ViewerDisplayConfigABC(ABC):
     def component_modes(self) -> Mapping[DisplayComponentToken, DisplayModeToken]:
         """Return mode assignments by display component."""
 
+    @abstractmethod
+    def display_payload_extra(self) -> ViewerWireMapping:
+        """Project backend-specific display fields onto the viewer wire payload."""
+
 
 class ViewerFilenameParserABC(ABC):
     """Filename parser surface needed by viewer streaming metadata."""