Switch preview throttling method for DetectorAssembler

src/calng/DetectorAssembler.py

@@ -4,7 +4,7 @@ import re
 
 import numpy as np
 from karabo.bound import (
-    FLOAT_ELEMENT,
+    DOUBLE_ELEMENT,
     IMAGEDATA_ELEMENT,
     NDARRAY_ELEMENT,
     NODE_ELEMENT,
@@ -71,7 +71,7 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             .setNewDefaultValue(["overview", "trainMatcherScene"])
             .commit(),
 
-            FLOAT_ELEMENT(expected)
+            DOUBLE_ELEMENT(expected)
             .key("timeOfFlight")
             .displayedName("Time of flight")
             .unit(Unit.SECOND)
@@ -98,9 +98,8 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             .key("preview")
             .description(
                 "The preview output is intended for Karabo GUI previews. It differs "
-                "from the main output in that it is lower rate (controlled by train "
-                "stride), can be downsampled, and is given the ImageData type for use "
-                "within Karabo."
+                "from the main output in that it is rate throttled, can be "
+                "downsampled, and is given the ImageData type for use within Karabo."
             )
             .commit(),
 
@@ -132,13 +131,17 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             .reconfigurable()
             .commit(),
 
-            UINT32_ELEMENT(expected)
-            .key("preview.trainStride")
-            .displayedName("Train stride")
-            .description("Only trains which are a multiple of this are sent to preview")
-            .unit(Unit.COUNT)
+            DOUBLE_ELEMENT(expected)
+            .key("preview.maxRate")
+            .displayedName("Max rate")
+            .description(
+                "Preview output is throttled to (at most) this speed. New trains "
+                "matched 'too soon' get dropped here (instead of sending to be dropped "
+                "by GUI)."
+            )
+            .unit(Unit.HERTZ)
             .assignmentOptional()
-            .defaultValue(10)
+            .defaultValue(2)
             .reconfigurable()
             .commit(),
 
@@ -178,6 +181,7 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
 
         # TODO: match inside device, fill multiple independent buffers
 
+        self._throttler = utils.SkippingThrottler(1 / self.get("preview.maxRate"))
         self._path_to_stack = self.get("pathToStack")
         self._geometry = None
         self._stack_input_buffer = None
@@ -294,7 +298,7 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             self.zmq_output.write(my_source, output_hash, my_timestamp)
             self.zmq_output.update()
 
-        if train_id % self.unsafe_get("preview.trainStride") == 0:
+        if self._throttler.test_and_set():
             downsampling_factor = self.unsafe_get("preview.downsamplingFactor")
             if downsampling_factor > 1:
                 assembled = downsample_2d(

src/calng/utils.py

@@ -369,6 +369,22 @@ class ChainHash:
         raise KeyError()
 
 
+class SkippingThrottler:
+    def __init__(self, min_period):
+        self.min_period = min_period
+        self.latest_ts = float("-inf")
+        self.lock = threading.Lock()
+
+    def test_and_set(self):
+        with self.lock:
+            now = default_timer()
+            if (now - self.latest_ts) >= self.min_period:
+                self.latest_ts = now
+                return True
+            else:
+                return False
+
+
 class BadPixelValues(enum.IntFlag):
     """The European XFEL Bad Pixel Encoding