mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2025-10-27 16:59:25 +00:00
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
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.") |