Merge upstream main: add DMR, WebSDR, HF SSTV, alerts, recordings, waterfall

Merges upstream changes into fork while preserving weather satellite
(NOAA APT/Meteor LRPT via SatDump), rtlamr, multi-arch build, and
decoder console features from our branch.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

routes/__init__.py

@@ -27,6 +27,11 @@ def register_blueprints(app):
     from .updater import updater_bp
     from .sstv import sstv_bp
     from .weather_sat import weather_sat_bp
+    from .sstv_general import sstv_general_bp
+    from .dmr import dmr_bp
+    from .websdr import websdr_bp
+    from .alerts import alerts_bp
+    from .recordings import recordings_bp
 
     app.register_blueprint(pager_bp)
     app.register_blueprint(sensor_bp)
@@ -53,6 +58,11 @@ def register_blueprints(app):
     app.register_blueprint(updater_bp)  # GitHub update checking
     app.register_blueprint(sstv_bp)  # ISS SSTV decoder
     app.register_blueprint(weather_sat_bp)  # NOAA/Meteor weather satellite decoder
+    app.register_blueprint(sstv_general_bp)  # General terrestrial SSTV
+    app.register_blueprint(dmr_bp)  # DMR / P25 / Digital Voice
+    app.register_blueprint(websdr_bp)  # HF/Shortwave WebSDR
+    app.register_blueprint(alerts_bp)  # Cross-mode alerts
+    app.register_blueprint(recordings_bp)  # Session recordings
 
     # Initialize TSCM state with queue and lock from app
     import app as app_module

routes/acars.py

@@ -20,13 +20,15 @@ from flask import Blueprint, jsonify, request, Response
 import app as app_module
 from utils.logging import sensor_logger as logger
 from utils.validation import validate_device_index, validate_gain, validate_ppm
-from utils.sse import format_sse
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.constants import (
     PROCESS_TERMINATE_TIMEOUT,
     SSE_KEEPALIVE_INTERVAL,
     SSE_QUEUE_TIMEOUT,
     PROCESS_START_WAIT,
 )
+from utils.process import register_process, unregister_process
 
 acars_bp = Blueprint('acars', __name__, url_prefix='/acars')
 
@@ -144,9 +146,24 @@ def stream_acars_output(process: subprocess.Popen, is_text_mode: bool = False) -
         logger.error(f"ACARS stream error: {e}")
         app_module.acars_queue.put({'type': 'error', 'message': str(e)})
     finally:
+        global acars_active_device
+        # Ensure process is terminated
+        try:
+            process.terminate()
+            process.wait(timeout=2)
+        except Exception:
+            try:
+                process.kill()
+            except Exception:
+                pass
+        unregister_process(process)
         app_module.acars_queue.put({'type': 'status', 'status': 'stopped'})
         with app_module.acars_lock:
             app_module.acars_process = None
+        # Release SDR device
+        if acars_active_device is not None:
+            app_module.release_sdr_device(acars_active_device)
+            acars_active_device = None
 
 
 @acars_bp.route('/tools')
@@ -311,6 +328,7 @@ def start_acars() -> Response:
             return jsonify({'status': 'error', 'message': error_msg}), 500
 
         app_module.acars_process = process
