import asyncio import logging import threading from datetime import datetime import pytz from pyhamtools import callinfo from core.config import SERVER_OWNER_CALLSIGN from core.constants import SOFTWARE_VERSION from core.data_store import DATA_STORE from data.spot import Spot logger = logging.getLogger(__name__) BANNER = ( """ == ==== ##### ##### #### == == ..### ..### ..### == ###==######## ###### ####### .####### ###### .### ###### ==###.. ..###..### ###..###...###. .###..### ###..### .### ###..### ==###.### .### .###.### .### .### .### .### .### .### .### .####### == .###.###.### .###.### .### .### ### .### .### .### .### .### .###... == ...### .####### ..###### ..##### #### #####..###### #####..###### == .###. .###=.. ...... ..... .... ..... ...... ..... ...... == %% ... %%%%###== == %%%%%%%%%%%#####== \r\n""" + "== ..... ==" + f"Welcome to Spothole v{SOFTWARE_VERSION}".rjust(56) + "\r\n" + "========================" + f"This server is run by {SERVER_OWNER_CALLSIGN}".rjust(56) ) MOTD = ( "Spothole's telnet server is a new feature and may not work properly in all\r\n" + "loggers. Please give it a try in your logger of choice and let me know if it\r\n" + "(or doesn't!) Please note that DXSpider-like commands are not yet supported\r\n" + "so you are not yet able to filter spots server-side or log in at all." ) class TelnetServer: """A telnet server designed to provide spots in the same format as DXSpider, for compatibility with desktop loggers.""" def __init__(self): self._port = None self._running = False self._clients = set() self._loop = None self._shutdown_event = asyncio.Event() def start(self, port=7373): """Starts the telnet server""" self._port = port # Start the telnet server. asyncio.run() needs a coroutine, and threading.Thread needs a plain callable, so # hand Thread the bridge between the two directly rather than writing a one-line wrapper method for it. t = threading.Thread(target=asyncio.run, args=(self._start_internal(),), name="TelnetServer", daemon=True) t.start() logger.debug("Telnet server background thread spawned") # Listen for new spots and alerts being added to the cache, so we can notify SSE clients immediately DATA_STORE.spots.add_listener(self.publish) self._running = True async def _start_internal(self): """Start method (async). Sets up the telnet server and waits for shutdown.""" self._loop = asyncio.get_running_loop() server = await asyncio.start_server(self._handle_client, "0.0.0.0", self._port) logger.info(f"Telnet server listening on port {self._port}") async with server: await self._shutdown_event.wait() await self._stop_internal() async def _handle_client(self, reader, writer): """Handles a new client connection""" logger.debug("Telnet client connected") self._clients.add(writer) # Print banner and MOTD try: text = ( BANNER + "\r\n\r\n" + "================================================================================" + "\r\n" + MOTD + "\r\n" + "================================================================================" + "\r\n\r\n" ) writer.write(text.encode("ascii")) await writer.drain() except Exception: logger.exception("Exception printing telnet motd") # Set up buffer for user input input_buffer = "" try: # Read forever, picking out any commands. Currently we just support "exit" while True: data = await reader.read(1024) if not data: break input_buffer, command = self._consume_input(input_buffer, data) if command == "exit": writer.write(b"Goodbye!\r\n") await writer.drain() # Exit the while read loop, this will disconnect the client. return except asyncio.CancelledError: pass except Exception: logger.exception("Exception handling telnet client") finally: logger.debug("Telnet client disconnected") self._clients.remove(writer) writer.close() await writer.wait_closed() def stop(self): """Stops the telnet server""" self._running = False if self._loop and self._loop.is_running(): logger.debug("Stopping telnet server...") self._loop.call_soon_threadsafe(self._shutdown_event.set) @property def client_count(self) -> int: return len(self._clients) async def _stop_internal(self): """Stops the telnet server""" for writer in list(self._clients): try: writer.close() await writer.wait_closed() except Exception: pass self._clients.clear() def publish(self, spot: Spot): """Callback from the data store when a spot is added""" if self._running and self._clients and self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._broadcast_spot_internal(spot), self._loop) async def _broadcast_spot_internal(self, spot: Spot): """Internal version, run on async loop for thread safety?""" # Ensure ASCII formatting for telnet clients encoded_line = self._format_dxspider_spot(spot).encode("ascii", errors="ignore") # Try to write to all clients, and in the process find the ones that are disconnected. Iterate over a copy # since a new client can connect (mutating self._clients) while we're awaiting a write below. disconnected_clients = set() for writer in list(self._clients): try: writer.write(encoded_line) await writer.drain() except Exception: disconnected_clients.add(writer) # Clean up any disconnected connections caught during writing for writer in disconnected_clients: self._clients.discard(writer) @staticmethod def _consume_input(input_buffer: str, data: bytes) -> tuple[str, str | None]: """Handle any input the user gives us, keeping a rolling buffer that we keep passing back through and adding to. Once we get a command, return that as well, so the caller can deal with it.""" command = None for char in data.decode("ascii", errors="ignore"): if char in ("\r", "\n"): stripped = input_buffer.strip().lower() if stripped: command = stripped input_buffer = "" elif char in ("\b", "\x7f"): # Handle backspaces input_buffer = input_buffer[:-1] else: input_buffer += char return input_buffer, command @staticmethod def _format_dxspider_spot(spot: Spot) -> str: """Formats a spot into the format DXspider uses: DX de CALLSIGN: FREQUENCY DX_CALLSIGN COMMENTS TIME_UTC. Always use the base call for the spotter to save space. Everything must align properly to parse in clients, and everything must be renderable in ASCII.""" de_call = f"{callinfo.Callinfo.get_homecall(spot.de_call)[:6] + ':' if spot.de_call else '???:'!s:<7}" frequency = f"{(spot.freq / 1000.0):10.1f}" dx_call = f"{spot.dx_call!s:<12}" comment = f"{spot.comment.encode('ascii', errors='ignore').decode()[:29]:<30}" if spot.time: timestamp = datetime.fromtimestamp(spot.time, tz=pytz.utc).strftime("%H%M") + "Z" else: timestamp = datetime.now(tz=pytz.utc).strftime("%H%M") + "Z" # Combine into classic DXSpider output string followed by network line breaks return f"DX de {de_call} {frequency} {dx_call} {comment} {timestamp}\r\n" # Global object TELNET_SERVER = TelnetServer()