Merge upstream/main: add gsm_spy blueprint

routes/__init__.py

@@ -32,6 +32,7 @@ def register_blueprints(app):
     from .websdr import websdr_bp
     from .alerts import alerts_bp
     from .recordings import recordings_bp
+    from .gsm_spy import gsm_spy_bp
 
     app.register_blueprint(pager_bp)
     app.register_blueprint(sensor_bp)
@@ -63,6 +64,7 @@ def register_blueprints(app):
     app.register_blueprint(websdr_bp)  # HF/Shortwave WebSDR
     app.register_blueprint(alerts_bp)  # Cross-mode alerts
     app.register_blueprint(recordings_bp)  # Session recordings
+    app.register_blueprint(gsm_spy_bp)  # GSM cellular intelligence
 
     # Initialize TSCM state with queue and lock from app
     import app as app_module

routes/audio_websocket.py

@@ -1,10 +1,11 @@
 """WebSocket-based audio streaming for SDR."""
 
+import json
+import shutil
+import socket
 import subprocess
 import threading
 import time
-import shutil
-import json
 from flask import Flask
 
 # Try to import flask-sock
@@ -251,4 +252,19 @@ def init_audio_websocket(app: Flask):
         finally:
             with process_lock:
                 kill_audio_processes()
+            # Complete WebSocket close handshake, then shut down the
+            # raw socket so Werkzeug cannot write its HTTP 200 response
+            # on top of the WebSocket stream.
+            try:
+                ws.close()
+            except Exception:
+                pass
+            try:
+                ws.sock.shutdown(socket.SHUT_RDWR)
+            except Exception:
+                pass
+            try:
+                ws.sock.close()
+            except Exception:
+                pass
             logger.info("WebSocket audio client disconnected")

routes/listening_post.py

@@ -19,8 +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.event_pipeline import process_event
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.constants import (
     SSE_QUEUE_TIMEOUT,
     SSE_KEEPALIVE_INTERVAL,
@@ -1181,13 +1181,13 @@ def stream_scanner_events() -> Response:
 
         while True:
             try:
-                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)
+                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:
@@ -1239,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:
@@ -1271,7 +1271,7 @@ def start_audio() -> Response:
             pass
         time.sleep(0.5)
 
-    data = request.json or {}
+    data = request.json or {}
 
     try:
         frequency = float(data.get('frequency', 0))
@@ -1286,11 +1286,11 @@ def start_audio() -> Response:
             'message': f'Invalid parameter: {e}'
         }), 400
 
-    if frequency <= 0:
-        return jsonify({
-            'status': 'error',
-            'message': 'frequency is required'
-        }), 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:
@@ -1299,19 +1299,28 @@ 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
-
-    # Stop waterfall if it's using the same SDR
-    if waterfall_running and waterfall_active_device == device:
-        _stop_waterfall_internal()
-        time.sleep(0.2)
+    # Update config for audio
+    scanner_config['squelch'] = squelch
+    scanner_config['gain'] = gain
+    scanner_config['device'] = device
+    scanner_config['sdr_type'] = sdr_type
 
-    # Claim device for listening audio
-    if listening_active_device is None or listening_active_device != device:
+    # Stop waterfall if it's using the same SDR (SSE path)
+    if waterfall_running and waterfall_active_device == device:
+        _stop_waterfall_internal()
+        time.sleep(0.2)
+
+    # Release waterfall device claim if the WebSocket waterfall is still
+    # holding it.  The JS client sends a stop command and closes the
+    # WebSocket before requesting audio, but the backend handler may not
+    # have finished its cleanup yet.
+    device_status = app_module.get_sdr_device_status()
+    if device_status.get(device) == 'waterfall':
+        app_module.release_sdr_device(device)
+        time.sleep(0.3)
+
+    # 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')
@@ -1524,204 +1533,204 @@ 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,
-}
+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
+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:
+def start_waterfall() -> Response:
     """Start the waterfall/spectrogram display."""
     global waterfall_thread, waterfall_running, waterfall_config, waterfall_active_device
 
