Add SSE client support for TELL updates and enhance resilience

Replaced periodic polling with an SSE-based listener for real-time TELL updates, improving efficiency and responsiveness. Enhanced WebSocket client logic by implementing automatic reconnection. Updated dependencies to include `sseclient` and upgraded `aaredb` for compatibility.

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]

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()

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