+        register_process(process)
 
         # Start output streaming thread
         thread = threading.Thread(
@@ -374,9 +392,13 @@ def stream_acars() -> Response:
 
         while True:
             try:
-                msg = app_module.acars_queue.get(timeout=SSE_QUEUE_TIMEOUT)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = app_module.acars_queue.get(timeout=SSE_QUEUE_TIMEOUT)
+                last_keepalive = time.time()
+                try:
+                    process_event('acars', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= SSE_KEEPALIVE_INTERVAL:

routes/adsb.py

@@ -43,6 +43,7 @@ from utils.validation import (
     validate_rtl_tcp_host, validate_rtl_tcp_port
 )
 from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.sdr import SDRFactory, SDRType
 from utils.constants import (
     ADSB_SBS_PORT,
@@ -843,6 +844,10 @@ def stream_adsb():
             try:
                 msg = app_module.adsb_queue.get(timeout=SSE_QUEUE_TIMEOUT)
                 last_keepalive = time.time()
+                try:
+                    process_event('adsb', msg, msg.get('type'))
+                except Exception:
+                    pass
                 yield format_sse(msg)
             except queue.Empty:
                 now = time.time()

routes/ais.py

@@ -19,6 +19,7 @@ from config import SHARED_OBSERVER_LOCATION_ENABLED
 from utils.logging import get_logger
 from utils.validation import validate_device_index, validate_gain
 from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.sdr import SDRFactory, SDRType
 from utils.constants import (
     AIS_TCP_PORT,
@@ -484,6 +485,10 @@ def stream_ais():
             try:
                 msg = app_module.ais_queue.get(timeout=SSE_QUEUE_TIMEOUT)
                 last_keepalive = time.time()
+                try:
+                    process_event('ais', msg, msg.get('type'))
+                except Exception:
+                    pass
                 yield format_sse(msg)
             except queue.Empty:
                 now = time.time()

routes/alerts.py

@@ -0,0 +1,76 @@
+"""Alerting API endpoints."""
+
+from __future__ import annotations
+
+import queue
+import time
+from typing import Generator
+
+from flask import Blueprint, Response, jsonify, request
+
+from utils.alerts import get_alert_manager
+from utils.sse import format_sse
+
+alerts_bp = Blueprint('alerts', __name__, url_prefix='/alerts')
+
+
+@alerts_bp.route('/rules', methods=['GET'])
+def list_rules():
+    manager = get_alert_manager()
+    include_disabled = request.args.get('all') in ('1', 'true', 'yes')
+    return jsonify({'status': 'success', 'rules': manager.list_rules(include_disabled=include_disabled)})
+
+
+@alerts_bp.route('/rules', methods=['POST'])
+def create_rule():
+    data = request.get_json() or {}
+    if not isinstance(data.get('match', {}), dict):
+        return jsonify({'status': 'error', 'message': 'match must be a JSON object'}), 400
+
+    manager = get_alert_manager()
+    rule_id = manager.add_rule(data)
+    return jsonify({'status': 'success', 'rule_id': rule_id})
+
+
+@alerts_bp.route('/rules/<int:rule_id>', methods=['PUT', 'PATCH'])
+def update_rule(rule_id: int):
+    data = request.get_json() or {}
+    manager = get_alert_manager()
+    ok = manager.update_rule(rule_id, data)
+    if not ok:
+        return jsonify({'status': 'error', 'message': 'Rule not found or no changes'}), 404
+    return jsonify({'status': 'success'})
+
+
+@alerts_bp.route('/rules/<int:rule_id>', methods=['DELETE'])
+def delete_rule(rule_id: int):
+    manager = get_alert_manager()
+    ok = manager.delete_rule(rule_id)
+    if not ok:
+        return jsonify({'status': 'error', 'message': 'Rule not found'}), 404
+    return jsonify({'status': 'success'})
+
+
+@alerts_bp.route('/events', methods=['GET'])
+def list_events():
+    manager = get_alert_manager()
+    limit = request.args.get('limit', default=100, type=int)
+    mode = request.args.get('mode')
+    severity = request.args.get('severity')
+    events = manager.list_events(limit=limit, mode=mode, severity=severity)
+    return jsonify({'status': 'success', 'events': events})
+
+
+@alerts_bp.route('/stream', methods=['GET'])
+def stream_alerts() -> Response:
+    manager = get_alert_manager()
+
+    def generate() -> Generator[str, None, None]:
+        for event in manager.stream_events(timeout=1.0):
+            yield format_sse(event)
+
+    response = Response(generate(), mimetype='text/event-stream')
+    response.headers['Cache-Control'] = 'no-cache'
+    response.headers['X-Accel-Buffering'] = 'no'
+    response.headers['Connection'] = 'keep-alive'
+    return response

routes/aprs.py

@@ -13,7 +13,7 @@ import tempfile
 import threading
 import time
 from datetime import datetime
-from subprocess import DEVNULL, PIPE, STDOUT
+from subprocess import PIPE, STDOUT
 from typing import Generator, Optional
 
 from flask import Blueprint, jsonify, request, Response
@@ -21,7 +21,8 @@ from flask import Blueprint, jsonify, request, Response
 import app as app_module
 from utils.logging import sensor_logger as logger
 from utils.validation import validate_device_index, validate_gain, validate_ppm
-from utils.sse import format_sse
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.constants import (
     PROCESS_TERMINATE_TIMEOUT,
     SSE_KEEPALIVE_INTERVAL,
@@ -31,6 +32,9 @@ from utils.constants import (
 
 aprs_bp = Blueprint('aprs', __name__, url_prefix='/aprs')
 
+# Track which SDR device is being used
+aprs_active_device: int | None = None
+
 # APRS frequencies by region (MHz)
 APRS_FREQUENCIES = {
     'north_america': '144.390',
@@ -1301,7 +1305,7 @@ def stream_aprs_output(rtl_process: subprocess.Popen, decoder_process: subproces
 
     This function reads from the decoder's stdout (text mode, line-buffered).
     The decoder's stderr is merged into stdout (STDOUT) to avoid deadlocks.
-    rtl_fm's stderr is sent to DEVNULL for the same reason.
+    rtl_fm's stderr is captured via PIPE with a monitor thread.
 
     Outputs two types of messages to the queue:
     - type='aprs': Decoded APRS packets
@@ -1383,6 +1387,7 @@ def stream_aprs_output(rtl_process: subprocess.Popen, decoder_process: subproces
         logger.error(f"APRS stream error: {e}")
         app_module.aprs_queue.put({'type': 'error', 'message': str(e)})
     finally:
+        global aprs_active_device
         app_module.aprs_queue.put({'type': 'status', 'status': 'stopped'})
         # Cleanup processes
         for proc in [rtl_process, decoder_process]:
@@ -1394,6 +1399,10 @@ def stream_aprs_output(rtl_process: subprocess.Popen, decoder_process: subproces
                     proc.kill()
                 except Exception:
                     pass
+        # Release SDR device
+        if aprs_active_device is not None:
+            app_module.release_sdr_device(aprs_active_device)
+            aprs_active_device = None
 
 
 @aprs_bp.route('/tools')
@@ -1441,6 +1450,7 @@ def get_stations() -> Response:
 def start_aprs() -> Response:
     """Start APRS decoder."""
     global aprs_packet_count, aprs_station_count, aprs_last_packet_time, aprs_stations
+    global aprs_active_device
 
     with app_module.aprs_lock:
         if app_module.aprs_process and app_module.aprs_process.poll() is None:
@@ -1477,6 +1487,16 @@ def start_aprs() -> Response:
     except ValueError as e:
         return jsonify({'status': 'error', 'message': str(e)}), 400
 
+    # Reserve SDR device to prevent conflicts with other modes
+    error = app_module.claim_sdr_device(device, 'aprs')
+    if error:
+        return jsonify({
+            'status': 'error',
+            'error_type': 'DEVICE_BUSY',
+            'message': error
+        }), 409
+    aprs_active_device = device
+
     # Get frequency for region
     region = data.get('region', 'north_america')
     frequency = APRS_FREQUENCIES.get(region, '144.390')
@@ -1552,15 +1572,25 @@ def start_aprs() -> Response:
 
     try:
         # Start rtl_fm with stdout piped to decoder.
-        # stderr goes to DEVNULL to prevent blocking (rtl_fm logs to stderr).
+        # stderr is captured via PIPE so errors are reported to the user.
         # NOTE: RTL-SDR Blog V4 may show offset-tuned frequency in logs - this is normal.
         rtl_process = subprocess.Popen(
             rtl_cmd,
             stdout=PIPE,
-            stderr=DEVNULL,
+            stderr=PIPE,
             start_new_session=True
         )
 
+        # Start a thread to monitor rtl_fm stderr for errors
+        def monitor_rtl_stderr():
+            for line in rtl_process.stderr:
+                err_text = line.decode('utf-8', errors='replace').strip()
+                if err_text:
+                    logger.debug(f"[RTL_FM] {err_text}")
+
+        rtl_stderr_thread = threading.Thread(target=monitor_rtl_stderr, daemon=True)
+        rtl_stderr_thread.start()
+
         # Start decoder with stdin wired to rtl_fm's stdout.
         # Use text mode with line buffering for reliable line-by-line reading.
         # Merge stderr into stdout to avoid blocking on unbuffered stderr.
@@ -1582,13 +1612,25 @@ def start_aprs() -> Response:
         time.sleep(PROCESS_START_WAIT)
 
         if rtl_process.poll() is not None:
-            # rtl_fm exited early - something went wrong
+            # rtl_fm exited early - capture stderr for diagnostics
+            stderr_output = ''
+            try:
+                remaining = rtl_process.stderr.read()
+                if remaining:
+                    stderr_output = remaining.decode('utf-8', errors='replace').strip()
+            except Exception:
+                pass
             error_msg = f'rtl_fm failed to start (exit code {rtl_process.returncode})'
+            if stderr_output:
+                error_msg += f': {stderr_output[:200]}'
             logger.error(error_msg)
             try:
                 decoder_process.kill()
             except Exception:
                 pass
+            if aprs_active_device is not None:
+                app_module.release_sdr_device(aprs_active_device)
+                aprs_active_device = None
             return jsonify({'status': 'error', 'message': error_msg}), 500
 
         if decoder_process.poll() is not None:
@@ -1602,6 +1644,9 @@ def start_aprs() -> Response:
                 rtl_process.kill()
             except Exception:
                 pass
+            if aprs_active_device is not None:
+                app_module.release_sdr_device(aprs_active_device)
+                aprs_active_device = None
             return jsonify({'status': 'error', 'message': error_msg}), 500
 
         # Store references for status checks and cleanup
@@ -1626,12 +1671,17 @@ def start_aprs() -> Response:
 
     except Exception as e:
         logger.error(f"Failed to start APRS decoder: {e}")
+        if aprs_active_device is not None:
+            app_module.release_sdr_device(aprs_active_device)
+            aprs_active_device = None
         return jsonify({'status': 'error', 'message': str(e)}), 500
 
 
 @aprs_bp.route('/stop', methods=['POST'])
 def stop_aprs() -> Response:
     """Stop APRS decoder."""
+    global aprs_active_device
+
     with app_module.aprs_lock:
         processes_to_stop = []
 
@@ -1660,6 +1710,11 @@ def stop_aprs() -> Response:
         if hasattr(app_module, 'aprs_rtl_process'):
             app_module.aprs_rtl_process = None
 
+        # Release SDR device
+        if aprs_active_device is not None:
+            app_module.release_sdr_device(aprs_active_device)
+            aprs_active_device = None
+
     return jsonify({'status': 'stopped'})
 
 
@@ -1671,9 +1726,13 @@ def stream_aprs() -> Response:
 
         while True:
             try:
-                msg = app_module.aprs_queue.get(timeout=SSE_QUEUE_TIMEOUT)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = app_module.aprs_queue.get(timeout=SSE_QUEUE_TIMEOUT)
+                last_keepalive = time.time()
+                try:
+                    process_event('aprs', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= SSE_KEEPALIVE_INTERVAL:

routes/audio_websocket.py

@@ -66,12 +66,6 @@ def kill_audio_processes():
                 pass
         rtl_process = None
 
-    # Kill any orphaned processes
-    try:
-        subprocess.run(['pkill', '-9', '-f', 'rtl_fm'], capture_output=True, timeout=1)
-    except:
-        pass
-
     time.sleep(0.3)
 
 
@@ -228,13 +222,13 @@ def init_audio_websocket(app: Flask):
 
                 except TimeoutError:
                     pass
-                except Exception as e:
-                    msg = str(e).lower()
-                    if "connection closed" in msg:
-                        logger.info("WebSocket closed by client")
-                        break
-                    if "timed out" not in msg:
-                        logger.error(f"WebSocket receive error: {e}")
+                except Exception as e:
+                    msg = str(e).lower()
+                    if "connection closed" in msg:
+                        logger.info("WebSocket closed by client")
+                        break
+                    if "timed out" not in msg:
+                        logger.error(f"WebSocket receive error: {e}")
 
                 # Stream audio data if active
                 if streaming and proc and proc.poll() is None:

routes/bluetooth.py

@@ -18,10 +18,11 @@ from typing import Any, Generator
 from flask import Blueprint, jsonify, request, Response
 
 import app as app_module
-from utils.dependencies import check_tool
-from utils.logging import bluetooth_logger as logger
-from utils.sse import format_sse
-from utils.validation import validate_bluetooth_interface
+from utils.dependencies import check_tool
+from utils.logging import bluetooth_logger as logger
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from utils.validation import validate_bluetooth_interface
 from data.oui import OUI_DATABASE, load_oui_database, get_manufacturer
 from data.patterns import AIRTAG_PREFIXES, TILE_PREFIXES, SAMSUNG_TRACKER
 from utils.constants import (
@@ -561,9 +562,13 @@ def stream_bt():
 
         while True:
             try:
-                msg = app_module.bt_queue.get(timeout=1)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = app_module.bt_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('bluetooth', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= keepalive_interval:

routes/bluetooth_v2.py

@@ -7,32 +7,40 @@ aggregation, and heuristics.
 
 from __future__ import annotations
 
-import csv
-import io
-import json
-import logging
+import csv
+import io
+import json
+import logging
+import threading
+import time
 from datetime import datetime
 from typing import Generator
 
 from flask import Blueprint, Response, jsonify, request, session
 
-from utils.bluetooth import (
-    BluetoothScanner,
-    BTDeviceAggregate,
-    get_bluetooth_scanner,
-    check_capabilities,
-    RANGE_UNKNOWN,
+from utils.bluetooth import (
+    BluetoothScanner,
+    BTDeviceAggregate,
+    get_bluetooth_scanner,
+    check_capabilities,
+    RANGE_UNKNOWN,
     TrackerType,
     TrackerConfidence,
     get_tracker_engine,
-)
-from utils.database import get_db
-from utils.sse import format_sse
+)
+from utils.database import get_db
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 
 logger = logging.getLogger('intercept.bluetooth_v2')
 
 # Blueprint
-bluetooth_v2_bp = Blueprint('bluetooth_v2', __name__, url_prefix='/api/bluetooth')
+bluetooth_v2_bp = Blueprint('bluetooth_v2', __name__, url_prefix='/api/bluetooth')
+
+# Seen-before tracking
+_bt_seen_cache: set[str] = set()
+_bt_session_seen: set[str] = set()
+_bt_seen_lock = threading.Lock()
 
 # =============================================================================
 # DATABASE FUNCTIONS
@@ -164,13 +172,20 @@ def get_all_baselines() -> list[dict]:
         return [dict(row) for row in cursor]
 
 
-def save_observation_history(device: BTDeviceAggregate) -> None:
-    """Save device observation to history."""
-    with get_db() as conn:
-        conn.execute('''
-            INSERT INTO bt_observation_history (device_id, rssi, seen_count)
-            VALUES (?, ?, ?)
-        ''', (device.device_id, device.rssi_current, device.seen_count))
+def save_observation_history(device: BTDeviceAggregate) -> None:
+    """Save device observation to history."""
+    with get_db() as conn:
+        conn.execute('''
+            INSERT INTO bt_observation_history (device_id, rssi, seen_count)
+            VALUES (?, ?, ?)
+        ''', (device.device_id, device.rssi_current, device.seen_count))
+
+
+def load_seen_device_ids() -> set[str]:
+    """Load distinct device IDs from history for seen-before tracking."""
+    with get_db() as conn:
+        cursor = conn.execute('SELECT DISTINCT device_id FROM bt_observation_history')
+        return {row['device_id'] for row in cursor}
 
 
 # =============================================================================
@@ -191,7 +206,7 @@ def get_capabilities():
 
 
 @bluetooth_v2_bp.route('/scan/start', methods=['POST'])
-def start_scan():
+def start_scan():
     """
     Start Bluetooth scanning.
 
@@ -221,17 +236,42 @@ def start_scan():
     # Get scanner instance
     scanner = get_bluetooth_scanner(adapter_id)
 
-    # Check if already scanning
-    if scanner.is_scanning:
-        return jsonify({
-            'status': 'already_running',
-            'scan_status': scanner.get_status().to_dict()
-        })
-
-    # Initialize database tables if needed
-    init_bt_tables()
-
-    # Load active baseline if exists
+    # Initialize database tables if needed
+    init_bt_tables()
+
+    def _handle_seen_before(device: BTDeviceAggregate) -> None:
+        try:
+            with _bt_seen_lock:
+                device.seen_before = device.device_id in _bt_seen_cache
+                if device.device_id not in _bt_session_seen:
+                    save_observation_history(device)
+                    _bt_session_seen.add(device.device_id)
+        except Exception as e:
+            logger.debug(f"BT seen-before update failed: {e}")
+
+    # Setup seen-before callback
+    if scanner._on_device_updated is None:
+        scanner._on_device_updated = _handle_seen_before
+
+    # Ensure cache is initialized
+    with _bt_seen_lock:
+        if not _bt_seen_cache:
+            _bt_seen_cache.update(load_seen_device_ids())
+
+    # Check if already scanning
+    if scanner.is_scanning:
+        return jsonify({
+            'status': 'already_running',
+            'scan_status': scanner.get_status().to_dict()
+        })
+
+    # Refresh seen-before cache and reset session set for a new scan
+    with _bt_seen_lock:
+        _bt_seen_cache.clear()
+        _bt_seen_cache.update(load_seen_device_ids())
+        _bt_session_seen.clear()
+
+    # Load active baseline if exists
     baseline_id = get_active_baseline_id()
     if baseline_id:
         device_ids = get_baseline_device_ids(baseline_id)
@@ -856,11 +896,15 @@ def stream_events():
         else:
             return event_type, event
 
-    def event_generator() -> Generator[str, None, None]:
-        """Generate SSE events from scanner."""
-        for event in scanner.stream_events(timeout=1.0):
-            event_name, event_data = map_event_type(event)
-            yield format_sse(event_data, event=event_name)
+    def event_generator() -> Generator[str, None, None]:
+        """Generate SSE events from scanner."""
+        for event in scanner.stream_events(timeout=1.0):
+            event_name, event_data = map_event_type(event)
+            try:
+                process_event('bluetooth', event_data, event_name)
+            except Exception:
+                pass
+            yield format_sse(event_data, event=event_name)
 
     return Response(
         event_generator(),
@@ -944,23 +988,34 @@ def get_tscm_bluetooth_snapshot(duration: int = 8) -> list[dict]:
     devices = scanner.get_devices()
     logger.info(f"TSCM snapshot: get_devices() returned {len(devices)} devices")
 
-    # Convert to TSCM format with tracker detection data
-    tscm_devices = []
-    for device in devices:
-        device_data = {
-            'mac': device.address,
-            'address_type': device.address_type,
-            'device_key': device.device_key,
-            'name': device.name or 'Unknown',
-            'rssi': device.rssi_current or -100,
-            'rssi_median': device.rssi_median,
-            'rssi_ema': round(device.rssi_ema, 1) if device.rssi_ema else None,
-            'type': _classify_device_type(device),
-            'manufacturer': device.manufacturer_name,
-            'manufacturer_id': device.manufacturer_id,
-            'manufacturer_data': device.manufacturer_bytes.hex() if device.manufacturer_bytes else None,
-            'protocol': device.protocol,
-            'first_seen': device.first_seen.isoformat(),
+    # Convert to TSCM format with tracker detection data
+    tscm_devices = []
+    for device in devices:
+        manufacturer_name = device.manufacturer_name
+        if (not manufacturer_name) or str(manufacturer_name).lower().startswith('unknown'):
+            if device.address and not device.is_randomized_mac:
+                try:
+                    from data.oui import get_manufacturer
+                    oui_vendor = get_manufacturer(device.address)
+                    if oui_vendor and oui_vendor != 'Unknown':
+                        manufacturer_name = oui_vendor
+                except Exception:
+                    pass
+
+        device_data = {
+            'mac': device.address,
+            'address_type': device.address_type,
+            'device_key': device.device_key,
+            'name': device.name or 'Unknown',
+            'rssi': device.rssi_current or -100,
+            'rssi_median': device.rssi_median,
+            'rssi_ema': round(device.rssi_ema, 1) if device.rssi_ema else None,
+            'type': _classify_device_type(device),
+            'manufacturer': manufacturer_name,
+            'manufacturer_id': device.manufacturer_id,
+            'manufacturer_data': device.manufacturer_bytes.hex() if device.manufacturer_bytes else None,
+            'protocol': device.protocol,
+            'first_seen': device.first_seen.isoformat(),
             'last_seen': device.last_seen.isoformat(),
             'seen_count': device.seen_count,
             'range_band': device.range_band,
@@ -1174,14 +1229,38 @@ def get_device_timeseries(device_key: str):
     return jsonify(result)
 
 
-def _classify_device_type(device: BTDeviceAggregate) -> str:
-    """Classify device type from available data."""
-    name_lower = (device.name or '').lower()
-    manufacturer_lower = (device.manufacturer_name or '').lower()
-
-    # Check by name patterns
-    if any(x in name_lower for x in ['airpods', 'headphone', 'earbuds', 'buds', 'beats']):
-        return 'audio'
+def _classify_device_type(device: BTDeviceAggregate) -> str:
+    """Classify device type from available data."""
+    name_lower = (device.name or '').lower()
+    manufacturer_lower = (device.manufacturer_name or '').lower()
+    service_uuids = device.service_uuids or []
+
+    if (not manufacturer_lower) or manufacturer_lower.startswith('unknown'):
+        if device.address and not device.is_randomized_mac:
+            try:
+                from data.oui import get_manufacturer
+                oui_vendor = get_manufacturer(device.address)
+                if oui_vendor and oui_vendor != 'Unknown':
+                    manufacturer_lower = oui_vendor.lower()
+            except Exception:
+                pass
+
+    def normalize_uuid(uuid: str) -> str:
+        if not uuid:
+            return ''
+        value = str(uuid).lower().strip()
+        if value.startswith('0x'):
+            value = value[2:]
+        # Bluetooth Base UUID normalization (16-bit UUIDs)
+        if value.endswith('-0000-1000-8000-00805f9b34fb') and len(value) >= 8:
+            return value[4:8]
+        if len(value) == 4:
+            return value
+        return value
+
+    # Check by name patterns
+    if any(x in name_lower for x in ['airpods', 'headphone', 'earbuds', 'buds', 'beats']):
+        return 'audio'
     if any(x in name_lower for x in ['watch', 'band', 'fitbit', 'garmin']):
         return 'wearable'
     if any(x in name_lower for x in ['iphone', 'pixel', 'galaxy', 'phone']):
@@ -1190,18 +1269,41 @@ def _classify_device_type(device: BTDeviceAggregate) -> str:
         return 'computer'
     if any(x in name_lower for x in ['mouse', 'keyboard', 'trackpad']):
         return 'peripheral'
-    if any(x in name_lower for x in ['tile', 'airtag', 'smarttag', 'chipolo']):
-        return 'tracker'
-    if any(x in name_lower for x in ['speaker', 'sonos', 'echo', 'home']):
-        return 'speaker'
-    if any(x in name_lower for x in ['tv', 'chromecast', 'roku', 'firestick']):
-        return 'media'
-
-    # Check by manufacturer
-    if 'apple' in manufacturer_lower:
-        return 'apple_device'
-    if 'samsung' in manufacturer_lower:
-        return 'samsung_device'
+    if any(x in name_lower for x in ['tile', 'airtag', 'smarttag', 'chipolo']):
+        return 'tracker'
+    if any(x in name_lower for x in ['speaker', 'sonos', 'echo', 'home']):
+        return 'speaker'
+    if any(x in name_lower for x in ['tv', 'chromecast', 'roku', 'firestick']):
+        return 'media'
+
+    # Tracker signals (metadata or Find My service)
+    if getattr(device, 'is_tracker', False) or getattr(device, 'tracker_type', None):
+        return 'tracker'
+
+    normalized_uuids = {normalize_uuid(u) for u in service_uuids if u}
+    if 'fd6f' in normalized_uuids:
+        return 'tracker'
+
+    # Service UUIDs (GATT / classic)
+    audio_uuids = {'110b', '110a', '111e', '111f', '1108', '1203'}
+    wearable_uuids = {'180d', '1814', '1816'}
+    hid_uuids = {'1812'}
+    beacon_uuids = {'feaa', 'feab', 'feb1', 'febe'}
+
+    if normalized_uuids & audio_uuids:
+        return 'audio'
+    if normalized_uuids & hid_uuids:
+        return 'peripheral'
+    if normalized_uuids & wearable_uuids:
+        return 'wearable'
+    if normalized_uuids & beacon_uuids:
+        return 'beacon'
+
+    # Check by manufacturer
+    if 'apple' in manufacturer_lower:
+        return 'apple_device'
+    if 'samsung' in manufacturer_lower:
+        return 'samsung_device'
 
     # Check by class of device
     if device.major_class:

routes/dmr.py

@@ -0,0 +1,513 @@
+"""DMR / P25 / Digital Voice decoding routes."""
+
+from __future__ import annotations
+
+import os
+import queue
+import re
+import select
+import shutil
+import subprocess
+import threading
+import time
+from datetime import datetime
+from typing import Generator, Optional
+
+from flask import Blueprint, jsonify, request, Response
+
+import app as app_module
+from utils.logging import get_logger
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from utils.process import register_process, unregister_process
+from utils.constants import (
+    SSE_QUEUE_TIMEOUT,
+    SSE_KEEPALIVE_INTERVAL,
+    QUEUE_MAX_SIZE,
+)
+
+logger = get_logger('intercept.dmr')
+
+dmr_bp = Blueprint('dmr', __name__, url_prefix='/dmr')
+
+# ============================================
+# GLOBAL STATE
+# ============================================
+
+dmr_rtl_process: Optional[subprocess.Popen] = None
+dmr_dsd_process: Optional[subprocess.Popen] = None
+dmr_thread: Optional[threading.Thread] = None
+dmr_running = False
+dmr_lock = threading.Lock()
+dmr_queue: queue.Queue = queue.Queue(maxsize=QUEUE_MAX_SIZE)
+dmr_active_device: Optional[int] = None
+
+VALID_PROTOCOLS = ['auto', 'dmr', 'p25', 'nxdn', 'dstar', 'provoice']
+
+# Classic dsd flags
+_DSD_PROTOCOL_FLAGS = {
+    'auto': [],
+    'dmr': ['-fd'],
+    'p25': ['-fp'],
+    'nxdn': ['-fn'],
+    'dstar': ['-fi'],
+    'provoice': ['-fv'],
+}
+
+# dsd-fme uses different flag names
+_DSD_FME_PROTOCOL_FLAGS = {
+    'auto': ['-ft'],
+    'dmr': ['-fs'],
+    'p25': ['-f1'],
+    'nxdn': ['-fi'],
+    'dstar': [],
+    'provoice': ['-fp'],
+}
+
+# ============================================
+# HELPERS
+# ============================================
+
+
+def find_dsd() -> tuple[str | None, bool]:
+    """Find DSD (Digital Speech Decoder) binary.
+
+    Checks for dsd-fme first (common fork), then falls back to dsd.
+    Returns (path, is_fme) tuple.
+    """
+    path = shutil.which('dsd-fme')
+    if path:
+        return path, True
+    path = shutil.which('dsd')
+    if path:
+        return path, False
+    return None, False
+
+
+def find_rtl_fm() -> str | None:
+    """Find rtl_fm binary."""
+    return shutil.which('rtl_fm')
+
+
+def parse_dsd_output(line: str) -> dict | None:
+    """Parse a line of DSD stderr output into a structured event.
+
+    Handles output from both classic ``dsd`` and ``dsd-fme`` which use
+    different formatting for talkgroup / source / voice frame lines.
+    """
+    line = line.strip()
+    if not line:
+        return None
+
+    # Skip DSD/dsd-fme startup banner lines (ASCII art, version info, etc.)
+    # These contain box-drawing characters or are pure decoration.
+    if re.search(r'[╔╗╚╝║═██▀▄╗╝╩╦╠╣╬│┤├┘└┐┌─┼█▓▒░]', line):
+        return None
+    if re.match(r'^\s*(Build Version|MBElib|CODEC2|Audio (Out|In)|Decoding )', line):
+        return None
+
+    ts = datetime.now().strftime('%H:%M:%S')
+
+    # Sync detection: "Sync: +DMR (data)" or "Sync: +P25 Phase 1"
+    sync_match = re.match(r'Sync:\s*\+?(\S+.*)', line)
+    if sync_match:
+        return {
+            'type': 'sync',
+            'protocol': sync_match.group(1).strip(),
+            'timestamp': ts,
+        }
+
+    # Talkgroup and Source — check BEFORE slot so "Slot 1 Voice LC, TG: …"
+    # is captured as a call event rather than a bare slot event.
+    # Classic dsd:   "TG: 12345  Src: 67890"
+    # dsd-fme:       "TG: 12345, Src: 67890"  or  "Talkgroup: 12345, Source: 67890"
+    tg_match = re.search(
+        r'(?:TG|Talkgroup)[:\s]+(\d+)[,\s]+(?:Src|Source)[:\s]+(\d+)', line, re.IGNORECASE
+    )
+    if tg_match:
+        result = {
+            'type': 'call',
+            'talkgroup': int(tg_match.group(1)),
+            'source_id': int(tg_match.group(2)),
+            'timestamp': ts,
+        }
+        # Extract slot if present on the same line
+        slot_inline = re.search(r'Slot\s*(\d+)', line)
+        if slot_inline:
+            result['slot'] = int(slot_inline.group(1))
+        return result
+
+    # P25 NAC (Network Access Code) — check before voice/slot
+    nac_match = re.search(r'NAC[:\s]+([0-9A-Fa-f]+)', line)
+    if nac_match:
+        return {
+            'type': 'nac',
+            'nac': nac_match.group(1),
+            'timestamp': ts,
+        }
+
+    # Voice frame detection — check BEFORE bare slot match
+    # Classic dsd: "Voice" keyword in frame lines
+    # dsd-fme: "voice" or "Voice LC" or "VOICE" in output
+    if re.search(r'\bvoice\b', line, re.IGNORECASE):
+        result = {
+            'type': 'voice',
+            'detail': line,
+            'timestamp': ts,
+        }
+        slot_inline = re.search(r'Slot\s*(\d+)', line)
+        if slot_inline:
+            result['slot'] = int(slot_inline.group(1))
+        return result
+
+    # Bare slot info (only when line is *just* slot info, not voice/call)
+    slot_match = re.match(r'\s*Slot\s*(\d+)\s*$', line)
+    if slot_match:
+        return {
+            'type': 'slot',
+            'slot': int(slot_match.group(1)),
+            'timestamp': ts,
+        }
+
+    # dsd-fme status lines we can surface: "TDMA", "CACH", "PI", "BS", etc.
+    # Also catches "Closing", "Input", and other lifecycle lines.
+    # Forward as raw so the frontend can show decoder is alive.
+    return {
+        'type': 'raw',
+        'text': line[:200],
+        'timestamp': ts,
+    }
+
+
+_HEARTBEAT_INTERVAL = 3.0  # seconds between heartbeats when decoder is idle
+
+
+def _queue_put(event: dict):
+    """Put an event on the DMR queue, dropping oldest if full."""
+    try:
+        dmr_queue.put_nowait(event)
+    except queue.Full:
+        try:
+            dmr_queue.get_nowait()
+        except queue.Empty:
+            pass
+        try:
+            dmr_queue.put_nowait(event)
+        except queue.Full:
+            pass
+
+
+def stream_dsd_output(rtl_process: subprocess.Popen, dsd_process: subprocess.Popen):
+    """Read DSD stderr output and push parsed events to the queue.
+
+    Uses select() with a timeout so we can send periodic heartbeat
+    events while readline() would otherwise block indefinitely during
+    silence (no signal being decoded).
+    """
+    global dmr_running
+
+    try:
+        _queue_put({'type': 'status', 'text': 'started'})
+        last_heartbeat = time.time()
+
+        while dmr_running:
+            if dsd_process.poll() is not None:
+                break
+
+            # Wait up to 1s for data on stderr instead of blocking forever
+            ready, _, _ = select.select([dsd_process.stderr], [], [], 1.0)
+
+            if ready:
+                line = dsd_process.stderr.readline()
+                if not line:
+                    if dsd_process.poll() is not None:
+                        break
+                    continue
+
+                text = line.decode('utf-8', errors='replace').strip()
+                if not text:
+                    continue
+
+                parsed = parse_dsd_output(text)
+                if parsed:
+                    _queue_put(parsed)
+                last_heartbeat = time.time()
+            else:
+                # No stderr output — send heartbeat so frontend knows
+                # decoder is still alive and listening
+                now = time.time()
+                if now - last_heartbeat >= _HEARTBEAT_INTERVAL:
+                    _queue_put({
+                        'type': 'heartbeat',
+                        'timestamp': datetime.now().strftime('%H:%M:%S'),
+                    })
+                    last_heartbeat = now
+
+    except Exception as e:
+        logger.error(f"DSD stream error: {e}")
+    finally:
+        global dmr_active_device, dmr_rtl_process, dmr_dsd_process
+        dmr_running = False
+        # Capture exit info for diagnostics
+        rc = dsd_process.poll()
+        reason = 'stopped'
+        detail = ''
+        if rc is not None and rc != 0:
+            reason = 'crashed'
+            try:
+                remaining = dsd_process.stderr.read(1024)
+                if remaining:
+                    detail = remaining.decode('utf-8', errors='replace').strip()[:200]
+            except Exception:
+                pass
+            logger.warning(f"DSD process exited with code {rc}: {detail}")
+        # Cleanup both processes
+        for proc in [dsd_process, rtl_process]:
+            if proc and proc.poll() is None:
+                try:
+                    proc.terminate()
+                    proc.wait(timeout=2)
+                except Exception:
+                    try:
+                        proc.kill()
+                    except Exception:
+                        pass
+            if proc:
+                unregister_process(proc)
+        dmr_rtl_process = None
+        dmr_dsd_process = None
+        _queue_put({'type': 'status', 'text': reason, 'exit_code': rc, 'detail': detail})
+        # Release SDR device
+        if dmr_active_device is not None:
+            app_module.release_sdr_device(dmr_active_device)
+            dmr_active_device = None
+        logger.info("DSD stream thread stopped")
+
+
+# ============================================
+# API ENDPOINTS
+# ============================================
+
+@dmr_bp.route('/tools')
+def check_tools() -> Response:
+    """Check for required tools."""
+    dsd_path, _ = find_dsd()
+    rtl_fm = find_rtl_fm()
+    return jsonify({
+        'dsd': dsd_path is not None,
+        'rtl_fm': rtl_fm is not None,
+        'available': dsd_path is not None and rtl_fm is not None,
+        'protocols': VALID_PROTOCOLS,
+    })
+
+
+@dmr_bp.route('/start', methods=['POST'])
+def start_dmr() -> Response:
+    """Start digital voice decoding."""
+    global dmr_rtl_process, dmr_dsd_process, dmr_thread, dmr_running, dmr_active_device
+
+    with dmr_lock:
+        if dmr_running:
+            return jsonify({'status': 'error', 'message': 'Already running'}), 409
+
+    dsd_path, is_fme = find_dsd()
+    if not dsd_path:
+        return jsonify({'status': 'error', 'message': 'dsd not found. Install dsd-fme or dsd.'}), 503
+
+    rtl_fm_path = find_rtl_fm()
+    if not rtl_fm_path:
+        return jsonify({'status': 'error', 'message': 'rtl_fm not found. Install rtl-sdr tools.'}), 503
+
+    data = request.json or {}
+
+    try:
+        frequency = float(data.get('frequency', 462.5625))
+        gain = int(data.get('gain', 40))
+        device = int(data.get('device', 0))
+        protocol = str(data.get('protocol', 'auto')).lower()
+    except (ValueError, TypeError) as e:
+        return jsonify({'status': 'error', 'message': f'Invalid parameter: {e}'}), 400
+
+    if frequency <= 0:
+        return jsonify({'status': 'error', 'message': 'Frequency must be positive'}), 400
+
+    if protocol not in VALID_PROTOCOLS:
+        return jsonify({'status': 'error', 'message': f'Invalid protocol. Use: {", ".join(VALID_PROTOCOLS)}'}), 400
+
+    # Clear stale queue
+    try:
+        while True:
+            dmr_queue.get_nowait()
+    except queue.Empty:
+        pass
+
+    # Claim SDR device
+    error = app_module.claim_sdr_device(device, 'dmr')
+    if error:
+        return jsonify({'status': 'error', 'error_type': 'DEVICE_BUSY', 'message': error}), 409
+
+    dmr_active_device = device
+
+    freq_hz = int(frequency * 1e6)
+
+    # Build rtl_fm command (48kHz sample rate for DSD)
+    rtl_cmd = [
+        rtl_fm_path,
+        '-M', 'fm',
+        '-f', str(freq_hz),
+        '-s', '48000',
+        '-g', str(gain),
+        '-d', str(device),
+        '-l', '1',  # squelch level
+    ]
+
+    # Build DSD command
+    # Use -o - to send decoded audio to stdout (piped to DEVNULL)
+    # instead of PulseAudio which may not be available under sudo
+    dsd_cmd = [dsd_path, '-i', '-', '-o', '-']
+    if is_fme:
+        dsd_cmd.extend(_DSD_FME_PROTOCOL_FLAGS.get(protocol, []))
+    else:
+        dsd_cmd.extend(_DSD_PROTOCOL_FLAGS.get(protocol, []))
+
+    try:
+        dmr_rtl_process = subprocess.Popen(
+            rtl_cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        register_process(dmr_rtl_process)
+
+        dmr_dsd_process = subprocess.Popen(
+            dsd_cmd,
+            stdin=dmr_rtl_process.stdout,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE,
+        )
+        register_process(dmr_dsd_process)
+
+        # Allow rtl_fm to send directly to dsd
+        dmr_rtl_process.stdout.close()
+
+        time.sleep(0.3)
+
+        rtl_rc = dmr_rtl_process.poll()
+        dsd_rc = dmr_dsd_process.poll()
+        if rtl_rc is not None or dsd_rc is not None:
+            # Process died — capture stderr for diagnostics
+            rtl_err = ''
+            if dmr_rtl_process.stderr:
+                rtl_err = dmr_rtl_process.stderr.read().decode('utf-8', errors='replace')[:500]
+            dsd_err = ''
+            if dmr_dsd_process.stderr:
+                dsd_err = dmr_dsd_process.stderr.read().decode('utf-8', errors='replace')[:500]
+            logger.error(f"DSD pipeline died: rtl_fm rc={rtl_rc} err={rtl_err!r}, dsd rc={dsd_rc} err={dsd_err!r}")
+            if dmr_active_device is not None:
+                app_module.release_sdr_device(dmr_active_device)
+                dmr_active_device = None
+            # Surface a clear error to the user
+            detail = rtl_err.strip() or dsd_err.strip()
+            if 'usb_claim_interface' in rtl_err or 'Failed to open' in rtl_err:
+                msg = f'SDR device {device} is busy — it may be in use by another mode or process. Try a different device.'
+            elif detail:
+                msg = f'Failed to start DSD pipeline: {detail}'
+            else:
+                msg = 'Failed to start DSD pipeline'
+            return jsonify({'status': 'error', 'message': msg}), 500
+
+        # Drain rtl_fm stderr in background to prevent pipe blocking
+        def _drain_rtl_stderr(proc):
+            try:
+                for line in proc.stderr:
+                    pass
+            except Exception:
+                pass
+
+        threading.Thread(target=_drain_rtl_stderr, args=(dmr_rtl_process,), daemon=True).start()
+
+        dmr_running = True
+        dmr_thread = threading.Thread(
+            target=stream_dsd_output,
+            args=(dmr_rtl_process, dmr_dsd_process),
+            daemon=True,
+        )
+        dmr_thread.start()
+
+        return jsonify({
+            'status': 'started',
+            'frequency': frequency,
+            'protocol': protocol,
+        })
+
+    except Exception as e:
+        logger.error(f"Failed to start DMR: {e}")
+        if dmr_active_device is not None:
+            app_module.release_sdr_device(dmr_active_device)
+            dmr_active_device = None
+        return jsonify({'status': 'error', 'message': str(e)}), 500
+
+
+@dmr_bp.route('/stop', methods=['POST'])
+def stop_dmr() -> Response:
+    """Stop digital voice decoding."""
+    global dmr_rtl_process, dmr_dsd_process, dmr_running, dmr_active_device
+
+    with dmr_lock:
+        dmr_running = False
+
+        for proc in [dmr_dsd_process, dmr_rtl_process]:
+            if proc and proc.poll() is None:
+                try:
+                    proc.terminate()
+                    proc.wait(timeout=2)
+                except Exception:
+                    try:
+                        proc.kill()
+                    except Exception:
+                        pass
+            if proc:
+                unregister_process(proc)
+
+        dmr_rtl_process = None
+        dmr_dsd_process = None
+
+    if dmr_active_device is not None:
+        app_module.release_sdr_device(dmr_active_device)
+        dmr_active_device = None
+
+    return jsonify({'status': 'stopped'})
+
+
+@dmr_bp.route('/status')
+def dmr_status() -> Response:
+    """Get DMR decoder status."""
+    return jsonify({
+        'running': dmr_running,
+        'device': dmr_active_device,
+    })
+
+
+@dmr_bp.route('/stream')
+def stream_dmr() -> Response:
+    """SSE stream for DMR decoder events."""
+    def generate() -> Generator[str, None, None]:
+        last_keepalive = time.time()
+        while True:
+            try:
+                msg = dmr_queue.get(timeout=SSE_QUEUE_TIMEOUT)
+                last_keepalive = time.time()
+                try:
+                    process_event('dmr', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
+            except queue.Empty:
+                now = time.time()
+                if now - last_keepalive >= SSE_KEEPALIVE_INTERVAL:
+                    yield format_sse({'type': 'keepalive'})
+                    last_keepalive = now
+
+    response = Response(generate(), mimetype='text/event-stream')
+    response.headers['Cache-Control'] = 'no-cache'
+    response.headers['X-Accel-Buffering'] = 'no'
+    return response

routes/dsc.py

@@ -36,9 +36,11 @@ from utils.database import (
 )
 from utils.dsc.parser import parse_dsc_message
 from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.validation import validate_device_index, validate_gain
 from utils.sdr import SDRFactory, SDRType
 from utils.dependencies import get_tool_path
+from utils.process import register_process, unregister_process
 
 logger = logging.getLogger('intercept.dsc')
 
@@ -169,17 +171,34 @@ def stream_dsc_decoder(master_fd: int, decoder_process: subprocess.Popen) -> Non
             'error': str(e)
         })
     finally:
+        global dsc_active_device
         try:
             os.close(master_fd)
         except OSError:
             pass
-        decoder_process.wait()
         dsc_running = False
+        # Cleanup both processes
+        with app_module.dsc_lock:
+            rtl_proc = app_module.dsc_rtl_process
+        for proc in [rtl_proc, decoder_process]:
+            if proc:
+                try:
+                    proc.terminate()
+                    proc.wait(timeout=2)
+                except Exception:
+                    try:
+                        proc.kill()
+                    except Exception:
+                        pass
+                unregister_process(proc)
         app_module.dsc_queue.put({'type': 'status', 'status': 'stopped'})
-
         with app_module.dsc_lock:
             app_module.dsc_process = None
             app_module.dsc_rtl_process = None
+        # Release SDR device
+        if dsc_active_device is not None:
+            app_module.release_sdr_device(dsc_active_device)
+            dsc_active_device = None
 
 
 def _store_critical_alert(msg: dict) -> None:
@@ -362,6 +381,7 @@ def start_decoding() -> Response:
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE
             )
+            register_process(rtl_process)
 
             # Start stderr monitor thread
             stderr_thread = threading.Thread(
@@ -382,6 +402,7 @@ def start_decoding() -> Response:
                 stderr=slave_fd,
                 close_fds=True
             )
+            register_process(decoder_process)
 
             os.close(slave_fd)
             rtl_process.stdout.close()
@@ -408,6 +429,15 @@ def start_decoding() -> Response:
             })
 
         except FileNotFoundError as e:
+            # Kill orphaned rtl_fm process
+            try:
+                rtl_process.terminate()
+                rtl_process.wait(timeout=2)
+            except Exception:
+                try:
+                    rtl_process.kill()
+                except Exception:
+                    pass
             # Release device on failure
             if dsc_active_device is not None:
                 app_module.release_sdr_device(dsc_active_device)
@@ -417,6 +447,15 @@ def start_decoding() -> Response:
                 'message': f'Tool not found: {e.filename}'
             }), 400
         except Exception as e:
+            # Kill orphaned rtl_fm process if it was started
+            try:
+                rtl_process.terminate()
+                rtl_process.wait(timeout=2)
+            except Exception:
+                try:
+                    rtl_process.kill()
+                except Exception:
+                    pass
             # Release device on failure
             if dsc_active_device is not None:
                 app_module.release_sdr_device(dsc_active_device)
@@ -487,6 +526,10 @@ def stream() -> Response:
             try:
                 msg = app_module.dsc_queue.get(timeout=1)
                 last_keepalive = time.time()
+                try:
+                    process_event('dsc', msg, msg.get('type'))
+                except Exception:
+                    pass
                 yield format_sse(msg)
             except queue.Empty:
                 now = time.time()

routes/listening_post.py

@@ -19,7 +19,8 @@ from flask import Blueprint, jsonify, request, Response
 
 import app as app_module
 from utils.logging import get_logger
-from utils.sse import format_sse
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.constants import (
     SSE_QUEUE_TIMEOUT,
     SSE_KEEPALIVE_INTERVAL,
@@ -101,6 +102,17 @@ def find_ffmpeg() -> str | None:
     return shutil.which('ffmpeg')
 
 
+VALID_MODULATIONS = ['fm', 'wfm', 'am', 'usb', 'lsb']
+
+
+def normalize_modulation(value: str) -> str:
+    """Normalize and validate modulation string."""
+    mod = str(value or '').lower().strip()
+    if mod not in VALID_MODULATIONS:
+        raise ValueError(f'Invalid modulation. Use: {", ".join(VALID_MODULATIONS)}')
+    return mod
+
+
 
 
 def add_activity_log(event_type: str, frequency: float, details: str = ''):
@@ -724,31 +736,52 @@ def _start_audio_stream(frequency: float, modulation: str):
         ]
 
         try:
-            # Use shell pipe for reliable streaming
-            # Log stderr to temp files for error diagnosis
+            # Use subprocess piping for reliable streaming.
+            # Log stderr to temp files for error diagnosis.
             rtl_stderr_log = '/tmp/rtl_fm_stderr.log'
             ffmpeg_stderr_log = '/tmp/ffmpeg_stderr.log'
-            shell_cmd = f"{' '.join(sdr_cmd)} 2>{rtl_stderr_log} | {' '.join(encoder_cmd)} 2>{ffmpeg_stderr_log}"
             logger.info(f"Starting audio: {frequency} MHz, mod={modulation}, device={scanner_config['device']}")
 
             # Retry loop for USB device contention (device may not be
             # released immediately after a previous process exits)
             max_attempts = 3
             for attempt in range(max_attempts):
-                audio_rtl_process = None  # Not used in shell mode
-                audio_process = subprocess.Popen(
-                    shell_cmd,
-                    shell=True,
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
-                    bufsize=0,
-                    start_new_session=True  # Create new process group for clean shutdown
-                )
+                audio_rtl_process = None
+                audio_process = None
+                rtl_err_handle = None
+                ffmpeg_err_handle = None
+                try:
+                    rtl_err_handle = open(rtl_stderr_log, 'w')
+                    ffmpeg_err_handle = open(ffmpeg_stderr_log, 'w')
+                    audio_rtl_process = subprocess.Popen(
+                        sdr_cmd,
+                        stdout=subprocess.PIPE,
+                        stderr=rtl_err_handle,
+                        bufsize=0,
+                        start_new_session=True  # Create new process group for clean shutdown
+                    )
+                    audio_process = subprocess.Popen(
+                        encoder_cmd,
+                        stdin=audio_rtl_process.stdout,
+                        stdout=subprocess.PIPE,
+                        stderr=ffmpeg_err_handle,
+                        bufsize=0,
+                        start_new_session=True  # Create new process group for clean shutdown
+                    )
+                    if audio_rtl_process.stdout:
+                        audio_rtl_process.stdout.close()
+                finally:
+                    if rtl_err_handle:
+                        rtl_err_handle.close()
+                    if ffmpeg_err_handle:
+                        ffmpeg_err_handle.close()
 
                 # Brief delay to check if process started successfully
                 time.sleep(0.3)
 
-                if audio_process.poll() is not None:
+                if (audio_rtl_process and audio_rtl_process.poll() is not None) or (
+                    audio_process and audio_process.poll() is not None
+                ):
                     # Read stderr from temp files
                     rtl_stderr = ''
                     ffmpeg_stderr = ''
@@ -765,10 +798,39 @@ def _start_audio_stream(frequency: float, modulation: str):
 
                     if 'usb_claim_interface' in rtl_stderr and attempt < max_attempts - 1:
                         logger.warning(f"USB device busy (attempt {attempt + 1}/{max_attempts}), waiting for release...")
+                        if audio_process:
+                            try:
+                                audio_process.terminate()
+                                audio_process.wait(timeout=0.5)
+                            except Exception:
+                                pass
+                        if audio_rtl_process:
+                            try:
+                                audio_rtl_process.terminate()
+                                audio_rtl_process.wait(timeout=0.5)
+                            except Exception:
+                                pass
                         time.sleep(1.0)
                         continue
 
-                    logger.error(f"Audio pipeline exited immediately. rtl_fm stderr: {rtl_stderr}, ffmpeg stderr: {ffmpeg_stderr}")
+                    if audio_process and audio_process.poll() is None:
+                        try:
+                            audio_process.terminate()
+                            audio_process.wait(timeout=0.5)
+                        except Exception:
+                            pass
+                    if audio_rtl_process and audio_rtl_process.poll() is None:
+                        try:
+                            audio_rtl_process.terminate()
+                            audio_rtl_process.wait(timeout=0.5)
+                        except Exception:
+                            pass
+                    audio_process = None
+                    audio_rtl_process = None
+
+                    logger.error(
+                        f"Audio pipeline exited immediately. rtl_fm stderr: {rtl_stderr}, ffmpeg stderr: {ffmpeg_stderr}"
+                    )
                     return
 
                 # Pipeline started successfully
@@ -778,9 +840,13 @@ def _start_audio_stream(frequency: float, modulation: str):
             try:
                 ready, _, _ = select.select([audio_process.stdout], [], [], 4.0)
                 if not ready:
-                    logger.warning("Audio pipeline produced no data in startup window")
+                    logger.warning("Audio pipeline produced no data in startup window — killing stalled pipeline")
+                    _stop_audio_stream_internal()
+                    return
             except Exception as e:
                 logger.warning(f"Audio startup check failed: {e}")
+                _stop_audio_stream_internal()
+                return
 
             audio_running = True
             audio_frequency = frequency
@@ -805,34 +871,36 @@ def _stop_audio_stream_internal():
     audio_running = False
     audio_frequency = 0.0
 
-    # Kill the shell process and its children
+    had_processes = audio_process is not None or audio_rtl_process is not None
+
+    # Kill the pipeline processes and their groups
     if audio_process:
         try:
-            # Kill entire process group (rtl_fm, ffmpeg, shell)
+            # Kill entire process group (SDR demod + ffmpeg)
             try:
                 os.killpg(os.getpgid(audio_process.pid), signal.SIGKILL)
             except (ProcessLookupError, PermissionError):
                 audio_process.kill()
             audio_process.wait(timeout=0.5)
-        except:
+        except Exception:
+            pass
+
+    if audio_rtl_process:
+        try:
+            try:
+                os.killpg(os.getpgid(audio_rtl_process.pid), signal.SIGKILL)
+            except (ProcessLookupError, PermissionError):
+                audio_rtl_process.kill()
+            audio_rtl_process.wait(timeout=0.5)
+        except Exception:
             pass
 
     audio_process = None
     audio_rtl_process = None
 
-    # Kill any orphaned rtl_fm, rtl_power, and ffmpeg processes
-    for proc_pattern in ['rtl_fm', 'rtl_power']:
-        try:
-            subprocess.run(['pkill', '-9', proc_pattern], capture_output=True, timeout=0.5)
-        except Exception:
-            pass
-    try:
-        subprocess.run(['pkill', '-9', '-f', 'ffmpeg.*pipe:0'], capture_output=True, timeout=0.5)
-    except Exception:
-        pass
-
     # Pause for SDR device USB interface to be released by kernel
-    time.sleep(1.0)
+    if had_processes:
+        time.sleep(1.0)
 
 
 # ============================================
@@ -891,7 +959,7 @@ def start_scanner() -> Response:
         scanner_config['start_freq'] = float(data.get('start_freq', 88.0))
         scanner_config['end_freq'] = float(data.get('end_freq', 108.0))
         scanner_config['step'] = float(data.get('step', 0.1))
-        scanner_config['modulation'] = str(data.get('modulation', 'wfm')).lower()
+        scanner_config['modulation'] = normalize_modulation(data.get('modulation', 'wfm'))
         scanner_config['squelch'] = int(data.get('squelch', 0))
         scanner_config['dwell_time'] = float(data.get('dwell_time', 3.0))
         scanner_config['scan_delay'] = float(data.get('scan_delay', 0.5))
@@ -1074,8 +1142,14 @@ def update_scanner_config() -> Response:
         updated.append(f"dwell={data['dwell_time']}s")
 
     if 'modulation' in data:
-        scanner_config['modulation'] = str(data['modulation']).lower()
-        updated.append(f"mod={data['modulation']}")
+        try:
+            scanner_config['modulation'] = normalize_modulation(data['modulation'])
+            updated.append(f"mod={data['modulation']}")
+        except (ValueError, TypeError) as e:
+            return jsonify({
+                'status': 'error',
+                'message': str(e)
+            }), 400
 
     if updated:
         logger.info(f"Scanner config updated: {', '.join(updated)}")
@@ -1107,9 +1181,13 @@ def stream_scanner_events() -> Response:
 
         while True:
             try:
-                msg = scanner_queue.get(timeout=SSE_QUEUE_TIMEOUT)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = scanner_queue.get(timeout=SSE_QUEUE_TIMEOUT)
+                last_keepalive = time.time()
+                try:
+                    process_event('listening_scanner', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= SSE_KEEPALIVE_INTERVAL:
@@ -1161,10 +1239,10 @@ def get_presets() -> Response:
 # MANUAL AUDIO ENDPOINTS (for direct listening)
 # ============================================
 
-@listening_post_bp.route('/audio/start', methods=['POST'])
-def start_audio() -> Response:
-    """Start audio at specific frequency (manual mode)."""
-    global scanner_running, scanner_active_device, listening_active_device, scanner_power_process, scanner_thread
+@listening_post_bp.route('/audio/start', methods=['POST'])
+def start_audio() -> Response:
+    """Start audio at specific frequency (manual mode)."""
+    global scanner_running, scanner_active_device, listening_active_device, scanner_power_process, scanner_thread
 
     # Stop scanner if running
     if scanner_running:
@@ -1193,11 +1271,11 @@ def start_audio() -> Response:
             pass
         time.sleep(0.5)
 
-    data = request.json or {}
+    data = request.json or {}
 
     try:
         frequency = float(data.get('frequency', 0))
-        modulation = str(data.get('modulation', 'wfm')).lower()
+        modulation = normalize_modulation(data.get('modulation', 'wfm'))
         squelch = int(data.get('squelch', 0))
         gain = int(data.get('gain', 40))
         device = int(data.get('device', 0))
@@ -1208,18 +1286,11 @@ def start_audio() -> Response:
             'message': f'Invalid parameter: {e}'
         }), 400
 
-    if frequency <= 0:
-        return jsonify({
-            'status': 'error',
-            'message': 'frequency is required'
-        }), 400
-
-    valid_mods = ['fm', 'wfm', 'am', 'usb', 'lsb']
-    if modulation not in valid_mods:
-        return jsonify({
-            'status': 'error',
-            'message': f'Invalid modulation. Use: {", ".join(valid_mods)}'
-        }), 400
+    if frequency <= 0:
+        return jsonify({
+            'status': 'error',
+            'message': 'frequency is required'
+        }), 400
 
     valid_sdr_types = ['rtlsdr', 'hackrf', 'airspy', 'limesdr', 'sdrplay']
     if sdr_type not in valid_sdr_types:
@@ -1228,14 +1299,19 @@ def start_audio() -> Response:
             'message': f'Invalid sdr_type. Use: {", ".join(valid_sdr_types)}'
         }), 400
 
-    # Update config for audio
-    scanner_config['squelch'] = squelch
-    scanner_config['gain'] = gain
-    scanner_config['device'] = device
-    scanner_config['sdr_type'] = sdr_type
+    # Update config for audio
+    scanner_config['squelch'] = squelch
+    scanner_config['gain'] = gain
+    scanner_config['device'] = device
+    scanner_config['sdr_type'] = sdr_type
+
+    # Stop waterfall if it's using the same SDR
+    if waterfall_running and waterfall_active_device == device:
+        _stop_waterfall_internal()
+        time.sleep(0.2)
 
-    # Claim device for listening audio
-    if listening_active_device is None or listening_active_device != device:
+    # Claim device for listening audio
+    if listening_active_device is None or listening_active_device != device:
         if listening_active_device is not None:
             app_module.release_sdr_device(listening_active_device)
         error = app_module.claim_sdr_device(device, 'listening')
@@ -1341,13 +1417,6 @@ def audio_probe() -> Response:
 @listening_post_bp.route('/audio/stream')
 def stream_audio() -> Response:
     """Stream WAV audio."""
-    # Optionally restart pipeline so the stream starts with a fresh header
-    if request.args.get('fresh') == '1' and audio_running:
-        try:
-            _start_audio_stream(audio_frequency or 0.0, audio_modulation or 'fm')
-        except Exception as e:
-            logger.error(f"Audio stream restart failed: {e}")
-
     # Wait for audio to be ready (up to 2 seconds for modulation/squelch changes)
     for _ in range(40):
         if audio_running and audio_process:
@@ -1397,3 +1466,362 @@ def stream_audio() -> Response:
             'Transfer-Encoding': 'chunked',
         }
     )
+
+
+# ============================================
+# SIGNAL IDENTIFICATION ENDPOINT
+# ============================================
+
+@listening_post_bp.route('/signal/guess', methods=['POST'])
+def guess_signal() -> Response:
+    """Identify a signal based on frequency, modulation, and other parameters."""
+    data = request.json or {}
+
+    freq_mhz = data.get('frequency_mhz')
+    if freq_mhz is None:
+        return jsonify({'status': 'error', 'message': 'frequency_mhz is required'}), 400
+
+    try:
+        freq_mhz = float(freq_mhz)
+    except (ValueError, TypeError):
+        return jsonify({'status': 'error', 'message': 'Invalid frequency_mhz'}), 400
+
+    if freq_mhz <= 0:
+        return jsonify({'status': 'error', 'message': 'frequency_mhz must be positive'}), 400
+
+    frequency_hz = int(freq_mhz * 1e6)
+
+    modulation = data.get('modulation')
+    bandwidth_hz = data.get('bandwidth_hz')
+    if bandwidth_hz is not None:
+        try:
+            bandwidth_hz = int(bandwidth_hz)
+        except (ValueError, TypeError):
+            bandwidth_hz = None
+
+    region = data.get('region', 'UK/EU')
+
+    try:
+        from utils.signal_guess import guess_signal_type_dict
+        result = guess_signal_type_dict(
+            frequency_hz=frequency_hz,
+            modulation=modulation,
+            bandwidth_hz=bandwidth_hz,
+            region=region,
+        )
+        return jsonify({'status': 'ok', **result})
+    except Exception as e:
+        logger.error(f"Signal guess error: {e}")
+        return jsonify({'status': 'error', 'message': str(e)}), 500
+
+
+# ============================================
+# WATERFALL / SPECTROGRAM ENDPOINTS
+# ============================================
+
+waterfall_process: Optional[subprocess.Popen] = None
+waterfall_thread: Optional[threading.Thread] = None
+waterfall_running = False
+waterfall_lock = threading.Lock()
+waterfall_queue: queue.Queue = queue.Queue(maxsize=200)
+waterfall_active_device: Optional[int] = None
+waterfall_config = {
+    'start_freq': 88.0,
+    'end_freq': 108.0,
+    'bin_size': 10000,
+    'gain': 40,
+    'device': 0,
+    'max_bins': 1024,
+    'interval': 0.4,
+}
+
+
+def _parse_rtl_power_line(line: str) -> tuple[str | None, float | None, float | None, list[float]]:
+    """Parse a single rtl_power CSV line into bins."""
+    if not line or line.startswith('#'):
+        return None, None, None, []
+
+    parts = [p.strip() for p in line.split(',')]
+    if len(parts) < 6:
+        return None, None, None, []
+
+    # Timestamp in first two fields (YYYY-MM-DD, HH:MM:SS)
+    timestamp = f"{parts[0]} {parts[1]}" if len(parts) >= 2 else parts[0]
+
+    start_idx = None
+    for i, tok in enumerate(parts):
+        try:
+            val = float(tok)
+        except ValueError:
+            continue
+        if val > 1e5:
+            start_idx = i
+            break
+    if start_idx is None or len(parts) < start_idx + 4:
+        return timestamp, None, None, []
+
+    try:
+        seg_start = float(parts[start_idx])
+        seg_end = float(parts[start_idx + 1])
+        raw_values = []
+        for v in parts[start_idx + 3:]:
+            try:
+                raw_values.append(float(v))
+            except ValueError:
+                continue
+        if raw_values and raw_values[0] >= 0 and any(val < 0 for val in raw_values[1:]):
+            raw_values = raw_values[1:]
+        return timestamp, seg_start, seg_end, raw_values
+    except ValueError:
+        return timestamp, None, None, []
+
+
+def _waterfall_loop():
+    """Continuous rtl_power sweep loop emitting waterfall data."""
+    global waterfall_running, waterfall_process
+
+    rtl_power_path = find_rtl_power()
+    if not rtl_power_path:
+        logger.error("rtl_power not found for waterfall")
+        waterfall_running = False
+        return
+
+    start_hz = int(waterfall_config['start_freq'] * 1e6)
+    end_hz = int(waterfall_config['end_freq'] * 1e6)
+    bin_hz = int(waterfall_config['bin_size'])
+    gain = waterfall_config['gain']
+    device = waterfall_config['device']
+    interval = float(waterfall_config.get('interval', 0.4))
+
+    cmd = [
+        rtl_power_path,
+        '-f', f'{start_hz}:{end_hz}:{bin_hz}',
+        '-i', str(interval),
+        '-g', str(gain),
+        '-d', str(device),
+    ]
+
+    try:
+        waterfall_process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            bufsize=1,
+            text=True,
+        )
+
+        current_ts = None
+        all_bins: list[float] = []
+        sweep_start_hz = start_hz
+        sweep_end_hz = end_hz
+
+        if not waterfall_process.stdout:
+            return
+
+        for line in waterfall_process.stdout:
+            if not waterfall_running:
+                break
+
+            ts, seg_start, seg_end, bins = _parse_rtl_power_line(line)
+            if ts is None or not bins:
+                continue
+
+            if current_ts is None:
+                current_ts = ts
+
+            if ts != current_ts and all_bins:
+                max_bins = int(waterfall_config.get('max_bins') or 0)
+                bins_to_send = all_bins
+                if max_bins > 0 and len(bins_to_send) > max_bins:
+                    bins_to_send = _downsample_bins(bins_to_send, max_bins)
+                msg = {
+                    'type': 'waterfall_sweep',
+                    'start_freq': sweep_start_hz / 1e6,
+                    'end_freq': sweep_end_hz / 1e6,
+                    'bins': bins_to_send,
+                    'timestamp': datetime.now().isoformat(),
+                }
+                try:
+                    waterfall_queue.put_nowait(msg)
+                except queue.Full:
+                    try:
+                        waterfall_queue.get_nowait()
+                    except queue.Empty:
+                        pass
+                    try:
+                        waterfall_queue.put_nowait(msg)
+                    except queue.Full:
+                        pass
+
+                all_bins = []
+                sweep_start_hz = start_hz
+                sweep_end_hz = end_hz
+                current_ts = ts
+
+            all_bins.extend(bins)
+            if seg_start is not None:
+                sweep_start_hz = min(sweep_start_hz, seg_start)
+            if seg_end is not None:
+                sweep_end_hz = max(sweep_end_hz, seg_end)
+
+        # Flush any remaining bins
+        if all_bins and waterfall_running:
+            max_bins = int(waterfall_config.get('max_bins') or 0)
+            bins_to_send = all_bins
+            if max_bins > 0 and len(bins_to_send) > max_bins:
+                bins_to_send = _downsample_bins(bins_to_send, max_bins)
+            msg = {
+                'type': 'waterfall_sweep',
+                'start_freq': sweep_start_hz / 1e6,
+                'end_freq': sweep_end_hz / 1e6,
+                'bins': bins_to_send,
+                'timestamp': datetime.now().isoformat(),
+            }
+            try:
+                waterfall_queue.put_nowait(msg)
+            except queue.Full:
+                pass
+
+    except Exception as e:
+        logger.error(f"Waterfall loop error: {e}")
+    finally:
+        waterfall_running = False
+        if waterfall_process and waterfall_process.poll() is None:
+            try:
+                waterfall_process.terminate()
+                waterfall_process.wait(timeout=1)
+            except Exception:
+                try:
+                    waterfall_process.kill()
+                except Exception:
+                    pass
+        waterfall_process = None
+        logger.info("Waterfall loop stopped")
+
+
+def _stop_waterfall_internal() -> None:
+    """Stop the waterfall display and release resources."""
+    global waterfall_running, waterfall_process, waterfall_active_device
+
+    waterfall_running = False
+    if waterfall_process and waterfall_process.poll() is None:
+        try:
+            waterfall_process.terminate()
+            waterfall_process.wait(timeout=1)
+        except Exception:
+            try:
+                waterfall_process.kill()
+            except Exception:
+                pass
+        waterfall_process = None
+
+    if waterfall_active_device is not None:
+        app_module.release_sdr_device(waterfall_active_device)
+        waterfall_active_device = None
+
+
+@listening_post_bp.route('/waterfall/start', methods=['POST'])
+def start_waterfall() -> Response:
+    """Start the waterfall/spectrogram display."""
+    global waterfall_thread, waterfall_running, waterfall_config, waterfall_active_device
+
+    with waterfall_lock:
+        if waterfall_running:
+            return jsonify({'status': 'error', 'message': 'Waterfall already running'}), 409
+
+    if not find_rtl_power():
+        return jsonify({'status': 'error', 'message': 'rtl_power not found'}), 503
+
+    data = request.json or {}
+
+    try:
+        waterfall_config['start_freq'] = float(data.get('start_freq', 88.0))
+        waterfall_config['end_freq'] = float(data.get('end_freq', 108.0))
+        waterfall_config['bin_size'] = int(data.get('bin_size', 10000))
+        waterfall_config['gain'] = int(data.get('gain', 40))
+        waterfall_config['device'] = int(data.get('device', 0))
+        if data.get('interval') is not None:
+            interval = float(data.get('interval', waterfall_config['interval']))
+            if interval < 0.1 or interval > 5:
+                return jsonify({'status': 'error', 'message': 'interval must be between 0.1 and 5 seconds'}), 400
+            waterfall_config['interval'] = interval
+        if data.get('max_bins') is not None:
+            max_bins = int(data.get('max_bins', waterfall_config['max_bins']))
+            if max_bins < 64 or max_bins > 4096:
+                return jsonify({'status': 'error', 'message': 'max_bins must be between 64 and 4096'}), 400
+            waterfall_config['max_bins'] = max_bins
+    except (ValueError, TypeError) as e:
+        return jsonify({'status': 'error', 'message': f'Invalid parameter: {e}'}), 400
+
+    if waterfall_config['start_freq'] >= waterfall_config['end_freq']:
+        return jsonify({'status': 'error', 'message': 'start_freq must be less than end_freq'}), 400
+
+    # Clear stale queue
+    try:
+        while True:
+            waterfall_queue.get_nowait()
+    except queue.Empty:
+        pass
+
+    # Claim SDR device
+    error = app_module.claim_sdr_device(waterfall_config['device'], 'waterfall')
+    if error:
+        return jsonify({'status': 'error', 'error_type': 'DEVICE_BUSY', 'message': error}), 409
+
+    waterfall_active_device = waterfall_config['device']
+    waterfall_running = True
+    waterfall_thread = threading.Thread(target=_waterfall_loop, daemon=True)
+    waterfall_thread.start()
+
+    return jsonify({'status': 'started', 'config': waterfall_config})
+
+
+@listening_post_bp.route('/waterfall/stop', methods=['POST'])
+def stop_waterfall() -> Response:
+    """Stop the waterfall display."""
+    _stop_waterfall_internal()
+
+    return jsonify({'status': 'stopped'})
+
+
+@listening_post_bp.route('/waterfall/stream')
+def stream_waterfall() -> Response:
+    """SSE stream for waterfall data."""
+    def generate() -> Generator[str, None, None]:
+        last_keepalive = time.time()
+        while True:
+            try:
+                msg = waterfall_queue.get(timeout=SSE_QUEUE_TIMEOUT)
+                last_keepalive = time.time()
+                try:
+                    process_event('waterfall', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
+            except queue.Empty:
+                now = time.time()
+                if now - last_keepalive >= SSE_KEEPALIVE_INTERVAL:
+                    yield format_sse({'type': 'keepalive'})
+                    last_keepalive = now
+
+    response = Response(generate(), mimetype='text/event-stream')
+    response.headers['Cache-Control'] = 'no-cache'
+    response.headers['X-Accel-Buffering'] = 'no'
+    return response
+def _downsample_bins(values: list[float], target: int) -> list[float]:
+    """Downsample bins to a target length using simple averaging."""
+    if target <= 0 or len(values) <= target:
+        return values
+
+    out: list[float] = []
+    step = len(values) / target
+    for i in range(target):
+        start = int(i * step)
+        end = int((i + 1) * step)
+        if end <= start:
+            end = min(start + 1, len(values))
+        chunk = values[start:end]
+        if not chunk:
+            continue
+        out.append(sum(chunk) / len(chunk))
+    return out

routes/pager.py

@@ -22,8 +22,9 @@ from utils.validation import (
     validate_frequency, validate_device_index, validate_gain, validate_ppm,
     validate_rtl_tcp_host, validate_rtl_tcp_port
 )
-from utils.sse import format_sse
-from utils.process import safe_terminate, register_process
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from utils.process import safe_terminate, register_process, unregister_process
 from utils.sdr import SDRFactory, SDRType, SDRValidationError
 from utils.dependencies import get_tool_path
 
@@ -146,14 +147,32 @@ def stream_decoder(master_fd: int, process: subprocess.Popen[bytes]) -> None:
     except Exception as e:
         app_module.output_queue.put({'type': 'error', 'text': str(e)})
     finally:
+        global pager_active_device
         try:
             os.close(master_fd)
         except OSError:
             pass
-        process.wait()
+        # Cleanup companion rtl_fm process and decoder
+        with app_module.process_lock:
+            rtl_proc = getattr(app_module.current_process, '_rtl_process', None)
+        for proc in [rtl_proc, process]:
+            if proc:
+                try:
+                    proc.terminate()
+                    proc.wait(timeout=2)
+                except Exception:
+                    try:
+                        proc.kill()
+                    except Exception:
+                        pass
+                unregister_process(proc)
         app_module.output_queue.put({'type': 'status', 'text': 'stopped'})
         with app_module.process_lock:
             app_module.current_process = None
+        # Release SDR device
+        if pager_active_device is not None:
+            app_module.release_sdr_device(pager_active_device)
+            pager_active_device = None
 
 
 @pager_bp.route('/start', methods=['POST'])
@@ -281,6 +300,7 @@ def start_decoding() -> Response:
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE
             )
+            register_process(rtl_process)
 
             # Start a thread to monitor rtl_fm stderr for errors
             def monitor_rtl_stderr():
@@ -304,6 +324,7 @@ def start_decoding() -> Response:
                 stderr=slave_fd,
                 close_fds=True
             )
+            register_process(multimon_process)
 
             os.close(slave_fd)
             rtl_process.stdout.close()
@@ -322,12 +343,30 @@ def start_decoding() -> Response:
             return jsonify({'status': 'started', 'command': full_cmd})
 
         except FileNotFoundError as e:
+            # Kill orphaned rtl_fm process
+            try:
+                rtl_process.terminate()
+                rtl_process.wait(timeout=2)
+            except Exception:
+                try:
+                    rtl_process.kill()
+                except Exception:
+                    pass
             # Release device on failure
             if pager_active_device is not None:
                 app_module.release_sdr_device(pager_active_device)
                 pager_active_device = None
             return jsonify({'status': 'error', 'message': f'Tool not found: {e.filename}'})
         except Exception as e:
+            # Kill orphaned rtl_fm process if it was started
+            try:
+                rtl_process.terminate()
+                rtl_process.wait(timeout=2)
+            except Exception:
+                try:
+                    rtl_process.kill()
+                except Exception:
+                    pass
             # Release device on failure
             if pager_active_device is not None:
                 app_module.release_sdr_device(pager_active_device)
@@ -430,10 +469,14 @@ def stream() -> Response:
         keepalive_interval = 30.0  # Send keepalive every 30 seconds instead of 1 second
 
         while True:
-            try:
-                msg = app_module.output_queue.get(timeout=1)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+            try:
+                msg = app_module.output_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('pager', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= keepalive_interval:

routes/recordings.py

@@ -0,0 +1,109 @@
+"""Session recording API endpoints."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from flask import Blueprint, jsonify, request, send_file
+
+from utils.recording import get_recording_manager, RECORDING_ROOT
+
+recordings_bp = Blueprint('recordings', __name__, url_prefix='/recordings')
+
+
+@recordings_bp.route('/start', methods=['POST'])
+def start_recording():
+    data = request.get_json() or {}
+    mode = (data.get('mode') or '').strip()
+    if not mode:
+        return jsonify({'status': 'error', 'message': 'mode is required'}), 400
+
+    label = data.get('label')
+    metadata = data.get('metadata') if isinstance(data.get('metadata'), dict) else {}
+
+    manager = get_recording_manager()
+    session = manager.start_recording(mode=mode, label=label, metadata=metadata)
+
+    return jsonify({
+        'status': 'success',
+        'session': {
+            'id': session.id,
+            'mode': session.mode,
+            'label': session.label,
+            'started_at': session.started_at.isoformat(),
+            'file_path': str(session.file_path),
+        }
+    })
+
+
+@recordings_bp.route('/stop', methods=['POST'])
+def stop_recording():
+    data = request.get_json() or {}
+    mode = data.get('mode')
+    session_id = data.get('id')
+
+    manager = get_recording_manager()
+    session = manager.stop_recording(mode=mode, session_id=session_id)
+    if not session:
+        return jsonify({'status': 'error', 'message': 'No active recording found'}), 404
+
+    return jsonify({
+        'status': 'success',
+        'session': {
+            'id': session.id,
+            'mode': session.mode,
+            'label': session.label,
+            'started_at': session.started_at.isoformat(),
+            'stopped_at': session.stopped_at.isoformat() if session.stopped_at else None,
+            'event_count': session.event_count,
+            'size_bytes': session.size_bytes,
+            'file_path': str(session.file_path),
+        }
+    })
+
+
+@recordings_bp.route('', methods=['GET'])
+def list_recordings():
+    manager = get_recording_manager()
+    limit = request.args.get('limit', default=50, type=int)
+    return jsonify({
+        'status': 'success',
+        'recordings': manager.list_recordings(limit=limit),
+        'active': manager.get_active(),
+    })
+
+
+@recordings_bp.route('/<session_id>', methods=['GET'])
+def get_recording(session_id: str):
+    manager = get_recording_manager()
+    rec = manager.get_recording(session_id)
+    if not rec:
+        return jsonify({'status': 'error', 'message': 'Recording not found'}), 404
+    return jsonify({'status': 'success', 'recording': rec})
+
+
+@recordings_bp.route('/<session_id>/download', methods=['GET'])
+def download_recording(session_id: str):
+    manager = get_recording_manager()
+    rec = manager.get_recording(session_id)
+    if not rec:
+        return jsonify({'status': 'error', 'message': 'Recording not found'}), 404
+
+    file_path = Path(rec['file_path'])
+    try:
+        resolved_root = RECORDING_ROOT.resolve()
+        resolved_file = file_path.resolve()
+        if resolved_root not in resolved_file.parents:
+            return jsonify({'status': 'error', 'message': 'Invalid recording path'}), 400
+    except Exception:
+        return jsonify({'status': 'error', 'message': 'Invalid recording path'}), 400
+
+    if not file_path.exists():
+        return jsonify({'status': 'error', 'message': 'Recording file missing'}), 404
+
+    return send_file(
+        file_path,
+        mimetype='application/x-ndjson',
+        as_attachment=True,
+        download_name=file_path.name,
+    )

routes/rtlamr.py

@@ -18,7 +18,8 @@ from utils.validation import (
     validate_frequency, validate_device_index, validate_gain, validate_ppm
 )
 from utils.sse import format_sse
-from utils.process import safe_terminate, register_process
+from utils.event_pipeline import process_event
+from utils.process import safe_terminate, register_process, unregister_process
 
 rtlamr_bp = Blueprint('rtlamr', __name__)
 
@@ -61,10 +62,37 @@ def stream_rtlamr_output(process: subprocess.Popen[bytes]) -> None:
     except Exception as e:
         app_module.rtlamr_queue.put({'type': 'error', 'text': str(e)})
     finally:
-        process.wait()
+        global rtl_tcp_process, rtlamr_active_device
+        # Ensure rtlamr process is terminated
+        try:
+            process.terminate()
+            process.wait(timeout=2)
+        except Exception:
+            try:
+                process.kill()
+            except Exception:
+                pass
+        unregister_process(process)
+        # Kill companion rtl_tcp process
+        with rtl_tcp_lock:
+            if rtl_tcp_process:
+                try:
+                    rtl_tcp_process.terminate()
+                    rtl_tcp_process.wait(timeout=2)
+                except Exception:
+                    try:
+                        rtl_tcp_process.kill()
+                    except Exception:
+                        pass
+                unregister_process(rtl_tcp_process)
+                rtl_tcp_process = None
         app_module.rtlamr_queue.put({'type': 'status', 'text': 'stopped'})
         with app_module.rtlamr_lock:
             app_module.rtlamr_process = None
+        # Release SDR device
+        if rtlamr_active_device is not None:
+            app_module.release_sdr_device(rtlamr_active_device)
+            rtlamr_active_device = None
 
 
 @rtlamr_bp.route('/start_rtlamr', methods=['POST'])
@@ -133,7 +161,8 @@ def start_rtlamr() -> Response:
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE
                     )
-                    
+                    register_process(rtl_tcp_process)
+
                     # Wait a moment for rtl_tcp to start
                     time.sleep(3)
                     
@@ -141,6 +170,10 @@ def start_rtlamr() -> Response:
                     app_module.rtlamr_queue.put({'type': 'info', 'text': f'rtl_tcp: {" ".join(rtl_tcp_cmd)}'})
                 except Exception as e:
                     logger.error(f"Failed to start rtl_tcp: {e}")
+                    # Release SDR device on rtl_tcp failure
+                    if rtlamr_active_device is not None:
+                        app_module.release_sdr_device(rtlamr_active_device)
+                        rtlamr_active_device = None
                     return jsonify({'status': 'error', 'message': f'Failed to start rtl_tcp: {e}'}), 500
 
         # Build rtlamr command
@@ -174,6 +207,7 @@ def start_rtlamr() -> Response:
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE
             )
+            register_process(app_module.rtlamr_process)
 
             # Start output thread
             thread = threading.Thread(target=stream_rtlamr_output, args=(app_module.rtlamr_process,))
@@ -262,6 +296,10 @@ def stream_rtlamr() -> Response:
             try:
                 msg = app_module.rtlamr_queue.get(timeout=1)
                 last_keepalive = time.time()
+                try:
+                    process_event('rtlamr', msg, msg.get('type'))
+                except Exception:
+                    pass
                 yield format_sse(msg)
             except queue.Empty:
                 now = time.time()

routes/sensor.py

@@ -18,8 +18,9 @@ from utils.validation import (
     validate_frequency, validate_device_index, validate_gain, validate_ppm,
     validate_rtl_tcp_host, validate_rtl_tcp_port
 )
-from utils.sse import format_sse
-from utils.process import safe_terminate, register_process
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from utils.process import safe_terminate, register_process, unregister_process
 from utils.sdr import SDRFactory, SDRType
 
 sensor_bp = Blueprint('sensor', __name__)
@@ -59,10 +60,24 @@ def stream_sensor_output(process: subprocess.Popen[bytes]) -> None:
     except Exception as e:
         app_module.sensor_queue.put({'type': 'error', 'text': str(e)})
     finally:
-        process.wait()
+        global sensor_active_device
+        # Ensure process is terminated
+        try:
+            process.terminate()
+            process.wait(timeout=2)
+        except Exception:
+            try:
+                process.kill()
+            except Exception:
+                pass
+        unregister_process(process)
         app_module.sensor_queue.put({'type': 'status', 'text': 'stopped'})
         with app_module.sensor_lock:
             app_module.sensor_process = None
+        # Release SDR device
+        if sensor_active_device is not None:
+            app_module.release_sdr_device(sensor_active_device)
+            sensor_active_device = None
 
 
 @sensor_bp.route('/start_sensor', methods=['POST'])
@@ -149,6 +164,7 @@ def start_sensor() -> Response:
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE
             )
+            register_process(app_module.sensor_process)
 
             # Start output thread
             thread = threading.Thread(target=stream_sensor_output, args=(app_module.sensor_process,))
@@ -216,9 +232,13 @@ def stream_sensor() -> Response:
 
         while True:
             try:
-                msg = app_module.sensor_queue.get(timeout=1)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = app_module.sensor_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('sensor', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= keepalive_interval:

routes/sstv.py

@@ -13,8 +13,10 @@ from typing import Generator
 
 from flask import Blueprint, jsonify, request, Response, send_file
 
+import app as app_module
 from utils.logging import get_logger
-from utils.sse import format_sse
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.sstv import (
     get_sstv_decoder,
     is_sstv_available,
@@ -30,6 +32,9 @@ sstv_bp = Blueprint('sstv', __name__, url_prefix='/sstv')
 # Queue for SSE progress streaming
 _sstv_queue: queue.Queue = queue.Queue(maxsize=100)
 
+# Track which device is being used
+sstv_active_device: int | None = None
+
 
 def _progress_callback(progress: DecodeProgress) -> None:
     """Callback to queue progress updates for SSE stream."""
@@ -94,7 +99,7 @@ def start_decoder():
     if not is_sstv_available():
         return jsonify({
             'status': 'error',
-            'message': 'SSTV decoder not available. Install slowrx: apt install slowrx'
+            'message': 'SSTV decoder not available. Install numpy and Pillow: pip install numpy Pillow'
         }), 400
 
     decoder = get_sstv_decoder()
@@ -158,6 +163,17 @@ def start_decoder():
         latitude = None
         longitude = None
 
+    # Claim SDR device
+    global sstv_active_device
+    device_int = int(device_index)
+    error = app_module.claim_sdr_device(device_int, 'sstv')
+    if error:
+        return jsonify({
+            'status': 'error',
+            'error_type': 'DEVICE_BUSY',
+            'message': error
+        }), 409
+
     # Set callback and start
     decoder.set_callback(_progress_callback)
     success = decoder.start(
@@ -168,6 +184,8 @@ def start_decoder():
     )
 
     if success:
+        sstv_active_device = device_int
+
         result = {
             'status': 'started',
             'frequency': frequency,
@@ -181,6 +199,8 @@ def start_decoder():
 
         return jsonify(result)
     else:
+        # Release device on failure
+        app_module.release_sdr_device(device_int)
         return jsonify({
             'status': 'error',
             'message': 'Failed to start decoder'
@@ -195,8 +215,15 @@ def stop_decoder():
     Returns:
         JSON confirmation.
     """
+    global sstv_active_device
     decoder = get_sstv_decoder()
     decoder.stop()
+
+    # Release device from registry
+    if sstv_active_device is not None:
+        app_module.release_sdr_device(sstv_active_device)
+        sstv_active_device = None
+
     return jsonify({'status': 'stopped'})
 
 
@@ -287,6 +314,73 @@ def get_image(filename: str):
     return send_file(image_path, mimetype='image/png')
 
 
+@sstv_bp.route('/images/<filename>/download')
+def download_image(filename: str):
+    """
+    Download a decoded SSTV image file.
+
+    Args:
+        filename: Image filename
+
+    Returns:
+        Image file as attachment or 404.
+    """
+    decoder = get_sstv_decoder()
+
+    # Security: only allow alphanumeric filenames with .png extension
+    if not filename.replace('_', '').replace('-', '').replace('.', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Invalid filename'}), 400
+
+    if not filename.endswith('.png'):
+        return jsonify({'status': 'error', 'message': 'Only PNG files supported'}), 400
+
+    image_path = decoder._output_dir / filename
+
+    if not image_path.exists():
+        return jsonify({'status': 'error', 'message': 'Image not found'}), 404
+
+    return send_file(image_path, mimetype='image/png', as_attachment=True, download_name=filename)
+
+
+@sstv_bp.route('/images/<filename>', methods=['DELETE'])
+def delete_image(filename: str):
+    """
+    Delete a decoded SSTV image.
+
+    Args:
+        filename: Image filename
+
+    Returns:
+        JSON confirmation.
+    """
+    decoder = get_sstv_decoder()
+
+    # Security: only allow alphanumeric filenames with .png extension
+    if not filename.replace('_', '').replace('-', '').replace('.', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Invalid filename'}), 400
+
+    if not filename.endswith('.png'):
+        return jsonify({'status': 'error', 'message': 'Only PNG files supported'}), 400
+
+    if decoder.delete_image(filename):
+        return jsonify({'status': 'ok'})
+    else:
+        return jsonify({'status': 'error', 'message': 'Image not found'}), 404
+
+
+@sstv_bp.route('/images', methods=['DELETE'])
+def delete_all_images():
+    """
+    Delete all decoded SSTV images.
+
+    Returns:
+        JSON with count of deleted images.
+    """
+    decoder = get_sstv_decoder()
+    count = decoder.delete_all_images()
+    return jsonify({'status': 'ok', 'deleted': count})
+
+
 @sstv_bp.route('/stream')
 def stream_progress():
     """
@@ -305,10 +399,14 @@ def stream_progress():
         keepalive_interval = 30.0
 
         while True:
-            try:
-                progress = _sstv_queue.get(timeout=1)
-                last_keepalive = time.time()
-                yield format_sse(progress)
+            try:
+                progress = _sstv_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('sstv', progress, progress.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(progress)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= keepalive_interval:

routes/sstv_general.py

@@ -0,0 +1,339 @@
+"""General SSTV (Slow-Scan Television) decoder routes.
+
+Provides endpoints for decoding terrestrial SSTV images on common HF/VHF/UHF
+frequencies used by amateur radio operators worldwide.
+"""
+
+from __future__ import annotations
+
+import queue
+import time
+from collections.abc import Generator
+from pathlib import Path
+
+from flask import Blueprint, Response, jsonify, request, send_file
+
+from utils.logging import get_logger
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from utils.sstv import (
+    DecodeProgress,
+    get_general_sstv_decoder,
+)
+
+logger = get_logger('intercept.sstv_general')
+
+sstv_general_bp = Blueprint('sstv_general', __name__, url_prefix='/sstv-general')
+
+# Queue for SSE progress streaming
+_sstv_general_queue: queue.Queue = queue.Queue(maxsize=100)
+
+# Predefined SSTV frequencies
+SSTV_FREQUENCIES = [
+    {'band': '80 m', 'frequency': 3.845, 'modulation': 'lsb', 'notes': 'Common US SSTV calling frequency', 'type': 'Terrestrial HF'},
+    {'band': '80 m', 'frequency': 3.730, 'modulation': 'lsb', 'notes': 'Europe primary (analog/digital variants)', 'type': 'Terrestrial HF'},
+    {'band': '40 m', 'frequency': 7.171, 'modulation': 'lsb', 'notes': 'Common international/US/EU SSTV activity', 'type': 'Terrestrial HF'},
+    {'band': '40 m', 'frequency': 7.040, 'modulation': 'lsb', 'notes': 'Alternative US/Europe calling', 'type': 'Terrestrial HF'},
+    {'band': '30 m', 'frequency': 10.132, 'modulation': 'usb', 'notes': 'Narrowband SSTV (e.g., MP73-N digital)', 'type': 'Terrestrial HF'},
+    {'band': '20 m', 'frequency': 14.230, 'modulation': 'usb', 'notes': 'Most popular international SSTV frequency', 'type': 'Terrestrial HF'},
+    {'band': '20 m', 'frequency': 14.233, 'modulation': 'usb', 'notes': 'Digital SSTV calling / alternative activity', 'type': 'Terrestrial HF'},
+    {'band': '20 m', 'frequency': 14.240, 'modulation': 'usb', 'notes': 'Europe alternative', 'type': 'Terrestrial HF'},
+    {'band': '15 m', 'frequency': 21.340, 'modulation': 'usb', 'notes': 'International calling frequency', 'type': 'Terrestrial HF'},
+    {'band': '10 m', 'frequency': 28.680, 'modulation': 'usb', 'notes': 'International calling frequency', 'type': 'Terrestrial HF'},
+    {'band': '6 m', 'frequency': 50.950, 'modulation': 'usb', 'notes': 'SSTV calling (less common)', 'type': 'Terrestrial VHF'},
+    {'band': '2 m', 'frequency': 145.625, 'modulation': 'fm', 'notes': 'Australia/common simplex (FM sometimes used)', 'type': 'Terrestrial VHF'},
+    {'band': '70 cm', 'frequency': 433.775, 'modulation': 'fm', 'notes': 'Australia/common simplex', 'type': 'Terrestrial UHF'},
+]
+
+# Build a lookup for auto-detecting modulation from frequency
+_FREQ_MODULATION_MAP = {entry['frequency']: entry['modulation'] for entry in SSTV_FREQUENCIES}
+
+
+def _progress_callback(progress: DecodeProgress) -> None:
+    """Callback to queue progress updates for SSE stream."""
+    try:
+        _sstv_general_queue.put_nowait(progress.to_dict())
+    except queue.Full:
+        try:
+            _sstv_general_queue.get_nowait()
+            _sstv_general_queue.put_nowait(progress.to_dict())
+        except queue.Empty:
+            pass
+
+
+@sstv_general_bp.route('/frequencies')
+def get_frequencies():
+    """Return the predefined SSTV frequency table."""
+    return jsonify({
+        'status': 'ok',
+        'frequencies': SSTV_FREQUENCIES,
+    })
+
+
+@sstv_general_bp.route('/status')
+def get_status():
+    """Get general SSTV decoder status."""
+    decoder = get_general_sstv_decoder()
+
+    return jsonify({
+        'available': decoder.decoder_available is not None,
+        'decoder': decoder.decoder_available,
+        'running': decoder.is_running,
+        'image_count': len(decoder.get_images()),
+    })
+
+
+@sstv_general_bp.route('/start', methods=['POST'])
+def start_decoder():
+    """
+    Start general SSTV decoder.
+
+    JSON body:
+        {
+            "frequency": 14.230,     // Frequency in MHz (required)
+            "modulation": "usb",     // fm, usb, or lsb (auto-detected from frequency table if omitted)
+            "device": 0              // RTL-SDR device index
+        }
+    """
+    decoder = get_general_sstv_decoder()
+
+    if decoder.decoder_available is None:
+        return jsonify({
+            'status': 'error',
+            'message': 'SSTV decoder not available. Install numpy and Pillow: pip install numpy Pillow',
+        }), 400
+
+    if decoder.is_running:
+        return jsonify({
+            'status': 'already_running',
+        })
+
+    # Clear queue
+    while not _sstv_general_queue.empty():
+        try:
+            _sstv_general_queue.get_nowait()
+        except queue.Empty:
+            break
+
+    data = request.get_json(silent=True) or {}
+    frequency = data.get('frequency')
+    modulation = data.get('modulation')
+    device_index = data.get('device', 0)
+
+    # Validate frequency
+    if frequency is None:
+        return jsonify({
+            'status': 'error',
+            'message': 'Frequency is required',
+        }), 400
+
+    try:
+        frequency = float(frequency)
+        if not (1 <= frequency <= 500):
+            return jsonify({
+                'status': 'error',
+                'message': 'Frequency must be between 1-500 MHz (HF requires upconverter for RTL-SDR)',
+            }), 400
+    except (TypeError, ValueError):
+        return jsonify({
+            'status': 'error',
+            'message': 'Invalid frequency',
+        }), 400
+
+    # Auto-detect modulation from frequency table if not specified
+    if not modulation:
+        modulation = _FREQ_MODULATION_MAP.get(frequency, 'usb')
+
+    # Validate modulation
+    if modulation not in ('fm', 'usb', 'lsb'):
+        return jsonify({
+            'status': 'error',
+            'message': 'Modulation must be fm, usb, or lsb',
+        }), 400
+
+    # Set callback and start
+    decoder.set_callback(_progress_callback)
+    success = decoder.start(
+        frequency=frequency,
+        device_index=device_index,
+        modulation=modulation,
+    )
+
+    if success:
+        return jsonify({
+            'status': 'started',
+            'frequency': frequency,
+            'modulation': modulation,
+            'device': device_index,
+        })
+    else:
+        return jsonify({
+            'status': 'error',
+            'message': 'Failed to start decoder',
+        }), 500
+
+
+@sstv_general_bp.route('/stop', methods=['POST'])
+def stop_decoder():
+    """Stop general SSTV decoder."""
+    decoder = get_general_sstv_decoder()
+    decoder.stop()
+    return jsonify({'status': 'stopped'})
+
+
+@sstv_general_bp.route('/images')
+def list_images():
+    """Get list of decoded SSTV images."""
+    decoder = get_general_sstv_decoder()
+    images = decoder.get_images()
+
+    limit = request.args.get('limit', type=int)
+    if limit and limit > 0:
+        images = images[-limit:]
+
+    return jsonify({
+        'status': 'ok',
+        'images': [img.to_dict() for img in images],
+        'count': len(images),
+    })
+
+
+@sstv_general_bp.route('/images/<filename>')
+def get_image(filename: str):
+    """Get a decoded SSTV image file."""
+    decoder = get_general_sstv_decoder()
+
+    # Security: only allow alphanumeric filenames with .png extension
+    if not filename.replace('_', '').replace('-', '').replace('.', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Invalid filename'}), 400
+
+    if not filename.endswith('.png'):
+        return jsonify({'status': 'error', 'message': 'Only PNG files supported'}), 400
+
+    image_path = decoder._output_dir / filename
+
+    if not image_path.exists():
+        return jsonify({'status': 'error', 'message': 'Image not found'}), 404
+
+    return send_file(image_path, mimetype='image/png')
+
+
+@sstv_general_bp.route('/images/<filename>/download')
+def download_image(filename: str):
+    """Download a decoded SSTV image file."""
+    decoder = get_general_sstv_decoder()
+
+    # Security: only allow alphanumeric filenames with .png extension
+    if not filename.replace('_', '').replace('-', '').replace('.', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Invalid filename'}), 400
+
+    if not filename.endswith('.png'):
+        return jsonify({'status': 'error', 'message': 'Only PNG files supported'}), 400
+
+    image_path = decoder._output_dir / filename
+
+    if not image_path.exists():
+        return jsonify({'status': 'error', 'message': 'Image not found'}), 404
+
+    return send_file(image_path, mimetype='image/png', as_attachment=True, download_name=filename)
+
+
+@sstv_general_bp.route('/images/<filename>', methods=['DELETE'])
+def delete_image(filename: str):
+    """Delete a decoded SSTV image."""
+    decoder = get_general_sstv_decoder()
+
+    # Security: only allow alphanumeric filenames with .png extension
+    if not filename.replace('_', '').replace('-', '').replace('.', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Invalid filename'}), 400
+
+    if not filename.endswith('.png'):
+        return jsonify({'status': 'error', 'message': 'Only PNG files supported'}), 400
+
+    if decoder.delete_image(filename):
+        return jsonify({'status': 'ok'})
+    else:
+        return jsonify({'status': 'error', 'message': 'Image not found'}), 404
+
+
+@sstv_general_bp.route('/images', methods=['DELETE'])
+def delete_all_images():
+    """Delete all decoded SSTV images."""
+    decoder = get_general_sstv_decoder()
+    count = decoder.delete_all_images()
+    return jsonify({'status': 'ok', 'deleted': count})
+
+
+@sstv_general_bp.route('/stream')
+def stream_progress():
+    """SSE stream of SSTV decode progress."""
+    def generate() -> Generator[str, None, None]:
+        last_keepalive = time.time()
+        keepalive_interval = 30.0
+
+        while True:
+            try:
+                progress = _sstv_general_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('sstv_general', progress, progress.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(progress)
+            except queue.Empty:
+                now = time.time()
+                if now - last_keepalive >= keepalive_interval:
+                    yield format_sse({'type': 'keepalive'})
+                    last_keepalive = now
+
+    response = Response(generate(), mimetype='text/event-stream')
+    response.headers['Cache-Control'] = 'no-cache'
+    response.headers['X-Accel-Buffering'] = 'no'
+    response.headers['Connection'] = 'keep-alive'
+    return response
+
+
+@sstv_general_bp.route('/decode-file', methods=['POST'])
+def decode_file():
+    """Decode SSTV from an uploaded audio file."""
+    if 'audio' not in request.files:
+        return jsonify({
+            'status': 'error',
+            'message': 'No audio file provided',
+        }), 400
+
+    audio_file = request.files['audio']
+
+    if not audio_file.filename:
+        return jsonify({
+            'status': 'error',
+            'message': 'No file selected',
+        }), 400
+
+    import tempfile
+    with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
+        audio_file.save(tmp.name)
+        tmp_path = tmp.name
+
+    try:
+        decoder = get_general_sstv_decoder()
+        images = decoder.decode_file(tmp_path)
+
+        return jsonify({
+            'status': 'ok',
+            'images': [img.to_dict() for img in images],
+            'count': len(images),
+        })
+
+    except Exception as e:
+        logger.error(f"Error decoding file: {e}")
+        return jsonify({
+            'status': 'error',
+            'message': str(e),
+        }), 500
+
+    finally:
+        try:
+            Path(tmp_path).unlink()
+        except Exception:
+            pass

routes/tscm.py

@@ -60,6 +60,7 @@ from utils.tscm.device_identity import (
     ingest_ble_dict,
     ingest_wifi_dict,
 )
+from utils.event_pipeline import process_event
 
 # Import unified Bluetooth scanner helper for TSCM integration
 try:
@@ -627,6 +628,10 @@ def sweep_stream():
             try:
                 if tscm_queue:
                     msg = tscm_queue.get(timeout=1)
+                    try:
+                        process_event('tscm', msg, msg.get('type'))
+                    except Exception:
+                        pass
                     yield f"data: {json.dumps(msg)}\n\n"
                 else:
                     time.sleep(1)
@@ -1072,6 +1077,32 @@ def _scan_wifi_networks(interface: str) -> list[dict]:
         return []
 
 
+def _scan_wifi_clients(interface: str) -> list[dict]:
+    """
+    Get WiFi client observations from the unified WiFi scanner.
+
+    Clients are only available when monitor-mode scanning is active.
+    """
+    try:
+        from utils.wifi import get_wifi_scanner
+
+        scanner = get_wifi_scanner()
+        if interface:
+            try:
+                if not scanner._is_monitor_mode_interface(interface):
+                    return []
+            except Exception:
+                return []
+
+        return [client.to_dict() for client in scanner.clients]
+    except ImportError as e:
+        logger.error(f"Failed to import wifi scanner: {e}")
+        return []
+    except Exception as e:
+        logger.exception(f"WiFi client scan failed: {e}")
+        return []
+
+
 def _scan_bluetooth_devices(interface: str, duration: int = 10) -> list[dict]:
     """
     Scan for Bluetooth devices with manufacturer data detection.
@@ -1606,6 +1637,7 @@ def _run_sweep(
         threats_found = 0
         severity_counts = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0}
         all_wifi = {}  # Use dict for deduplication by BSSID
+        all_wifi_clients = {}  # Use dict for deduplication by client MAC
         all_bt = {}    # Use dict for deduplication by MAC
         all_rf = []
 
@@ -1702,6 +1734,7 @@ def _run_sweep(
                                     'channel': network.get('channel', ''),
                                     'signal': network.get('power', ''),
                                     'security': network.get('privacy', ''),
+                                    'vendor': network.get('vendor'),
                                     'is_threat': is_threat,
                                     'is_new': not classification.get('in_baseline', False),
                                     'classification': profile.risk_level.value,
@@ -1715,6 +1748,77 @@ def _run_sweep(
                                 })
                         except Exception as e:
                             logger.error(f"WiFi device processing error for {network.get('bssid', '?')}: {e}")
+
+                    # WiFi clients (monitor mode only)
+                    try:
+                        wifi_clients = _scan_wifi_clients(wifi_interface)
+                        for client in wifi_clients:
+                            mac = (client.get('mac') or '').upper()
+                            if not mac or mac in all_wifi_clients:
+                                continue
+                            all_wifi_clients[mac] = client
+
+                            rssi_val = client.get('rssi_current')
+                            if rssi_val is None:
+                                rssi_val = client.get('rssi_median') or client.get('rssi_ema')
+
+                            client_device = {
+                                'mac': mac,
+                                'vendor': client.get('vendor'),
+                                'name': client.get('vendor') or 'WiFi Client',
+                                'rssi': rssi_val,
+                                'associated_bssid': client.get('associated_bssid'),
+                                'probed_ssids': client.get('probed_ssids', []),
+                                'probe_count': client.get('probe_count', len(client.get('probed_ssids', []))),
+                                'is_client': True,
+                            }
+
+                            try:
+                                timeline_manager.add_observation(
+                                    identifier=mac,
+                                    protocol='wifi',
+                                    rssi=rssi_val,
+                                    name=client_device.get('vendor') or f'WiFi Client {mac[-5:]}',
+                                    attributes={'client': True, 'associated_bssid': client_device.get('associated_bssid')}
+                                )
+                            except Exception as e:
+                                logger.debug(f"WiFi client timeline observation error: {e}")
+                            _maybe_store_timeline(
+                                identifier=mac,
+                                protocol='wifi',
+                                rssi=rssi_val,
+                                attributes={'client': True, 'associated_bssid': client_device.get('associated_bssid')}
+                            )
+
+                            profile = correlation.analyze_wifi_device(client_device)
+                            client_device['classification'] = profile.risk_level.value
+                            client_device['score'] = profile.total_score
+                            client_device['score_modifier'] = profile.score_modifier
+                            client_device['known_device'] = profile.known_device
+                            client_device['known_device_name'] = profile.known_device_name
+                            client_device['indicators'] = [
+                                {'type': i.type.value, 'desc': i.description}
+                                for i in profile.indicators
+                            ]
+                            client_device['recommended_action'] = profile.recommended_action
+
+                            # Feed to identity engine for MAC-randomization resistant clustering
+                            try:
+                                wifi_obs = {
+                                    'timestamp': datetime.now().isoformat(),
+                                    'src_mac': mac,
+                                    'bssid': client_device.get('associated_bssid'),
+                                    'rssi': rssi_val,
+                                    'frame_type': 'probe_request',
+                                    'probed_ssids': client_device.get('probed_ssids', []),
+                                }
+                                ingest_wifi_dict(wifi_obs)
+                            except Exception as e:
+                                logger.debug(f"Identity engine WiFi client ingest error: {e}")
+
+                            _emit_event('wifi_client', client_device)
+                    except Exception as e:
+                        logger.debug(f"WiFi client scan error: {e}")
                 except Exception as e:
                     last_wifi_scan = current_time
                     logger.error(f"WiFi scan error: {e}")
@@ -1793,6 +1897,9 @@ def _run_sweep(
                                     'name': device.get('name', 'Unknown'),
                                     'device_type': device.get('type', ''),
                                     'rssi': device.get('rssi', ''),
+                                    'manufacturer': device.get('manufacturer'),
+                                    'tracker': device.get('tracker'),
+                                    'tracker_type': device.get('tracker_type'),
                                     'is_threat': is_threat,
                                     'is_new': not classification.get('in_baseline', False),
                                     'classification': profile.risk_level.value,
@@ -1921,6 +2028,7 @@ def _run_sweep(
                 comparator = BaselineComparator(baseline)
                 baseline_comparison = comparator.compare_all(
                     wifi_devices=list(all_wifi.values()),
+                    wifi_clients=list(all_wifi_clients.values()),
                     bt_devices=list(all_bt.values()),
                     rf_signals=all_rf
                 )
@@ -1936,6 +2044,7 @@ def _run_sweep(
 
             if verbose_results:
                 wifi_payload = list(all_wifi.values())
+                wifi_client_payload = list(all_wifi_clients.values())
                 bt_payload = list(all_bt.values())
                 rf_payload = list(all_rf)
             else:
@@ -1951,6 +2060,28 @@ def _run_sweep(
                     }
                     for d in all_wifi.values()
                 ]
+                wifi_client_payload = []
+                for client in all_wifi_clients.values():
+                    mac = client.get('mac') or client.get('address')
+                    if isinstance(mac, str):
+                        mac = mac.upper()
+                    probed_ssids = client.get('probed_ssids') or []
+                    rssi = client.get('rssi')
+                    if rssi is None:
+                        rssi = client.get('rssi_current')
+                    if rssi is None:
+                        rssi = client.get('rssi_median')
+                    if rssi is None:
+                        rssi = client.get('rssi_ema')
+                    wifi_client_payload.append({
+                        'mac': mac,
+                        'vendor': client.get('vendor'),
+                        'rssi': rssi,
+                        'associated_bssid': client.get('associated_bssid'),
+                        'is_associated': client.get('is_associated'),
+                        'probed_ssids': probed_ssids,
+                        'probe_count': client.get('probe_count', len(probed_ssids)),
+                    })
                 bt_payload = [
                     {
                         'mac': d.get('mac') or d.get('address'),
@@ -1975,9 +2106,11 @@ def _run_sweep(
                 status='completed',
                 results={
                     'wifi_devices': wifi_payload,
+                    'wifi_clients': wifi_client_payload,
                     'bt_devices': bt_payload,
                     'rf_signals': rf_payload,
                     'wifi_count': len(all_wifi),
+                    'wifi_client_count': len(all_wifi_clients),
                     'bt_count': len(all_bt),
                     'rf_count': len(all_rf),
                     'severity_counts': severity_counts,
@@ -2005,6 +2138,7 @@ def _run_sweep(
                     'total_new': baseline_comparison['total_new'],
                     'total_missing': baseline_comparison['total_missing'],
                     'wifi': baseline_comparison.get('wifi'),
+                    'wifi_clients': baseline_comparison.get('wifi_clients'),
                     'bluetooth': baseline_comparison.get('bluetooth'),
                     'rf': baseline_comparison.get('rf'),
                 })
@@ -2022,6 +2156,7 @@ def _run_sweep(
                 'sweep_id': _current_sweep_id,
                 'threats_found': threats_found,
                 'wifi_count': len(all_wifi),
+                'wifi_client_count': len(all_wifi_clients),
                 'bt_count': len(all_bt),
                 'rf_count': len(all_rf),
                 'severity_counts': severity_counts,
@@ -2169,6 +2304,7 @@ def compare_against_baseline():
 
     Expects JSON body with:
     - wifi_devices: list of WiFi devices (optional)
+    - wifi_clients: list of WiFi clients (optional)
     - bt_devices: list of Bluetooth devices (optional)
     - rf_signals: list of RF signals (optional)
 
@@ -2177,12 +2313,14 @@ def compare_against_baseline():
     data = request.get_json() or {}
 
     wifi_devices = data.get('wifi_devices')
+    wifi_clients = data.get('wifi_clients')
     bt_devices = data.get('bt_devices')
     rf_signals = data.get('rf_signals')
 
     # Use the convenience function that gets active baseline
     comparison = get_comparison_for_active_baseline(
         wifi_devices=wifi_devices,
+        wifi_clients=wifi_clients,
         bt_devices=bt_devices,
         rf_signals=rf_signals
     )
@@ -2276,7 +2414,10 @@ def feed_wifi():
     """Feed WiFi device data for baseline recording."""
     data = request.get_json()
     if data:
-        _baseline_recorder.add_wifi_device(data)
+        if data.get('is_client'):
+            _baseline_recorder.add_wifi_client(data)
+        else:
+            _baseline_recorder.add_wifi_device(data)
     return jsonify({'status': 'success'})
 
 
@@ -2928,12 +3069,14 @@ def get_baseline_diff(baseline_id: int, sweep_id: int):
             results = json.loads(results)
 
         current_wifi = results.get('wifi_devices', [])
+        current_wifi_clients = results.get('wifi_clients', [])
         current_bt = results.get('bt_devices', [])
         current_rf = results.get('rf_signals', [])
 
         diff = calculate_baseline_diff(
             baseline=baseline,
             current_wifi=current_wifi,
+            current_wifi_clients=current_wifi_clients,
             current_bt=current_bt,
             current_rf=current_rf,
             sweep_id=sweep_id

routes/websdr.py

@@ -0,0 +1,504 @@
+"""HF/Shortwave WebSDR Integration - KiwiSDR network access."""
+
+from __future__ import annotations
+
+import json
+import math
+import queue
+import re
+import struct
+import threading
+import time
+from typing import Optional
+
+from flask import Blueprint, Flask, jsonify, request, Response
+
+try:
+    from flask_sock import Sock
+    WEBSOCKET_AVAILABLE = True
+except ImportError:
+    WEBSOCKET_AVAILABLE = False
+
+from utils.kiwisdr import KiwiSDRClient, KIWI_SAMPLE_RATE, VALID_MODES, parse_host_port
+from utils.logging import get_logger
+
+logger = get_logger('intercept.websdr')
+
+websdr_bp = Blueprint('websdr', __name__, url_prefix='/websdr')
+
+# ============================================
+# RECEIVER CACHE
+# ============================================
+
+_receiver_cache: list[dict] = []
+_cache_lock = threading.Lock()
+_cache_timestamp: float = 0
+CACHE_TTL = 3600  # 1 hour
+
+
+def _parse_gps_coord(coord_str: str) -> Optional[float]:
+    """Parse a GPS coordinate string like '51.5074' or '(-33.87)' into a float."""
+    if not coord_str:
+        return None
+    # Remove parentheses and whitespace
+    cleaned = coord_str.strip().strip('()').strip()
+    try:
+        return float(cleaned)
+    except (ValueError, TypeError):
+        return None
+
+
+def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """Calculate distance in km between two GPS coordinates."""
+    R = 6371  # Earth radius in km
+    dlat = math.radians(lat2 - lat1)
+    dlon = math.radians(lon2 - lon1)
+    a = (math.sin(dlat / 2) ** 2 +
+         math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *
+         math.sin(dlon / 2) ** 2)
+    c = 2 * math.asin(math.sqrt(a))
+    return R * c
+
+
+KIWI_DATA_URLS = [
+    'https://rx.skywavelinux.com/kiwisdr_com.js',
+    'http://rx.linkfanel.net/kiwisdr_com.js',
+]
+
+
+def _fetch_kiwi_receivers() -> list[dict]:
+    """Fetch the KiwiSDR receiver list from the public directory."""
+    import urllib.request
+    import json
+
+    receivers = []
+    raw = None
+
+    # Try each data source until one works
+    for data_url in KIWI_DATA_URLS:
+        try:
+            req = urllib.request.Request(data_url, headers={
+                'User-Agent': 'INTERCEPT-SIGINT/1.0',
+            })
+            with urllib.request.urlopen(req, timeout=20) as resp:
+                raw = resp.read().decode('utf-8', errors='replace')
+            if raw and len(raw) > 100:
+                logger.info(f"Fetched KiwiSDR data from {data_url}")
+                break
+            raw = None
+        except Exception as e:
+            logger.warning(f"Failed to fetch from {data_url}: {e}")
+            continue
+
+    if not raw:
+        logger.error("All KiwiSDR data sources failed")
+        return receivers
+
+    # The JS file contains: var kiwisdr_com = [ {...}, {...}, ... ];
+    # Extract the JSON array
+    match = re.search(r'var\s+kiwisdr_com\s*=\s*(\[.*\])\s*;?', raw, re.DOTALL)
+    if not match:
+        # Try bare array
+        match = re.search(r'(\[\s*\{.*\}\s*\])', raw, re.DOTALL)
+        if not match:
+            logger.warning("Could not find receiver array in KiwiSDR data")
+            return receivers
+
+    arr_str = match.group(1)
+
+    # Parse JSON
+    try:
+        raw_list = json.loads(arr_str)
+    except json.JSONDecodeError:
+        # Fix common JS → JSON issues (trailing commas)
+        fixed = re.sub(r',\s*}', '}', arr_str)
+        fixed = re.sub(r',\s*]', ']', fixed)
+        try:
+            raw_list = json.loads(fixed)
+        except json.JSONDecodeError:
+            logger.error("Failed to parse KiwiSDR JSON")
+            return receivers
+
+    for entry in raw_list:
+        if not isinstance(entry, dict):
+            continue
+
+        # Skip offline receivers
+        if entry.get('offline') == 'yes' or entry.get('status') != 'active':
+            continue
+
+        name = entry.get('name', 'Unknown')
+        url = entry.get('url', '')
+        gps = entry.get('gps', '')
+        antenna = entry.get('antenna', '')
+        location = entry.get('loc', '')
+
+        # Parse users (strings in actual data)
+        try:
+            users = int(entry.get('users', 0))
+        except (ValueError, TypeError):
+            users = 0
+        try:
+            users_max = int(entry.get('users_max', 4))
+        except (ValueError, TypeError):
+            users_max = 4
+
+        # Parse bands field: "0-30000000" (Hz) → freq_lo/freq_hi in kHz
+        bands_str = entry.get('bands', '0-30000000')
+        freq_lo = 0
+        freq_hi = 30000
+        if bands_str and '-' in str(bands_str):
+            try:
+                parts = str(bands_str).split('-')
+                freq_lo = int(parts[0]) / 1000  # Hz to kHz
+                freq_hi = int(parts[1]) / 1000  # Hz to kHz
+            except (ValueError, IndexError):
+                pass
+
+        # Parse GPS: "(51.317266, -2.950479)" format
+        lat, lon = None, None
+        if gps:
+            parts = str(gps).replace('(', '').replace(')', '').split(',')
+            if len(parts) >= 2:
+                lat = _parse_gps_coord(parts[0])
+                lon = _parse_gps_coord(parts[1])
+
+        if not url:
+            continue
+
+        # Ensure URL has protocol
+        if not url.startswith('http'):
+            url = 'http://' + url
+
+        receivers.append({
+            'name': name,
+            'url': url.rstrip('/'),
+            'lat': lat,
+            'lon': lon,
+            'location': location,
+            'users': users,
+            'users_max': users_max,
+            'antenna': antenna,
+            'bands': bands_str,
+            'freq_lo': freq_lo,
+            'freq_hi': freq_hi,
+            'available': users < users_max,
+        })
+
+    return receivers
+
+
+def get_receivers(force_refresh: bool = False) -> list[dict]:
+    """Get cached receiver list, refreshing if stale."""
+    global _receiver_cache, _cache_timestamp
+
+    with _cache_lock:
+        now = time.time()
+        if force_refresh or not _receiver_cache or (now - _cache_timestamp) > CACHE_TTL:
+            logger.info("Refreshing KiwiSDR receiver list...")
+            _receiver_cache = _fetch_kiwi_receivers()
+            _cache_timestamp = now
+            logger.info(f"Loaded {len(_receiver_cache)} KiwiSDR receivers")
+
+    return _receiver_cache
+
+
+# ============================================
+# API ENDPOINTS
+# ============================================
+
+@websdr_bp.route('/receivers')
+def list_receivers() -> Response:
+    """List KiwiSDR receivers, with optional filters."""
+    freq_khz = request.args.get('freq_khz', type=float)
+    available = request.args.get('available', type=str)
+    refresh = request.args.get('refresh', type=str)
+
+    receivers = get_receivers(force_refresh=(refresh == 'true'))
+
+    filtered = receivers
+    if available == 'true':
+        filtered = [r for r in filtered if r.get('available', True)]
+
+    if freq_khz is not None:
+        filtered = [
+            r for r in filtered
+            if r.get('freq_lo', 0) <= freq_khz <= r.get('freq_hi', 30000)
+        ]
+
+    return jsonify({
+        'status': 'success',
+        'receivers': filtered[:100],
+        'total': len(filtered),
+        'cached_total': len(receivers),
+    })
+
+
+@websdr_bp.route('/receivers/nearest')
+def nearest_receivers() -> Response:
+    """Find receivers nearest to a given location."""
+    lat = request.args.get('lat', type=float)
+    lon = request.args.get('lon', type=float)
+    freq_khz = request.args.get('freq_khz', type=float)
+
+    if lat is None or lon is None:
+        return jsonify({'status': 'error', 'message': 'lat and lon are required'}), 400
+
+    receivers = get_receivers()
+
+    # Filter by frequency if specified
+    if freq_khz is not None:
+        receivers = [
+            r for r in receivers
+            if r.get('freq_lo', 0) <= freq_khz <= r.get('freq_hi', 30000)
+        ]
+
+    # Calculate distances and sort
+    with_distance = []
+    for r in receivers:
+        if r.get('lat') is not None and r.get('lon') is not None:
+            dist = _haversine(lat, lon, r['lat'], r['lon'])
+            entry = dict(r)
+            entry['distance_km'] = round(dist, 1)
+            with_distance.append(entry)
+
+    with_distance.sort(key=lambda x: x['distance_km'])
+
+    return jsonify({
+        'status': 'success',
+        'receivers': with_distance[:10],
+    })
+
+
+@websdr_bp.route('/spy-station/<station_id>/receivers')
+def spy_station_receivers(station_id: str) -> Response:
+    """Find receivers that can tune to a spy station's frequency."""
+    try:
+        from routes.spy_stations import STATIONS
+    except ImportError:
+        return jsonify({'status': 'error', 'message': 'Spy stations module not available'}), 503
+
+    # Find the station
+    station = None
+    for s in STATIONS:
+        if s.get('id') == station_id:
+            station = s
+            break
+
+    if not station:
+        return jsonify({'status': 'error', 'message': 'Station not found'}), 404
+
+    # Get primary frequency
+    freq_khz = None
+    for f in station.get('frequencies', []):
+        if f.get('primary'):
+            freq_khz = f.get('freq_khz')
+            break
+    if freq_khz is None and station.get('frequencies'):
+        freq_khz = station['frequencies'][0].get('freq_khz')
+
+    if freq_khz is None:
+        return jsonify({'status': 'error', 'message': 'No frequency found for station'}), 404
+
+    receivers = get_receivers()
+
+    # Filter receivers that cover this frequency and are available
+    matching = [
+        r for r in receivers
+        if r.get('freq_lo', 0) <= freq_khz <= r.get('freq_hi', 30000) and r.get('available', True)
+    ]
+
+    return jsonify({
+        'status': 'success',
+        'station': {
+            'id': station['id'],
+            'name': station.get('name', ''),
+            'nickname': station.get('nickname', ''),
+            'freq_khz': freq_khz,
+            'mode': station.get('mode', 'USB'),
+        },
+        'receivers': matching[:20],
+        'total': len(matching),
+    })
+
+
+@websdr_bp.route('/status')
+def websdr_status() -> Response:
+    """Get WebSDR connection and cache status."""
+    return jsonify({
+        'status': 'ok',
+        'cached_receivers': len(_receiver_cache),
+        'cache_age_seconds': round(time.time() - _cache_timestamp, 0) if _cache_timestamp > 0 else None,
+        'cache_ttl': CACHE_TTL,
+        'audio_connected': _kiwi_client is not None and _kiwi_client.connected if _kiwi_client else False,
+    })
+
+
+# ============================================
+# KIWISDR AUDIO PROXY
+# ============================================
+
+_kiwi_client: Optional[KiwiSDRClient] = None
+_kiwi_lock = threading.Lock()
+_kiwi_audio_queue: queue.Queue = queue.Queue(maxsize=200)
+
+
+def _disconnect_kiwi() -> None:
+    """Disconnect active KiwiSDR client."""
+    global _kiwi_client
+    with _kiwi_lock:
+        if _kiwi_client:
+            _kiwi_client.disconnect()
+            _kiwi_client = None
+    # Drain audio queue
+    while not _kiwi_audio_queue.empty():
+        try:
+            _kiwi_audio_queue.get_nowait()
+        except queue.Empty:
+            break
+
+
+def _handle_kiwi_command(ws, cmd: str, data: dict) -> None:
+    """Handle a command from the browser client."""
+    global _kiwi_client
+
+    if cmd == 'connect':
+        receiver_url = data.get('url', '')
+        host = data.get('host', '')
+        port = int(data.get('port', 8073))
+        freq_khz = float(data.get('freq_khz', 7000))
+        mode = data.get('mode', 'am').lower()
+        password = data.get('password', '')
+
+        # Parse host/port from URL if provided
+        if receiver_url and not host:
+            host, port = parse_host_port(receiver_url)
+
+        if mode not in VALID_MODES:
+            ws.send(json.dumps({'type': 'error', 'message': f'Invalid mode: {mode}'}))
+            return
+
+        if not host or ';' in host or '&' in host or '|' in host:
+            ws.send(json.dumps({'type': 'error', 'message': 'Invalid host'}))
+            return
+
+        _disconnect_kiwi()
+
+        def on_audio(pcm_bytes, smeter):
+            # Package: 2 bytes smeter (big-endian int16) + PCM data
+            header = struct.pack('>h', smeter)
+            try:
+                _kiwi_audio_queue.put_nowait(header + pcm_bytes)
+            except queue.Full:
+                try:
+                    _kiwi_audio_queue.get_nowait()
+                except queue.Empty:
+                    pass
+                try:
+                    _kiwi_audio_queue.put_nowait(header + pcm_bytes)
+                except queue.Full:
+                    pass
+
+        def on_error(msg):
+            try:
+                ws.send(json.dumps({'type': 'error', 'message': msg}))
+            except Exception:
+                pass
+
+        def on_disconnect():
+            try:
+                ws.send(json.dumps({'type': 'disconnected'}))
+            except Exception:
+                pass
+
+        with _kiwi_lock:
+            _kiwi_client = KiwiSDRClient(
+                host=host, port=port,
+                on_audio=on_audio,
+                on_error=on_error,
+                on_disconnect=on_disconnect,
+                password=password,
+            )
+            success = _kiwi_client.connect(freq_khz, mode)
+
+        if success:
+            ws.send(json.dumps({
+                'type': 'connected',
+                'host': host,
+                'port': port,
+                'freq_khz': freq_khz,
+                'mode': mode,
+                'sample_rate': KIWI_SAMPLE_RATE,
+            }))
+        else:
+            ws.send(json.dumps({'type': 'error', 'message': 'Connection to KiwiSDR failed'}))
+            _disconnect_kiwi()
+
+    elif cmd == 'tune':
+        freq_khz = float(data.get('freq_khz', 0))
+        mode = data.get('mode', '').lower() or None
+
+        with _kiwi_lock:
+            if _kiwi_client and _kiwi_client.connected:
+                success = _kiwi_client.tune(
+                    freq_khz,
+                    mode or _kiwi_client.mode
+                )
+                if success:
+                    ws.send(json.dumps({
+                        'type': 'tuned',
+                        'freq_khz': freq_khz,
+                        'mode': mode or _kiwi_client.mode,
+                    }))
+                else:
+                    ws.send(json.dumps({'type': 'error', 'message': 'Retune failed'}))
+            else:
+                ws.send(json.dumps({'type': 'error', 'message': 'Not connected'}))
+
+    elif cmd == 'disconnect':
+        _disconnect_kiwi()
+        ws.send(json.dumps({'type': 'disconnected'}))
+
+
+def init_websdr_audio(app: Flask) -> None:
+    """Initialize WebSocket audio proxy for KiwiSDR. Called from app.py."""
+    if not WEBSOCKET_AVAILABLE:
+        logger.warning("flask-sock not installed, KiwiSDR audio proxy disabled")
+        return
+
+    sock = Sock(app)
+
+    @sock.route('/ws/kiwi-audio')
+    def kiwi_audio_stream(ws):
+        """WebSocket endpoint: proxy audio between browser and KiwiSDR."""
+        logger.info("KiwiSDR audio client connected")
+
+        try:
+            while True:
+                # Check for commands from browser
+                try:
+                    msg = ws.receive(timeout=0.005)
+                    if msg:
+                        data = json.loads(msg)
+                        cmd = data.get('cmd', '')
+                        _handle_kiwi_command(ws, cmd, data)
+                except TimeoutError:
+                    pass
+                except Exception as e:
+                    if 'closed' in str(e).lower():
+                        break
+                    if 'timed out' not in str(e).lower():
+                        logger.error(f"KiwiSDR WS receive error: {e}")
+
+                # Forward audio from KiwiSDR to browser
+                try:
+                    audio_data = _kiwi_audio_queue.get_nowait()
+                    ws.send(audio_data)
+                except queue.Empty:
+                    time.sleep(0.005)
+
+        except Exception as e:
+            logger.info(f"KiwiSDR WS closed: {e}")
+        finally:
+            _disconnect_kiwi()
+            logger.info("KiwiSDR audio client disconnected")

routes/wifi.py

@@ -17,11 +17,12 @@ from flask import Blueprint, jsonify, request, Response
 
 import app as app_module
 from utils.dependencies import check_tool, get_tool_path
-from utils.logging import wifi_logger as logger
-from utils.process import is_valid_mac, is_valid_channel
-from utils.validation import validate_wifi_channel, validate_mac_address, validate_network_interface
-from utils.sse import format_sse
-from data.oui import get_manufacturer
+from utils.logging import wifi_logger as logger
+from utils.process import is_valid_mac, is_valid_channel
+from utils.validation import validate_wifi_channel, validate_mac_address, validate_network_interface
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
+from data.oui import get_manufacturer
 from utils.constants import (
     WIFI_TERMINATE_TIMEOUT,
     PMKID_TERMINATE_TIMEOUT,
@@ -46,8 +47,33 @@ from utils.constants import (
 wifi_bp = Blueprint('wifi', __name__, url_prefix='/wifi')
 
 # PMKID process state
-pmkid_process = None
-pmkid_lock = threading.Lock()
+pmkid_process = None
+pmkid_lock = threading.Lock()
+
+
+def _parse_channel_list(raw_channels: Any) -> list[int] | None:
+    """Parse a channel list from string/list input."""
+    if raw_channels in (None, '', []):
+        return None
+
+    if isinstance(raw_channels, str):
+        parts = [p.strip() for p in re.split(r'[\s,]+', raw_channels) if p.strip()]
+    elif isinstance(raw_channels, (list, tuple, set)):
+        parts = list(raw_channels)
+    else:
+        parts = [raw_channels]
+
+    channels: list[int] = []
+    seen = set()
+    for part in parts:
+        if part in (None, ''):
+            continue
+        ch = validate_wifi_channel(part)
+        if ch not in seen:
+            channels.append(ch)
+            seen.add(ch)
+
+    return channels or None
 
 
 def detect_wifi_interfaces():
@@ -607,8 +633,9 @@ def start_wifi_scan():
             return jsonify({'status': 'error', 'message': 'Scan already running'})
 
         data = request.json
-        channel = data.get('channel')
-        band = data.get('band', 'abg')
+        channel = data.get('channel')
+        channels = data.get('channels')
+        band = data.get('band', 'abg')
 
         # Use provided interface or fall back to stored monitor interface
         interface = data.get('interface')
@@ -658,8 +685,17 @@ def start_wifi_scan():
             interface
         ]
 
-        if channel:
-            cmd.extend(['-c', str(channel)])
+        channel_list = None
+        if channels:
+            try:
+                channel_list = _parse_channel_list(channels)
+            except ValueError as e:
+                return jsonify({'status': 'error', 'message': str(e)}), 400
+
+        if channel_list:
+            cmd.extend(['-c', ','.join(str(c) for c in channel_list)])
+        elif channel:
+            cmd.extend(['-c', str(channel)])
 
         logger.info(f"Running: {' '.join(cmd)}")
 
@@ -851,32 +887,53 @@ def check_handshake_status():
                 return jsonify({'status': 'stopped', 'file_exists': False, 'handshake_found': False})
 
     file_size = os.path.getsize(capture_file)
-    handshake_found = False
+    handshake_found = False
+    handshake_valid: bool | None = None
+    handshake_checked = False
+    handshake_reason: str | None = None
 
     try:
-        if target_bssid and is_valid_mac(target_bssid):
-            aircrack_path = get_tool_path('aircrack-ng')
-            if aircrack_path:
-                result = subprocess.run(
-                    [aircrack_path, '-a', '2', '-b', target_bssid, capture_file],
-                    capture_output=True, text=True, timeout=10
-                )
-                output = result.stdout + result.stderr
-                if '1 handshake' in output or ('handshake' in output.lower() and 'wpa' in output.lower()):
-                    if '0 handshake' not in output:
-                        handshake_found = True
+        if target_bssid and is_valid_mac(target_bssid):
+            aircrack_path = get_tool_path('aircrack-ng')
+            if aircrack_path:
+                result = subprocess.run(
+                    [aircrack_path, '-a', '2', '-b', target_bssid, capture_file],
+                    capture_output=True, text=True, timeout=10
+                )
+                output = result.stdout + result.stderr
+                output_lower = output.lower()
+                handshake_checked = True
+
+                if 'no valid wpa handshakes found' in output_lower:
+                    handshake_valid = False
+                    handshake_reason = 'No valid WPA handshake found'
+                elif '0 handshake' in output_lower:
+                    handshake_valid = False
+                elif '1 handshake' in output_lower or ('handshake' in output_lower and 'wpa' in output_lower):
+                    handshake_valid = True
+                else:
+                    handshake_valid = False
     except subprocess.TimeoutExpired:
         pass
-    except Exception as e:
-        logger.error(f"Error checking handshake: {e}")
-
-    return jsonify({
-        'status': 'running' if app_module.wifi_process and app_module.wifi_process.poll() is None else 'stopped',
-        'file_exists': True,
-        'file_size': file_size,
-        'file': capture_file,
-        'handshake_found': handshake_found
-    })
+    except Exception as e:
+        logger.error(f"Error checking handshake: {e}")
+
+    if handshake_valid:
+        handshake_found = True
+        normalized_bssid = target_bssid.upper() if target_bssid else None
+        if normalized_bssid and normalized_bssid not in app_module.wifi_handshakes:
+            app_module.wifi_handshakes.append(normalized_bssid)
+
+    return jsonify({
+        'status': 'running' if app_module.wifi_process and app_module.wifi_process.poll() is None else 'stopped',
+        'file_exists': True,
+        'file_size': file_size,
+        'file': capture_file,
+        'handshake_found': handshake_found,
+        'handshake_valid': handshake_valid,
+        'handshake_checked': handshake_checked,
+        'handshake_reason': handshake_reason
+    })
 
 
 @wifi_bp.route('/pmkid/capture', methods=['POST'])
@@ -1084,9 +1141,13 @@ def stream_wifi():
 
         while True:
             try:
-                msg = app_module.wifi_queue.get(timeout=1)
-                last_keepalive = time.time()
-                yield format_sse(msg)
+                msg = app_module.wifi_queue.get(timeout=1)
+                last_keepalive = time.time()
+                try:
+                    process_event('wifi', msg, msg.get('type'))
+                except Exception:
+                    pass
+                yield format_sse(msg)
             except queue.Empty:
                 now = time.time()
                 if now - last_keepalive >= keepalive_interval:

routes/wifi_v2.py

@@ -16,14 +16,16 @@ from typing import Generator
 
 from flask import Blueprint, jsonify, request, Response
 
-from utils.wifi import (
-    get_wifi_scanner,
-    analyze_channels,
-    get_hidden_correlator,
-    SCAN_MODE_QUICK,
-    SCAN_MODE_DEEP,
-)
-from utils.sse import format_sse
+from utils.wifi import (
+    get_wifi_scanner,
+    analyze_channels,
+    get_hidden_correlator,
+    SCAN_MODE_QUICK,
+    SCAN_MODE_DEEP,
+)
+from utils.sse import format_sse
+from utils.validation import validate_wifi_channel
+from utils.event_pipeline import process_event
 
 logger = logging.getLogger(__name__)
 
@@ -85,28 +87,44 @@ def start_deep_scan():
 
     Requires monitor mode interface and root privileges.
 
-    Request body:
-        interface: Monitor mode interface (e.g., 'wlan0mon')
-        band: Band to scan ('2.4', '5', 'all')
-        channel: Optional specific channel to monitor
+    Request body:
+        interface: Monitor mode interface (e.g., 'wlan0mon')
+        band: Band to scan ('2.4', '5', 'all')
+        channel: Optional specific channel to monitor
+        channels: Optional list or comma-separated channels to monitor
     """
     data = request.get_json() or {}
     interface = data.get('interface')
     band = data.get('band', 'all')
-    channel = data.get('channel')
-
-    if channel:
-        try:
-            channel = int(channel)
-        except ValueError:
-            return jsonify({'error': 'Invalid channel'}), 400
+    channel = data.get('channel')
+    channels = data.get('channels')
+
+    channel_list = None
+    if channels:
+        if isinstance(channels, str):
+            channel_list = [c.strip() for c in channels.split(',') if c.strip()]
+        elif isinstance(channels, (list, tuple, set)):
+            channel_list = list(channels)
+        else:
+            channel_list = [channels]
+        try:
+            channel_list = [validate_wifi_channel(c) for c in channel_list]
+        except (TypeError, ValueError):
+            return jsonify({'error': 'Invalid channels'}), 400
+
+    if channel:
+        try:
+            channel = validate_wifi_channel(channel)
+        except ValueError:
+            return jsonify({'error': 'Invalid channel'}), 400
 
     scanner = get_wifi_scanner()
-    success = scanner.start_deep_scan(
-        interface=interface,
-        band=band,
-        channel=channel,
-    )
+    success = scanner.start_deep_scan(
+        interface=interface,
+        band=band,
+        channel=channel,
+        channels=channel_list,
+    )
 
     if success:
         return jsonify({
@@ -388,10 +406,14 @@ def event_stream():
         - keepalive: Periodic keepalive
     """
     def generate() -> Generator[str, None, None]:
-        scanner = get_wifi_scanner()
-
-        for event in scanner.get_event_stream():
-            yield format_sse(event)
+        scanner = get_wifi_scanner()
+
+        for event in scanner.get_event_stream():
+            try:
+                process_event('wifi', event, event.get('type'))
+            except Exception:
+                pass
+            yield format_sse(event)
 
     response = Response(generate(), mimetype='text/event-stream')
     response.headers['Cache-Control'] = 'no-cache'