diff --git a/.env.example b/.env.example index c49a1a7..f1079e9 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,9 @@ NODE_ID= 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/.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/Dockerfile b/Dockerfile index c933c07..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 @@ -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 f9e2d95..c38d745 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") @@ -112,6 +115,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. @@ -153,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 @@ -197,6 +231,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 MQTT_BROKER: return None + local_path = f"/calls/{call_id}.mp3" + if not os.path.exists(local_path): return None + + try: + with open(local_path, "rb") as f: + files = {"file": (f"{call_id}.mp3", f, "audio/mpeg")} + 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: + 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) @@ -212,7 +279,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 = {} @@ -240,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: @@ -248,12 +315,51 @@ 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, + "call_id": current_call_id + } 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')})") - payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_start", "metadata": current_meta} + + # 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"/calls/{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, + "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 @@ -263,12 +369,26 @@ 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, + "call_id": current_call_id + } 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..6994136 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,6 +21,10 @@ services: - NODE_LONG=${NODE_LONG} - MQTT_BROKER=${MQTT_BROKER} - ICECAST_SERVER=${ICECAST_SERVER} + - AUDIO_BUCKET=${AUDIO_BUCKET} + - HTTP_SERVER_PROTOCOL=${HTTP_SERVER_PROTOCOL} + - HTTP_SERVER_ADDRESS=${HTTP_SERVER_ADDRESS} + - HTTP_SERVER_PORT=${HTTP_SERVER_PORT} networks: - radio-shared-net