w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-20 14:27:42 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Code review fixes

Browse Source
This commit is contained in:
Ian Renton
2026-09-18 07:56:08 +01:00
parent 0fa8cd763d
commit a03e1336c8
18 changed files with 113 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/spot/aprsis.py
+52 -33
View File
@@ -1,6 +1,6 @@
import logging
from datetime import datetime
from threading import Thread
from threading import Event, Thread
import aprslib
import pytz
@@ -17,49 +17,68 @@ class APRSIS(SpotProvider):
def __init__(self, provider_config):
super().__init__("APRS-IS", provider_config)
self._thread = Thread(target=self._connect, name="APRSISSpotProvider")
self._thread = Thread(target=self._run, name="APRSISSpotProvider")
self._thread.daemon = True
self._aprsis = None
self._running = True
self._stop_event = Event()
def start(self):
self._thread.start()
def _connect(self):
self._aprsis = aprslib.IS(SERVER_OWNER_CALLSIGN)
self.status = "Connecting"
logger.info("APRS-IS connecting...")
self._aprsis.connect()
self._aprsis.consumer(self._handle)
logger.info("APRS-IS connected.")
def _run(self):
while self._running:
try:
self._aprsis = aprslib.IS(SERVER_OWNER_CALLSIGN)
self.status = "Connecting"
logger.info("APRS-IS connecting...")
self._aprsis.connect()
logger.info("APRS-IS connected.")
self._aprsis.consumer(self._handle, immortal=True)
except Exception:
if self._running:
self.status = "Error"
logger.exception("Exception in APRS-IS provider")
if self._running:
self._stop_event.wait(timeout=5)
def stop(self):
self._running = False
self.status = "Shutting down"
self._aprsis.close()
self._stop_event.set()
if self._aprsis:
self._aprsis.close()
self._thread.join()
def _handle(self, data):
# Split SSID in "from" call and store separately
from_parts = str(data["from"]).split("-")
dx_call = from_parts[0].upper()
dx_ssid = from_parts[1].upper() if len(from_parts) > 1 else None
via_parts = str(data["via"]).split("-")
de_call = via_parts[0].upper()
de_ssid = via_parts[1].upper() if len(via_parts) > 1 else None
spot = Spot(
source="APRS-IS",
dx_call=dx_call,
dx_ssid=dx_ssid,
de_call=de_call,
de_ssid=de_ssid,
comment=str(data["comment"]) if "comment" in data else None,
dx_latitude=float(data["latitude"]) if "latitude" in data else None,
dx_longitude=float(data["longitude"]) if "longitude" in data else None,
time=datetime.now(pytz.UTC).timestamp(),
) # APRS-IS spots are live so we can assume spot time is "now"
try:
# Split SSID in "from" call and store separately
from_parts = str(data["from"]).split("-")
dx_call = from_parts[0].upper()
dx_ssid = from_parts[1].upper() if len(from_parts) > 1 else None
via_parts = str(data["via"]).split("-")
de_call = via_parts[0].upper()
de_ssid = via_parts[1].upper() if len(via_parts) > 1 else None
spot = Spot(
source="APRS-IS",
dx_call=dx_call,
dx_ssid=dx_ssid,
de_call=de_call,
de_ssid=de_ssid,
comment=str(data["comment"]) if "comment" in data else None,
dx_latitude=float(data["latitude"]) if "latitude" in data else None,
dx_longitude=float(data["longitude"]) if "longitude" in data else None,
time=datetime.now(pytz.UTC).timestamp(),
) # APRS-IS spots are live so we can assume spot time is "now"
# Add to our list
self._submit(spot)
# Add to our list
self._submit(spot)
self.status = "OK"
self.last_update_time = datetime.now(pytz.UTC)
logger.debug("Data received from APRS-IS.")
self.status = "OK"
self.last_update_time = datetime.now(pytz.UTC)
logger.debug("Data received from APRS-IS.")
except Exception:
logger.exception("Exception handling APRS-IS packet")