diff --git a/daq/pyproject.toml b/daq/pyproject.toml index 7c7c95a8..0dc881de 100644 --- a/daq/pyproject.toml +++ b/daq/pyproject.toml @@ -13,10 +13,11 @@ dependencies = [ "fastapi==0.115.12", "uvicorn==0.34.2", "ultralytics==8.3.133", - "aareDB==0.1.1a8", + "aaredb==0.1.1a9", "opencv-python-headless==4.11.0.86", "python_multipart==0.0.20", - "websocket-client" + "websocket-client", + "sseclient>=0.0.27", ] [lint] diff --git a/daq/src/aaredaq/tellupdater.py b/daq/src/aaredaq/tellupdater.py index 5d712ce5..78608b07 100644 --- a/daq/src/aaredaq/tellupdater.py +++ b/daq/src/aaredaq/tellupdater.py @@ -1,6 +1,8 @@ import os import json import websocket +import sseclient +import requests import threading import time from aareDBclient.models import PuckWithTellPosition @@ -12,7 +14,6 @@ from aaredaqlib.beamline import MXBeamline SLOT_IDENTIFIER = "X06DA" WS_URL = f"wss://mx-db-01.psi.ch/dispatcher/protected_router/wstell/ws/slot/{SLOT_IDENTIFIER}" WS_HEADERS = [f"X-Shared-Password: {os.getenv('AAREDB_SHARED_PASSWORD')}"] -POLL_INTERVAL = 30 # seconds # Initialize TELL client and DB wrapper beamline = MXBeamline.X06DA # Use your beamline enum/value @@ -23,6 +24,13 @@ aare_db = AareWrapper(bl=beamline) current_pucks = [] last_pucks_sent = [] +def listen_to_sse(): + sse_url = tell_client.url + "/events" + with requests.get(sse_url, stream=True) as response: + client = sseclient.SSEClient(response) + for event in client: + on_sse_event(event) + def compare_and_report_change(old, new, key_func): old_ids = set(key_func(p) for p in old) new_ids = set(key_func(p) for p in new) @@ -53,14 +61,26 @@ def pucks_equal(pucks1, pucks2): return all(p1.id == p2.id for p1, p2 in zip(sorted(pucks1, key=lambda p: p.id), sorted(pucks2, key=lambda p: p.id))) -def timer_update_pucks_beamline(): - global last_pucks_sent +def handle_tell_change_event(): + """Fetch the latest detected pucks from TELL and update the database.""" try: detected_pucks = tell_client.get_detected_pucks() aare_db.set_pucks_beamline(detected_pucks) - + print("[SSE][INFO] Updated pucks in database after TELL change.") except Exception as exc: - print(f"[ERROR][Timer] {exc}") + print(f"[SSE][ERROR] {exc}") + +def on_sse_event(event): + # event is an sseclient.Event object + print(f"Received SSE event: {event.event}") + if event.event == "DewarContentUpdate": + print("Dewar content changed, updating...") + # Option 1: Use event.data directly if it is complete puck info: + # puck_info = json.loads(event.data) + # do something with puck_info + + # Option 2: Ask TELL again for fresh data + handle_tell_change_event() def on_message(ws, message): global current_pucks @@ -92,25 +112,24 @@ def on_close(ws, close_status_code, close_msg): def on_open(ws): print("[WS][OPEN] WebSocket opened.") -def periodic_polling(): - while True: - timer_update_pucks_beamline() - time.sleep(POLL_INTERVAL) - def main(): - ws = websocket.WebSocketApp( - WS_URL, - header=WS_HEADERS, - on_message=on_message, - on_error=on_error, - on_close=on_close, - on_open=on_open, - ) + # Start SSE listener in a separate background thread + sse_thread = threading.Thread(target=listen_to_sse, daemon=True) + sse_thread.start() - poll_thread = threading.Thread(target=periodic_polling, daemon=True) - poll_thread.start() - - ws.run_forever(sslopt={"cert_reqs": 0}) + # Main thread runs websocket client loop + while True: + ws = websocket.WebSocketApp( + WS_URL, + header=WS_HEADERS, + on_message=on_message, + on_error=on_error, + on_close=on_close, + on_open=on_open, + ) + ws.run_forever(sslopt={"cert_reqs": 0}) + print("[WS][INFO] WebSocket connection lost. Reconnecting in 5 seconds...") + time.sleep(5) if __name__ == "__main__": main() \ No newline at end of file diff --git a/daq/src/mxlibs3/tell_client.py b/daq/src/mxlibs3/tell_client.py index 0cccece6..21cb98e5 100755 --- a/daq/src/mxlibs3/tell_client.py +++ b/daq/src/mxlibs3/tell_client.py @@ -43,6 +43,7 @@ POSITION_HEATER = "pHeatB" class TellClient: def __init__(self, bl: MXBeamline): + self.__url = None if bl == MXBeamline.SIMULATED: self.__simulation = True elif bl == MXBeamline.X06DA: @@ -75,6 +76,10 @@ class TellClient: self.debug = False self._last_cmd_id = -1 + @property + def url(self): + return self.__url + def smc_get_current(self) -> float: if self.__simulation: return 30.0