spothole/telnetserver/telnetserver.py

import asyncio
import logging
import threading
from datetime import datetime

import pytz

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__)


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

    def start(self, port=7373):
        """Starts the telnet server"""

        self._port = port

        # Start telnet server on the async loop
        def run_loop():
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)
            self._loop.run_until_complete(self._start_internal())
            self._loop.run_forever()

        # Start the network thread as a daemon so it exits cleanly when the main script stops
        t = threading.Thread(target=run_loop, 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):
        logger.info(f"Telnet server listening on port {self._port}")
        await asyncio.start_server(self._handle_client, "0.0.0.0", self._port)

    async def _handle_client(self, reader, writer):
        """Handles a new client connection"""

        logger.debug("Telnet client connected")
        self._clients.add(writer)

        # Print MOTD
        try:
            motd = (
                f"Welcome to Spothole v{SOFTWARE_VERSION}.\r\nThis server is run by {SERVER_OWNER_CALLSIGN}.\r\n===\r\n"
            )
            writer.write(motd.encode("ascii"))
            await writer.drain()
        except Exception:
            pass

        # 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

                text = data.decode("ascii", errors="ignore")

                for char in text:
                    if char in ("\r", "\n"):
                        # User pressed Enter, evaluate the command
                        command = input_buffer.strip().lower()
                        input_buffer = ""

                        if command == "exit":
                            writer.write(b"Goodbye!\r\n")
                            await writer.drain()
                            # Exit the while read loop, this will disconnect the client.
                            return

                    elif char in ("\b", "\x7f"):
                        # Handle backspaces
                        input_buffer = input_buffer[:-1]
                    else:
                        input_buffer += char

        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"""

        if self._loop and self._loop.is_running():
            logger.debug("Stopping telnet server...")
            self._loop.call_soon_threadsafe(self._loop.stop)

    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
        disconnected_clients = set()
        for writer in 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 _format_dxspider_spot(spot: Spot) -> str:
        """Formats a spot into the format DXspider uses:
        DX de CALLSIGN:  FREQUENCY  DX_CALLSIGN  COMMENTS   TIME_UTC"""

        de_call = f"{spot.de_call + ':' if spot.de_call else '???:'!s:<9}"
        frequency = f"{(spot.freq / 1000.0):8.1f}"
        dx_call = f"{spot.dx_call!s:<12}"
        comment = f"{spot.comment[: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()