@@ -1734,24 +1743,24 @@ def start_waterfall() -> Response:
 
     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
+    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
@@ -1777,11 +1786,11 @@ def start_waterfall() -> Response:
 
 
 @listening_post_bp.route('/waterfall/stop', methods=['POST'])
-def stop_waterfall() -> Response:
-    """Stop the waterfall display."""
-    _stop_waterfall_internal()
-
-    return jsonify({'status': 'stopped'})
+def stop_waterfall() -> Response:
+    """Stop the waterfall display."""
+    _stop_waterfall_internal()
+
+    return jsonify({'status': 'stopped'})
 
 
 @listening_post_bp.route('/waterfall/stream')
@@ -1790,14 +1799,14 @@ def stream_waterfall() -> Response:
     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)
+            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:
@@ -1808,20 +1817,20 @@ def stream_waterfall() -> Response:
     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
+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

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
+import math
 import os
 import pathlib
 import re
 import pty
 import queue
 import select
+import struct
 import subprocess
 import threading
 import time
@@ -22,8 +24,8 @@ 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.event_pipeline import process_event
+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
@@ -106,6 +108,62 @@ def log_message(msg: dict[str, Any]) -> None:
         logger.error(f"Failed to log message: {e}")
 
 
+def audio_relay_thread(
+    rtl_stdout,
+    multimon_stdin,
+    output_queue: queue.Queue,
+    stop_event: threading.Event,
+) -> None:
+    """Relay audio from rtl_fm to multimon-ng while computing signal levels.
+
+    Reads raw 16-bit LE PCM from *rtl_stdout*, writes every chunk straight
+    through to *multimon_stdin*, and every ~100 ms pushes an RMS / peak scope
+    event onto *output_queue*.
+    """
+    CHUNK = 4096  # bytes – 2048 samples at 16-bit mono
+    INTERVAL = 0.1  # seconds between scope updates
+    last_scope = time.monotonic()
+
+    try:
+        while not stop_event.is_set():
+            data = rtl_stdout.read(CHUNK)
+            if not data:
+                break
+
+            # Forward audio untouched
+            try:
+                multimon_stdin.write(data)
+                multimon_stdin.flush()
+            except (BrokenPipeError, OSError):
+                break
+
+            # Compute scope levels every ~100 ms
+            now = time.monotonic()
+            if now - last_scope >= INTERVAL:
+                last_scope = now
+                try:
+                    n_samples = len(data) // 2
+                    if n_samples == 0:
+                        continue
+                    samples = struct.unpack(f'<{n_samples}h', data[:n_samples * 2])
+                    peak = max(abs(s) for s in samples)
+                    rms = int(math.sqrt(sum(s * s for s in samples) / n_samples))
+                    output_queue.put_nowait({
+                        'type': 'scope',
+                        'rms': rms,
+                        'peak': peak,
+                    })
+                except (struct.error, ValueError, queue.Full):
+                    pass
+    except Exception as e:
+        logger.debug(f"Audio relay error: {e}")
+    finally:
+        try:
+            multimon_stdin.close()
+        except OSError:
+            pass
+
+
 def stream_decoder(master_fd: int, process: subprocess.Popen[bytes]) -> None:
     """Stream decoder output to queue using PTY for unbuffered output."""
     try:
@@ -152,6 +210,11 @@ def stream_decoder(master_fd: int, process: subprocess.Popen[bytes]) -> None:
             os.close(master_fd)
         except OSError:
             pass
+        # Signal relay thread to stop
+        with app_module.process_lock:
+            stop_relay = getattr(app_module.current_process, '_stop_relay', None)
+        if stop_relay:
+            stop_relay.set()
         # Cleanup companion rtl_fm process and decoder
         with app_module.process_lock:
             rtl_proc = getattr(app_module.current_process, '_rtl_process', None)
