init

app/internal/op25_config_utls.py

@@ -0,0 +1,122 @@
+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"