drb-edge-node/app/internal/op25_config_utls.py

import csv
import json
import os
import shutil
from pathlib import Path
from models.models import TalkgroupTag, IcecastConfig
from typing import List, Dict
from internal.logger import create_logger
from internal.liquidsoap_config_utils import generate_liquid_script

LOGGER = create_logger(__name__)

CONFIG_DIR = "/configs"

def scan_local_library() -> List[Dict]:
    """
    Scans the /configs directory for JSON files to build the 'nearby_systems' list.
    """
    library = []
    if not os.path.exists(CONFIG_DIR):
        return library

    for filename in os.listdir(CONFIG_DIR):
        # We don't want to include the active config or the sidecar files in the library scan
        if filename.endswith(".json") and filename != "active.cfg.json":
            try:
                path = os.path.join(CONFIG_DIR, filename)
                with open(path, 'r') as f:
                    data = json.load(f)
                    # Use trunking sysname or filename as the identifier
                    sys_name = data.get("trunking", {}).get("sysname", filename.replace(".json", ""))
                    library.append({
                        "system_name": sys_name,
                        "filename": filename,
                        "mode": "P25" if "trunking" in data else "NBFM"
                    })
            except Exception as e:
                LOGGER.error(f"Failed to parse library file {filename}: {e}")
    
    return library

def activate_config_from_library(system_name: str) -> bool:
    """
    Copies a config from the library to the active slot.
    """
    if not system_name.endswith(".json"):
        system_name += ".json"

    config_path = Path(CONFIG_DIR)
    src = config_path / system_name
    dst = config_path / "active.cfg.json"
    
    if not src.exists():
        LOGGER.error(f"Source config {system_name} not found in library.")
        return False
        
    try:
        shutil.copy2(src, dst)
        LOGGER.info(f"Activated config: {system_name}")
        
        # Copy sidecar files (tags/whitelist) if they exist
        src_tags = src.with_suffix(".tags.tsv")
        if src_tags.exists():
            shutil.copy2(src_tags, config_path / "active.cfg.tags.tsv")
            
        src_whitelist = src.with_suffix(".whitelist.tsv")
        if src_whitelist.exists():
            shutil.copy2(src_whitelist, config_path / "active.cfg.whitelist.tsv")

        # Generate Liquidsoap Script by reading the activated config
        with open(dst, 'r') as f:
            data = json.load(f)

        if "trunking" in data and "metadata" in data:
            streams = data.get("metadata", {}).get("streams", [])
            if streams:
                stream = streams[0]
                address = stream.get("icecastServerAddress", "127.0.0.1:8000")
                host, port = address.split(":") if ":" in address else (address, 8000)
                
                ice_config = IcecastConfig(
                    icecast_host=host,
                    icecast_port=int(port),
                    icecast_mountpoint=stream.get("icecastMountpoint", "/stream"),
                    icecast_password=stream.get("icecastPass", "hackme"),
                    icecast_description="OP25 Stream",
                    icecast_genre="Scanner"
                )
                generate_liquid_script(ice_config)

        return True
    except Exception as e:
        LOGGER.error(f"Failed to copy config: {e}")
        return False

def save_config_to_library(system_name: str, config: Dict) -> bool:
    """
    Saves a configuration dictionary to the local library.
    """
    if not system_name.endswith(".json"):
        system_name += ".json"
        
    path = os.path.join(CONFIG_DIR, system_name)
    
    try:
        with open(path, 'w') as f:
            json.dump(config, f, indent=2)
        LOGGER.info(f"Saved config to library: {system_name}")
        return True
    except Exception as e:
        LOGGER.error(f"Failed to save config to library: {e}")
        return False

def get_current_active_config() -> Dict:
    """Reads the current active.cfg.json if it exists."""
    path = os.path.join(CONFIG_DIR, "active.cfg.json")
    if os.path.exists(path):
        try:
            with open(path, 'r') as f:
                return json.load(f)
        except:
            return {}
    return {}

def save_talkgroup_tags(talkgroup_tags: List[TalkgroupTag], prefix: str = "active.cfg") -> None:
    filename = f"{prefix}.tags.tsv"
    with open(os.path.join(CONFIG_DIR, filename), 'w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
        for tag in talkgroup_tags:
            writer.writerow([tag.tagDec, tag.talkgroup])

def save_whitelist(talkgroup_tags: List[int], prefix: str = "active.cfg") -> None:
    filename = f"{prefix}.whitelist.tsv"
    with open(os.path.join(CONFIG_DIR, filename), 'w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
        for tag in talkgroup_tags:
            writer.writerow([tag])

def del_none_in_dict(d):
    for key, value in list(d.items()):
        if value is None:
            del d[key]
        elif isinstance(value, dict):
            del_none_in_dict(value)
        elif isinstance(value, list):
            for iterative_value in value:
                if isinstance(iterative_value, dict):
                    del_none_in_dict(iterative_value)
    return d

def get_current_system_from_config() -> str:
    data = get_current_active_config()
    if not data:
        return None
    try:
        return data.get("trunking", {}).get("chans", [{}])[0].get("sysname", "Unknown System")
    except:
        return "Unknown System"