Refactor stacking for reuse and overlappability

Currently, the stacking feature adds a lot of code to various parts of ShmemTrainMatcher. It would be nice if this could be cleaner, for:

• more maintainable ShmemTrainMatcher
• reusability: the DetectorAssembler also does stacking except in its own less good way
• reusability: we will at some point probably split ShmemTrainMatcher and some of the resulting matching-and-processing devices will want to stack
• improvability: the current implementation is pretty naive and though the handle_source is intended to be run in a thread pool, only one train can be processed at a time; this requires rethinking of buffer handling to change
  • bonus concern: the stacking buffers, if we start overlapping processing and decoupling this from sending, would not satisfy the safeNDArray flag in the current version

Plan for this MR:

• Move stacking functionality to "friend" class
• Get rid of global stacking buffers; idea: create, per train, a stacking context with fresh buffers
• Add some testing of stacking (not just utils, but the friend class working on tables)
• (Out of scope) Test that it works in a overlapping trains setting
• (Out of scope) Start overlapping train processing
• Refactor DetectorAssembler to use StackingFriend to stack before assembly
• Change the configuration that the manager passes to DetectorAssembler correspondingly
• Nevermind, DetectorAssembler is not a great fit actually; maybe use xarray and exploit extra-geom's partial data assembly functionality (Out of scope; see !74 (closed))
• Test everything with actual Karabo (pending node allocation)

src/calng/ShmemTrainMatcher.py

@@ -207,23 +207,14 @@ class ShmemTrainMatcher(TrainMatcher.TrainMatcher):
             ).astype(np.bool, copy=False),
 
         # prepare stacking
-        (
-            stacking_data_shapes,
-            ignore_stacking,
-            fill_missed_data,
-        ) = self._stacking_friend.check_stacking_data(sources, frame_selection_mask)
-        self._stacking_friend.maybe_update_stacking_buffers(
-            stacking_data_shapes, fill_missed_data, new_sources_map
+        self._stacking_friend.prepare_stacking_for_train(
+            sources, frame_selection_mask, new_sources_map
         )
 
-        for (new_source, key), msg in ignore_stacking.items():
-            self.log.WARN(f"Failed to stack {new_source}.{key}: {msg}")
-
         if self._thread_pool is None:
             for source, (data, timestamp) in sources.items():
                 self._handle_source(
                     source, data, timestamp, new_sources_map, frame_selection_mask,
-                    ignore_stacking
                 )
         else:
             concurrent.futures.wait(
@@ -235,7 +226,6 @@ class ShmemTrainMatcher(TrainMatcher.TrainMatcher):
                         timestamp,
                         new_sources_map,
                         frame_selection_mask,
-                        ignore_stacking,
                     )
                     for source, (data, timestamp) in sources.items()
                 ]