Improved usability

Author: Karol-G

src/napari_stream/__init__.py

@@ -3,6 +3,6 @@
 except ImportError:
     __version__ = "unknown"
 
-from .sender import send
+from .sender import StreamSender, send
 
 __all__ = []

src/napari_stream/_receiver_widget.py

@@ -40,9 +40,7 @@ def __init__(self, napari_viewer: Viewer):
         self.autocontrast = QCheckBox("Auto-contrast on new images")
         self.autocontrast.setChecked(True)
 
-        self.btn_start = QPushButton("Start")
-        self.btn_stop = QPushButton("Stop")
-        self.btn_stop.setEnabled(False)
+        self.btn_run = QPushButton("Start")
         self.btn_copy = QPushButton("Copy Endpoint")
 
         top = QVBoxLayout(self)
@@ -55,16 +53,22 @@ def __init__(self, napari_viewer: Viewer):
         top.addWidget(self.autocontrast)
         top.addWidget(self.status_label)
         row2 = QHBoxLayout()
-        row2.addWidget(self.btn_start)
-        row2.addWidget(self.btn_stop)
+        row2.addWidget(self.btn_run)
         top.addLayout(row2)
 
-        self.btn_start.clicked.connect(self._on_start)
-        self.btn_stop.clicked.connect(self._on_stop)
+        self.btn_run.clicked.connect(self._on_toggle_clicked)
         self.btn_copy.clicked.connect(self._copy_endpoint)
         self.public_access.stateChanged.connect(self._on_public_toggled)
 
+    def _on_toggle_clicked(self):
+        if self._is_running():
+            self._on_stop()
+        else:
+            self._on_start()
+
     def _on_start(self):
+        if self._is_running():
+            return
         endpoint = self.endpoint_edit.text().strip()
         bind_endpoint = self._resolve_endpoint_for_worker(endpoint)
         self._thread = QThread()
@@ -76,8 +80,8 @@ def _on_start(self):
         self._worker.status.connect(self.status_label.setText)
         self._worker.error.connect(self._on_error)
 
-        self.btn_start.setEnabled(False)
-        self.btn_stop.setEnabled(True)
+        self.btn_run.setText("Stop")
+        self.btn_run.setEnabled(True)
         self._thread.start()
 
     def _on_stop(self):
@@ -86,8 +90,10 @@ def _on_stop(self):
         if self._thread is not None:
             self._thread.quit()
             self._thread.wait()
-        self.btn_start.setEnabled(True)
-        self.btn_stop.setEnabled(False)
+        self._thread = None
+        self._worker = None
+        self.btn_run.setText("Start")
+        self.btn_run.setEnabled(True)
 
     def _on_public_toggled(self, checked: int):
         is_public = bool(checked)
@@ -96,6 +102,8 @@ def _on_public_toggled(self, checked: int):
         if current == self._last_auto_endpoint:
             self.endpoint_edit.setText(suggested)
         self._last_auto_endpoint = suggested
+        if self._is_running():
+            self._restart_listener()
 
     def _copy_endpoint(self):
         endpoint = self.endpoint_edit.text().strip()
@@ -112,6 +120,13 @@ def _resolve_endpoint_for_worker(self, endpoint: str) -> str:
         self._last_auto_endpoint = fallback
         return bind_endpoint_for_public(fallback)
 
+    def _is_running(self) -> bool:
+        return self._thread is not None and self._thread.isRunning()
+
+    def _restart_listener(self):
+        self._on_stop()
+        self._on_start()
+
     def _on_received(self, arr: np.ndarray, meta: dict):
         name = meta.get("name", "array")
         is_labels = bool(meta.get("is_labels", False))

src/napari_stream/sender.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 import json
+import os
+import time
 from typing import Optional, Sequence, Any, Iterable
 from collections.abc import Mapping
 
@@ -28,12 +30,16 @@ def __init__(
         linger_ms: int = 2000,
         ensure_delivery: bool = True,
     ):
-        self.endpoint = endpoint or default_endpoint()
+        if endpoint is None:
+            endpoint = os.environ.get("NAPARI_STREAM_ENDPOINT") or default_endpoint()
+        self.endpoint = endpoint
         self.ensure_delivery = ensure_delivery
         self._ctx = zmq.Context.instance()
         self._sock = self._ctx.socket(zmq.PUSH)
         self._sock.setsockopt(zmq.SNDHWM, high_water_mark)
         self._sock.setsockopt(zmq.LINGER, linger_ms)
+        # Fail fast if there is no receiver instead of silently queueing.
+        self._sock.setsockopt(zmq.IMMEDIATE, 1)
         self._sock.connect(self.endpoint)
 
     # ------------------------------ public API ------------------------------
@@ -152,7 +158,24 @@ def _send_numpy(
 
         header = json.dumps(meta).encode("utf-8")
         buf = memoryview(arr)  # zero-copy
-        tracker = self._sock.send_multipart([header, buf], copy=False, track=self.ensure_delivery)
+        tracker = None
+        last_err = None
+        for _ in range(3):
+            try:
+                tracker = self._sock.send_multipart(
+                    [header, buf],
+                    flags=zmq.NOBLOCK,
+                    copy=False,
+                    track=self.ensure_delivery,
+                )
+                last_err = None
+                break
+            except zmq.Again as e:
+                last_err = e
+                # Allow a brief window for sockets to finish connecting
+                time.sleep(0.05)
+        if last_err is not None:
+            raise RuntimeError(f"No receiver available at endpoint {self.endpoint}") from last_err
         if self.ensure_delivery and tracker is not None:
             tracker.wait()
 
@@ -298,5 +321,17 @@ def is_arraylike(self, x):
 
 
 def send(*args, **kwargs) -> None:
-    sender = StreamSender()
+    """Send data to a napari receiver.
+
+    Accepts:
+      - numpy.ndarray
+      - torch.Tensor  (detach().cpu().numpy())
+      - blosc2.NDArray
+      - zarr.Array
+      - Python lists/tuples and dicts (recursively searched for arraylikes;
+        nested Python lists of numbers are also converted to NumPy)
+    """
+    sender_keys = ("endpoint", "high_water_mark", "linger_ms", "ensure_delivery")
+    sender_kwargs = {k: kwargs.pop(k) for k in sender_keys if k in kwargs}
+    sender = StreamSender(**sender_kwargs)
     sender.send(*args, **kwargs)