spothole/telnetserver/telnetserver.py

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._thread = 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.
        self._thread = threading.Thread(
            target=asyncio.run, args=(self._start_internal(),), name="TelnetServer", daemon=True
        )
        self._thread.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)
        if self._thread:
            self._thread.join(timeout=15)
            if self._thread.is_alive():
                logger.warning("Telnet server background thread did not exit on time and will be killed.")

    @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:
                # No problem, we were disconnecting this client anyway.
                logger.debug("Exception while closing telnet client connection", exc_info=True)
        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:
                logger.debug("Exception while writing to telnet client, assuming it disconnected", exc_info=True)
                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()