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/stacking_utils.py

@@ -8,6 +8,7 @@ from karabo.bound import (
     TABLE_ELEMENT,
     Hash,
     Schema,
+    State,
 )
 import numpy as np
 
@@ -87,7 +88,7 @@ class StackingFriend:
     @staticmethod
     def add_schema(self, schema):
         (
-            TABLE_ELEMENT(expected)
+            TABLE_ELEMENT(schema)
             .key("merge")
             .displayedName("Array stacking")
             .allowedStates(State.PASSIVE)