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