spothole/providers/spot/dxcluster.py

import logging
import re
from datetime import datetime

import pytz

from core.config import SERVER_OWNER_CALLSIGN
from data.spot import Spot
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,
    )

    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(),
        )