fix(eiger): add dectris-decompression and dependencies.

csaxs_bec/devices/jungfraujoch/eiger.py

@@ -84,7 +84,7 @@ class EigerError(Exception):
 
 class Eiger(PSIDeviceBase):
     """
-    Base integration of the Eiger1.5M and Eiger9M at cSAXS. All relevant
+    Base integration of the Eiger1.5M and Eiger9M at cSAXS.
     """
 
     USER_ACCESS = ["detector_distance", "beam_center"]
@@ -208,12 +208,13 @@ class Eiger(PSIDeviceBase):
 
         # Setup Detector settings, here we may also set the energy already as this might be time consuming
         settings = DetectorSettings(frame_time_us=int(500), timing=DetectorTiming.TRIGGER)
-        self.jfj_client.set_detector_settings(settings, timeout=10)
+        self.jfj_client.set_detector_settings(settings, timeout=5)
         # Set the file writer to the appropriate output for the HDF5 file
         file_writer_settings = FileWriterSettings(overwrite=True, format=FileWriterFormat.NXMXVDS)
         logger.debug(
             f"Setting writer_settings: {yaml.dump(file_writer_settings.to_dict(), indent=4)}"
         )
+        # Setup the file writer settings
         self.jfj_client.api.config_file_writer_put(
             file_writer_settings=file_writer_settings, _request_timeout=10
         )
@@ -230,6 +231,7 @@ class Eiger(PSIDeviceBase):
         """
         start_time = time.time()
         scan_msg = self.scan_info.msg
+
         # Set acquisition parameter
         # TODO add check of mono energy, this can then also be passed to DatasetSettings
         incident_energy = 12.0
@@ -250,8 +252,10 @@ class Eiger(PSIDeviceBase):
             successful=False,
             hinted_h5_entries={"data": "entry/data/data"},
         )
+
         # JFJ adds _master.h5 automatically
         path = os.path.relpath(self._full_path, start="/sls/x12sa/data").removesuffix("_master.h5")
+
         data_settings = DatasetSettings(
             image_time_us=int(frame_time_us * 1e6),  # This is currently ignored
             ntrigger=ntrigger,

csaxs_bec/devices/jungfraujoch/jungfraujoch_preview.py

@@ -15,6 +15,68 @@ logger = bec_logger.logger
 
 ZMQ_TOPIC_FILTER = b""
 
+import cbor2
+import numpy as np
+from dectris.compression import decompress
+
+
+def decode_multi_dim_array(tag, column_major):
+    dimensions, contents = tag.value
+    if isinstance(contents, list):
+        array = np.empty((len(contents),), dtype=object)
+        array[:] = contents
+    elif isinstance(contents, (np.ndarray, np.generic)):
+        array = contents
+    else:
+        raise cbor2.CBORDecodeValueError("expected array or typed array")
+    return array.reshape(dimensions, order="F" if column_major else "C")
+
+
+def decode_typed_array(tag, dtype):
+    if not isinstance(tag.value, bytes):
+        raise cbor2.CBORDecodeValueError("expected byte string in typed array")
+    return np.frombuffer(tag.value, dtype=dtype)
+
+
+def decode_dectris_compression(tag):
+    algorithm, elem_size, encoded = tag.value
+    return decompress(encoded, algorithm, elem_size=elem_size)
+
+
+tag_decoders = {
+    40: lambda tag: decode_multi_dim_array(tag, column_major=False),
+    64: lambda tag: decode_typed_array(tag, dtype="u1"),
+    65: lambda tag: decode_typed_array(tag, dtype=">u2"),
+    66: lambda tag: decode_typed_array(tag, dtype=">u4"),
+    67: lambda tag: decode_typed_array(tag, dtype=">u8"),
+    68: lambda tag: decode_typed_array(tag, dtype="u1"),
+    69: lambda tag: decode_typed_array(tag, dtype="<u2"),
+    70: lambda tag: decode_typed_array(tag, dtype="<u4"),
+    71: lambda tag: decode_typed_array(tag, dtype="<u8"),
+    72: lambda tag: decode_typed_array(tag, dtype="i1"),
+    73: lambda tag: decode_typed_array(tag, dtype=">i2"),
+    74: lambda tag: decode_typed_array(tag, dtype=">i4"),
+    75: lambda tag: decode_typed_array(tag, dtype=">i8"),
+    77: lambda tag: decode_typed_array(tag, dtype="<i2"),
+    78: lambda tag: decode_typed_array(tag, dtype="<i4"),
+    79: lambda tag: decode_typed_array(tag, dtype="<i8"),
+    80: lambda tag: decode_typed_array(tag, dtype=">f2"),
+    81: lambda tag: decode_typed_array(tag, dtype=">f4"),
+    82: lambda tag: decode_typed_array(tag, dtype=">f8"),
+    83: lambda tag: decode_typed_array(tag, dtype=">f16"),
+    84: lambda tag: decode_typed_array(tag, dtype="<f2"),
+    85: lambda tag: decode_typed_array(tag, dtype="<f4"),
+    86: lambda tag: decode_typed_array(tag, dtype="<f8"),
+    87: lambda tag: decode_typed_array(tag, dtype="<f16"),
+    1040: lambda tag: decode_multi_dim_array(tag, column_major=True),
+    56500: lambda tag: decode_dectris_compression(tag),
+}
+
+
+def tag_hook(decoder, tag):
+    tag_decoder = tag_decoders.get(tag.tag)
+    return tag_decoder(tag) if tag_decoder else tag
+
 
 class JungfrauJochPreview:
     USER_ACCESS = ["start", "stop"]
@@ -91,6 +153,7 @@ class JungfrauJochPreview:
             self._socket.setsockopt(zmq.UNSUBSCRIBE, ZMQ_TOPIC_FILTER)
 
     def _parse_data(self, data):
-        # TODO decode and parse the data
-        # self._on_update_callback(data)
-        pass
+        message = cbor2.loads(data, tag_hook=tag_hook)
+        # Parse message to data, call callback with data
+        for cb in self._on_update_callback:
+            cb(message)

pyproject.toml

@@ -25,6 +25,8 @@ dependencies = [
     "bec_widgets",
     "zmq",
     "opencv-python",
+    "dectris-compression", # for JFJ preview stream decompression
+    "cbor2",
 ]
 
 [project.optional-dependencies]