device_tracker.py

Go to the documentation of this file.

"""Support for APRS device tracking."""
 
from __future__ import annotations
 
import logging
import threading
from typing import Any
 
import aprslib
from aprslib import ConnectionError as AprsConnectionError, LoginError
import geopy.distance
import voluptuous as vol
 
from homeassistant.components.device_tracker import (
    PLATFORM_SCHEMA as DEVICE_TRACKER_PLATFORM_SCHEMA,
    SeeCallback,
)
from homeassistant.const import (
    ATTR_GPS_ACCURACY,
    ATTR_LATITUDE,
    ATTR_LONGITUDE,
    CONF_HOST,
    CONF_PASSWORD,
    CONF_TIMEOUT,
    CONF_USERNAME,
    EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import Event, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import slugify
 
DOMAIN = "aprs"
 
_LOGGER = logging.getLogger(__name__)
 
ATTR_ALTITUDE = "altitude"
ATTR_COURSE = "course"
ATTR_COMMENT = "comment"
ATTR_FROM = "from"
ATTR_FORMAT = "format"
ATTR_OBJECT_NAME = "object_name"
ATTR_POS_AMBIGUITY = "posambiguity"
ATTR_SPEED = "speed"
 
CONF_CALLSIGNS = "callsigns"
 
DEFAULT_HOST = "rotate.aprs2.net"
DEFAULT_PASSWORD = "-1"
DEFAULT_TIMEOUT = 30.0
 
FILTER_PORT = 14580
 
MSG_FORMATS = ["compressed", "uncompressed", "mic-e", "object"]
 
PLATFORM_SCHEMA = DEVICE_TRACKER_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_CALLSIGNS): cv.ensure_list,
        vol.Required(CONF_USERNAME): cv.string,
        vol.Optional(CONF_PASSWORD, default=DEFAULT_PASSWORD): cv.string,
        vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
        vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): vol.Coerce(float),
    }
)
 
 
def make_filter(callsigns: list) -> str:
    """Make a server-side filter from a list of callsigns."""
    return " ".join(f"b/{sign.upper()}" for sign in callsigns)
 
 
def gps_accuracy(gps: tuple[float, float], posambiguity: int) -> int:
    """Calculate the GPS accuracy based on APRS posambiguity."""
 
    pos_a_map = {0: 0, 1: 1 / 600, 2: 1 / 60, 3: 1 / 6, 4: 1}
    if posambiguity in pos_a_map:
        degrees = pos_a_map[posambiguity]
 
        gps2 = (gps[0], gps[1] + degrees)
        dist_m: float = geopy.distance.distance(gps, gps2).m
 
        accuracy = round(dist_m)
    else:
        message = f"APRS position ambiguity must be 0-4, not '{posambiguity}'."
        raise ValueError(message)
 
    return accuracy
 
 
def setup_scanner(
    hass: HomeAssistant,
    config: ConfigType,
    see: SeeCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> bool:
    """Set up the APRS tracker."""
    callsigns = config[CONF_CALLSIGNS]
    server_filter = make_filter(callsigns)
 
    callsign = config[CONF_USERNAME]
    password = config[CONF_PASSWORD]
    host = config[CONF_HOST]
    timeout = config[CONF_TIMEOUT]
    aprs_listener = AprsListenerThread(callsign, password, host, server_filter, see)
 
    def aprs_disconnect(event: Event) -> None:
        """Stop the APRS connection."""
        aprs_listener.stop()
 
    aprs_listener.start()
    hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, aprs_disconnect)
 
    if not aprs_listener.start_event.wait(timeout):
        _LOGGER.error("Timeout waiting for APRS to connect")
        return False
 
    if not aprs_listener.start_success:
        _LOGGER.error(aprs_listener.start_message)
        return False
 
    _LOGGER.debug(aprs_listener.start_message)
    return True
 
 
class AprsListenerThread(threading.Thread):
    """APRS message listener."""
 
    def __init__(
        self,
        callsign: str,
        password: str,
        host: str,
        server_filter: str,
        see: SeeCallback,
    ) -> None:
        """Initialize the class."""
        super().__init__()
 
        self.callsigncallsign = callsign
        self.hosthost = host
        self.start_eventstart_event = threading.Event()
        self.seesee = see
        self.server_filterserver_filter = server_filter
        self.start_messagestart_message = ""
        self.start_successstart_success = False
 
        self.aisais = aprslib.IS(
            self.callsigncallsign, passwd=password, host=self.hosthost, port=FILTER_PORT
        )
 
    def start_complete(self, success: bool, message: str) -> None:
        """Complete startup process."""
        self.start_messagestart_message = message
        self.start_successstart_success = success
        self.start_eventstart_event.set()
 
    def run(self) -> None:
        """Connect to APRS and listen for data."""
        self.aisais.set_filter(self.server_filterserver_filter)
 
        try:
            _LOGGER.debug(
                "Opening connection to %s with callsign %s", self.hosthost, self.callsigncallsign
            )
            self.aisais.connect()
            self.start_completestart_complete(
                True, f"Connected to {self.host} with callsign {self.callsign}."
            )
            self.aisais.consumer(callback=self.rx_msgrx_msg, immortal=True)
        except (AprsConnectionError, LoginError) as err:
            self.start_completestart_complete(False, str(err))
        except OSError:
            _LOGGER.debug(
                "Closing connection to %s with callsign %s", self.hosthost, self.callsigncallsign
            )
 
    def stop(self) -> None:
        """Close the connection to the APRS network."""
        self.aisais.close()
 
    def rx_msg(self, msg: dict[str, Any]) -> None:
        """Receive message and process if position."""
        _LOGGER.debug("APRS message received: %s", str(msg))
        if msg[ATTR_FORMAT] in MSG_FORMATS:
            if msg[ATTR_FORMAT] == "object":
                dev_id = slugify(msg[ATTR_OBJECT_NAME])
            else:
                dev_id = slugify(msg[ATTR_FROM])
            lat = msg[ATTR_LATITUDE]
            lon = msg[ATTR_LONGITUDE]
 
            attrs = {}
            if ATTR_POS_AMBIGUITY in msg:
                pos_amb = msg[ATTR_POS_AMBIGUITY]
                try:
                    attrs[ATTR_GPS_ACCURACY] = gps_accuracy((lat, lon), pos_amb)
                except ValueError:
                    _LOGGER.warning(
                        "APRS message contained invalid posambiguity: %s", str(pos_amb)
                    )
            for attr in (ATTR_ALTITUDE, ATTR_COMMENT, ATTR_COURSE, ATTR_SPEED):
                if attr in msg:
                    attrs[attr] = msg[attr]
 
            self.seesee(dev_id=dev_id, gps=(lat, lon), attributes=attrs)