diff --git a/pyproject.toml b/pyproject.toml index bf6dac4..2261a1e 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/polystore/__init__.py b/src/polystore/__init__.py index 9a61863..61f816c 100644 --- a/src/polystore/__init__.py +++ b/src/polystore/__init__.py @@ -2,7 +2,7 @@ Polystore package exports. """ -__version__ = "0.1.24" +__version__ = "0.1.25" from .atomic import ( FileLockError, diff --git a/src/polystore/fiji_stream.py b/src/polystore/fiji_stream.py index 24a56b9..11750af 100644 --- a/src/polystore/fiji_stream.py +++ b/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, diff --git a/src/polystore/napari_stream.py b/src/polystore/napari_stream.py index 87ce4f2..a28c6e7 100644 --- a/src/polystore/napari_stream.py +++ b/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,26 +40,6 @@ 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 @@ -68,14 +47,6 @@ class NapariStreamingBackend(StreamingBackend): 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, diff --git a/src/polystore/streaming/_streaming_backend.py b/src/polystore/streaming/_streaming_backend.py index 7eb066b..0419cbf 100644 --- a/src/polystore/streaming/_streaming_backend.py +++ b/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, diff --git a/src/polystore/streaming/receivers/fiji/fiji_batch_processor.py b/src/polystore/streaming/receivers/fiji/fiji_batch_processor.py index b1a95f8..88431da 100644 --- a/src/polystore/streaming/receivers/fiji/fiji_batch_processor.py +++ b/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"] diff --git a/src/polystore/streaming/receivers/napari/napari_batch_processor.py b/src/polystore/streaming/receivers/napari/napari_batch_processor.py index c143056..6dafafa 100644 --- a/src/polystore/streaming/receivers/napari/napari_batch_processor.py +++ b/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, diff --git a/src/polystore/streaming/viewer_transport.py b/src/polystore/streaming/viewer_transport.py index b7f6944..eff7dbc 100644 --- a/src/polystore/streaming/viewer_transport.py +++ b/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.""" diff --git a/tests/test_streaming_metadata.py b/tests/test_streaming_metadata.py index 339e10d..fc467a4 100644 --- a/tests/test_streaming_metadata.py +++ b/tests/test_streaming_metadata.py @@ -1,30 +1,33 @@ from types import SimpleNamespace import pytest - -from polystore.streaming._streaming_backend import StreamingBackend -from polystore.streaming._streaming_backend import StreamingBatchItemPreparationAuthority -from polystore.streaming._streaming_backend import StreamingBatchMessageBuilder -from polystore.streaming._streaming_backend import StreamingBatchMessageRequest -from polystore.streaming._streaming_backend import StreamingComponentNamesRequest -from polystore.streaming._streaming_backend import StreamingItemPath -from polystore.streaming._streaming_backend import StreamingItemPreparationRequest -from polystore.streaming_constants import StreamingDataType -from polystore.streaming.identity import StreamProducerIdentity -from polystore.streaming.viewer_transport import BatchViewerStreamSourceMetadata -from polystore.streaming.viewer_transport import IndexedViewerStreamSourceMetadata -from polystore.streaming.viewer_transport import PathMappedViewerStreamSourceMetadata -from polystore.streaming.viewer_transport import ViewerDisplayConfigABC -from polystore.streaming.viewer_transport import ViewerMicroscopeHandlerABC -from polystore.streaming.viewer_transport import ViewerStreamProducer -from polystore.streaming.viewer_transport import ViewerStreamItemPayload -from polystore.streaming.viewer_transport import ViewerStreamRequest -from polystore.streaming.viewer_transport import ViewerStreamSource -from polystore.streaming.viewer_transport import ViewerStreamSourceIdentity -from polystore.streaming.viewer_transport import ViewerStreamSourceMetadata from zmqruntime.config import TransportMode -from zmqruntime.viewer_protocol import ViewerAckPolicy -from zmqruntime.viewer_protocol import ViewerTransportEndpoint +from zmqruntime.viewer_protocol import ViewerAckPolicy, ViewerTransportEndpoint + +from polystore.streaming._streaming_backend import ( + StreamingBackend, + StreamingBatchItemPreparationAuthority, + StreamingBatchMessageBuilder, + StreamingBatchMessageRequest, + StreamingComponentNamesRequest, + StreamingItemPath, + StreamingItemPreparationRequest, +) +from polystore.streaming.identity import StreamProducerIdentity +from polystore.streaming.viewer_transport import ( + BatchViewerStreamSourceMetadata, + IndexedViewerStreamSourceMetadata, + PathMappedViewerStreamSourceMetadata, + ViewerDisplayConfigABC, + ViewerMicroscopeHandlerABC, + ViewerStreamItemPayload, + ViewerStreamProducer, + ViewerStreamRequest, + ViewerStreamSource, + ViewerStreamSourceIdentity, + ViewerStreamSourceMetadata, +) +from polystore.streaming_constants import StreamingDataType class MetadataProbeStreamingBackend(StreamingBackend): @@ -51,6 +54,27 @@ def component_modes(self): "channel": "stack", } + def display_payload_extra(self): + return {} + + +def test_streaming_backend_projects_display_payload_through_config_owner() -> None: + stream_request = SimpleNamespace( + display_config=SimpleNamespace( + display_payload_extra=lambda: { + "colormap": "gray", + "variable_size_handling": "pad", + } + ) + ) + + projected = StreamingBackend.display_payload_extra(object(), stream_request) + + assert projected.to_wire_mapping() == { + "colormap": "gray", + "variable_size_handling": "pad", + } + PRODUCER_IDENTITY = StreamProducerIdentity( origin="pipeline", diff --git a/tests/test_viewer_transport.py b/tests/test_viewer_transport.py index 4ea1e79..bc73388 100644 --- a/tests/test_viewer_transport.py +++ b/tests/test_viewer_transport.py @@ -1,23 +1,25 @@ import pytest - -from polystore.streaming.identity import StreamProducerIdentity -from polystore.streaming.viewer_transport import BatchViewerStreamSourceMetadata -from polystore.streaming.viewer_transport import ExplicitViewerTransportConfig -from polystore.streaming.viewer_transport import IndexedViewerStreamSourceMetadata -from polystore.streaming.viewer_transport import ViewerStreamBackendKwargs -from polystore.streaming.viewer_transport import ViewerStreamKwarg -from polystore.streaming.viewer_transport import ViewerDisplayConfigABC -from polystore.streaming.viewer_transport import ViewerFilenameParserABC -from polystore.streaming.viewer_transport import ViewerMetadataHandlerABC -from polystore.streaming.viewer_transport import ViewerMicroscopeHandlerABC -from polystore.streaming.viewer_transport import ViewerStreamRequest -from polystore.streaming.viewer_transport import ViewerStreamProducer -from polystore.streaming.viewer_transport import ViewerStreamSource -from polystore.streaming.viewer_transport import ViewerStreamSourceIdentity -from polystore.streaming.viewer_transport import ViewerStreamSourceMetadata from zmqruntime.config import TransportMode, ZMQConfig from zmqruntime.viewer_protocol import ViewerTransportEndpoint +from polystore.streaming.identity import StreamProducerIdentity +from polystore.streaming.viewer_transport import ( + BatchViewerStreamSourceMetadata, + ExplicitViewerTransportConfig, + IndexedViewerStreamSourceMetadata, + ViewerDisplayConfigABC, + ViewerFilenameParserABC, + ViewerMetadataHandlerABC, + ViewerMicroscopeHandlerABC, + ViewerStreamBackendKwargs, + ViewerStreamKwarg, + ViewerStreamProducer, + ViewerStreamRequest, + ViewerStreamSource, + ViewerStreamSourceIdentity, + ViewerStreamSourceMetadata, +) + class DisplayConfigFixture(ViewerDisplayConfigABC): COMPONENT_ORDER = ("well", "site", "channel") @@ -29,6 +31,9 @@ def component_modes(self): "channel": "channel", } + def display_payload_extra(self): + return {} + class FilenameParserFixture(ViewerFilenameParserABC): def parse_filename(self, filename):