From 67776928b42cd82bd012177c1f2fea94265fdf6c Mon Sep 17 00:00:00 2001
From: Benedict Diederich <bene.d@gmx.de>
Date: Wed, 18 Dec 2024 05:12:10 -0800
Subject: [PATCH] improving socket-based speed on raspi

---
 imswitch/__init__.py                          |  1 +
 imswitch/imcommon/framework/noqt.py           | 36 +++++++++++++++++--
 .../controllers/FlowStopController.py         |  2 +-
 3 files changed, 36 insertions(+), 3 deletions(-)

diff --git a/imswitch/__init__.py b/imswitch/__init__.py
index 9f5eb0ad..2a0e90bd 100644
--- a/imswitch/__init__.py
+++ b/imswitch/__init__.py
@@ -10,6 +10,7 @@
 DEFAULT_SETUP_FILE = None
 DEFAULT_CONFIG_PATH = None
 DEFAULT_DATA_PATH = None
+SOCKET_STREAM = True           # Stream Images via socket ?
 
 # Copyright (C) 2020-2024 ImSwitch developers
 # This file is part of ImSwitch.
diff --git a/imswitch/imcommon/framework/noqt.py b/imswitch/imcommon/framework/noqt.py
index 0009dc57..34ee7fc8 100644
--- a/imswitch/imcommon/framework/noqt.py
+++ b/imswitch/imcommon/framework/noqt.py
@@ -13,6 +13,7 @@
 from imswitch import __ssl__
 import cv2 
 import base64
+from imswitch import SOCKET_STREAM
 
 if TYPE_CHECKING:
     from typing import Tuple, Callable, Union
@@ -39,6 +40,9 @@ def __init__(self) -> None:
         ...
 
 class SignalInstance(psygnal.SignalInstance):
+    last_emit_time = 0
+    emit_interval = 0.05  # Emit at most every 100ms
+
     def emit(
         self, *args: Any, check_nargs: bool = False, check_types: bool = False
     ) -> None:
@@ -50,7 +54,8 @@ def emit(
 
         # Skip large data signals
         if self.name in ["sigUpdateImage", "sigImageUpdated"]:
-            self._handle_image_signal(args)
+            if SOCKET_STREAM:
+                self._handle_image_signal(args)
             return
 
         try:
@@ -60,14 +65,28 @@ def emit(
             return
 
         self._safe_broadcast_message(message)
+        del message
         
     def _handle_image_signal(self, args):
         """Compress and broadcast image signals."""
         try:
             for arg in args:
                 if isinstance(arg, np.ndarray):
+                    output_frame = np.ascontiguousarray(arg)  # Avoid memory fragmentation
+                    if output_frame.shape[0] > 640 or output_frame.shape[1] > 480:
+                        everyNthsPixel = np.min([output_frame.shape[0]//480, output_frame.shape[1]//640])
+                    else:
+                        everyNthsPixel = 1
+                                
+                    try:
+                        output_frame = output_frame[::everyNthsPixel, ::everyNthsPixel]
+                    except: 
+                        output_frame = np.zeros((640,460))
+                    # adjust the parameters of the jpeg compression
+                    quality = 80  # Set the desired quality level (0-100)
+                    encode_params = [cv2.IMWRITE_JPEG_QUALITY, quality]
                     # Compress image using JPEG format
-                    _, compressed = cv2.imencode('.jpg', arg, [cv2.IMWRITE_JPEG_QUALITY, 80])
+                    flag, compressed = cv2.imencode(".jpg", output_frame, encode_params)
                     encoded_image = base64.b64encode(compressed).decode('utf-8')
 
                     # Create a minimal message
@@ -77,6 +96,7 @@ def _handle_image_signal(self, args):
                         "format": "jpeg"
                     }
                     self._safe_broadcast_message(message)
+                    del message
         except Exception as e:
             print(f"Error processing image signal: {e}")
                     
@@ -115,6 +135,18 @@ def _safe_broadcast_message(self, message: dict) -> None:
                     sio.start_background_task(sio.emit, "signal", json.dumps(message))
             except Exception as e:
             '''
+            
+        def _safe_broadcast_message(self, message: dict) -> None:
+            """Throttle the emit to avoid task buildup."""
+            now = time.time()
+            if now - self.last_emit_time < self.emit_interval:
+                return  # Skip if emit interval hasn't passed
+            self.last_emit_time = now
+
+            try:
+                sio.start_background_task(sio.emit, "signal", json.dumps(message))
+            except Exception as e:
+                print(f"Error broadcasting message via Socket.IO: {e}")            
             try:
                 print(f"Error broadcasting message via Socket.IO: {e}")
                 asyncio.run_coroutine_threadsafe(sio.emit("signal", json.dumps(message)), asyncio.new_event_loop())
diff --git a/imswitch/imcontrol/controller/controllers/FlowStopController.py b/imswitch/imcontrol/controller/controllers/FlowStopController.py
index f72aaa86..f49711dc 100644
--- a/imswitch/imcontrol/controller/controllers/FlowStopController.py
+++ b/imswitch/imcontrol/controller/controllers/FlowStopController.py
@@ -292,7 +292,7 @@ def flowExperimentThread(self, timeStamp: str, experimentName: str,
             if self.imagesTaken > numImages: break
             if self.is_measure:
                 stepsToMove = volumePerImage
-                self.positioner.move(value=stepsToMove, speed=pumpSpeed, axis=self.pumpAxis, is_absolute=False, is_blocking=True)
+                self.positioner.move(value=stepsToMove, speed=pumpSpeed, axis=self.pumpAxis, is_absolute=False, is_blocking=True, timeout=1)
                 time.sleep(timeToStabilize)
                 metaData = {
                     'timeStamp': timeStamp,