@@ -319,7 +382,7 @@ def start_decoding() -> Response:
 
             multimon_process = subprocess.Popen(
                 multimon_cmd,
-                stdin=rtl_process.stdout,
+                stdin=subprocess.PIPE,
                 stdout=slave_fd,
                 stderr=slave_fd,
                 close_fds=True
@@ -327,11 +390,22 @@ def start_decoding() -> Response:
             register_process(multimon_process)
 
             os.close(slave_fd)
-            rtl_process.stdout.close()
+
+            # Spawn audio relay thread between rtl_fm and multimon-ng
+            stop_relay = threading.Event()
+            relay = threading.Thread(
+                target=audio_relay_thread,
+                args=(rtl_process.stdout, multimon_process.stdin,
+                      app_module.output_queue, stop_relay),
+            )
+            relay.daemon = True
+            relay.start()
 
             app_module.current_process = multimon_process
             app_module.current_process._rtl_process = rtl_process
             app_module.current_process._master_fd = master_fd
+            app_module.current_process._stop_relay = stop_relay
+            app_module.current_process._relay_thread = relay
 
             # Start output thread with PTY master fd
             thread = threading.Thread(target=stream_decoder, args=(master_fd, multimon_process))
@@ -380,6 +454,10 @@ def stop_decoding() -> Response:
 
     with app_module.process_lock:
         if app_module.current_process:
+            # Signal audio relay thread to stop
+            if hasattr(app_module.current_process, '_stop_relay'):
+                app_module.current_process._stop_relay.set()
+
             # Kill rtl_fm process first
             if hasattr(app_module.current_process, '_rtl_process'):
                 try:
@@ -469,14 +547,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()
-                try:
-                    process_event('pager', msg, msg.get('type'))
-                except Exception:
-                    pass
-                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/sensor.py

@@ -18,8 +18,8 @@ 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.event_pipeline import process_event
+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
 
@@ -45,6 +45,21 @@ def stream_sensor_output(process: subprocess.Popen[bytes]) -> None:
                 data['type'] = 'sensor'
                 app_module.sensor_queue.put(data)
 
+                # Push scope event when signal level data is present
+                rssi = data.get('rssi')
+                snr = data.get('snr')
+                noise = data.get('noise')
+                if rssi is not None or snr is not None:
+                    try:
+                        app_module.sensor_queue.put_nowait({
+                            'type': 'scope',
+                            'rssi': rssi if rssi is not None else 0,
+                            'snr': snr if snr is not None else 0,
+                            'noise': noise if noise is not None else 0,
+                        })
+                    except queue.Full:
+                        pass
+
                 # Log if enabled
                 if app_module.logging_enabled:
                     try:
@@ -80,6 +95,14 @@ def stream_sensor_output(process: subprocess.Popen[bytes]) -> None:
             sensor_active_device = None
 
 
+@sensor_bp.route('/sensor/status')
+def sensor_status() -> Response:
+    """Check if sensor decoder is currently running."""
+    with app_module.sensor_lock:
+        running = app_module.sensor_process is not None and app_module.sensor_process.poll() is None
+    return jsonify({'running': running})
+
+
 @sensor_bp.route('/start_sensor', methods=['POST'])
 def start_sensor() -> Response:
     global sensor_active_device
@@ -158,6 +181,10 @@ def start_sensor() -> Response:
         full_cmd = ' '.join(cmd)
         logger.info(f"Running: {full_cmd}")
 
+        # Add signal level metadata so the frontend scope can display RSSI/SNR
+        # Disable stats reporting to suppress "row count limit 50 reached" warnings
+        cmd.extend(['-M', 'level', '-M', 'stats:0'])
+
         try:
             app_module.sensor_process = subprocess.Popen(
                 cmd,
@@ -232,13 +259,13 @@ def stream_sensor() -> Response:
 
         while True:
             try:
-                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)
+                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

@@ -15,14 +15,12 @@ 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.event_pipeline import process_event
+from utils.sse import format_sse
+from utils.event_pipeline import process_event
 from utils.sstv import (
     get_sstv_decoder,
     is_sstv_available,
     ISS_SSTV_FREQ,
-    DecodeProgress,
-    DopplerInfo,
 )
 
 logger = get_logger('intercept.sstv')
@@ -36,14 +34,14 @@ _sstv_queue: queue.Queue = queue.Queue(maxsize=100)
 sstv_active_device: int | None = None
 
 
-def _progress_callback(progress: DecodeProgress) -> None:
-    """Callback to queue progress updates for SSE stream."""
+def _progress_callback(data: dict) -> None:
+    """Callback to queue progress/scope updates for SSE stream."""
     try:
-        _sstv_queue.put_nowait(progress.to_dict())
+        _sstv_queue.put_nowait(data)
     except queue.Full:
         try:
             _sstv_queue.get_nowait()
-            _sstv_queue.put_nowait(progress.to_dict())
+            _sstv_queue.put_nowait(data)
         except queue.Empty:
             pass
 
@@ -399,14 +397,14 @@ def stream_progress():
         keepalive_interval = 30.0
 
         while True:
-            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)
+            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

