config.py

from __future__ import annotations

import ast
import importlib.util
import logging
import pprint
import sys
from dataclasses import dataclass
from pathlib import Path

from aprs.agwpe import AGWPEConfig
from aprs.agwpe_server import AGWPEServerConfig
from aprs.is_server import APRSISServerConfig
from aprs.kiss import KISSConfig
from cot.cot import CoTConfig
from gui import GuiConfig
from mesh.adapter import MeshtasticConfig
from sartrack.aprs_server_client import SARTrackAPRSServerConfig
from sartrack.tcp_server import SARTrackTCPConfig


# Edit the values below when you want to change how the gateway connects.
# This file is the single, human-friendly place for IPs, ports, names,
# callsigns and feature switches used by the CLI start path.

AGWPE = {'enabled': False,
 'host': '127.0.0.1',
 'port': 8000,
 'channel': 0,
 'reconnect_delay': 3.0,
 'rx_monitor_enabled': False,
 'tx_callsign': 'N2CALL-10',
 'tx_destination': 'APSAR',
 'tx_path': 'WIDE2-2'}

AGWPE_SERVER = {'enabled': False,
 'host': '0.0.0.0',
 'port': 8002,
 'channel': 0,
 'rx_monitor_enabled': True,
 'tx_send_monitor_t': True,
 'tx_send_sartrack_v': True,
 'tx_callsign': 'N2CALL-10',
 'tx_destination': 'APSAR',
 'tx_path': 'WIDE2-2'}

APRS_IS_SERVER = {'enabled': True, 'host': '0.0.0.0', 'port': 10152, 'server_name': 'Ham Router'}

KISS = {
    "host": "127.0.0.1",
    "port": 8001,
}

SARTRACK_TCP = {'enabled': False, 'host': '0.0.0.0', 'port': 8005}

SARTRACK_APRS = {"enabled": False, "host": "127.0.0.1", "port": 10152, "reconnect_delay": 5.0}

COT = {'enabled': True,
 'cot_url': 'udp://127.0.0.1:4242',
 'rx_enabled': True,
 'listen_url': 'udp://0.0.0.0:4243',
 'chat_listen_url': 'udp://224.10.10.1:17012',
 'pytak_no_hello': True,
 'tak_proto': 0,
 'stale_seconds': 300}

GUI = {
    "enabled": True,
    "host": "127.0.0.1",
    "port": 8080,
    "title": "Gateway Ops Panel",
}

LOGGING = {
    "level": "DEBUG",
    "enabled_loggers": {
        "main": True,
        "router": True,
        "status": True,
        "gui": False,
        "agwpe": True,
        "agwpe.server": False,
        "aprs.parser": True,
        "sartrack.tcp": False,
        "meshtastic": True,
        "meshtastic.mesh_interface": False,
        "meshtastic.stream_interface": False,
        "meshtastic.tcp_interface": False,
        "cot": True,
        "kiss": False,
    },
    "disabled_exact_loggers": [
        "meshtastic",
    ],
}

DEMO = {
    "demo_enabled": False,
    "demo_delay_seconds": 10.0,
    "demo_callsign": "N2CALL-10",
    "demo_lat": 50.06143,
    "demo_lon": 19.93658,
    "demo_position_message": "demo position",
    "demo_message_target": "N0CALL-10",
    "demo_text_message": "demo message to mesh",
}

MESH_CALLSIGN_ALIASES = {}

MESH_ADAPTERS = [{'name': 'mesh4405',
  'enabled': True,
  'host': '127.0.0.1',
  'port': 4404,
  'destination_id': '^all',
  'hop_limit': None,
  'aprs_channel_index': 0,
  'transport_mode': 'tak_packet',
  'position_transport_mode': 'tak_packet',
  'tak_rx_enabled': True,
  'tak_tx_enabled': True,
  'tak_detail_enabled': True,
  'tak_detail_compress': True,
  'tak_detail_compress_threshold': 180,
  'waypoint_expire_seconds': 1800,
  'waypoint_icon': 0,
  'outbound_echo_ttl_seconds': 30,
  'connect_quiet_seconds': 2.5,
  'read_timeout_seconds': 2.0,
  'reconnect_backoff_seconds': (2.0, 5.0, 10.0)}]


_PERSISTED_CONFIG_NAMES = (
    "AGWPE",
    "AGWPE_SERVER",
    "APRS_IS_SERVER",
    "KISS",
    "SARTRACK_TCP",
    "SARTRACK_APRS",
    "COT",
    "GUI",
    "LOGGING",
    "DEMO",
    "MESH_CALLSIGN_ALIASES",
    "MESH_ADAPTERS",
)


