Add APRS-IS support

README.md

@@ -14,10 +14,7 @@ Currently supports:
 * UKBOTA
 * Parks n Peaks
 * RBN
-
+* APRS
-Future plans:
-* APRS?
-* Packet?
 
 Suggested names so far:
 * All in 1 Spots

config-example.yml

@@ -19,6 +19,7 @@ providers:
   - type: "GMA"
   - type: "HEMA"
   - type: "ParksNPeaks"
+#  - type: "APRS-IS"
 # Some, like DX Clusters, require extra config. You can add multiple DX clusters if you want!
   -
     type: "DXCluster"
@@ -31,7 +32,6 @@ providers:
 #  -
 #    type: "RBN"
 #    port: 7001
-#  - type: "APRS-IS"
 
 # Port to open the local web server on
 web-server-port: 8080

data/spot.py

@@ -40,6 +40,9 @@ class Spot:
     dx_cq_zone: int = None
     # ITU zone of the DX operator
     dx_itu_zone: int = None
+    # If this is an APRS spot, what SSID was the DX operator using?
+    # This is a string not an int for now, as I often see non-numeric ones somehow
+    dx_aprs_ssid: str = None
     # Reported mode, such as SSB, PHONE, CW, FT8...
     mode: str = None
     # Inferred mode "family". One of "CW", "PHONE" or "DIGI".

main.py

@@ -6,6 +6,7 @@ from time import sleep
 
 from core.cleanup import CleanupTimer
 from core.config import config
+from providers.aprsis import APRSIS
 from providers.dxcluster import DXCluster
 from providers.gma import GMA
 from providers.hema import HEMA
@@ -17,7 +18,11 @@ from providers.wwbota import WWBOTA
 from providers.wwff import WWFF
 from server.webserver import WebServer
 
-# Main control flag, switch False to stop main application thread
+# Globals
+spot_list = []
+status_data = {}
+providers = []
+cleanup_timer = None
 run = True
 
 # Shutdown function
@@ -51,6 +56,8 @@ def get_provider_from_config(config_providers_entry):
             return DXCluster(config_providers_entry["host"], config_providers_entry["port"])
         case "RBN":
             return RBN(config_providers_entry["port"])
+        case "APRS-IS":
+            return APRSIS()
     return None
 
 
@@ -69,12 +76,6 @@ if __name__ == '__main__':
     # Shut down gracefully on SIGINT
     signal.signal(signal.SIGINT, shutdown)
 
-    # Set up spot list & status data areas
-    spot_list = []
-    status_data = {}
-
-    # Create data providers
-    providers = []
     for entry in config["providers"]:
         providers.append(get_provider_from_config(entry))
     # Set up data providers

providers/aprsis.py

@@ -0,0 +1,62 @@
+import logging
+from datetime import datetime, timezone
+from threading import Thread
+
+import aprslib
+import pytz
+
+from core.config import config
+from data.spot import Spot
+from providers.provider import Provider
+
+
+# Provider for the APRS-IS.
+class APRSIS(Provider):
+
+    def __init__(self):
+        super().__init__()
+        self.thread = Thread(target=self.connect)
+        self.thread.daemon = True
+        self.aprsis = None
+
+    def name(self):
+        return "APRS-IS"
+
+    def start(self):
+        self.thread.start()
+
+    def connect(self):
+        self.aprsis = aprslib.IS(config["server-owner-callsign"])
+        self.status = "Connecting"
+        logging.info("APRS-IS connecting...")
+        self.aprsis.connect()
+        self.aprsis.consumer(self.handle)
+        logging.info("APRS-IS connected.")
+
+    def stop(self):
+        self.status = "Shutting down"
+        self.aprsis.close()
+        self.thread.join()
+
+    def handle(self, data):
+        # Split SSID in "from" call and store separately
+        from_parts = data["from"].split("-")
+        dx_call = from_parts[0]
+        dx_aprs_ssid = from_parts[1] if len(from_parts) > 1 else None
+        spot = Spot(source="APRS-IS",
+                    dx_call=dx_call,
+                    dx_aprs_ssid=dx_aprs_ssid,
+                    de_call=data["via"],
+                    comment=data["comment"] if "comment" in data else None,
+                    latitude=data["latitude"] if "latitude" in data else None,
+                    longitude=data["longitude"] if "longitude" in data else None,
+                    time=datetime.now(pytz.UTC)) # APRS-IS spots are live so we can assume spot time is "now"
+        # Fill in any blanks
+        spot.infer_missing()
+        # Add to our list
+        self.submit(spot)
+        print(spot)
+
+        self.status = "OK"
+        self.last_update_time = datetime.now(timezone.utc)
+        logging.debug("Data received from APRS-IS.")

requirements.txt

@@ -5,3 +5,4 @@ pyhamtools~=0.12.0
 telnetlib3~=2.0.8
 pytz~=2025.2
 requests~=2.32.5
+aprslib~=0.7.2