Merge pull request 'Implement Call Recording for STT and Replay' (#3) from implement-call-recording into main

Reviewed-on: #3

.env.example

@@ -1,5 +1,9 @@
 NODE_ID=
 MQTT_BROKER=
 ICECAST_SERVER=
+AUDIO_BUCKET=
 NODE_LAT=
 NODE_LONG=
+HTTP_SERVER_PROTOCOL=
+HTTP_SERVER_ADDRESS=
+HTTP_SERVER_PORT=

.gitea/workflows/run-tests.yml

@@ -1,37 +0,0 @@
-name: Python Application Tests
-
-on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "*" ]
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        python-version: ["3.13"]
-
-    steps:
-    - uses: actions/checkout@v4
-
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v4
-      with:
-        python-version: ${{ matrix.python-version }}
-
-    - name: Install Dependencies
-      run: |
-        python -m pip install --upgrade pip
-        # Install test dependencies
-        pip install pytest pytest-asyncio httpx
-        # Install application dependencies (assuming you have a requirements.txt)
-        # If you don't have one, create it with `pip freeze > requirements.txt`
-        # For now, we'll install the dependencies we know are needed from context
-        pip install fastapi "uvicorn[standard]" paho-mqtt requests
-
-    - name: Test with pytest
-      run: |
-        pytest

.gitignore

