spothole/providers/spot/dxcluster.py

import logging
import re
import socket
from datetime import datetime
from threading import Event, Lock, Thread

import pytz
import telnetlib3

from core.config import SERVER_OWNER_CALLSIGN
from data.spot import Spot
from providers.spot.spot_provider import SpotProvider

logger = logging.getLogger(__name__)


class DXCluster(SpotProvider):
    """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,
    )

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

        name = provider_config.get("name", "Cluster")
        super().__init__(name, provider_config)
        self._hostname = provider_config["host"]
        self._port = provider_config["port"]
        self._login_prompt = provider_config.get("login_prompt", "login:")
        self._login_callsign = provider_config.get("login_callsign", SERVER_OWNER_CALLSIGN)
        self._allow_rbn_spots = provider_config.get("allow_rbn_spots", False)
        self._spot_line_pattern = (
            self._LINE_PATTERN_ALLOW_RBN if self._allow_rbn_spots else self._LINE_PATTERN_EXCLUDE_RBN
        )
        self._telnet = None
        self._telnet_lock = Lock()
        self._thread = None
        self._stop_event = Event()

    def start(self):
        self._thread = Thread(target=self._handle, name=f"DXClusterSpotProvider-{self.name}", daemon=True)
        self._thread.start()

    def stop(self):
        self._stop_event.set()
        with self._telnet_lock:
            if self._telnet:
                try:
                    self._telnet.sock.shutdown(socket.SHUT_RDWR)
                except (AttributeError, OSError):
                    pass
                self._telnet.close()
        if self._thread:
            self._thread.join(timeout=5)
            if self._thread.is_alive():
                logger.warning(f"DX Cluster {self._hostname} worker thread did not exit on time and will be killed.")

    def _handle(self):
        while not self._stop_event.is_set():
            connected = False
            while not connected and not self._stop_event.is_set():
                try:
                    self.status = "Connecting"
                    logger.info(f"DX Cluster {self._hostname} connecting...")
                    new_telnet = telnetlib3.Telnet(self._hostname, self._port)
                    with self._telnet_lock:
                        self._telnet = new_telnet
                        if self._stop_event.is_set():
                            # stop() was called while we were connecting, close the connection rather than trying to
                            # read when we know it won't work
                            new_telnet.close()
                            break
                    self._telnet.read_until(self._login_prompt.encode("latin-1"))
                    self._telnet.write(f"{self._login_callsign}\n".encode("latin-1"))
                    connected = True
                    logger.info(f"DX Cluster {self._hostname} connected.")
                except ConnectionRefusedError:
                    self.status = "Error"
                    logger.warning(f"Connection refused to DX cluster {self._hostname}")
                    self._stop_event.wait(timeout=300)
                except Exception:
                    self.status = "Error"
                    logger.exception(f"Exception while connecting to DX Cluster Provider ({self._hostname}).")
                    self._stop_event.wait(timeout=5)

            self.status = "Waiting for Data"
            while connected and not self._stop_event.is_set():
                try:
                    # Check new telnet info against regular expression
                    telnet_output = self._telnet.read_until("\n".encode("latin-1"))
                    match = self._spot_line_pattern.match(telnet_output.decode("latin-1"))
                    if match:
                        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,
                        )
                        spot = 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(),
                        )

                        # Add to our list
                        self._submit(spot)

                        self.status = "OK"
                        self.last_update_time = datetime.now(pytz.UTC)
                        logger.debug(f"Data received from DX Cluster {self._hostname}.")

                except EOFError:
                    connected = False
                    if not self._stop_event.is_set():
                        self.status = "Restarting"
                        logger.warning(f"Disconnected from DX Cluster {self._hostname}. Reconnecting...")
                        self._stop_event.wait(timeout=5)
                    else:
                        logger.info(f"DX Cluster {self._hostname} shutting down...")
                        self.status = "Shutting down"
                except Exception:
                    connected = False
                    if not self._stop_event.is_set():
                        self.status = "Error"
                        logger.exception(f"Exception in DX Cluster Provider ({self._hostname})")
                        self._stop_event.wait(timeout=5)
                    else:
                        logger.info(f"DX Cluster {self._hostname} shutting down...")
                        self.status = "Shutting down"

            self.status = "Disconnected"