73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Awaitable, Callable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_KEEPALIVE_INTERVAL = 60
|
|
_RECONNECT_DELAY = 30
|
|
|
|
|
|
async def run_aprs_is_collector(
|
|
host: str,
|
|
port: int,
|
|
callsign: str,
|
|
passcode: str,
|
|
filter_str: str,
|
|
on_frame: Callable[..., Awaitable[None]],
|
|
) -> None:
|
|
writer = None
|
|
while True:
|
|
try:
|
|
logger.info("APRS-IS: connecting to %s:%d", host, port)
|
|
reader, writer = await asyncio.open_connection(host, port)
|
|
|
|
banner = await asyncio.wait_for(reader.readline(), timeout=15)
|
|
logger.info("APRS-IS: %s", banner.decode(errors="replace").strip())
|
|
|
|
login_line = (
|
|
f"user {callsign} pass {passcode} "
|
|
f"vers aprs-collector 0.1 "
|
|
f"filter {filter_str}\r\n"
|
|
)
|
|
writer.write(login_line.encode())
|
|
await writer.drain()
|
|
|
|
ack = await asyncio.wait_for(reader.readline(), timeout=15)
|
|
logger.info("APRS-IS login: %s", ack.decode(errors="replace").strip())
|
|
|
|
async def _keepalive() -> None:
|
|
while True:
|
|
await asyncio.sleep(_KEEPALIVE_INTERVAL)
|
|
writer.write(b"#ping\r\n")
|
|
await writer.drain()
|
|
|
|
ka_task = asyncio.create_task(_keepalive())
|
|
try:
|
|
while True:
|
|
line = await reader.readline()
|
|
if not line:
|
|
logger.warning("APRS-IS: server closed connection")
|
|
break
|
|
decoded = line.decode("utf-8", errors="replace").strip()
|
|
if not decoded or decoded.startswith("#"):
|
|
continue
|
|
await on_frame(ts=datetime.now(timezone.utc), raw=decoded)
|
|
finally:
|
|
ka_task.cancel()
|
|
|
|
except asyncio.TimeoutError:
|
|
logger.warning("APRS-IS: timeout during handshake")
|
|
except Exception as exc:
|
|
logger.error("APRS-IS: %s", exc)
|
|
finally:
|
|
if writer:
|
|
try:
|
|
writer.close()
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info("APRS-IS: reconnecting in %ds...", _RECONNECT_DELAY)
|
|
await asyncio.sleep(_RECONNECT_DELAY)
|