@@ -17,7 +17,6 @@ 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,
 )
 
@@ -49,14 +48,14 @@ SSTV_FREQUENCIES = [
 _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."""
+def _progress_callback(data: dict) -> None:
+    """Callback to queue progress/scope updates for SSE stream."""
     try:
-        _sstv_general_queue.put_nowait(progress.to_dict())
+        _sstv_general_queue.put_nowait(data)
     except queue.Full:
         try:
             _sstv_general_queue.get_nowait()
-            _sstv_general_queue.put_nowait(progress.to_dict())
+            _sstv_general_queue.put_nowait(data)
         except queue.Empty:
             pass
 

routes/tscm.py

@@ -551,6 +551,12 @@ def _start_sweep_internal(
     }
 
 
+@tscm_bp.route('/status')
+def tscm_status():
+    """Check if any TSCM operation is currently running."""
+    return jsonify({'running': _sweep_running})
+
+
 @tscm_bp.route('/sweep/start', methods=['POST'])
 def start_sweep():
     """Start a TSCM sweep."""

routes/waterfall_websocket.py

@@ -0,0 +1,386 @@
+"""WebSocket-based waterfall streaming with I/Q capture and server-side FFT."""
+
+import json
+import queue
+import socket
+import subprocess
+import threading
+import time
+
+from flask import Flask
+
+try:
+    from flask_sock import Sock
+    WEBSOCKET_AVAILABLE = True
+except ImportError:
+    WEBSOCKET_AVAILABLE = False
+    Sock = None
+
+from utils.logging import get_logger
+from utils.process import safe_terminate, register_process, unregister_process
+from utils.waterfall_fft import (
+    build_binary_frame,
+    compute_power_spectrum,
+    cu8_to_complex,
+    quantize_to_uint8,
+)
+from utils.sdr import SDRFactory, SDRType
+from utils.sdr.base import SDRCapabilities, SDRDevice
+
+logger = get_logger('intercept.waterfall_ws')
+
+# Maximum bandwidth per SDR type (Hz)
+MAX_BANDWIDTH = {
+    SDRType.RTL_SDR: 2400000,
+    SDRType.HACKRF: 20000000,
+    SDRType.LIME_SDR: 20000000,
+    SDRType.AIRSPY: 10000000,
+    SDRType.SDRPLAY: 2000000,
+}
+
+
+def _resolve_sdr_type(sdr_type_str: str) -> SDRType:
+    """Convert client sdr_type string to SDRType enum."""
+    mapping = {
+        'rtlsdr': SDRType.RTL_SDR,
+        'rtl_sdr': SDRType.RTL_SDR,
+        'hackrf': SDRType.HACKRF,
+        'limesdr': SDRType.LIME_SDR,
+        'lime_sdr': SDRType.LIME_SDR,
+        'airspy': SDRType.AIRSPY,
+        'sdrplay': SDRType.SDRPLAY,
+    }
+    return mapping.get(sdr_type_str.lower(), SDRType.RTL_SDR)
+
+
+def _build_dummy_device(device_index: int, sdr_type: SDRType) -> SDRDevice:
+    """Build a minimal SDRDevice for command building."""
+    builder = SDRFactory.get_builder(sdr_type)
+    caps = builder.get_capabilities()
+    return SDRDevice(
+        sdr_type=sdr_type,
+        index=device_index,
+        name=f'{sdr_type.value}-{device_index}',
+        serial='N/A',
+        driver=sdr_type.value,
+        capabilities=caps,
+    )
+
+
+def init_waterfall_websocket(app: Flask):
+    """Initialize WebSocket waterfall streaming."""
+    if not WEBSOCKET_AVAILABLE:
+        logger.warning("flask-sock not installed, WebSocket waterfall disabled")
+        return
+
+    sock = Sock(app)
+
+    @sock.route('/ws/waterfall')
+    def waterfall_stream(ws):
+        """WebSocket endpoint for real-time waterfall streaming."""
+        logger.info("WebSocket waterfall client connected")
+
+        # Import app module for device claiming
+        import app as app_module
+
+        iq_process = None
+        reader_thread = None
+        stop_event = threading.Event()
+        claimed_device = None
+        # Queue for outgoing messages — only the main loop touches ws.send()
+        send_queue = queue.Queue(maxsize=120)
+
+        try:
+            while True:
+                # Drain send queue first (non-blocking)
+                while True:
+                    try:
+                        outgoing = send_queue.get_nowait()
+                    except queue.Empty:
+                        break
+                    try:
+                        ws.send(outgoing)
+                    except Exception:
+                        stop_event.set()
+                        break
+
+                try:
+                    msg = ws.receive(timeout=0.1)
+                except Exception as e:
+                    err = str(e).lower()
+                    if "closed" in err:
+                        break
+                    if "timed out" not in err:
+                        logger.error(f"WebSocket receive error: {e}")
+                    continue
+
+                if msg is None:
+                    # simple-websocket returns None on timeout AND on
+                    # close; check ws.connected to tell them apart.
+                    if not ws.connected:
+                        break
+                    if stop_event.is_set():
+                        break
+                    continue
+
+                try:
+                    data = json.loads(msg)
+                except (json.JSONDecodeError, TypeError):
+                    continue
+
+                cmd = data.get('cmd')
+
+                if cmd == 'start':
+                    # Stop any existing capture
+                    was_restarting = iq_process is not None
+                    stop_event.set()
+                    if reader_thread and reader_thread.is_alive():
+                        reader_thread.join(timeout=2)
+                    if iq_process:
+                        safe_terminate(iq_process)
+                        unregister_process(iq_process)
+                        iq_process = None
+                    if claimed_device is not None:
+                        app_module.release_sdr_device(claimed_device)
+                        claimed_device = None
+                    stop_event.clear()
+                    # Flush stale frames from previous capture
+                    while not send_queue.empty():
+                        try:
+                            send_queue.get_nowait()
+                        except queue.Empty:
+                            break
+                    # Allow USB device to be released by the kernel
+                    if was_restarting:
+                        time.sleep(0.5)
+
+                    # Parse config
+                    center_freq = float(data.get('center_freq', 100.0))
+                    span_mhz = float(data.get('span_mhz', 2.0))
+                    gain = data.get('gain')
+                    if gain is not None:
+                        gain = float(gain)
+                    device_index = int(data.get('device', 0))
+                    sdr_type_str = data.get('sdr_type', 'rtlsdr')
+                    fft_size = int(data.get('fft_size', 1024))
+                    fps = int(data.get('fps', 25))
+                    avg_count = int(data.get('avg_count', 4))
+                    ppm = data.get('ppm')
+                    if ppm is not None:
+                        ppm = int(ppm)
+                    bias_t = bool(data.get('bias_t', False))
+
+                    # Clamp FFT size to valid powers of 2
+                    fft_size = max(256, min(8192, fft_size))
+
+                    # Resolve SDR type and bandwidth
+                    sdr_type = _resolve_sdr_type(sdr_type_str)
+                    max_bw = MAX_BANDWIDTH.get(sdr_type, 2400000)
+                    span_hz = int(span_mhz * 1e6)
+                    sample_rate = min(span_hz, max_bw)
+
+                    # Compute effective frequency range
+                    effective_span_mhz = sample_rate / 1e6
+                    start_freq = center_freq - effective_span_mhz / 2
+                    end_freq = center_freq + effective_span_mhz / 2
+
+                    # Claim the device
+                    claim_err = app_module.claim_sdr_device(device_index, 'waterfall')
+                    if claim_err:
+                        ws.send(json.dumps({
+                            'status': 'error',
+                            'message': claim_err,
+                            'error_type': 'DEVICE_BUSY',
+                        }))
+                        continue
+                    claimed_device = device_index
+
+                    # Build I/Q capture command
+                    try:
+                        builder = SDRFactory.get_builder(sdr_type)
+                        device = _build_dummy_device(device_index, sdr_type)
+                        iq_cmd = builder.build_iq_capture_command(
+                            device=device,
+                            frequency_mhz=center_freq,
+                            sample_rate=sample_rate,
+                            gain=gain,
+                            ppm=ppm,
+                            bias_t=bias_t,
+                        )
+                    except NotImplementedError as e:
+                        app_module.release_sdr_device(device_index)
+                        claimed_device = None
+                        ws.send(json.dumps({
+                            'status': 'error',
+                            'message': str(e),
+                        }))
+                        continue
+
+                    # Spawn I/Q capture process (retry to handle USB release lag)
+                    max_attempts = 3 if was_restarting else 1
+                    try:
+                        for attempt in range(max_attempts):
+                            logger.info(
+                                f"Starting I/Q capture: {center_freq} MHz, "
+                                f"span={effective_span_mhz:.1f} MHz, "
+                                f"sr={sample_rate}, fft={fft_size}"
+                            )
+                            iq_process = subprocess.Popen(
+                                iq_cmd,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.DEVNULL,
+                                bufsize=0,
+                            )
+                            register_process(iq_process)
+
+                            # Brief check that process started
+                            time.sleep(0.3)
+                            if iq_process.poll() is not None:
+                                unregister_process(iq_process)
+                                iq_process = None
+                                if attempt < max_attempts - 1:
+                                    logger.info(
+                                        f"I/Q process exited immediately, "
+                                        f"retrying ({attempt + 1}/{max_attempts})..."
+                                    )
+                                    time.sleep(0.5)
+                                    continue
+                                raise RuntimeError(
+                                    "I/Q capture process exited immediately"
+                                )
+                            break  # Process started successfully
+                    except Exception as e:
+                        logger.error(f"Failed to start I/Q capture: {e}")
+                        if iq_process:
+                            safe_terminate(iq_process)
+                            unregister_process(iq_process)
+                            iq_process = None
+                        app_module.release_sdr_device(device_index)
+                        claimed_device = None
+                        ws.send(json.dumps({
+                            'status': 'error',
+                            'message': f'Failed to start I/Q capture: {e}',
+                        }))
+                        continue
+
+                    # Send started confirmation
+                    ws.send(json.dumps({
+                        'status': 'started',
+                        'start_freq': start_freq,
+                        'end_freq': end_freq,
+                        'fft_size': fft_size,
+                        'sample_rate': sample_rate,
+                    }))
+
+                    # Start reader thread — puts frames on queue, never calls ws.send()
+                    def fft_reader(
+                        proc, _send_q, stop_evt,
+                        _fft_size, _avg_count, _fps,
+                        _start_freq, _end_freq,
+                    ):
+                        """Read I/Q from subprocess, compute FFT, enqueue binary frames."""
+                        bytes_per_frame = _fft_size * _avg_count * 2
+                        frame_interval = 1.0 / _fps
+
+                        try:
+                            while not stop_evt.is_set():
+                                if proc.poll() is not None:
+                                    break
+
+                                frame_start = time.monotonic()
+
+                                # Read raw I/Q bytes
+                                raw = b''
+                                remaining = bytes_per_frame
+                                while remaining > 0 and not stop_evt.is_set():
+                                    chunk = proc.stdout.read(min(remaining, 65536))
+                                    if not chunk:
+                                        break
+                                    raw += chunk
+                                    remaining -= len(chunk)
+
+                                if len(raw) < _fft_size * 2:
+                                    break
+
+                                # Process FFT pipeline
+                                samples = cu8_to_complex(raw)
+                                power_db = compute_power_spectrum(
+                                    samples,
+                                    fft_size=_fft_size,
+                                    avg_count=_avg_count,
+                                )
+                                quantized = quantize_to_uint8(power_db)
+                                frame = build_binary_frame(
+                                    _start_freq, _end_freq, quantized,
+                                )
+
+                                try:
+                                    _send_q.put_nowait(frame)
+                                except queue.Full:
+                                    # Drop frame if main loop can't keep up
+                                    pass
+
+                                # Pace to target FPS
+                                elapsed = time.monotonic() - frame_start
+                                sleep_time = frame_interval - elapsed
+                                if sleep_time > 0:
+                                    stop_evt.wait(sleep_time)
+
+                        except Exception as e:
+                            logger.debug(f"FFT reader stopped: {e}")
+
+                    reader_thread = threading.Thread(
+                        target=fft_reader,
+                        args=(
+                            iq_process, send_queue, stop_event,
+                            fft_size, avg_count, fps,
+                            start_freq, end_freq,
+                        ),
+                        daemon=True,
+                    )
+                    reader_thread.start()
+
+                elif cmd == 'stop':
+                    stop_event.set()
+                    if reader_thread and reader_thread.is_alive():
+                        reader_thread.join(timeout=2)
+                        reader_thread = None
+                    if iq_process:
+                        safe_terminate(iq_process)
+                        unregister_process(iq_process)
+                        iq_process = None
+                    if claimed_device is not None:
+                        app_module.release_sdr_device(claimed_device)
+                        claimed_device = None
+                    stop_event.clear()
+                    ws.send(json.dumps({'status': 'stopped'}))
+
+        except Exception as e:
+            logger.info(f"WebSocket waterfall closed: {e}")
+        finally:
+            # Cleanup
+            stop_event.set()
+            if reader_thread and reader_thread.is_alive():
+                reader_thread.join(timeout=2)
+            if iq_process:
+                safe_terminate(iq_process)
+                unregister_process(iq_process)
+            if claimed_device is not None:
+                app_module.release_sdr_device(claimed_device)
+            # Complete WebSocket close handshake, then shut down the
+            # raw socket so Werkzeug cannot write its HTTP 200 response
+            # on top of the WebSocket stream (which browsers see as
+            # "Invalid frame header").
+            try:
+                ws.close()
+            except Exception:
+                pass
+            try:
+                ws.sock.shutdown(socket.SHUT_RDWR)
+            except Exception:
+                pass
+            try:
+                ws.sock.close()
+            except Exception:
+                pass
+            logger.info("WebSocket waterfall client disconnected")