Re-implement xOTA using Websocket client

config-example.yml

@@ -85,7 +85,7 @@ spot-providers:
     class: "XOTA"
     name: "39C3 TOTA"
     enabled: false
-    url: "https://dev.39c3.totawatch.de/"
+    url: "wss://dev.39c3.totawatch.de/api/spot/live"
     # Fixed SIG/latitude/longitude for all spots from a provider is currently only a feature for the "XOTA" provider,
     # the software found at https://github.com/nischu/xOTA/. This is because this is a generic backend for xOTA
     # programmes and so different URLs provide different programmes.

requirements.txt

@@ -12,4 +12,5 @@ requests-sse~=0.5.2
 rss-parser~=2.1.1
 pyproj~=3.7.2
 prometheus_client~=0.23.1
-beautifulsoup4~=4.14.2
+beautifulsoup4~=4.14.2
+websocket-client~=1.9.0

spotproviders/websocket_spot_provider.py

@@ -0,0 +1,74 @@
+import logging
+from datetime import datetime
+from threading import Thread
+from time import sleep
+
+import pytz
+from websocket import create_connection
+
+from core.constants import HTTP_HEADERS
+from spotproviders.spot_provider import SpotProvider
+
+
+# Spot provider using websockets.
+class WebsocketSpotProvider(SpotProvider):
+
+    def __init__(self, provider_config, url):
+        super().__init__(provider_config)
+        self.url = url
+        self.ws = None
+        self.thread = None
+        self.stopped = False
+        self.last_event_id = None
+
+    def start(self):
+        logging.info("Set up websocket connection to " + self.name + " spot API.")
+        self.stopped = False
+        self.thread = Thread(target=self.run)
+        self.thread.daemon = True
+        self.thread.start()
+
+    def stop(self):
+        self.stopped = True
+        if self.ws:
+            self.ws.close()
+        if self.thread:
+            self.thread.join()
+
+    def _on_open(self):
+        self.status = "Waiting for Data"
+
+    def _on_error(self):
+        self.status = "Connecting"
+
+    def run(self):
+        while not self.stopped:
+            try:
+                logging.debug("Connecting to " + self.name + " spot API...")
+                self.status = "Connecting"
+                self.ws = create_connection(self.url, header=HTTP_HEADERS)
+                data = self.ws.recv()
+                if data:
+                    try:
+                        new_spot = self.ws_message_to_spot(data)
+                        if new_spot:
+                            self.submit(new_spot)
+
+                        self.status = "OK"
+                        self.last_update_time = datetime.now(pytz.UTC)
+                        logging.debug("Received data from " + self.name + " spot API.")
+
+                    except Exception as e:
+                        logging.exception("Exception processing message from Websocket Spot Provider (" + self.name + ")")
+
+            except Exception as e:
+                self.status = "Error"
+                logging.exception("Exception in Websocket Spot Provider (" + self.name + ")", e)
+            else:
+                self.status = "Disconnected"
+            sleep(5)  # Wait before trying to reconnect
+
+    # Convert a WS message received from the API into a spot. The exact message data (in bytes) is provided here so the
+    # subclass implementations can handle the message as string, JSON, XML, whatever the API actually provides.
+    def ws_message_to_spot(self, bytes):
+        raise NotImplementedError("Subclasses must implement this method")

spotproviders/xota.py

@@ -1,43 +1,39 @@
+import json
 from datetime import datetime
 
+import pytz
+
 from data.sig_ref import SIGRef
 from data.spot import Spot
-from spotproviders.http_spot_provider import HTTPSpotProvider
+from spotproviders.websocket_spot_provider import WebsocketSpotProvider
 
 
 # Spot provider for servers based on the "xOTA" software at https://github.com/nischu/xOTA/
 # The provider typically doesn't give us a lat/lon or SIG explicitly, so our own config provides this information. This
 # functionality is implemented for TOTA events.
-class XOTA(HTTPSpotProvider):
-    POLL_INTERVAL_SEC = 120
+class XOTA(WebsocketSpotProvider):
     FIXED_LATITUDE = None
     FIXED_LONGITUDE = None
     SIG = None
 
     def __init__(self, provider_config):
-        super().__init__(provider_config, provider_config["url"] + "/api/spot/all", self.POLL_INTERVAL_SEC)
+        super().__init__(provider_config, provider_config["url"])
         self.FIXED_LATITUDE = provider_config["latitude"] if "latitude" in provider_config else None
         self.FIXED_LONGITUDE = provider_config["longitude"] if "longitude" in provider_config else None
         self.SIG = provider_config["sig"] if "sig" in provider_config else None
 
-    def http_response_to_spots(self, http_response):
-        new_spots = []
-        # Iterate through source data
-        for source_spot in http_response.json():
-            # Convert to our spot format
-            spot = Spot(source=self.name,
-                        source_id=source_spot["id"],
-                        dx_call=source_spot["stationCallSign"].upper(),
-                        freq=float(source_spot["freq"]) * 1000,
-                        mode=source_spot["mode"].upper(),
-                        sig=self.SIG,
-                        sig_refs=[SIGRef(id=source_spot["reference"]["title"], sig=self.SIG, url=source_spot["reference"]["website"])],
-                        time=datetime.fromisoformat(source_spot["modificationDate"]).timestamp(),
-                        dx_latitude=self.FIXED_LATITUDE,
-                        dx_longitude=self.FIXED_LONGITUDE,
-                        qrt=source_spot["state"] != "active")
-
-            # Add to our list. Don't worry about de-duping, removing old spots etc. at this point; other code will do
-            # that for us.
-            new_spots.append(spot)
-        return new_spots
+    def ws_message_to_spot(self, bytes):
+        string = bytes.decode("utf-8")
+        source_spot = json.loads(string)
+        spot = Spot(source=self.name,
+                    source_id=source_spot["id"],
+                    dx_call=source_spot["stationCallSign"].upper(),
+                    freq=float(source_spot["freq"]) * 1000,
+                    mode=source_spot["mode"].upper(),
+                    sig=self.SIG,
+                    sig_refs=[SIGRef(id=source_spot["reference"]["title"], sig=self.SIG, url=source_spot["reference"]["website"])],
+                    time=datetime.now(pytz.UTC).timestamp(),
+                    dx_latitude=self.FIXED_LATITUDE,
+                    dx_longitude=self.FIXED_LONGITUDE,
+                    qrt=source_spot["state"] != "active")
+        return spot