@dataclass(slots=True)
class DemoConfig:
    demo_enabled: bool
    demo_delay_seconds: float
    demo_callsign: str
    demo_lat: float
    demo_lon: float
    demo_position_message: str
    demo_message_target: str
    demo_text_message: str


@dataclass(slots=True)
class GatewayConfig:
    agwpe: AGWPEConfig
    agwpe_server: AGWPEServerConfig
    aprs_is_server: APRSISServerConfig
    kiss: KISSConfig
    sartrack: SARTrackTCPConfig
    sartrack_aprs: SARTrackAPRSServerConfig
    cot: CoTConfig
    gui: GuiConfig
    demo: DemoConfig
    mesh_adapters: list[MeshtasticConfig]
    mesh_callsign_aliases: dict[str, str]
    log_level: int
    disabled_logger_prefixes: list[str]
    disabled_logger_names: list[str]


def load_config() -> GatewayConfig:
    _apply_runtime_overrides()

    mesh_adapters = [
        MeshtasticConfig(
            name=str(item["name"]),
            host=str(item["host"]),
            port=int(item["port"]),
            enabled=bool(item["enabled"]),
            destination_id=str(item.get("destination_id", "^all")),
            hop_limit=item.get("hop_limit"),
            aprs_channel_index=int(item.get("aprs_channel_index", 0)),
            transport_mode=str(item.get("transport_mode", item.get("position_transport_mode", "tak_packet"))),
            position_transport_mode=str(item.get("position_transport_mode", "waypoint")),
            tak_rx_enabled=bool(item.get("tak_rx_enabled", True)),
            tak_tx_enabled=bool(item.get("tak_tx_enabled", True)),
            tak_detail_enabled=bool(item.get("tak_detail_enabled", True)),
            tak_detail_compress=bool(item.get("tak_detail_compress", True)),
            tak_detail_compress_threshold=int(item.get("tak_detail_compress_threshold", 180)),
            waypoint_expire_seconds=int(item.get("waypoint_expire_seconds", 1800)),
            waypoint_icon=int(item.get("waypoint_icon", 0)),
            outbound_echo_ttl_seconds=int(item.get("outbound_echo_ttl_seconds", 30)),
            connect_quiet_seconds=float(item.get("connect_quiet_seconds", 2.5)),
            read_timeout_seconds=float(item.get("read_timeout_seconds", 2.0)),
            reconnect_backoff_seconds=tuple(item.get("reconnect_backoff_seconds", (2.0, 5.0, 10.0))),
            callsign_aliases=dict(MESH_CALLSIGN_ALIASES),
        )
        for item in MESH_ADAPTERS
    ]

    return GatewayConfig(
        agwpe=AGWPEConfig(
            enabled=bool(AGWPE["enabled"]),
            host=str(AGWPE["host"]),
            port=int(AGWPE["port"]),
            channel=int(AGWPE["channel"]),
            reconnect_delay=float(AGWPE["reconnect_delay"]),
            rx_monitor_enabled=bool(AGWPE.get("rx_monitor_enabled", True)),
            tx_callsign=str(AGWPE["tx_callsign"]),
            tx_destination=str(AGWPE.get("tx_destination", "APSAR")),
            tx_path=str(AGWPE.get("tx_path", "WIDE2-2")),
        ),
        agwpe_server=AGWPEServerConfig(
            enabled=bool(AGWPE_SERVER["enabled"]),
            host=str(AGWPE_SERVER["host"]),
            port=int(AGWPE_SERVER["port"]),
            channel=int(AGWPE_SERVER["channel"]),
            rx_monitor_enabled=bool(AGWPE_SERVER.get("rx_monitor_enabled", True)),
            tx_send_monitor_t=bool(AGWPE_SERVER.get("tx_send_monitor_t", True)),
            tx_send_sartrack_v=bool(AGWPE_SERVER.get("tx_send_sartrack_v", True)),
            tx_callsign=str(AGWPE_SERVER["tx_callsign"]),
            tx_destination=str(AGWPE_SERVER["tx_destination"]),
            tx_path=str(AGWPE_SERVER["tx_path"]),
        ),
        aprs_is_server=APRSISServerConfig(
            enabled=bool(APRS_IS_SERVER["enabled"]),
            host=str(APRS_IS_SERVER["host"]),
            port=int(APRS_IS_SERVER["port"]),
            server_name=str(APRS_IS_SERVER.get("server_name", "Ham Router")),
        ),
        kiss=KISSConfig(
            host=str(KISS["host"]),
            port=int(KISS["port"]),
        ),
        sartrack=SARTrackTCPConfig(
            enabled=bool(SARTRACK_TCP["enabled"]),
            host=str(SARTRACK_TCP["host"]),
            port=int(SARTRACK_TCP["port"]),
        ),
        sartrack_aprs=SARTrackAPRSServerConfig(
            enabled=bool(SARTRACK_APRS["enabled"]),
            host=str(SARTRACK_APRS["host"]),
            port=int(SARTRACK_APRS["port"]),
            reconnect_delay=float(SARTRACK_APRS.get("reconnect_delay", 5.0)),
        ),
        cot=CoTConfig(
            enabled=bool(COT["enabled"]),
            cot_url=str(COT.get("cot_url", "udp://127.0.0.1:4242")),
            rx_enabled=bool(COT.get("rx_enabled", True)),
            listen_url=str(COT.get("listen_url", "udp://0.0.0.0:4243")),
            chat_listen_url=str(COT.get("chat_listen_url", "udp://224.10.10.1:17012")),
            pytak_no_hello=bool(COT.get("pytak_no_hello", True)),
            tak_proto=int(COT.get("tak_proto", 0)),
            stale_seconds=int(COT.get("stale_seconds", 300)),
        ),
        gui=GuiConfig(
            enabled=bool(GUI["enabled"]),
            host=str(GUI["host"]),
            port=int(GUI["port"]),
            title=str(GUI["title"]),
        ),
        demo=DemoConfig(
            demo_enabled=bool(DEMO["demo_enabled"]),
            demo_delay_seconds=float(DEMO["demo_delay_seconds"]),
            demo_callsign=str(DEMO["demo_callsign"]),
            demo_lat=float(DEMO["demo_lat"]),
            demo_lon=float(DEMO["demo_lon"]),
            demo_position_message=str(DEMO["demo_position_message"]),
            demo_message_target=str(DEMO["demo_message_target"]),
            demo_text_message=str(DEMO["demo_text_message"]),
        ),
        mesh_adapters=mesh_adapters,
        mesh_callsign_aliases=dict(MESH_CALLSIGN_ALIASES),
        log_level=_parse_log_level(str(LOGGING.get("level", "DEBUG"))),
        disabled_logger_prefixes=_disabled_logger_prefixes(LOGGING.get("enabled_loggers", {})),
        disabled_logger_names=_disabled_logger_names(LOGGING.get("disabled_exact_loggers", [])),
    )


