From 41075a5950b4be17c4b0d6dcb9603064c34f3226 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 29 Dec 2025 22:18:58 -0500 Subject: [PATCH] init --- .env.example | 1 + app/node_main.py | 115 ++++++++++++++++++++++++++++++++++++++++++++- docker-compose.yml | 1 + 3 files changed, 115 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index c49a1a7..440f778 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,6 @@ NODE_ID= MQTT_BROKER= ICECAST_SERVER= +AUDIO_BUCKET= NODE_LAT= NODE_LONG= \ No newline at end of file diff --git a/app/node_main.py b/app/node_main.py index f9e2d95..82229bd 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -9,6 +9,7 @@ from internal.logger import create_logger from internal.op25_config_utls import scan_local_library import paho.mqtt.client as mqtt import requests +from google.cloud import storage # Initialize logging LOGGER = create_logger(__name__) @@ -22,6 +23,7 @@ NODE_ID = os.getenv("NODE_ID", "standalone-node") MQTT_BROKER = os.getenv("MQTT_BROKER", None) NODE_LAT = os.getenv("NODE_LAT") NODE_LONG = os.getenv("NODE_LONG") +AUDIO_BUCKET = os.getenv("AUDIO_BUCKET") # Global flag to track MQTT connection state MQTT_CONNECTED = False @@ -112,6 +114,37 @@ def handle_c2_command(topic, payload): except Exception as e: LOGGER.error(f"Error processing C2 command: {e}") +def get_current_stream_url(): + """ + Dynamically resolves the audio stream URL from the active OP25 configuration. + Falls back to env var or default if config is missing/invalid. + """ + default_url = os.getenv("STREAM_URL", "http://127.0.0.1:8000/stream_0") + config_path = "/configs/active.cfg.json" + + if not os.path.exists(config_path): + return default_url + + try: + with open(config_path, "r") as f: + config = json.load(f) + + streams = config.get("metadata", {}).get("streams", []) + if not streams: + return default_url + + stream = streams[0] + address = stream.get("icecastServerAddress", "127.0.0.1:8000") + mount = stream.get("icecastMountpoint", "stream_0") + + if not mount.startswith("/"): + mount = f"/{mount}" + + return f"http://{address}{mount}" + except Exception as e: + LOGGER.warning(f"Failed to resolve stream URL from config: {e}") + return default_url + async def mqtt_lifecycle_manager(): """ Manages the application-level logic: Check-in, Heartbeats, and Shutdown. @@ -197,6 +230,39 @@ async def mqtt_lifecycle_manager(): # This is the specific payload the OP25 web interface uses [cite: 45562, 45563] COMMAND_PAYLOAD = [{"command": "update", "arg1": 0, "arg2": 0}] + # Audio Recording State + recorder_proc = None + current_call_id = None + + async def stop_recording(): + nonlocal recorder_proc + if recorder_proc: + if recorder_proc.returncode is None: + recorder_proc.terminate() + try: + await asyncio.wait_for(recorder_proc.wait(), timeout=2.0) + except asyncio.TimeoutError: + recorder_proc.kill() + recorder_proc = None + + def upload_audio(call_id): + if not AUDIO_BUCKET: return None + local_path = f"/tmp/{call_id}.mp3" + if not os.path.exists(local_path): return None + + try: + client = storage.Client() + bucket = client.bucket(AUDIO_BUCKET) + blob = bucket.blob(f"audio/{call_id}.mp3") + blob.upload_from_filename(local_path, content_type="audio/mpeg") + return f"gs://{AUDIO_BUCKET}/audio/{call_id}.mp3" + except Exception as e: + LOGGER.error(f"Upload failed: {e}") + return None + finally: + if os.path.exists(local_path): + os.remove(local_path) + while True: if not MQTT_CONNECTED: await asyncio.sleep(1) @@ -248,11 +314,43 @@ async def mqtt_lifecycle_manager(): if current_tgid != last_tgid: if last_tgid != 0: + # --- END PREVIOUS CALL --- + await stop_recording() + audio_url = None + if current_call_id: + audio_url = await loop.run_in_executor(None, upload_audio, current_call_id) + LOGGER.debug(f"Switching TGID: {last_tgid} -> {current_tgid}") - payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata} + payload = { + "node_id": NODE_ID, + "timestamp": now.isoformat(), + "event": "call_end", + "metadata": last_metadata, + "audio_url": audio_url + } client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) + # --- START NEW CALL --- LOGGER.debug(f"Call Start: TGID {current_tgid} ({current_meta.get('alpha_tag')})") + + # Generate ID + start_ts = int(now.timestamp()) + sysname = current_meta.get('sysname', 'unknown') + tgid = current_meta.get('tgid', '0') + current_call_id = f"{NODE_ID}_{sysname}_{tgid}_{start_ts}" + + # Start Recording (FFmpeg) + try: + stream_url = get_current_stream_url() + recorder_proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-i", stream_url, "-y", "-t", "300", + f"/tmp/{current_call_id}.mp3", + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL + ) + except Exception as e: + LOGGER.error(f"Failed to start recorder: {e}") + payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_start", "metadata": current_meta} client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) last_tgid = current_tgid @@ -263,12 +361,25 @@ async def mqtt_lifecycle_manager(): LOGGER.debug(f"Signal lost for TGID {last_tgid}. Starting debounce.") potential_end_time = now elif (now - potential_end_time).total_seconds() > DEBOUNCE_SECONDS: + # --- END CALL (Debounce Expired) --- + await stop_recording() + audio_url = None + if current_call_id: + audio_url = await loop.run_in_executor(None, upload_audio, current_call_id) + LOGGER.debug(f"Call End (Debounce expired): TGID {last_tgid}") - payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata} + payload = { + "node_id": NODE_ID, + "timestamp": now.isoformat(), + "event": "call_end", + "metadata": last_metadata, + "audio_url": audio_url + } client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) last_tgid = 0 last_metadata = {} potential_end_time = None + current_call_id = None else: LOGGER.debug(f"OP25 API returned status: {response.status_code}") diff --git a/docker-compose.yml b/docker-compose.yml index 84e860a..e046813 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,6 +21,7 @@ services: - NODE_LONG=${NODE_LONG} - MQTT_BROKER=${MQTT_BROKER} - ICECAST_SERVER=${ICECAST_SERVER} + - AUDIO_BUCKET=${AUDIO_BUCKET} networks: - radio-shared-net