From 679f8cc085d9064846a6e9f535e37e4b65cb1cdd Mon Sep 17 00:00:00 2001 From: appel_c Date: Tue, 27 Jan 2026 09:40:31 +0100 Subject: [PATCH] cleanup eiger/jungfraujoch wip --- csaxs_bec/devices/jungfraujoch/eiger.py | 14 +- .../jungfraujoch/jungfrau_joch_client.py | 32 +++-- .../jungfraujoch/jungfraujoch_preview.py | 121 ++++++++++++++---- 3 files changed, 126 insertions(+), 41 deletions(-) diff --git a/csaxs_bec/devices/jungfraujoch/eiger.py b/csaxs_bec/devices/jungfraujoch/eiger.py index 8b2e231..15294fa 100644 --- a/csaxs_bec/devices/jungfraujoch/eiger.py +++ b/csaxs_bec/devices/jungfraujoch/eiger.py @@ -28,7 +28,7 @@ Some extra notes for setting up the detector: Hardware related notes: - If there is an HW issue with the detector, power cycling may help. -- The sls_detector package is available on console on /sls/X12SA/data/gac-x12sa/erik/micromamba +- The sls_detector package is available on console on /sls/x12sa/applications/erik/micromamba - Run: source setup_9m.sh # Be careful, this connects to the detector, so it should not be used during operation - Useful commands: @@ -120,8 +120,9 @@ class Eiger(PSIDeviceBase): super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) self._host = f"{host}:{port}" self.jfj_client = JungfrauJochClient(host=self._host, parent=self) + # NOTE: fetch this information from JungfrauJochClient during on_connected! self.jfj_preview_client = JungfrauJochPreview( - url="tcp://129.129.95.114:5400", cb=self.preview_image.put + url="tcp://129.129.95.114:5400", cb=self._preview_callback ) # IP of sls-jfjoch-001.psi.ch on port 5400 for ZMQ stream self.device_manager = device_manager self.detector_name = detector_name @@ -132,6 +133,15 @@ class Eiger(PSIDeviceBase): if self.device_manager is not None: self.device_manager: DeviceManagerDS + def _preview_callback(self, message: dict) -> None: + if message.get("type", "") == "image": + data = message.get("data", {}).get("default", None) + if data is None: + logger.error(f"Received image message on device {self.name} without data.") + return + logger.info(f"Received preview image on device {self.name}") + self.preview_image.put(data) + @property def detector_distance(self) -> float: """The detector distance in mm.""" diff --git a/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py b/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py index 6bd678f..bddc4eb 100644 --- a/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py +++ b/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py @@ -3,6 +3,7 @@ from __future__ import annotations import enum +import threading import time import traceback from typing import TYPE_CHECKING @@ -152,18 +153,25 @@ class JungfrauJochClient: def stop(self, request_timeout: float = 0.5) -> None: """Stop the acquisition, this only logs errors and is not raising.""" - try: - self.api.cancel_post_with_http_info(_request_timeout=request_timeout) - except requests.exceptions.Timeout: - content = traceback.format_exc() - logger.error( - f"Timeout in JungFrauJochClient for device {self._parent_name} during stop: {content}" - ) - except Exception: - content = traceback.format_exc() - logger.error( - f"Error in JungFrauJochClient for device {self._parent_name} during stop: {content}" - ) + + def _stop_call(self): + try: + self.api.cancel_post_with_http_info() # (_request_timeout=request_timeout) + except requests.exceptions.Timeout: + content = traceback.format_exc() + logger.error( + f"Timeout in JungFrauJochClient for device {self._parent_name} during stop: {content}" + ) + except Exception: + content = traceback.format_exc() + logger.error( + f"Error in JungFrauJochClient for device {self._parent_name} during stop: {content}" + ) + + thread = threading.Thread( + target=_stop_call, daemon=True, args=(self,), name="stop_jungfraujoch_thread" + ) + thread.start() def wait_for_idle( self, timeout: int = 10, request_timeout: float | None = None, raise_on_timeout: bool = True diff --git a/csaxs_bec/devices/jungfraujoch/jungfraujoch_preview.py b/csaxs_bec/devices/jungfraujoch/jungfraujoch_preview.py index 9d31d2b..3aeb6f4 100644 --- a/csaxs_bec/devices/jungfraujoch/jungfraujoch_preview.py +++ b/csaxs_bec/devices/jungfraujoch/jungfraujoch_preview.py @@ -1,26 +1,39 @@ -"""Module for the Eiger preview ZMQ stream.""" +""" +Module for the JungfrauJoch preview ZMQ stream for the Eiger detector at cSAXS. +The Preview client is implemented for the JungfrauJoch ZMQ PUB-SUB interface, and +should be independent of the EIGER detector type. + +The client connects to the ZMQ PUB-SUB preview stream and calls a user provided callback +function with the decompressed messages received from the stream. The callback needs to be +able to deal with the different message types sent by the JungfrauJoch server ("start", +"image", "end") as described in the JungfrauJoch ZEROMQ preview stream documentation. +(https://jungfraujoch.readthedocs.io/en/latest/ZEROMQ_STREAM.html#preview-stream). +""" from __future__ import annotations -import json import threading import time from typing import Callable +import cbor2 import numpy as np import zmq from bec_lib.logger import bec_logger +from dectris.compression import decompress logger = bec_logger.logger -ZMQ_TOPIC_FILTER = b"" - -import cbor2 -import numpy as np -from dectris.compression import decompress +############################### +###### CBOR TAG DECODERS ###### +############################### +# Dectris specific CBOR tags and decoders for Jungfrau data +# Reference: +# https://github.com/dectris/documentation/blob/main/stream_v2/examples/client.py -def decode_multi_dim_array(tag, column_major): +def decode_multi_dim_array(tag: cbor2.CBORTag, column_major: bool = False): + """Decode a multi-dimensional array from a CBOR tag.""" dimensions, contents = tag.value if isinstance(contents, list): array = np.empty((len(contents),), dtype=object) @@ -32,17 +45,23 @@ def decode_multi_dim_array(tag, column_major): return array.reshape(dimensions, order="F" if column_major else "C") -def decode_typed_array(tag, dtype): +def decode_typed_array(tag: cbor2.CBORTag, dtype: str): + """Decode a typed array from a CBOR tag.""" 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): +def decode_dectris_compression(tag: cbor2.CBORTag): + """Decode a Dectris compressed array from a CBOR tag.""" algorithm, elem_size, encoded = tag.value return decompress(encoded, algorithm, elem_size=elem_size) +######################################### +#### Dectris CBOR TAG Extensions ######## +######################################### + tag_decoders = { 40: lambda tag: decode_multi_dim_array(tag, column_major=False), 64: lambda tag: decode_typed_array(tag, dtype="u1"), @@ -73,12 +92,42 @@ tag_decoders = { } -def tag_hook(decoder, tag): +def tag_hook(tag: int): + """Get the decoder for Dectris specific CBOR tags. tag must be in tag_decoders.""" tag_decoder = tag_decoders.get(tag.tag) return tag_decoder(tag) if tag_decoder else tag +###################### +#### ZMQ Settings #### +###################### + +ZMQ_TOPIC_FILTER = b"" # Subscribe to all topics +ZMQ_CONFLATE_SETTING = 1 # Keep only the most recent message +ZMQ_RCVHWM_SETTING = 1 # Set high water mark to 1, this configures the max number of queue messages + + +################################# +#### Jungfrau Preview Client #### +################################# + + class JungfrauJochPreview: + """ + Preview client for the JungfrauJoch ZMQ preview stream. The client is started with + a URL to receive the data from the JungfrauJoch PUB-SUB preview interface, and a + callback function that is called with messages received from the preview stream. + The callback needs to be able to deal with the different message types sent + by the JungfrauJoch server ("start", "image", "end") as described in the + JungfrauJoch ZEROMQ preview stream documentation. Messages are dictionary dumps. + (https://jungfraujoch.readthedocs.io/en/latest/ZEROMQ_STREAM.html#preview-stream). + + Args: + url (str): ZMQ PUB-SUB preview stream URL. + cb (Callable): Callback function called with messages received from the stream. + + """ + USER_ACCESS = ["start", "stop"] def __init__(self, url: str, cb: Callable): @@ -89,16 +138,18 @@ class JungfrauJochPreview: self._on_update_callback = cb def connect(self): - """Connect to the JungfrauJoch PUB-SUB streaming interface - - JungfrauJoch may reject connection for a few seconds when it restarts, - so if it fails, wait a bit and try to connect again. + """ + Connect to the JungfrauJoch PUB-SUB streaming interface. If the connection is refused + it will reattempt a second time after a one second delay. """ # pylint: disable=no-member context = zmq.Context() self._socket = context.socket(zmq.SUB) + self._socket.setsockopt(zmq.CONFLATE, ZMQ_CONFLATE_SETTING) self._socket.setsockopt(zmq.SUBSCRIBE, ZMQ_TOPIC_FILTER) + self._socket.setsockopt(zmq.RCVHWM, ZMQ_RCVHWM_SETTING) + try: self._socket.connect(self.url) except ConnectionRefusedError: @@ -106,17 +157,20 @@ class JungfrauJochPreview: self._socket.connect(self.url) def start(self): + """Start the ZMQ update loop in a background thread.""" self._zmq_thread = threading.Thread( target=self._zmq_update_loop, daemon=True, name="JungfrauJoch_live_preview" ) self._zmq_thread.start() def stop(self): + """Stop the ZMQ update loop and wait for the thread to finish.""" self._shutdown_event.set() if self._zmq_thread: self._zmq_thread.join() def _zmq_update_loop(self): + """Zmq update loop with polling for new data. The loop runs at maximum 10 Hz.""" while not self._shutdown_event.is_set(): if self._socket is None: self.connect() @@ -127,17 +181,22 @@ class JungfrauJochPreview: pass except zmq.error.Again: # Happens when receive queue is empty - time.sleep(0.1) + time.sleep(0.1) # NOTE: Change sleep time to control polling rate def _poll(self): """ - Poll the ZMQ socket for new data. It will throttle the data update and - only subscribe to the topic for a single update. This is not very nice - but it seems like there is currently no option to set the update rate on - the backend. + Poll the ZMQ socket for new data. We are currently subscribing and unsubscribing + for each poll loop to avoid receiving too many messages. The zmq update loop is + running with a maximum of 10 Hz, and we wait an additional 0.2 seconds here to + avoid busy waiting. Therefore, the maximum data rate is around 3 Hz. + + NOTE: + This can be optimized in the future as JungfrauJoch supports controlling the update + rate from the server side. We still safeguard here to avoid that the device-server + becomes overloaded with messages. """ - if self._shutdown_event.wait(0.2): + if self._shutdown_event.wait(0.2): # NOTE: Change to adjust throttle rate in polling. return try: @@ -153,10 +212,18 @@ class JungfrauJochPreview: self._socket.setsockopt(zmq.UNSUBSCRIBE, ZMQ_TOPIC_FILTER) def _parse_data(self, bytes_list: list[bytes]): + """ + Parse the received ZMQ data from the JungfrauJoch preview stream. + We will call the _on_update_callback with the decompressed messages. + + The callback needs to be able to deal with the different message types sent + by the JungfrauJoch server ("start", "image", "end") as described in the + JungfrauJoch ZEROMQ preview stream documentation. Messages are dictionary dumps. + (https://jungfraujoch.readthedocs.io/en/latest/ZEROMQ_STREAM.html#preview-stream). + + Args: + bytes_list (list[bytes]): List of byte messages received from ZMQ recv_multipart. + """ for byte_msg in bytes_list: - messages = cbor2.loads(byte_msg, tag_hook=tag_hook) - # Parse message to data, call callback with data - for message in messages: - message = cbor2.loads(message, tag_hook=tag_hook) - for cb in self._on_update_callback: - cb(message) + msg = cbor2.loads(byte_msg, tag_hook=tag_hook) + self._on_update_callback(msg)