From 41075a5950b4be17c4b0d6dcb9603064c34f3226 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 29 Dec 2025 22:18:58 -0500 Subject: [PATCH 01/11] 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 -- 2.49.1 From 3b98e3a72a4a9f5995fd24ea2e957801ee45893d Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 29 Dec 2025 22:21:58 -0500 Subject: [PATCH 02/11] Add GCP to the requirements --- requirements.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index e49e782..b6472b2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,5 @@ uvicorn[standard] paho-mqtt pydantic python-multipart -requests \ No newline at end of file +requests +google-cloud-storage \ No newline at end of file -- 2.49.1 From a7de6bfb0435a54f2c7e3c00c0d1f9e3fee9eed0 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 29 Dec 2025 22:47:18 -0500 Subject: [PATCH 03/11] Fix the calls directory bug --- Dockerfile | 3 +++ app/node_main.py | 6 +++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index c933c07..172bb7d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,6 +34,9 @@ EXPOSE 8001 8081 # Create and set up the configuration directory VOLUME ["/configs"] +# Create the calls local cache directory +VOLUME ["/calls"] + # Set the working directory in the container WORKDIR /app diff --git a/app/node_main.py b/app/node_main.py index 82229bd..6f6af6b 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -247,7 +247,7 @@ async def mqtt_lifecycle_manager(): def upload_audio(call_id): if not AUDIO_BUCKET: return None - local_path = f"/tmp/{call_id}.mp3" + local_path = f"/calls/{call_id}.mp3" if not os.path.exists(local_path): return None try: @@ -278,7 +278,7 @@ async def mqtt_lifecycle_manager(): if response.status_code == 200: data = response.json() - LOGGER.debug(f"Response from OP25 API: {data}") + # LOGGER.debug(f"Response from OP25 API: {data}") current_tgid = 0 current_meta = {} @@ -344,7 +344,7 @@ async def mqtt_lifecycle_manager(): 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", + f"/calls/{current_call_id}.mp3", stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL ) -- 2.49.1 From a5d5fa9de705ddfc4597bb6fb5e565a9239c715c Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 29 Dec 2025 22:55:18 -0500 Subject: [PATCH 04/11] Install ffmpeg to test if that resolves issue with recording --- Dockerfile | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/Dockerfile b/Dockerfile index 172bb7d..bc11526 100644 --- a/Dockerfile +++ b/Dockerfile @@ -37,6 +37,11 @@ VOLUME ["/configs"] # Create the calls local cache directory VOLUME ["/calls"] + +##### REMOVE ME OR PROPERLY ADD IF WORKS +RUN apt-get install ffmpeg -y +##### REMOVE ME OR PROPERLY ADD IF WORKS + # Set the working directory in the container WORKDIR /app -- 2.49.1 From 8c106473cfc32d9620b1530d3b91d145a9893e37 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 30 Dec 2025 03:01:31 -0500 Subject: [PATCH 05/11] Move bucket upload to the c2 server and replaced with upload to c2 server --- .gitignore | 3 ++- app/node_main.py | 14 ++++++-------- requirements.txt | 3 +-- 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/.gitignore b/.gitignore index 2a65b26..d78b9e1 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ *.log *.db *.conf -configs/* \ No newline at end of file +configs/* +*.json \ No newline at end of file diff --git a/app/node_main.py b/app/node_main.py index 6f6af6b..9e0662d 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -9,7 +9,6 @@ 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__) @@ -23,7 +22,6 @@ 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 @@ -246,16 +244,16 @@ async def mqtt_lifecycle_manager(): recorder_proc = None def upload_audio(call_id): - if not AUDIO_BUCKET: return None + if not MQTT_BROKER: return None local_path = f"/calls/{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" + with open(local_path, "rb") as f: + files = {"file": (f"{call_id}.mp3", f, "audio/mpeg")} + response = requests.post(f"{MQTT_BROKER}/upload", files=files, data={"node_id": NODE_ID, "call_id": call_id}, timeout=30) + response.raise_for_status() + return response.json().get("url") except Exception as e: LOGGER.error(f"Upload failed: {e}") return None diff --git a/requirements.txt b/requirements.txt index b6472b2..e49e782 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,4 @@ uvicorn[standard] paho-mqtt pydantic python-multipart -requests -google-cloud-storage \ No newline at end of file +requests \ No newline at end of file -- 2.49.1 From 0fe8194c395146dfe5633f34b07a9a35cf8ed79a Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Fri, 2 Jan 2026 00:17:36 -0500 Subject: [PATCH 06/11] fix upload url --- app/node_main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/node_main.py b/app/node_main.py index 9e0662d..d9e838c 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -251,7 +251,7 @@ async def mqtt_lifecycle_manager(): try: with open(local_path, "rb") as f: files = {"file": (f"{call_id}.mp3", f, "audio/mpeg")} - response = requests.post(f"{MQTT_BROKER}/upload", files=files, data={"node_id": NODE_ID, "call_id": call_id}, timeout=30) + response = requests.post(f"http://{MQTT_BROKER}/upload", files=files, data={"node_id": NODE_ID, "call_id": call_id}, timeout=30) response.raise_for_status() return response.json().get("url") except Exception as e: -- 2.49.1 From 9e92da4e58be88430afd3b0622aa07840c60d704 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 3 Jan 2026 03:10:45 -0500 Subject: [PATCH 07/11] Replace http server vars with dedicated vars --- .env.example | 5 ++++- app/node_main.py | 5 ++++- docker-compose.yml | 3 +++ 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index 440f778..f1079e9 100644 --- a/.env.example +++ b/.env.example @@ -3,4 +3,7 @@ MQTT_BROKER= ICECAST_SERVER= AUDIO_BUCKET= NODE_LAT= -NODE_LONG= \ No newline at end of file +NODE_LONG= +HTTP_SERVER_PROTOCOL= +HTTP_SERVER_ADDRESS= +HTTP_SERVER_PORT= \ No newline at end of file diff --git a/app/node_main.py b/app/node_main.py index d9e838c..24a05b0 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -20,6 +20,9 @@ app.include_router(create_op25_router(), prefix="/op25") # Configuration NODE_ID = os.getenv("NODE_ID", "standalone-node") MQTT_BROKER = os.getenv("MQTT_BROKER", None) +HTTP_SERVER_PROTOCOL = os.getenv("HTTP_SERVER_PROTOCOL", "http") +HTTP_SERVER_ADDRESS = os.getenv("HTTP_SERVER_ADDRESS", "127.0.0.1") +HTTP_SERVER_PORT = os.getenv("HTTP_SERVER_PORT", 8000) NODE_LAT = os.getenv("NODE_LAT") NODE_LONG = os.getenv("NODE_LONG") @@ -251,7 +254,7 @@ async def mqtt_lifecycle_manager(): try: with open(local_path, "rb") as f: files = {"file": (f"{call_id}.mp3", f, "audio/mpeg")} - response = requests.post(f"http://{MQTT_BROKER}/upload", files=files, data={"node_id": NODE_ID, "call_id": call_id}, timeout=30) + response = requests.post(f"{HTTP_SERVER_PROTOCOL}://{HTTP_SERVER_ADDRESS}:{HTTP_SERVER_PORT}/upload", files=files, data={"node_id": NODE_ID, "call_id": call_id}, timeout=30) response.raise_for_status() return response.json().get("url") except Exception as e: diff --git a/docker-compose.yml b/docker-compose.yml index e046813..f15b012 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -22,6 +22,9 @@ services: - MQTT_BROKER=${MQTT_BROKER} - ICECAST_SERVER=${ICECAST_SERVER} - AUDIO_BUCKET=${AUDIO_BUCKET} + - HTTP_SERVER_PROTOCOL=${HTTP_SERVER_PROTOCOL} + - HTTP_SERVER_HOST=${HTTP_SERVER_HOST} + - HTTP_SERVER_PORT=${HTTP_SERVER_PORT} networks: - radio-shared-net -- 2.49.1 From 83b995bfa51348e60fffb3e2349bd06d61c8e141 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 3 Jan 2026 03:12:57 -0500 Subject: [PATCH 08/11] Fix bootleg AI mistake --- docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker-compose.yml b/docker-compose.yml index f15b012..6994136 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -23,7 +23,7 @@ services: - ICECAST_SERVER=${ICECAST_SERVER} - AUDIO_BUCKET=${AUDIO_BUCKET} - HTTP_SERVER_PROTOCOL=${HTTP_SERVER_PROTOCOL} - - HTTP_SERVER_HOST=${HTTP_SERVER_HOST} + - HTTP_SERVER_ADDRESS=${HTTP_SERVER_ADDRESS} - HTTP_SERVER_PORT=${HTTP_SERVER_PORT} networks: - radio-shared-net -- 2.49.1 From d8190e307c555f3ec744917fd630d9a6c1223993 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 3 Jan 2026 11:41:27 -0500 Subject: [PATCH 09/11] Standardize timestamps to UTC --- app/node_main.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/node_main.py b/app/node_main.py index 24a05b0..48041fb 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -187,7 +187,7 @@ async def mqtt_lifecycle_manager(): payload = { "node_id": NODE_ID, "status": "online", - "timestamp": datetime.now().isoformat(), + "timestamp": datetime.utcnow().isoformat(), "is_listening": op25_status.get("is_running", False), "active_system": op25_status.get("active_system"), # Only scan library if needed, otherwise it's heavy I/O @@ -307,7 +307,7 @@ async def mqtt_lifecycle_manager(): break if current_tgid: break - now = datetime.now() + now = datetime.utcnow() # Logic for handling call start/end events if current_tgid != 0: -- 2.49.1 From 051eac88b08bb8d045eb2431b7c34ce3ddf97960 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 3 Jan 2026 19:18:30 -0500 Subject: [PATCH 10/11] Add call ID to the call metadata --- app/node_main.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/app/node_main.py b/app/node_main.py index 48041fb..c38d745 100644 --- a/app/node_main.py +++ b/app/node_main.py @@ -327,7 +327,8 @@ async def mqtt_lifecycle_manager(): "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata, - "audio_url": audio_url + "audio_url": audio_url, + "call_id": current_call_id } client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) @@ -352,7 +353,13 @@ async def mqtt_lifecycle_manager(): 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} + payload = { + "node_id": NODE_ID, + "timestamp": now.isoformat(), + "event": "call_start", + "metadata": current_meta, + "call_id": current_call_id + } client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) last_tgid = current_tgid last_metadata = current_meta @@ -374,7 +381,8 @@ async def mqtt_lifecycle_manager(): "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata, - "audio_url": audio_url + "audio_url": audio_url, + "call_id": current_call_id } client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) last_tgid = 0 -- 2.49.1 From 10554a2ff4b74461811ca2a825c35e79c1f2d720 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 3 Jan 2026 19:32:20 -0500 Subject: [PATCH 11/11] Properly add ffmpeg to the dockerfile install sequence --- Dockerfile | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/Dockerfile b/Dockerfile index bc11526..ba631e4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ ENV DEBIAN_FRONTEND=noninteractive # Install system dependencies RUN apt-get update && \ apt-get upgrade -y && \ - apt-get install git pulseaudio pulseaudio-utils liquidsoap -y + apt-get install git pulseaudio pulseaudio-utils liquidsoap ffmpeg -y # Clone the boatbod op25 repository RUN git clone -b gr310 https://github.com/boatbod/op25 /op25 @@ -37,11 +37,6 @@ VOLUME ["/configs"] # Create the calls local cache directory VOLUME ["/calls"] - -##### REMOVE ME OR PROPERLY ADD IF WORKS -RUN apt-get install ffmpeg -y -##### REMOVE ME OR PROPERLY ADD IF WORKS - # Set the working directory in the container WORKDIR /app -- 2.49.1