bec/superxas_bec
0
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(timepix): add ws status updates from backend

Browse Source
This commit is contained in:
appel_c
2026-05-07 12:47:18 +02:00
parent bb14841a3a
commit 962dbf8607
7 changed files with 695 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+3 -1
View File
@@ -12,7 +12,9 @@ classifiers = [
"Programming Language :: Python :: 3",
"Topic :: Scientific/Engineering",
]
dependencies = []
dependencies = [
"websockets",
]
[project.optional-dependencies]
dev = [
superxas_bec/devices/timepix/beamline_test.py
+57
View File
@@ -0,0 +1,57 @@
from ophyd_devices.devices.areadetector.cam import ASItpxCam
from ophyd import ADBase
from ophyd import Component as Cpt
from ophyd_devices import TransitionStatus
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import NetAddresses
class SuperXasTimePix(ADBase):
cam = Cpt(ASItpxCam, 'cam1:')
if __name__ == """__main__""":
import time
from superxas_bec.devices.timepix.timepix import Timepix
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_mock_server import (
TimePixFlyMockServer,
)
timepix_control = SuperXasTimePix(name='timepix_control', prefix='X10DA-ES-TPX1:')
# timepix_control.wait_for_connection(all_signals=True, timeout=10)
# Create a Timepix object
rest_url = 'p4-0017.psi.ch:8452'
ws_url = 'p4-0017:8452/ws'
timepix = Timepix(name="TimePixDetector", rest_url=rest_url, ws_url=ws_url)
# timepix = Timepix(name="TimePixDetector")
timepix.on_connected()
## LOOP for a scan
timepix.stage()
net_addresses:NetAddresses = timepix.timepix_fly_client.get_net_addresses()
print(f"Net addresses: {net_addresses}")
file_path = f"tcp://connect@{net_addresses.address}"
print(f"Using file path: {file_path}")
period = 1
num_images =5
timepix_control.cam.trigger_mode.set(0).wait(timeout=10)
timepix_control.cam.acquire_period.set(period).wait(timeout=10)
timepix_control.cam.acquire_time.set(period-2e-3).wait(timeout=10)
timepix_control.cam.num_images.set(num_images).wait(timeout=10)
timepix_control.cam.raw_enable.set(True).wait(timeout=10)
timepix_control.cam.raw_file_path.put(file_path)#.wait(timeout=10)
timepix_control.cam.raw_file_template.put('test_raw')#.wait(timeout=10)
timepix.pre_scan() # This sends the address to the ASI server from TimePixFly
status = TransitionStatus(timepix_control.cam.acquire, [1, 0])
timepix_control.cam.acquire.put(1)
status.wait(timeout=20)
# time.sleep(5)
time.sleep(5)
print(timepix._data_buffer)
print(timepix.timepix_fly_client.last_error())
# status = timepix.complete()
# status.wait(timeout=10)
# for ii, msg in enumerate(timepix._data_buffer):
# print(f"Received data {ii} with message type {msg['type']}")
timepix.unstage()
timepix.destroy()
superxas_bec/devices/timepix/test_script.py
+65 -20
View File
@@ -1,35 +1,80 @@
import enum
from ophyd import Component as Cpt
from ophyd import Device, EpicsSignal, Kind
from ophyd_devices import CompareStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
# class TimepixServerStatus(int, enum.Enum):
# DONE = 0
# ACQUIRE = 1
# class TimepixServerControl(Device):
# """Prefix of the detector 'X10DA-ES-TPX1:cam1:' """
# acquire = Cpt(EpicsSignal, 'Acquire', kind=Kind.omitted, doc="Detector acquiring signal")
# preview_enable= Cpt(EpicsSignal, 'PreviewEnable', kind=Kind.omitted, doc='Preview enable y/n')
# class TimepixServer(PSIDeviceBase, TimepixServerControl):
# """Prefix of the detector 'X10DA-ES-TPX1:cam1:' """
# def on_stage(self)-> None:
# self.preview_enable.set(1).wait(timeout=3)
# if self.acquire.get() != TimepixServerStatus.DONE:
# status = CompareStatus(self.acquire, TimepixServerStatus.DONE)
# self.cancel_on_stop(status)
# self.acquire.put(TimepixServerStatus.DONE)
# status.wait(timeout=3)
# def on_pre_scan(self):
# status = CompareStatus(self.acquire, TimepixServerStatus.ACQUIRE)
# self.cancel_on_stop(status)
# self.acquire.put(1)
# status.wait(timeout=3)
if __name__ == """__main__""":
import time
from superxas_bec.devices.timepix.timepix import Timepix
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_mock_server import (
TimePixFlyMockServer,
)
mock_server = TimePixFlyMockServer()
# from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_mock_server import (
# TimePixFlyMockServer,
# )
# # mock_server = TimePixFlyMockServer()
# server = TimepixServer(name='server', prefix='X10DA-ES-TPX1:cam1:')
# Create a Timepix object
timepix = Timepix(name="TimePixDetector")
rest_url = "p4-0017.psi.ch:8453"
ws_url = "p4-0017:8453/ws"
# data_server_host = "x10da-bec-001.psi.ch"
# port = 3030
timepix = Timepix(
name="TimePixDetector", prefix="X10DA-ES-TPX1:", rest_url=rest_url, ws_url=ws_url
)
timepix.on_connected()
timepix.wait_for_connection(all_signals=True, timeout=10)
# server.wait_for_connection(all_signals=True, timeout=10)
timepix.on_connected()
## LOOP for a scan
timepix.stage()
timepix.pre_scan()
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
mock_server.start_acquisition()
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
time.sleep(0.001)
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
time.sleep(0.1)
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
time.sleep(0.1)
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
time.sleep(1)
print(f"State of timepix_fly_client {timepix.timepix_fly_client.state().state}")
print("Timepix staged.")
# server.stage()
# timepix.pre_scan()
status = timepix.on_pre_scan()
print("Timepix staged in pre scan.")
timepix.trigger()
print("Timepix triggered.")
# time.sleep(5)
timepix.complete()
status = timepix.complete()
while not status.done:
time.sleep(0.1)
print(timepix.cam.acquire_busy.get())
status.wait(timeout=20)
for ii, msg in enumerate(timepix._data_buffer):
print(f"Received data {ii} with message type {msg['type']}")
timepix.timepix_fly_client.stop()
timepix.unstage()
timepix.destroy()
superxas_bec/devices/timepix/test_script_backup.py
+45
View File
@@ -0,0 +1,45 @@
if __name__ == """__main__""":
import threading
import time
from unittest import mock
import ophyd
from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
from superxas_bec.devices.timepix.timepix import Timepix
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_mock_server import (
TimePixFlyMockServer,
)
mock_server = TimePixFlyMockServer()
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
timepix = Timepix(name="timepix", prefix="")
patch_dual_pvs(timepix)
timepix.timepix_fly_client.on_connected()
timepix._reset_buffers()
timepix.start_data_server()
# timepix.on_connected()
## LOOP for a scan
timepix.stage()
status = timepix.pre_scan()
timepix.cam.acquire_busy._read_pv.mock_data = 1
status.wait(timeout=10)
print(f"Pre-scan status: {status}")
mock_server.start_acquisition()
print("Acquisition started on mock server.")
# time.sleep(5)
status = timepix.complete()
time.sleep(8)
timepix.cam.acquire_busy._read_pv.mock_data = 0
status.wait(timeout=10)
for ii, msg in enumerate(timepix._data_buffer):
print(f"Message {ii}: {msg.keys()}")
print(f"Received data {ii} with message type {msg['type']}")
if msg["type"] == "EndFrame":
print(f"Received error message: {msg}")
# print(f"Data: {msg}") # Print first 50 characters of data
timepix.unstage()
timepix.destroy()
superxas_bec/devices/timepix/timepix.py
+248 -46
View File
@@ -3,6 +3,8 @@ TimePix Detector class for interfacing with the TimePix detector. The timepix_si
implements the HTTP communication to the REST API for the tpx3app app.
"""
import atexit
import enum
import json
import signal
import socket
@@ -11,10 +13,17 @@ import time
from typing import Literal
from bec_lib.logger import bec_logger
from ophyd import ADBase
from ophyd import Component as Cpt
from ophyd import Device, DeviceStatus, StatusBase
from ophyd_devices import AndStatus, CompareStatus
from ophyd_devices.devices.areadetector.cam import ASItpxCam
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import TimepixFlyClient
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import (
TimepixFlyClient,
TimePixFlyStatus,
)
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
OtherConfigModel,
PixelMap,
@@ -22,56 +31,155 @@ from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface impor
logger = bec_logger.logger
DATA_SERVER_HOST = "localhost" # Default data server host for TimePix detector
DATA_SERVER_PORT = 3015 # Default data server port for TimePix detector
class TDCEdge(int, enum.Enum):
"""TDC Edge enum options for TimePix detector."""
RISING = 0
FALLING = 1
BOTH = 2
class TDCOuput(int, enum.Enum):
"""TDC Output enum options for TimePix detector."""
ALL_CHANNELS = 0
CHANNEL_0 = 1
CHANNEL_1 = 2
CHANNEL_2 = 3
CHANNEL_3 = 4
class ACQUIRESTATUS(int, enum.Enum):
"""Acquire status enum options for TimePix detector."""
DONE = 0
ACQUIRING = 1
class DETECTORSTATE(int, enum.Enum):
"""Detector state enum options for TimePix detector."""
IDLE = 0
ACQUIRE = 1
READOUT = 2
CORRECT = 3
SAVING = 4
ABORTING = 5
ERROR = 6
WAITING = 7
INITIALIZING = 8
DISCONNECTED = 9
ABORTED = 10
class TRIGGERMODE(int, enum.Enum):
"""Trigger mode enum options for TimePix detector."""
INTERNAL = 0
EXTERNAL = 1
SOFTWARE = 2
class TRIGGERSOURCE(int, enum.Enum):
"""Trigger source enum options for TimePix detector."""
HDMI1_1 = 0
HDMI1_2 = 1
HDMI1_3 = 2
HDMI2_1 = 3
HDMI2_2 = 4
HDMI2_3 = 5
class EXPOSUREMODE(int, enum.Enum):
"""Exposure mode enum options for TimePix detector."""
TIMED = 0
TRIGGER_WIDTH = 1
# pylint: disable=too-many-instance-attributes, too-many-arguments, too-many-locals
class TimePixControl(ADBase):
"""Interface for the TimePix EPICS control of the TimePix detector."""
cam = Cpt(ASItpxCam, "cam1:")
class Timepix(PSIDeviceBase, Device):
class Timepix(PSIDeviceBase, TimePixControl):
"""
TimePix Detector class for interfacing with the TimePix detector.
The timepix_signals module implements the HTTP communication to the REST API for the tpx3app app
The Timepix detector REST API service for the TimePixFly backend runs on SERVER_ADDRESS defined
in the timepix_signals module. This can certainly be improved to be configurable by the config
The Timepix detector REST API service module implements the HTTP communication to the REST API for the tpx3app app
DATA_SERVER_HOST = "x10da-bec-001.psi.ch" # Default data server host for TimePix detector
DATA_SERVER_PORT = 3015 # Default data server port for TimePix detector
rest_url = 'p4-0017.psi.ch:8452'
ws_url = 'p4-0017:8452/ws'
Prefix of EPICS Control is 'X10DA-ES-TPX1:'
"""
def __init__(
self,
*,
name,
prefix: str,
rest_url: str = "localhost:8452",
ws_url: str = "localhost:8452/ws",
scan_info=None,
device_manager=None,
backend_host: str | None = None,
data_server_host: str | None = None,
data_server_port: int | None = None,
hostname: str | None = None,
host_port: int | None = None,
**kwargs,
):
"""
#TODO addd docstring
Initialize the Timepix detector.
Args:
name (str): Name of the device.
rest_url (str): Host address of the TimePixFly backend server.
ws_url (str): WebSocket URL for the TimePixFly backend.
scan_info (dict, optional): Scan information. Defaults to None.
device_manager (DeviceManager, optional): Device manager instance. Defaults to None.
data_server_host (str, optional): Host address for the data server. Defaults to DATA
SERVER_HOST.
data_server_port (int, optional): Port for the data server. Defaults to DATA_SERVER_PORT.
**kwargs: Additional keyword arguments for the PSIDeviceBase class.
"""
super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs)
self.timepix_fly_client = TimepixFlyClient(
api_server_address=backend_host, logger=logger, parent=self
)
self._data_server_host = data_server_host if data_server_host else DATA_SERVER_HOST
self._data_server_port = (
data_server_port if data_server_port is not None else DATA_SERVER_PORT
super().__init__(
name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
)
self._ws_url = ws_url
self._rest_url = rest_url
self.timepix_fly_client = TimepixFlyClient(rest_url=rest_url, ws_url=ws_url, parent=self)
if hostname is None:
hostname = socket.getfqdn()
if not hostname.endswith(".psi.ch"):
logger.info(f"Found host without psi.ch domain {hostname}")
if host_port is None:
host_port = 3015
self._hostname = hostname
self._data_server_host: str | None = None
self._data_server_port = host_port
self._rlock = threading.RLock()
# Data server
self._data_server_thread = None
self._data_server_thread_event = None
self._data_server_started = threading.Event()
# Socket server
self._socket_server = None
self._socket_server_allowed_connections = 1 # How many ?
self._socket_server_timeout = 0.1
self._socket_server_buffer_size = 4096
#
# Data buffers
self._data_buffer = []
self._global_buffer = "" # Global buffer to store received data
# Decoding
self._decoder = json.JSONDecoder()
# Wait timeout
self._pv_timeout = 5.0 # Default timeout for PV operations
# Register cleanup
atexit.register(self.on_destroy)
### Beamline specifi methods for the TimePix Detector integration ###
@@ -95,7 +203,18 @@ class Timepix(PSIDeviceBase, Device):
Called after the device is connected and its signals are connected.
Default values for signals should be set here.
"""
# Prepare TimePix Detector
self.cam.tdc1_enable.set(1).wait(timeout=self._pv_timeout)
self.cam.tdc1_edge.set(TDCEdge.RISING).wait(timeout=self._pv_timeout)
self.cam.tdc1_output.set(TDCOuput.ALL_CHANNELS).wait(timeout=self._pv_timeout)
self.cam.tdc2_enable.set(0).wait(timeout=self._pv_timeout) # to be checked
self.cam.trigger_mode.set(TRIGGERMODE.INTERNAL).wait(timeout=self._pv_timeout)
self.cam.trigger_source.set(TRIGGERSOURCE.HDMI1_1).wait(timeout=self._pv_timeout)
self.cam.exposure_mode.set(EXPOSUREMODE.TIMED).wait(timeout=self._pv_timeout)
# Prepare backend for TimePixFly
self.timepix_fly_client.on_connected()
self._reset_buffers()
self.start_data_server()
def on_stage(self) -> DeviceStatus | StatusBase | None:
@@ -104,13 +223,30 @@ class Timepix(PSIDeviceBase, Device):
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
"""
# currently hardcode acquire time.. -> discuss logic here
self.cam.acquire_time.set(0.998).wait(timeout=self._pv_timeout)
self.cam.acquire_period.set(1.0).wait(timeout=self._pv_timeout)
self.cam.num_images.set(10).wait(timeout=self._pv_timeout)
# -------------------------
# Prepare TimePixFly
self._reset_buffers()
status = DeviceStatus(self)
self.cancel_on_stop(status)
self.timepix_fly_client.add_status_callback(
status,
success=[TimePixFlyStatus.CONFIG],
error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN],
)
if self._data_server_host is None:
raise RuntimeError(f"Data server host is not set for device {self.name}.")
# Parse scan info for OtherConfig
config = OtherConfigModel(
output_uri=f"tcp:{self._data_server_host}:{self._data_server_port}",
TRoiStep=1,
TRoiN=5000,
)
logger.info(config)
# Parse pixel map from scan info if needed, otherwise use some default pixel map.
pixel_map = PixelMap(
chips=[
@@ -120,34 +256,60 @@ class Timepix(PSIDeviceBase, Device):
[{"i": 0, "p": [1, 2], "f": [0.5, 0.5]}],
]
)
status.wait(timeout=5.0)
self.timepix_fly_client.set_other_config(config)
self.timepix_fly_client.set_pixel_map(pixel_map)
self._wait_for_state_condition("config", timeout=5.0)
def on_unstage(self) -> DeviceStatus | StatusBase | None:
def on_unstage(self) -> None:
"""Called while unstaging the device."""
self._reset_buffers()
def on_pre_scan(self) -> DeviceStatus | StatusBase | None:
def on_pre_scan(self) -> StatusBase:
"""Called right before the scan starts on all devices automatically."""
self.timepix_fly_client.start() # --> State goes to setup
self._wait_for_state_condition("setup", timeout=5.0)
status_acquire = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE)
status_detector_state = CompareStatus(self.cam.detector_state, DETECTORSTATE.IDLE)
status_detector = AndStatus(status_acquire, status_detector_state)
self.cancel_on_stop(status_detector)
status = DeviceStatus(self)
self.cancel_on_stop(status)
self.timepix_fly_client.add_status_callback(
status,
success=[TimePixFlyStatus.AWAIT_CONNECTION],
error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN],
)
self.timepix_fly_client.start()
pre_scan_status = AndStatus(status, status_detector)
self.cancel_on_stop(pre_scan_status) # --> State goes to setup
return pre_scan_status
def on_trigger(self) -> DeviceStatus | StatusBase | None:
"""Called when the device is triggered."""
self.cam.acquire.put(1)
def on_complete(self) -> DeviceStatus | StatusBase | None:
"""Called to inquire if a device has completed a scans."""
self._wait_for_state_condition(
"config", timeout=5.0
) # During measurment, the state is set to collect. Goes back to config after the measurement is done.
# Detector Control
status_detector = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE)
self.cancel_on_stop(status_detector)
# TimepixFFly
status = DeviceStatus(self)
logger.info(f"Registering status callback {id(status)} for {self.name} on complete.")
self.cancel_on_stop(status)
self.timepix_fly_client.add_status_callback(
status,
success=[TimePixFlyStatus.CONFIG],
error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN],
)
complete_status = AndStatus(status, status_detector)
return complete_status
def on_kickoff(self) -> DeviceStatus | StatusBase | None:
"""Called to kickoff a device for a fly scan. Has to be called explicitly."""
def on_stop(self) -> None:
"""Called when the device is stopped."""
self.timepix_fly_client.stop()
self.cam.acquire.put(0)
self.timepix_fly_client.stop_running_collection()
self._reset_buffers()
### Custom methods for the TimePix Data server ###
@@ -187,11 +349,15 @@ class Timepix(PSIDeviceBase, Device):
def _stop_data_receiver(self):
if self._data_server_thread is not None and self._data_server_thread.is_alive():
self._data_server_thread_event.set()
self._data_server_thread.join(timeout=5.0)
self._data_server_thread.join(timeout=1.0)
if self._data_server_thread is not None and self._data_server_thread.is_alive():
logger.warning(f"Data server thread did not stop gracefully.")
logger.warning(
f"Data server thread did not stop gracefully for device {self.name}."
)
else:
logger.warning(f"Data server thread is not running or has already stopped.")
logger.warning(
f"Data server thread is not running or has already stopped for device {self.name}."
)
def _receive_data_on_socket(self):
"""Receive data on socket connection."""
@@ -199,6 +365,8 @@ class Timepix(PSIDeviceBase, Device):
# for testing purposes, move logic into separate function for each while loop
while not self._data_server_thread_event.is_set():
try:
self._data_server_started.set()
# logger.info(self._socket_server)
conn, addr = self._socket_server.accept()
logger.info(f"Accepted connection from {addr}")
with conn:
@@ -209,12 +377,17 @@ class Timepix(PSIDeviceBase, Device):
time.sleep(0.1) # No data received, wait a bit before next attempt
continue
logger.info(f"Received data: {len(chunk)} bytes")
buffer += chunk.decode("utf-8") # Trailing byte, i.e. -> "\n"
buffer += chunk.decode("utf-8") # Trailing byte, i.e. -> "}\n"
self._global_buffer += (
buffer # Append to global buffer, #TODO check why endfram misses
)
if buffer.endswith("}\n"):
self._decode_received_data(buffer)
logger.info(
f"Buffer starts with {buffer[:50]}... and ends with {buffer[-50:]}"
)
for entry in buffer.split("}\n"):
if entry:
self._decode_received_data(entry + "}")
buffer = "" # Reset buffer after processing
# Ignore timeout exception on socket aslong as server_thread_event is not set
@@ -227,7 +400,7 @@ class Timepix(PSIDeviceBase, Device):
This method should be overridden to implement the actual decoding logic.
"""
try:
obj, idx = self._decoder.raw_decode(buffer)
obj, _ = self._decoder.raw_decode(buffer)
self._data_buffer.append(obj)
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON from buffer: {buffer}")
@@ -244,16 +417,19 @@ class Timepix(PSIDeviceBase, Device):
def on_destroy(self):
"""Cleanup method to stop the data server thread, clean up the socket server"""
self.timepix_fly_client.shutdown()
with self._rlock:
if self._socket_server:
try:
self._stop_data_receiver()
except Exception as e:
logger.info(
f"Failed to stop data receiver for device {self.name} with exception {e}."
)
try:
self._socket_server.close()
except Exception as e: # pylint: disable=broad-except
logger.warning(f"Failed to shutdown socket server gracefully. Error: {e}")
self._stop_data_receiver()
self._socket_server = None
self._data_server_thread = None
self._data_server_thread_event = None
logger.warning(f"Failed to shutdown socket server. Error: {e}")
def restart_data_receiver(self):
"""Restart the data receiver thread."""
@@ -273,13 +449,39 @@ class Timepix(PSIDeviceBase, Device):
"""
# AF_INET6 for IPv6, use AF_INET for IPv4; for localhost this may be different depending on the system
# TODO add an os check if self._data_server_host is localhost
self._socket_server = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
self._socket_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._socket_server.bind((self._data_server_host, self._data_server_port))
self._socket_server.listen(self._socket_server_allowed_connections)
# self._socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) #BEAMLINE USAGE!
# addr_info = socket.getaddrinfo(
# self._data_server_host, self._data_server_port, socket.AF_UNSPEC, socket.SOCK_STREAM
# )
# # Use the first valid address returned by getaddrinfo
# af, socktype, proto, _, _ = addr_info[0]
# self._socket_server = socket.socket(af, socktype, proto)
info = socket.getaddrinfo(
self._hostname,
port=self._data_server_port,
family=socket.AF_INET,
type=socket.SOCK_STREAM,
)
if len(info) == 0:
raise RuntimeError(f"No socket info found")
if len(info) > 1:
logger.warning(f"Found multiple socket interfaces {info}, using the first one")
af, socktype, proto, _, host_port_info = info[0]
self._data_server_host = host_port_info[0] # Hostname or IP address
self._data_server_port = host_port_info[1] # Port number, should be the same as before
self._socket_server = socket.create_server(
host_port_info, family=af, backlog=1, reuse_port=True
)
self._data_server_host, self._data_server_port = self._socket_server.getsockname()
logger.info(
f"Starting data server on {self._data_server_host}:{self._data_server_port} for {self.name}."
)
# self._socket_server = socket.socket(af, socktype, proto)
# self._socket_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# self._socket_server.bind(host_port_info)
# self._socket_server.listen(1) # Only allow one connection
self._socket_server.settimeout(self._socket_server_timeout)
self._start_data_receiver()
logger.info(f"Data server started on {self._data_server_host}:{self._data_server_port}")
# pylint: disable=protected-access
@@ -291,7 +493,7 @@ if __name__ == "__main__":
mock_server = TimePixFlyMockServer()
# Create a Timepix object
timepix = Timepix(name="TimePixDetector")
timepix = Timepix(name="TimePixDetector", prefix="")
timepix.on_connected()
timepix.stage()
timepix.pre_scan()
superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py
+246 -32
View File
@@ -2,12 +2,26 @@
This module implements a python client ot the REST interface of the tpx3app.
"""
from __future__ import annotations
import copy
import enum
import json
import threading
import time
import traceback
from typing import Any, Type
import requests
from bec_lib.logger import bec_logger
from ophyd import Device, StatusBase
from websockets import State
from websockets.exceptions import WebSocketException
from websockets.sync.client import ClientConnection, connect
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
LastError,
NetAddresses,
OtherConfigModel,
PixelMap,
PixelMapFromFile,
@@ -16,9 +30,30 @@ from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface impor
Version,
)
logger = bec_logger.logger
# TODO remove!
SERVER_ADDRESS = "localhost:8452" # Default server address for TimePix REST API
# pylint: disable=arguments-differ
class TimePixStatusError(Exception):
"""Exception raised when the TimePix detector status was in an unexpected state."""
class TimePixFlyStatus(str, enum.Enum):
"""
Enum representing the status of the TimePix detector.
"""
INIT = "init"
CONFIG = "config"
SETUP = "setup"
COLLECT = "collect"
SHUTDOWN = "shutdown"
UNDEFINED = "undefined"
AWAIT_CONNECTION = "await_connection"
EXCEPT = "except"
class TimepixFlyClient:
@@ -28,31 +63,205 @@ class TimepixFlyClient:
It provides methods to send GET and PUT requests to the TimePix server.
"""
def __init__(self, api_server_address: str | None = None, logger=None, parent=None):
def __init__(self, rest_url: str, ws_url: str, parent: Device):
"""
Initialize the TimePixFlyClient with a server address.
Args:
server_address (str): The address of the TimePix REST API: "tpx3app".
"""
self._api_server_address = api_server_address if api_server_address else SERVER_ADDRESS
self.rest_url = rest_url
self.ws_url = ws_url
self.parent = parent
self.ws_client: ClientConnection | None = None
self._rlock = threading.RLock()
self._timeout = 5 # Default timeout for requests
self._logger = logger
self._parent = parent
self._status: TimePixFlyStatus = TimePixFlyStatus.UNDEFINED
self._ws_update_thread: threading.Thread | None = None
self._shutdown_event = threading.Event()
self._status_callbacks: dict[
str, tuple[StatusBase, list[TimePixFlyStatus], list[TimePixFlyStatus]]
] = {}
self._started: bool = False # Flag to indicate if the client has started sending data
def _add_log(self, message: str) -> None:
#############################
### Utility Methods ###
#############################
def on_connected(self) -> None:
"""
Add a log message to the logger if available.
Called when the client is connected to the TimePix server.
This method can be overridden to perform actions when the client connects.
"""
try:
self.stop_running_collection()
self.connect()
self.wait_for_connection(timeout=5)
except Exception as e:
logger.error(
f"Error while checking the state of the TimePix server: {e}. "
f"Please check the server address and ensure the server is running."
)
raise e
def stop_running_collection(self):
"""
Resets the TimePix backend to the configuration state.
"""
state = self.state()
if state.state != TimePixFlyStatus.CONFIG or self._started is True:
logger.info(
f"Resetting TimePix backend to config state. Current state: {state.state}, started: {self._started}"
)
self.stop_collect()
##############################
### WebSocket Methods ###
### Status Update Handling ###
##############################
@property
def status(self) -> TimePixFlyStatus:
"""
Get the current status of the TimePix detector.
Returns:
TimePixFlyStatus: The current status of the TimePix detector.
"""
return self._status
def add_status_callback(
self, status: StatusBase, success: list[TimePixFlyStatus], error: list[TimePixFlyStatus]
):
"""
Add a StatusBase callback for the TimePix detector. The status will be updated when the detector status
changes and set to finished when the status matches one of the specified success statuses and to exception
when the status matches one of the specified error statuses.
Args:
message (str): The message to log.
status (StatusBase): StatusBase object
success (list[StdDaqStatus]): list of statuses that indicate success
error (list[StdDaqStatus]): list of statuses that indicate error
"""
if self._logger is not None:
if self._parent is not None and hasattr(self._parent, "name"):
message = f"{self._parent.name}: {message}"
self._logger.info(message)
else:
self._add_log(message)
if self.status in success:
status.set_finished()
return
self._status_callbacks[id(status)] = (status, success, error)
def connect(self):
"""
Connect to the TimePix WebSocket server.
"""
if self._ws_update_thread is not None and self._ws_update_thread.is_alive():
return
self._ws_update_thread = threading.Thread(
target=self._ws_update_loop, name=f"{self.parent.name}_stddaq_ws_loop", daemon=True
)
self._ws_update_thread.start()
def wait_for_connection(self, timeout: float = 10) -> None:
"""
Wait for the connection to the StdDAQ to be established.
Args:
timeout (float): timeout for the request
"""
start_time = time.time()
while True:
if self.ws_client is not None and self.ws_client.state == State.OPEN:
return
try:
self.ws_client = connect(f"ws://{self.ws_url}")
break
except ConnectionRefusedError as exc:
if time.time() - start_time > timeout:
raise TimeoutError("Timeout while waiting for connection to StdDAQ") from exc
time.sleep(0.5) # Try to reconnect every 2 seconds
def _ws_update_loop(self):
"""
Loop to update the status property of the StdDAQ.
"""
while not self._shutdown_event.is_set():
self._ws_send_and_receive()
def _ws_send_and_receive(self):
if not self.ws_client:
self.wait_for_connection()
try:
try:
recv_msgs = self.ws_client.recv(timeout=0.1)
except TimeoutError:
return
logger.trace(f"Received from timepixfly ws: {recv_msgs}")
if recv_msgs is not None:
self._on_received_ws_message(recv_msgs)
except WebSocketException:
content = traceback.format_exc()
logger.warning(f"Websocket connection closed unexpectedly: {content}")
self.wait_for_connection()
def _on_received_ws_message(self, msg: str):
"""
Handle a message received from the StdDAQ.
"""
try:
logger.info(f"Received message from TimePixFly: {msg}")
self._status = TimePixFlyStatus(msg)
except Exception:
content = traceback.format_exc()
logger.warning(f"Failed to decode websocket message: {content}")
return
self._run_status_callbacks()
def _run_status_callbacks(self):
"""
Update the StatusBase objects based on the current status of the StdDAQ.
If the status matches one of the success or error statuses, the StatusBase object will be set to finished
or exception, respectively and removed from the list of callbacks.
"""
status = self._status
completed_callbacks = []
logger.info(f"Running status callbacks for TimePixFly status: {status.value}")
callback_ids = list(self._status_callbacks.keys())
for cb_id in callback_ids:
dev_status, success, error = self._status_callbacks[cb_id]
logger.info(f"Checking status callback {cb_id} for TimePixFly status: {status.value}")
if dev_status.done:
logger.warning("Status object already resolved. Skipping Timepix callback.")
self._status_callbacks.pop(cb_id)
continue
if status in success:
dev_status.set_finished()
logger.info(f"Timepix status in succes is {status.value}")
self._status_callbacks.pop(cb_id)
elif status in error:
logger.warning(f"Timepix status in error is {status.value}")
dev_status.set_exception(
TimePixStatusError(
f"TimePixStatus status is '{status.value},' last error: {self.last_error()}"
)
)
self._status_callbacks.pop(cb_id)
# If config is reached, started can be reset to False
if status == TimePixFlyStatus.CONFIG:
self._started = False
def shutdown(self):
"""
Shutdown the StdDAQ client.
"""
self._shutdown_event.set()
if self._ws_update_thread is not None:
self._ws_update_thread.join()
if self.ws_client is not None:
self.ws_client.close()
self.ws_client = None
############################
##### REST API Methods #####
############################
def _get(
self, get_cmd: str, get_response_model: Type[TimePixResponse] | None = None
@@ -67,15 +276,13 @@ class TimepixFlyClient:
Returns:
Any: The parsed response if a model is provided, else the raw response.
"""
response = requests.get(
f"http://{self._api_server_address}/{get_cmd}", timeout=self._timeout
)
response = requests.get(f"http://{self.rest_url}/{get_cmd}", timeout=self._timeout)
response.raise_for_status() # Raise an error for bad responses
if get_response_model is not None:
try:
return get_response_model(**response.json())
except Exception as e:
self._add_log(f"Error parsing response for {get_cmd}: Response: {response.text}")
logger.info(f"Error parsing response for {get_cmd}: Response: {response.text}")
raise e
else:
return response.text
@@ -95,32 +302,19 @@ class TimepixFlyClient:
Any: The parsed response if a model is provided, else None.
"""
response = requests.put(
f"http://{self._api_server_address}/{put_cmd}", json=value, timeout=self._timeout
f"http://{self.rest_url}/{put_cmd}", json=value, timeout=self._timeout
)
response.raise_for_status()
if put_response_model is not None:
return put_response_model(**response.json())
def on_connected(self) -> None:
"""
Called when the client is connected to the TimePix server.
This method can be overridden to perform actions when the client connects.
"""
try:
self.state()
except Exception as e:
self._add_log(
f"An error occurred while connecting to the TimePix server: {e}. "
f"Please check the server address and ensure the server is running."
)
raise e
def start(self) -> None:
"""
Start the TimePix detector by sending a GET request to the start endpoint.
This method is a wrapper around the REST API call to start the detector.
"""
self._get(get_cmd="?start=true")
self._started = True
def stop(self) -> None:
"""
@@ -128,6 +322,15 @@ class TimepixFlyClient:
This method is a wrapper around the REST API call to stop the detector.
"""
self._get(get_cmd="?stop=true")
self._started = False
def stop_collect(self) -> None:
"""
Stop the data collection of the TimePix detector by sending a GET request to the stop-collect endpoint.
This method is a wrapper around the REST API call to stop data collection.
"""
self._get(get_cmd="?stop_collect=true")
self._started = False
def kill(self) -> None:
"""
@@ -135,6 +338,7 @@ class TimepixFlyClient:
This method is a wrapper around the REST API call to kill the detector.
"""
self._get(get_cmd="?kill=true")
self._started = False
def last_error(self) -> LastError:
"""
@@ -244,3 +448,13 @@ class TimepixFlyClient:
f"Value must be an instance of OtherConfigModel. Received {type(other_config)}, {other_config}."
)
self._put(put_cmd="other-config", value=other_config.model_dump(), put_response_model=None)
def get_net_addresses(self) -> dict[str, str]:
"""
Get the network addresses of the TimePix detector by sending a GET request
to the net-addresses endpoint.
Returns:
dict[str, str]: A dictionary containing the network addresses of the TimePix detector.
"""
return self._get(get_cmd="net-addresses", get_response_model=NetAddresses)
superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_interface.py
+31 -1
View File
@@ -67,7 +67,7 @@ class ProgramState(TimePixResponse):
"""
type: str = "ProgramState"
state: Literal["init", "config", "setup", "collect", "shutdown"]
state: Literal["init", "config", "setup", "await_connection", "collect", "shutdown"]
class Version(TimePixResponse):
@@ -119,3 +119,33 @@ class TimepixDataFrame(TimePixResponse):
class TimepixEndFrame(TimePixResponse):
pass
# Habe ein GET /net-addresses call implementiert
# // /net-addresses GET applicable net addresses
# // GET return:
# // - status 200
# // - data
# // {
# // "type":"NetAddresses",
# // "control":"127.0.0.1:8452", // own rest interface
# // "address":"127.0.0.1:8451", // own address, the destination of ASI server raw data
# // "server":"127.0.0.1:8080" // ASI server rest interface address
# // }
class NetAddresses(TimePixResponse):
"""
NetAddresses is a Pydantic model that represents the network addresses used by the TimePix detector.
Attributes:
- type: str - The type of the response, default is "NetAddresses".
- control: str - The address of the REST interface for control commands.
- address: str - The address where the ASI server sends raw data.
- server: str - The address of the ASI server's REST interface.
"""
type: str = "NetAddresses"
control: str # timepix_rest_host
address: str # data_socket_for_asi
server: str # asi_rest_host