diff --git a/pyproject.toml b/pyproject.toml index f0fb0a4..67c4e70 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,9 @@ classifiers = [ "Programming Language :: Python :: 3", "Topic :: Scientific/Engineering", ] -dependencies = [] +dependencies = [ + "websockets", +] [project.optional-dependencies] dev = [ diff --git a/superxas_bec/devices/timepix/beamline_test.py b/superxas_bec/devices/timepix/beamline_test.py new file mode 100644 index 0000000..a748ddf --- /dev/null +++ b/superxas_bec/devices/timepix/beamline_test.py @@ -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() diff --git a/superxas_bec/devices/timepix/test_script.py b/superxas_bec/devices/timepix/test_script.py index fac1fd5..c11709f 100644 --- a/superxas_bec/devices/timepix/test_script.py +++ b/superxas_bec/devices/timepix/test_script.py @@ -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() diff --git a/superxas_bec/devices/timepix/test_script_backup.py b/superxas_bec/devices/timepix/test_script_backup.py new file mode 100644 index 0000000..de931ef --- /dev/null +++ b/superxas_bec/devices/timepix/test_script_backup.py @@ -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() diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index 65f42ad..c622190 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/superxas_bec/devices/timepix/timepix.py @@ -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() diff --git a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py index 5081543..b2d1743 100644 --- a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py +++ b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py @@ -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) diff --git a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_interface.py b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_interface.py index b128c6c..0f5debb 100644 --- a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_interface.py +++ b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_interface.py @@ -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