DetectorAssembler: go to ERROR if trying to match without geometry

#48 (closed) this came up again recently, let's do something about it.

• geometry is a property again, so if geometry device is up, we don't even need to wait for signal to get it
• DRY a bit by putting utility to get property + connect to signal in calngUtils
• start warning harder if no geometry

Options:

1. stay in INIT until we get geometry
2. go to ERROR after some timeout if we don't get geometry
3. go to ERROR in match handler if we don't have geometry by the time we need it (current implementation)

As the diff shows, at least for option 3, we need to fight the TrainMatcher state logic a bit; it's going from PASSIVE to CHANGING to ACTIVE outside of the methods we're currently overriding. State transitions for an assembler started while geometry device is down and then geometry device comes up later:

src/calng/DetectorAssembler.py

@@ -4,8 +4,7 @@ from timeit import default_timer
 
 import numpy as np
 import xarray as xr
-from calngUtils import device as device_utils, trackers
-from geometryDevices import utils as geom_utils
+from calngUtils import device as device_utils, geom_utils, trackers
 from karabo.bound import (
     BOOL_ELEMENT,
     DOUBLE_ELEMENT,
@@ -150,6 +149,19 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             .assignmentMandatory()
             .commit(),
 
+            STRING_ELEMENT(expected)
+            .key("geometryState")
+            .displayedName("Geometry state")
+            .description(
+                "The assembler needs a detector geometry to function. This is gotten "
+                "from a geometry device (see the geometryDevice parameter). If this "
+                "state is ERROR, no assembly will take place - check the geometry "
+                "device!"
+            )
+            .readOnly()
+            .initialValue("INIT")
+            .commit(),
+
             STRING_ELEMENT(expected)
             .key("sourceIndexPattern")
             .displayedName("Source index regex")
@@ -192,7 +204,6 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
         self._processing_time_tracker = trackers.ExponentialMovingAverage(
             alpha=0.3
         )
-        self.registerSlot(self.slotReceiveGeometry)
         self._need_to_update_source_index_mapping = True
 
     def initialization(self):
@@ -201,19 +212,24 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
         self._preview_friend = PreviewFriend(self, self._preview_outputs, "preview")
         self._image_data_path = self.get("imageDataPath")
         self._image_mask_path = self.get("imageMaskPath")
-        self._geometry = None
         # set up source to index mapping
         self._merge_source_to_index_from_regex()
         self._set_source_to_index_from_table()
 
         self.KARABO_SLOT(self.requestScene)
 
-        self.signalSlotable.connect(
-            self.get("geometryDevice"),
-            "signalNewGeometry",
-            "",  # slot device ID (default: self)
-            "slotReceiveGeometry",
+        self._geometry = None
+        def _on_geometry(geometry):
+            if self._geometry is None:
+                self.set("geometryState", "ACTIVE")
+            self._geometry = geometry
+            self.log.INFO("Received new geometry")
+        geom_utils.subscribe_to_geometry_bound(
+            self, self.get("geometryDevice"), _on_geometry
         )
+        if self._geometry is None:
+            # initial get failed
+            self.set("geometryState", "ERROR")
 
         self.assembled_output = self.signalSlotable.getOutputChannel("assembledOutput")
         self.start()
@@ -357,6 +373,7 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
         payload["data"] = scenes.detector_assembler_overview(
             device_id=self.getInstanceId(),
             schema=self.getFullSchema(),
+            geometry_device_id=self.get("geometryDevice"),
         )
         self.reply(
             Hash(
@@ -369,21 +386,13 @@ class DetectorAssembler(TrainMatcher.TrainMatcher):
             )
         )
 
-    def slotReceiveGeometry(self, device_id, serialized_geometry):
-        self.log.INFO(f"Received geometry from {device_id}")
-        try:
-            self._geometry = geom_utils.deserialize_geometry(serialized_geometry)
-        except Exception as e:
-            self.log.WARN(f"Failed to deserialize geometry; {e}")
-            return
-        # TODO: test with multiple memory cells (extra geom notion of extra dimensions)
-        # NOTE: could almost test n_modules against _source_to_index JF inconsistent
-
     def on_matched_data(self, train_id, sources):
         ts_start = default_timer()
+
         if self._geometry is None:
-            self.log.WARN("Have not received a geometry yet, will not send anything")
+            # should have already set up warning
             return
+
         my_timestamp = Timestamp(Epochstamp(), Trainstamp(train_id))
         my_device_id = self.getInstanceId()
         bridge_output_choice = BridgeOutputOptions(