Allow listening with waterfall and speed up updates

routes/listening_post.py

@@ -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,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')
@@ -1527,125 +1532,196 @@ waterfall_config = {
     'gain': 40,
     'device': 0,
     'max_bins': 1024,
+    'interval': 0.4,
 }
 
 
-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
-
-    try:
-        while waterfall_running:
-            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']
-
-            cmd = [
-                rtl_power_path,
-                '-f', f'{start_hz}:{end_hz}:{bin_hz}',
-                '-i', '0.5',
-                '-1',
-                '-g', str(gain),
-                '-d', str(device),
-            ]
-
-            try:
-                proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
-                waterfall_process = proc
-                stdout, _ = proc.communicate(timeout=15)
-            except subprocess.TimeoutExpired:
-                proc.kill()
-                stdout = b''
-            finally:
-                waterfall_process = None
-
-            if not waterfall_running:
-                break
-
-            if not stdout:
-                time.sleep(0.2)
-                continue
-
-            # Parse rtl_power CSV output
-            all_bins = []
-            sweep_start_hz = start_hz
-            sweep_end_hz = end_hz
-
-            for line in stdout.decode(errors='ignore').splitlines():
-                if not line or line.startswith('#'):
-                    continue
-                parts = [p.strip() for p in line.split(',')]
-                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:
-                    continue
-                try:
-                    seg_start = float(parts[start_idx])
-                    seg_end = float(parts[start_idx + 1])
-                    seg_bin = float(parts[start_idx + 2])
-                    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:]
-                    all_bins.extend(raw_values)
-                    sweep_start_hz = min(sweep_start_hz, seg_start)
-                    sweep_end_hz = max(sweep_end_hz, seg_end)
-                except ValueError:
-                    continue
-
-            if all_bins:
+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)
-                if max_bins > 0 and len(all_bins) > max_bins:
-                    all_bins = _downsample_bins(all_bins, max_bins)
+                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': all_bins,
-                    '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
-
-            time.sleep(0.1)
-
-    except Exception as e:
-        logger.error(f"Waterfall loop error: {e}")
-    finally:
-        waterfall_running = False
-        logger.info("Waterfall loop stopped")
+                    '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
 
@@ -1664,6 +1740,11 @@ def start_waterfall() -> Response:
         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:
@@ -1696,27 +1777,11 @@ def start_waterfall() -> Response:
 
 
 @listening_post_bp.route('/waterfall/stop', methods=['POST'])
-def stop_waterfall() -> Response:
-    """Stop the waterfall display."""
-    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
-
-    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')