Performance improvements

python/mcsdaqtools/DaqTimingPattern.py

@@ -14,7 +14,7 @@
 
 
 from collections.abc import Sequence
-from enum import Enum, IntEnum, IntFlag, unique
+from enum import IntEnum, IntFlag, unique
 import numpy as np
 from ctypes import LittleEndianStructure, c_float, c_uint8, c_uint16, POINTER, cast
 
@@ -138,8 +138,6 @@ class TimingPatternPacked(LittleEndianStructure):
     ]
 
 
-TimingPatternPacked.seed_user_laser_triggers
-
 timing_pattern_type = {linac: np.dtype([('bunch_charge_setting', BunchChargeSetting),
                                         ('injector_laser_triggers', InjectorLaserTriggers[linac]),
                                         ('seed_user_laser_triggers', SeedUserLaserTriggers[linac]),
@@ -151,24 +149,41 @@ def unpack_timing_pattern(value, linac=LINACS[0]):
     if not isinstance(linac, str) or linac.lower() not in LINACS:
         raise TypeError(f"`mode` must be either of {LINACS}")
 
-    if isinstance(value, (Sequence, np.ndarray)):
+    if isinstance(value, Sequence):
+        value = np.asarray(value)
+
+    if isinstance(value, np.ndarray):
         values = value
-        patterns = [cast((c_float * 1)(values[i]), POINTER(TimingPatternPacked)).contents for i in range(len(values))]
-        return np.rec.array([
-            (
-                BunchChargeSetting(pattern.bunch_charge_setting),
-                InjectorLaserTriggers[linac](pattern.injector_laser_triggers),
-                SeedUserLaserTriggers[linac](pattern.seed_user_laser_triggers),
-                Destination[linac](pattern.destination),
-                SpecialFlags[linac](pattern.special_flags)
-            ) for pattern in patterns], dtype=timing_pattern_type[linac])
+        patterns = cast(values.ctypes.data_as(POINTER(c_float * values.size)),
+                        POINTER(TimingPatternPacked * values.size)).contents
+        patterns_unpacked = np.empty(values.shape + (5,), dtype=np.uint16)
+
+        for i, pattern in enumerate(patterns):
+            patterns_unpacked.flat[i * 5:(i + 1) * 5] = (
+                pattern.bunch_charge_setting,
+                pattern.injector_laser_triggers,
+                pattern.seed_user_laser_triggers,
+                pattern.destination,
+                pattern.special_flags)
+
+        return patterns_unpacked
     else:
         pattern = cast(value, POINTER(TimingPatternPacked)).contents
-        return np.rec.array(
-            (
-                BunchChargeSetting(pattern.bunch_charge_setting),
-                InjectorLaserTriggers[linac](pattern.injector_laser_triggers),
-                SeedUserLaserTriggers[linac](pattern.seed_user_laser_triggers),
-                Destination[linac](pattern.destination),
-                SpecialFlags[linac](pattern.special_flags)
-            ), dtype=timing_pattern_type[linac])
+        return np.asarray((
+            pattern.bunch_charge_setting,
+            pattern.injector_laser_triggers,
+            pattern.seed_user_laser_triggers,
+            pattern.destination,
+            pattern.special_flags), dtype=np.uint16)
+
+
+def decode_timing_pattern(value, linac=LINACS[0]):
+    raise NotImplementedError('TODO')
+    # TODO
+    # patterns_unpacked.flat[i]['bunch_charge_setting'] = BunchChargeSetting(pattern.bunch_charge_setting)
+    # patterns_unpacked.flat[i]['injector_laser_triggers'] = InjectorLaserTriggers[linac](
+    #     pattern.injector_laser_triggers)
+    # patterns_unpacked.flat[i]['seed_user_laser_triggers'] = SeedUserLaserTriggers[linac](
+    #     pattern.seed_user_laser_triggers)
+    # patterns_unpacked.flat[i]['destination'] = Destination[linac](pattern.destination)
+    # patterns_unpacked.flat[i]['special_flags'] = SpecialFlags[linac](pattern.special_flags)

python/test/test_DaqTimingPattern.py

+import unittest
+import numpy as np
+
+from mcsdaqtools.DaqTimingPattern import unpack_timing_pattern
+
+
+class TestDaqTimingPattern(unittest.TestCase):
+    def test_unpack(self):
+        bunch_pattern_packed = np.load('test.npy')
+        bunch_pattern = unpack_timing_pattern(bunch_pattern_packed)
+
+        return
+
+
+if __name__ == '__main__':
+    unittest.main()