Dev42 - parser batching now settable via UI

src/splash_timepix/app.py

@@ -9,6 +9,7 @@
 import math
 import os
 import queue
+import sys
 import threading
 import time
 import uuid
@@ -34,6 +35,24 @@
 logger = logging.getLogger(__name__)
 
 
+def _clear_screen_safely() -> None:
+    """Clear console only when it is a real terminal.
+
+    Under QProcess or other captures, ``clear`` may run with ``TERM`` unset and
+    spam ``tput: TERM variable not set`` on stderr.
+    """
+    try:
+        out = sys.stdout
+        if out is None or not out.isatty():
+            return
+    except (AttributeError, ValueError):
+        return
+    if os.name == "nt":
+        os.system("cls")
+    elif os.environ.get("TERM"):
+        os.system("clear")
+
+
 @app.command()
 def main(
     host: str = "localhost",
@@ -76,7 +95,7 @@ def main(
         collapse_y: Send x,y,t (False) or x,t data (True)
         heartbeat_port: Port for ZMQ heartbeat messages (default: 5658)
     """
-    os.system("cls" if os.name == "nt" else "clear")
+    _clear_screen_safely()
     print("Starting TimPix3 Streaming Application")
     print("=" * 50)
     if exit_on_disconnect:
@@ -138,6 +157,21 @@ def main(
     # Create message queue for start/stop control messages (only used with ZMQ worker)
     message_queue = queue.Queue(maxsize=10) if not plot else None
 
+    def _queue_stats_for_heartbeat():
+        """Snapshot for heartbeat (called from heartbeat thread; keep fast)."""
+        stats = {
+            "q_ingest_sz": server.get_queue_size(),
+            "q_ingest_max": server.buffer_size,
+            "q_xyt_sz": xyt_queue.qsize(),
+            "q_xyt_max": xyt_queue.maxsize,
+        }
+        if message_queue is not None:
+            stats["q_ctrl_sz"] = message_queue.qsize()
+            stats["q_ctrl_max"] = message_queue.maxsize
+        return stats
+
+    heartbeat.set_queue_stats_provider(_queue_stats_for_heartbeat)
+
     # Generate unique scan name
     # Generate initial scan_name (will be regenerated for each new client connection)
     # UTC, ISO 8601 format (YYYYMMDDTHHMMSSZ) for unambiguous, sortable identifiers
@@ -660,8 +694,8 @@ def handle_tdc(tdc_ts):
                 # Update number of total data points
                 last_total_data_points = event_count
 
-                # Clear the terminal and print the stats
-                os.system("cls" if os.name == "nt" else "clear")
+                # Clear the terminal and print the stats (skip when not a TTY)
+                _clear_screen_safely()
                 print(info)
                 print()
                 print("Overall Stats")

src/splash_timepix/heartbeat.py

@@ -8,7 +8,7 @@
 import threading
 import time
 from enum import Enum
-from typing import Optional
+from typing import Any, Callable, Dict, Optional
 
 import msgpack
 import zmq
@@ -36,6 +36,10 @@ class HeartbeatPublisher:
             'pid': int,             # Process ID
             'data_port': int,       # Port for data ZMQ PUB socket
             'tcp_port': int,        # Port for TCP socket (live-cli connection)
+            # Optional pipeline queue depths (when set via set_queue_stats_provider):
+            'q_ingest_sz': int, 'q_ingest_max': int,   # TCP raw-batch queue
+            'q_xyt_sz': int, 'q_xyt_max': int,         # 3D flush queue → ZMQ worker
+            'q_ctrl_sz': int, 'q_ctrl_max': int,       # ZMQ start/stop control queue
         }
 
     Usage:
@@ -79,6 +83,14 @@ def __init__(
         # ZMQ context and socket created in thread
         self._context: Optional[zmq.Context] = None
         self._socket: Optional[zmq.Socket] = None
+        self._queue_stats_provider: Optional[Callable[[], Dict[str, Any]]] = None
+
+    def set_queue_stats_provider(self, provider: Optional[Callable[[], Dict[str, Any]]]) -> None:
+        """Provide a callable that returns extra heartbeat keys (e.g. queue depths).
+
+        Called from the heartbeat thread once per publish; must be thread-safe and fast.
+        """
+        self._queue_stats_provider = provider
 
     def set_state(self, state: ServerState) -> None:
         """Update the current server state (thread-safe)."""
@@ -157,6 +169,13 @@ def _send_heartbeat(self) -> None:
             "data_port": self.data_port,
             "tcp_port": self.tcp_port,
         }
+        if self._queue_stats_provider is not None:
+            try:
+                extra = self._queue_stats_provider()
+                if extra:
+                    message.update(extra)
+            except Exception:
+                logger.debug("queue_stats_provider failed", exc_info=True)
 
         try:
             self._socket.send(msgpack.packb(message), zmq.DONTWAIT)

src/splash_timepix/ui/main.py

@@ -179,13 +179,25 @@ def _on_start_requested(self, mode: str, params: dict):
         tdc_freq = params["tdc_frequency"]
         tdc_channel = params["tdc_channel"]
         tdc_edge = params["tdc_edge"]
+        callback_batch_size = params.get("callback_batch_size", 10_000)
         duration = params["duration"]
 
-        logger.info(f"Starting {mode}: TDC={tdc_freq}Hz, ch={tdc_channel}, edge={tdc_edge}, duration={duration}s")
-        self._engineering_tab.append_system_log(f"Starting {mode}: TDC={tdc_freq}Hz, duration={duration}s")
+        logger.info(
+            f"Starting {mode}: TDC={tdc_freq}Hz, ch={tdc_channel}, edge={tdc_edge}, "
+            f"callback_batch_size={callback_batch_size}, duration={duration}s"
+        )
+        self._engineering_tab.append_system_log(
+            f"Starting {mode}: TDC={tdc_freq}Hz, parse_batch={callback_batch_size}, duration={duration}s"
+        )
 
         # Start streaming server (needed for all modes)
-        if not self._process_manager.start_streaming_server(tdc_freq, tdc_channel, tdc_edge, exit_on_disconnect=True):
+        if not self._process_manager.start_streaming_server(
+            tdc_freq,
+            tdc_channel,
+            tdc_edge,
+            callback_batch_size=callback_batch_size,
+            exit_on_disconnect=True,
+        ):
             QMessageBox.warning(self, "Error", "Failed to start streaming server")
             return