init

.env.example

@@ -1,5 +1,6 @@
 NODE_ID=
 MQTT_BROKER=
 ICECAST_SERVER=
+AUDIO_BUCKET=
 NODE_LAT=
 NODE_LONG=

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}")
 

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