def persist_agwpe_config(config: AGWPEConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
        "channel": config.channel,
        "reconnect_delay": config.reconnect_delay,
        "rx_monitor_enabled": config.rx_monitor_enabled,
        "tx_callsign": config.tx_callsign,
        "tx_destination": config.tx_destination,
        "tx_path": config.tx_path,
    }
    _replace_top_level_assignments({"AGWPE": values})


def persist_agwpe_server_config(config: AGWPEServerConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
        "channel": config.channel,
        "rx_monitor_enabled": config.rx_monitor_enabled,
        "tx_send_monitor_t": config.tx_send_monitor_t,
        "tx_send_sartrack_v": config.tx_send_sartrack_v,
        "tx_callsign": config.tx_callsign,
        "tx_destination": config.tx_destination,
        "tx_path": config.tx_path,
    }
    _replace_top_level_assignments({"AGWPE_SERVER": values})


def persist_aprs_is_server_config(config: APRSISServerConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
        "server_name": config.server_name,
    }
    _replace_top_level_assignments({"APRS_IS_SERVER": values})


def persist_sartrack_config(config: SARTrackTCPConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
    }
    _replace_top_level_assignments({"SARTRACK_TCP": values})


def persist_sartrack_aprs_config(config: SARTrackAPRSServerConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
        "reconnect_delay": config.reconnect_delay,
    }
    _replace_top_level_assignments({"SARTRACK_APRS": values})


def persist_cot_config(config: CoTConfig) -> None:
    values = {
        "enabled": config.enabled,
        "cot_url": config.cot_url,
        "rx_enabled": config.rx_enabled,
        "listen_url": config.listen_url,
        "chat_listen_url": config.chat_listen_url,
        "pytak_no_hello": config.pytak_no_hello,
        "tak_proto": config.tak_proto,
        "stale_seconds": config.stale_seconds,
    }
    _replace_top_level_assignments({"COT": values})


def persist_gui_config(config: GuiConfig) -> None:
    values = {
        "enabled": config.enabled,
        "host": config.host,
        "port": config.port,
        "title": config.title,
    }
    _replace_top_level_assignments({"GUI": values})


def _parse_log_level(value: str) -> int:
    normalized = value.strip().upper()
    return getattr(logging, normalized, logging.DEBUG)


def _disabled_logger_prefixes(enabled_loggers: dict[str, object]) -> list[str]:
    disabled: list[str] = []
    for logger_name, enabled in enabled_loggers.items():
        if not bool(enabled):
            disabled.append(str(logger_name))
    return disabled


def _disabled_logger_names(names: list[object]) -> list[str]:
    return [str(name) for name in names if str(name).strip()]