@@ -3,3 +3,4 @@
 *.db
 *.conf
 configs/*
+*.json

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
 

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

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
 

tests/test_op25_controller.py

@@ -1,316 +0,0 @@
-import pytest
-from fastapi.testclient import TestClient
-from unittest.mock import patch, MagicMock, mock_open, ANY
-import json
-import os
-import types
-from typing import List, Optional
-from pydantic import BaseModel
-
-# The router is included in the main app, so we test through it.
-# We need to adjust the python path for imports to work correctly
-import sys
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'app')))
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-
-# --- MOCK MODELS ---
-# The actual models.models file has a NameError (IcecastConfig used before definition).
-# Since we cannot edit the source code, we mock the module here to allow tests to run.
-mock_models = types.ModuleType("models.models")
-
-class MockTerminalConfig(BaseModel):
-    pass
-
-class MockTalkgroupTag(BaseModel):
-    tagDec: int
-    tagName: str
-
-class MockDecodeMode:
-    P25 = "P25"
-    ANALOG = "ANALOG"
-
-class MockIcecastConfig(BaseModel):
-    icecast_host: str
-    icecast_port: int
-    icecast_mountpoint: str
-    icecast_password: str
-
-class MockAnalogConfig(BaseModel):
-    systemName: str
-    frequency: str
-    nbfmSquelch: int
-
-class MockConfigGenerator(BaseModel):
-    type: str
-    systemName: str
-    channels: Optional[List[str]] = None
-    tags: Optional[List[MockTalkgroupTag]] = None
-    whitelist: Optional[str] = None
-    icecastConfig: Optional[MockIcecastConfig] = None
-    config: Optional[MockAnalogConfig] = None
-
-class MockChannelConfig(BaseModel):
-    name: Optional[str] = None
-    trunking_sysname: Optional[str] = None
-    enable_analog: Optional[str] = None
-    demod_type: Optional[str] = None
-    cqpsk_tracking: Optional[bool] = None
-    filter_type: Optional[str] = None
-    meta_stream_name: Optional[str] = None
-    channelName: Optional[str] = None
-    enableAnalog: Optional[str] = None
-    demodType: Optional[str] = None
-    frequency: Optional[str] = None
-    filterType: Optional[str] = None
-    nbfmSquelch: Optional[int] = None
-
-class MockDeviceConfig(BaseModel):
-    gain: Optional[str] = None
-
-class MockTrunkingChannelConfig(BaseModel):
-    sysname: str
-    control_channel_list: str
-    tagsFile: str
-    whitelist: str
-
-class MockTrunkingConfig(BaseModel):
-    module: str
-    chans: List[MockTrunkingChannelConfig]
-
-class MockMetadataStreamConfig(BaseModel):
-    stream_name: str
-    icecastServerAddress: str
-    icecastMountpoint: str
-    icecastPass: str
-
-class MockMetadataConfig(BaseModel):
-    streams: List[MockMetadataStreamConfig]
-
-
-mock_models.ConfigGenerator = MockConfigGenerator
-mock_models.DecodeMode = MockDecodeMode
-mock_models.ChannelConfig = MockChannelConfig
-mock_models.DeviceConfig = MockDeviceConfig
-mock_models.TrunkingConfig = MockTrunkingConfig
-mock_models.TrunkingChannelConfig = MockTrunkingChannelConfig
-mock_models.TerminalConfig = MockTerminalConfig
-mock_models.MetadataConfig = MockMetadataConfig
-mock_models.MetadataStreamConfig = MockMetadataStreamConfig
-mock_models.IcecastConfig = MockIcecastConfig
-mock_models.TalkgroupTag = MockTalkgroupTag
-
-sys.modules["models.models"] = mock_models
-sys.modules["models"] = types.ModuleType("models")
-sys.modules["models"].models = mock_models
-# -------------------
-
-from app.node_main import app
-
-# Use a client to make requests to the app
-client = TestClient(app)
-
-# Define a sample P25 config payload for testing
-SAMPLE_P25_CONFIG = {
-    "type": "P25",
-    "systemName": "TestSystem",
-    "channels": ["851.12345", "852.67890"],
-    "tags": [{"tagDec": 101, "tagName": "Group A"}, {"tagDec": 102, "tagName": "Group B"}],
-    "whitelist": "101",
-    "icecastConfig": {
-        "icecast_host": "localhost",
-        "icecast_port": 8000,
-        "icecast_mountpoint": "test",
-        "icecast_password": "hackme"
-    }
-}
-
-@pytest.fixture(autouse=True)
-def reset_and_mock_globals(monkeypatch):
-    """
-    Fixture to reset the global op25_process state and mock dependencies
-    before each test, ensuring test isolation.
-    """
-    # Reset the global process variable in the controller module
-    monkeypatch.setattr("routers.op25_controller.op25_process", None)
-
-    # Mock asyncio.sleep to prevent tests from actually waiting
-    mock_sleep = MagicMock()
-    monkeypatch.setattr("asyncio.sleep", mock_sleep)
-
-    # Mock os functions related to process groups
-    monkeypatch.setattr("os.killpg", MagicMock())
-    monkeypatch.setattr("os.getpgid", MagicMock(return_value=12345))
-
-
-@patch("routers.op25_controller.subprocess.Popen")
-def test_start_op25_success(mock_popen):
-    """Test the /start endpoint successfully starts the process."""
-    mock_process = MagicMock()
-    mock_process.pid = 12345
-    mock_popen.return_value = mock_process
-
-    response = client.post("/op25/start")
-    assert response.status_code == 200
-    assert response.json() == {"status": "OP25 started"}
-    mock_popen.assert_called_once()
-
-
-@patch("routers.op25_controller.subprocess.Popen", side_effect=Exception("Popen failed"))
-def test_start_op25_failure(mock_popen):
-    """Test the /start endpoint when Popen raises an exception."""
-    response = client.post("/op25/start")
-    assert response.status_code == 500
-    assert "Failed to start OP25" in response.json()["detail"]
-
-
-def test_stop_op25_not_running():
-    """Test the /stop endpoint when the process is not running."""
-    response = client.post("/op25/stop")
-    assert response.status_code == 200
-    assert response.json() == {"status": "OP25 was not running"}
-
-
-@patch("routers.op25_controller.subprocess.Popen")
-def test_stop_op25_success(mock_popen, monkeypatch):
-    """Test the /stop endpoint successfully stops a running process."""
-    mock_process = MagicMock()
-    mock_process.pid = 12345
-    mock_process.poll.return_value = None  # Indicates it's running
-    monkeypatch.setattr("routers.op25_controller.op25_process", mock_process)
-
-    response = client.post("/op25/stop")
-    assert response.status_code == 200
-    assert response.json() == {"status": "OP25 stopped"}
-    os.killpg.assert_called_with(os.getpgid(mock_process.pid), ANY)
-
-
-def test_get_status_not_running():
-    """Test the /status endpoint when the process is not running."""
-    response = client.get("/op25/status")
-    assert response.status_code == 200
-    data = response.json()
-    assert data["is_running"] is False
-    assert data["pid"] is None
-    assert data["active_system"] is None
-
-
-@patch("routers.op25_controller.get_current_system_from_config", return_value="TestSystem")
-@patch("routers.op25_controller.subprocess.Popen")
-def test_get_status_running(mock_popen, mock_get_system, monkeypatch):
-    """Test the /status endpoint when the process is running."""
-    mock_process = MagicMock()
-    mock_process.pid = 12345
-    mock_process.poll.return_value = None # Running
-    monkeypatch.setattr("routers.op25_controller.op25_process", mock_process)
-
-    response = client.get("/op25/status")
-    assert response.status_code == 200
-    data = response.json()
-    assert data["is_running"] is True
-    assert data["pid"] == 12345
-    assert data["active_system"] == "TestSystem"
-    mock_get_system.assert_called_once()
-
-
-@patch("builtins.open", new_callable=mock_open)
-@patch("routers.op25_controller.json.dump")
-@patch("routers.op25_controller.save_talkgroup_tags")
-@patch("routers.op25_controller.save_whitelist")
-@patch("routers.op25_controller.generate_liquid_script")
-@patch("routers.op25_controller.subprocess.Popen")
-def test_set_active_config_no_restart(mock_popen, mock_liquid, mock_white, mock_tags, mock_dump, mock_file):
-    """Test setting active config without restarting the radio."""
-    response = client.post("/op25/set_active_config?restart=false", json=SAMPLE_P25_CONFIG)
-
-    assert response.status_code == 200
-    assert response.json() == {"message": "Active configuration updated", "radio_restarted": False}
-
-    # Verify config files were written
-    mock_file.assert_called_with('/configs/active.cfg.json', 'w')
-    mock_dump.assert_called_once()
-    mock_tags.assert_called_with([MockTalkgroupTag(**t) for t in SAMPLE_P25_CONFIG["tags"]])
-    mock_white.assert_called_with(SAMPLE_P25_CONFIG["whitelist"])
-    mock_liquid.assert_called_with(MockIcecastConfig(**SAMPLE_P25_CONFIG["icecastConfig"]))
-
-    # Verify radio was NOT started/stopped
-    mock_popen.assert_not_called()
-    os.killpg.assert_not_called()
-
-
-@patch("routers.op25_controller.activate_config_from_library", return_value=True)
-@patch("routers.op25_controller.save_config_to_library")
-@patch("routers.op25_controller.save_library_sidecars")
-@patch("routers.op25_controller.subprocess.Popen")
-def test_set_active_config_with_save_to_library(mock_popen, mock_save_sidecars, mock_save_lib, mock_activate):
-    """Test setting active config and saving it to the library."""
-    library_name = "MyNewSystem"
-    response = client.post(
-        f"/op25/set_active_config?restart=true&save_to_library_name={library_name}",
-        json=SAMPLE_P25_CONFIG
-    )
-
-    assert response.status_code == 200
-    assert response.json()["radio_restarted"] is True
-
-    # Verify it was saved and then activated from the library
-    mock_save_lib.assert_called_with(library_name, ANY)
-    mock_save_sidecars.assert_called_with(library_name, ANY)
-    mock_activate.assert_called_with(library_name)
-
-    # Verify radio was restarted
-    assert mock_popen.call_count == 1
-
-
-@patch("routers.op25_controller.activate_config_from_library", return_value=True)
-@patch("routers.op25_controller.subprocess.Popen")
-def test_load_from_library_success(mock_popen, mock_activate):
-    """Test loading a configuration from the library."""
-    system_name = "ExistingSystem"
-    response = client.post(f"/op25/load_from_library?system_name={system_name}")
-
-    assert response.status_code == 200
-    assert response.json() == {"status": f"Loaded and started library config: {system_name}"}
-
-    # Verify activation and restart
-    mock_activate.assert_called_with(system_name)
-    assert mock_popen.call_count == 1
-
-
-@patch("routers.op25_controller.activate_config_from_library", return_value=False)
-def test_load_from_library_not_found(mock_activate):
-    """Test loading a non-existent configuration from the library."""
-    system_name = "NotFoundSystem"
-    response = client.post(f"/op25/load_from_library?system_name={system_name}")
-
-    assert response.status_code == 404
-    assert "not found in library" in response.json()["detail"]
-
-
-@patch("routers.op25_controller.save_config_to_library", return_value=True)
-@patch("routers.op25_controller.save_library_sidecars")
-def test_save_to_library(mock_save_sidecars, mock_save_lib):
-    """Test saving a configuration directly to the library."""
-    system_name = "NewLibSystem"
-    response = client.post(f"/op25/save_to_library?system_name={system_name}", json=SAMPLE_P25_CONFIG)
-
-    assert response.status_code == 200
-    assert response.json() == {"status": f"Config saved as {system_name}"}
-    mock_save_lib.assert_called_with(system_name, ANY)
-    mock_save_sidecars.assert_called_with(system_name, ANY)
-
-
-@patch("routers.op25_controller.scan_local_library", return_value=["System1.json", "System2.json"])
-def test_get_library(mock_scan):
-    """Test the /library endpoint."""
-    response = client.get("/op25/library")
-    assert response.status_code == 200
-    assert response.json() == ["System1.json", "System2.json"]
-    mock_scan.assert_called_once()
-
-
-@patch("routers.op25_controller.build_op25_config", side_effect=Exception("Build failed"))
-def test_set_active_config_build_failure(mock_build):
-    """Test error handling when config building fails."""
-    response = client.post("/op25/set_active_config", json=SAMPLE_P25_CONFIG)
-    assert response.status_code == 500
-    assert "Configuration error: Build failed" in response.json()["detail"]