import csv import json import os import shutil from models.models import TalkgroupTag from typing import List, Dict from internal.logger import create_logger 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({ "name": sys_name, "system_name": 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" src = os.path.join(CONFIG_DIR, system_name) dst = os.path.join(CONFIG_DIR, "active.cfg.json") if not os.path.exists(src): 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}") 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]) -> None: with open(os.path.join(CONFIG_DIR, "active.cfg.tags.tsv"), '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]) -> None: with open(os.path.join(CONFIG_DIR, "active.cfg.whitelist.tsv"), '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("sysname", "Unknown System") except: return "Unknown System"