def persist_mesh_adapter_config(config: MeshtasticConfig) -> None:
    updated = []
    found = False
    for item in MESH_ADAPTERS:
        if item.get("name") == config.name:
            updated.append(
                {
                    "name": config.name,
                    "enabled": config.enabled,
                    "host": config.host,
                    "port": config.port,
                    "destination_id": config.destination_id,
                    "hop_limit": config.hop_limit,
                    "aprs_channel_index": config.aprs_channel_index,
                    "transport_mode": config.transport_mode,
                    "position_transport_mode": config.position_transport_mode,
                    "tak_rx_enabled": config.tak_rx_enabled,
                    "tak_tx_enabled": config.tak_tx_enabled,
                    "tak_detail_enabled": config.tak_detail_enabled,
                    "tak_detail_compress": config.tak_detail_compress,
                    "tak_detail_compress_threshold": config.tak_detail_compress_threshold,
                    "waypoint_expire_seconds": config.waypoint_expire_seconds,
                    "waypoint_icon": config.waypoint_icon,
                    "outbound_echo_ttl_seconds": config.outbound_echo_ttl_seconds,
                    "connect_quiet_seconds": config.connect_quiet_seconds,
                    "read_timeout_seconds": config.read_timeout_seconds,
                    "reconnect_backoff_seconds": tuple(config.reconnect_backoff_seconds),
                }
            )
            found = True
            continue
        updated.append(item)

    if not found:
        updated.append(
            {
                "name": config.name,
                "enabled": config.enabled,
                "host": config.host,
                "port": config.port,
                "destination_id": config.destination_id,
                "hop_limit": config.hop_limit,
                "aprs_channel_index": config.aprs_channel_index,
                "transport_mode": config.transport_mode,
                "position_transport_mode": config.position_transport_mode,
                "tak_rx_enabled": config.tak_rx_enabled,
                "tak_tx_enabled": config.tak_tx_enabled,
                "tak_detail_enabled": config.tak_detail_enabled,
                "tak_detail_compress": config.tak_detail_compress,
                "tak_detail_compress_threshold": config.tak_detail_compress_threshold,
                "waypoint_expire_seconds": config.waypoint_expire_seconds,
                "waypoint_icon": config.waypoint_icon,
                "outbound_echo_ttl_seconds": config.outbound_echo_ttl_seconds,
                "connect_quiet_seconds": config.connect_quiet_seconds,
                "read_timeout_seconds": config.read_timeout_seconds,
                "reconnect_backoff_seconds": tuple(config.reconnect_backoff_seconds),
            }
        )

    _replace_top_level_assignments({"MESH_ADAPTERS": updated})


def _replace_top_level_assignments(values: dict[str, object]) -> None:
    path = _persist_target_path()
    if not path.exists():
        _initialize_runtime_config(path)
    source = path.read_text(encoding="utf-8")
    tree = ast.parse(source)
    lines = source.splitlines(keepends=True)
    replacements: list[tuple[int, int, str, str]] = []

    for node in tree.body:
        if not isinstance(node, ast.Assign):
            continue
        if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name):
            continue
        name = node.targets[0].id
        if name not in values:
            continue
        rendered = pprint.pformat(values[name], width=100, sort_dicts=False)
        replacements.append((node.lineno - 1, node.end_lineno, f"{name} = {rendered}\n", name))

    missing = set(values) - {name for _, _, _, name in replacements}
    if missing:
        raise KeyError(f"Config entries not found in source: {', '.join(sorted(missing))}")

    for start, end, replacement, name in sorted(replacements, reverse=True):
        lines[start:end] = [replacement]
        globals()[name] = values[name]

    path.write_text("".join(lines), encoding="utf-8")


def _apply_runtime_overrides() -> None:
    if not getattr(sys, "frozen", False):
        return

    path = _persist_target_path()
    if not path.exists():
        return

    spec = importlib.util.spec_from_file_location("runtime_config", path)
    if spec is None or spec.loader is None:
        return

    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    for name in _PERSISTED_CONFIG_NAMES:
        if hasattr(module, name):
            globals()[name] = getattr(module, name)


def _persist_target_path() -> Path:
    if getattr(sys, "frozen", False):
        return Path(sys.executable).resolve().with_name("runtime_config.py")
    return Path(__file__)


def _initialize_runtime_config(path: Path) -> None:
    rendered_sections = [
        "# Runtime overrides for the packaged gateway.\n",
        "# This file is created next to the executable on first config save.\n\n",
    ]
    for name in _PERSISTED_CONFIG_NAMES:
        rendered = pprint.pformat(globals()[name], width=100, sort_dicts=False)
        rendered_sections.append(f"{name} = {rendered}\n\n")
    path.write_text("".join(rendered_sections), encoding="utf-8")