diff --git a/app/internal/op25_config_utls.py b/app/internal/op25_config_utls.py index 30d0609..a4cedf3 100644 --- a/app/internal/op25_config_utls.py +++ b/app/internal/op25_config_utls.py @@ -2,6 +2,7 @@ import csv import json import os import shutil +from pathlib import Path from models.models import TalkgroupTag from typing import List, Dict from internal.logger import create_logger @@ -28,8 +29,8 @@ def scan_local_library() -> List[Dict]: # 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, + "system_name": sys_name, + "filename": filename, "mode": "P25" if "trunking" in data else "NBFM" }) except Exception as e: @@ -44,16 +45,27 @@ def activate_config_from_library(system_name: str) -> bool: 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") + config_path = Path(CONFIG_DIR) + src = config_path / system_name + dst = config_path / "active.cfg.json" - if not os.path.exists(src): + 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") + return True except Exception as e: LOGGER.error(f"Failed to copy config: {e}") @@ -88,14 +100,16 @@ def get_current_active_config() -> Dict: 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: +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]) -> None: - with open(os.path.join(CONFIG_DIR, "active.cfg.whitelist.tsv"), 'w', newline='', encoding='utf-8') as file: +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]) diff --git a/app/routers/op25_controller.py b/app/routers/op25_controller.py index 5accdbf..06c2786 100644 --- a/app/routers/op25_controller.py +++ b/app/routers/op25_controller.py @@ -61,6 +61,77 @@ async def start_op25_logic(): return False return False +def build_op25_config(generator: ConfigGenerator) -> dict: + if generator.type == DecodeMode.P25: + channels = [ChannelConfig( + name=generator.systemName, + trunking_sysname=generator.systemName, + enable_analog="off", + demod_type="cqpsk", + cqpsk_tracking=True, + filter_type="rc", + meta_stream_name="stream_0" + )] + devices = [DeviceConfig()] + + trunking = TrunkingConfig( + module="tk_p25.py", + chans=[TrunkingChannelConfig( + sysname=generator.systemName, + control_channel_list=','.join(generator.channels), + tagsFile="/configs/active.cfg.tags.tsv", + whitelist="/configs/active.cfg.whitelist.tsv" + )] + ) + + metadata = MetadataConfig( + streams=[ + MetadataStreamConfig( + stream_name="stream_0", + icecastServerAddress = f"{generator.icecastConfig.icecast_host}:{generator.icecastConfig.icecast_port}", + icecastMountpoint = generator.icecastConfig.icecast_mountpoint, + icecastPass = generator.icecastConfig.icecast_password + ) + ] + ) + + terminal = TerminalConfig() + + return { + "channels": [channel.dict() for channel in channels], + "devices": [device.dict() for device in devices], + "trunking": trunking.dict(), + "metadata": metadata.dict(), + "terminal": terminal.dict() + } + + elif generator.type == DecodeMode.ANALOG: + analog_config = generator.config + channels = [ChannelConfig( + channelName=analog_config.systemName, + enableAnalog="on", + demodType="fsk4", + frequency=analog_config.frequency, + filterType="widepulse", + nbfmSquelch=analog_config.nbfmSquelch + )] + devices = [DeviceConfig(gain="LNA:32")] + + return { + "channels": [channel.dict() for channel in channels], + "devices": [device.dict() for device in devices] + } + else: + raise HTTPException(status_code=400, detail="Invalid decode mode") + +def save_library_sidecars(system_name: str, generator: ConfigGenerator): + if generator.type == DecodeMode.P25: + prefix = system_name + if prefix.endswith(".json"): + prefix = prefix[:-5] + save_talkgroup_tags(generator.tags, prefix) + save_whitelist(generator.whitelist, prefix) + def create_op25_router(): router = APIRouter() @@ -93,46 +164,31 @@ def create_op25_router(): active.cfg.json, and optionally restarts the radio. """ try: - if generator.type == DecodeMode.P25: - # 1. Handle sidecar files (Tags/Whitelists) - if generator.config.talkgroupTags: - save_talkgroup_tags(generator.config.talkgroupTags) - if generator.config.whitelist: - save_whitelist(generator.config.whitelist) - - # 2. Build the main OP25 dictionary structure - config_dict = { - "channels": [c.dict() for c in generator.config.channels], - "devices": [d.dict() for d in generator.config.devices], - "trunking": generator.config.trunking.dict(), - "metadata": generator.config.metadata.dict(), - "terminal": generator.config.terminal.dict() - } - - elif generator.type == DecodeMode.ANALOG: - # Simple Analog NBFM Setup for quick testing - channels = [ChannelConfig( - channelName=generator.config.systemName, - enableAnalog="on", - frequency=generator.config.frequency, - demodType="fsk4", - filterType="widepulse" - )] - config_dict = { - "channels": [c.dict() for c in channels], - "devices": [{"gain": "LNA:32"}] # Default gain for analog test - } - else: - raise HTTPException(status_code=400, detail="Invalid decode mode") - - # 3. Clean 'None' values to prevent OP25 parsing errors and save + # 1. Build the configuration dictionary + config_dict = build_op25_config(generator) final_json = del_none_in_dict(config_dict) - - if save_to_library_name: - save_config_to_library(save_to_library_name, final_json) - with open('/configs/active.cfg.json', 'w') as f: - json.dump(final_json, f, indent=2) + # 2. Handle Storage and Activation + if save_to_library_name: + # Save to library + save_config_to_library(save_to_library_name, final_json) + save_library_sidecars(save_to_library_name, generator) + + # Activate from library (Copies json + sidecars) + if not activate_config_from_library(save_to_library_name): + raise HTTPException(status_code=500, detail="Failed to activate saved configuration") + else: + # Save directly to active + with open('/configs/active.cfg.json', 'w') as f: + json.dump(final_json, f, indent=2) + + if generator.type == DecodeMode.P25: + save_talkgroup_tags(generator.tags) + save_whitelist(generator.whitelist) + + # 3. Generate Liquidsoap Script (Always required for active P25 session) + if generator.type == DecodeMode.P25: + generate_liquid_script(generator.icecastConfig) LOGGER.info("Saved new configuration to active.cfg.json") @@ -162,13 +218,19 @@ def create_op25_router(): raise HTTPException(status_code=404, detail=f"Config '{system_name}' not found in library volume") @router.post("/save_to_library") - async def save_to_library(system_name: str, config: dict): + async def save_to_library(system_name: str, config: ConfigGenerator): """ Directly saves a JSON configuration to the library. """ - if save_config_to_library(system_name, config): - return {"status": f"Config saved as {system_name}"} - raise HTTPException(status_code=500, detail="Failed to save configuration") + try: + config_dict = build_op25_config(config) + final_json = del_none_in_dict(config_dict) + if save_config_to_library(system_name, final_json): + save_library_sidecars(system_name, config) + return {"status": f"Config saved as {system_name}"} + raise HTTPException(status_code=500, detail="Failed to save configuration") + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) @router.get("/library") async def get_library():