spothole/providers/spot/dxcluster.py

import logging
import re
from datetime import datetime

import pytz
import telnetlib3

from core.config import SERVER_OWNER_CALLSIGN
from core.utils import decode_telnet_bytes
from data.spot import Spot
from providers.spot.spot_provider import SpotSubmissionError
from providers.spot.telnet_spot_provider import TelnetSpotProvider

logger = logging.getLogger(__name__)


class DXCluster(TelnetSpotProvider):
    """Spot provider for a DX Cluster. Hostname, port, login_prompt, login_callsign and allow_rbn_spots are provided in config.
    See config-example.yml for examples."""

    _LINE_PATTERN_EXCLUDE_RBN = re.compile(
        r"^DX de ([a-z0-9/]+):\s+([0-9.]+)\s+([a-z0-9/]+)\s+(.*)\s+(\d{4}Z)",
        re.IGNORECASE,
    )
    _LINE_PATTERN_ALLOW_RBN = re.compile(
        r"^DX de ([a-z0-9/]+)-?#?:\s+([0-9.]+)\s+([a-z0-9/]+)\s+(.*)\s+(\d{4}Z)",
        re.IGNORECASE,
    )

    # Timeout for each step of the telnet conversation when submitting a spot
    _SUBMIT_TIMEOUT_SEC = 10

    def __init__(self, provider_config):
        """Constructor requires hostname and port"""

        name = provider_config.get("name", "Cluster")
        allow_rbn_spots = provider_config.get("allow_rbn_spots", False)
        self._spot_line_pattern = self._LINE_PATTERN_ALLOW_RBN if allow_rbn_spots else self._LINE_PATTERN_EXCLUDE_RBN
        super().__init__(
            name,
            provider_config,
            host=provider_config["host"],
            port=provider_config["port"],
            login_prompt=provider_config.get("login_prompt", "login:"),
            login_response=provider_config.get("login_callsign", SERVER_OWNER_CALLSIGN),
        )

    def _parse_line(self, line):
        match = self._spot_line_pattern.match(line)
        if not match:
            return None

        spot_time = datetime.strptime(match.group(5), "%H%MZ").replace(tzinfo=pytz.UTC)
        spot_datetime = datetime.combine(
            datetime.now(pytz.UTC).date(),
            spot_time.time(),
            tzinfo=pytz.UTC,
        )
        return Spot(
            source=self.name,
            dx_call=match.group(3),
            de_call=match.group(1),
            freq=float(match.group(2)) * 1000,
            comment=match.group(4).strip(),
            time=spot_datetime.timestamp(),
        )

    def can_submit_spot(self, activity):
        # Clusters accept spots for any activity, or for no activity at all.
        return True

    def submit_spot(self, spot, credentials):
        """Submit a spot to the cluster. This makes a new telnet connection to the same node that we receive spots from,
        logs in as the spotter, sends the spot using the DX command, then logs out. If anything goes wrong we raise an
        error which can then be logged by the caller and returned to the user as an error message in the API JSON / web
        UI."""

        command = self._build_dx_command(spot)
        telnet = telnetlib3.Telnet(self._host, self._port, timeout=self._SUBMIT_TIMEOUT_SEC)
        try:
            # Wait for login prompt and log in
            output = telnet.read_until(self._login_prompt.encode("latin-1"), timeout=self._SUBMIT_TIMEOUT_SEC)
            if self._login_prompt.encode("latin-1") not in output:
                logger.debug(f"{self.name} response while waiting for login prompt: {decode_telnet_bytes(output)}")
                raise SpotSubmissionError(f"{self.name} did not give the expected login prompt.")
            telnet.write(f"{spot.de_call}\n".encode("latin-1"))

            # Wait for the command prompt. This is a line like "M0TRT de W3LPL 27-Sep-2026 0945Z dxspider >". We can't
            # just wait for ">" as the welcome message might contain that character.
            call = re.escape(spot.de_call).encode("latin-1")
            prompt = re.compile(rb"(?im)^" + call + rb" de \S+.*>\s*$")
            index, _, output = telnet.expect([prompt], timeout=self._SUBMIT_TIMEOUT_SEC)
            if index < 0:
                logger.debug(f"{self.name} response while waiting for command prompt: {decode_telnet_bytes(output)}")
                raise SpotSubmissionError(f"{self.name} did not accept a login as {spot.de_call}.")

            # Send the spot.
            logger.debug(f"Submitting spot to {self.name} as {spot.de_call}: {command}")
            telnet.write(f"{command}\n".encode("latin-1"))

            # Log out
            telnet.write(b"bye\n")
        finally:
            telnet.close()

    @staticmethod
    def _build_dx_command(spot):
        """Build the "DX <freq kHz> <dx call> <comment>" command used to submit a spot to the cluster. The mode, and
        any activities and references e.g. "SSB POTA GB-1234 tnx 73"."""

        freq = spot.freq
        if not freq or not spot.dx_call:
            raise ValueError("A frequency and DX callsign are required to submit a spot to a cluster.")

        comment_parts = []
        # Mode
        if spot.mode:
            comment_parts.append(spot.mode)
        # Build list of activities with refs like "POTA GB-0001 SOTA G/SC-001"
        activities_with_refs = set()
        for activity_ref in spot.activity_refs:
            if activity_ref.id:
                comment_parts.append(f"{activity_ref.activity} {activity_ref.id}")
                activities_with_refs.add(activity_ref.activity)
        # If we have activities that don't have refs, add them to the comment
        comment_parts.extend(str(a) for a in spot.activities if a not in activities_with_refs)
        # Add the actual spot comment
        if spot.comment:
            comment_parts.append(spot.comment)

        # Build command
        command = f"DX {freq / 1000.0:.1f} {spot.dx_call} {' '.join(comment_parts)}"

        # Strip anything that isn't printable ASCII, so that e.g. a newline in the comment can't be used to send
        # exploit the system and send additional commands to the cluster as the spotter.
        return re.sub(r"\s+", " ", re.sub(r"[^\x20-\x7E]", " ", command)).strip()