Add real-time WebSocket waterfall with I/Q capture and server-side FFT

Replace the batch rtl_power SSE pipeline with continuous I/Q streaming
via WebSocket for smooth ~25fps waterfall display. The server captures
raw I/Q samples (rtl_sdr/rx_sdr), computes Hann-windowed FFT, and
sends compact binary frames (1035 bytes vs ~15KB JSON, 93% reduction).
Client falls back to existing SSE path if WebSocket is unavailable.

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

utils/sdr/airspy.py

@@ -185,6 +185,43 @@ class AirspyCommandBuilder(CommandBuilder):
 
         return cmd
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build rx_sdr command for raw I/Q capture with Airspy.
+
+        Outputs unsigned 8-bit I/Q pairs to stdout for waterfall display.
+        """
+        device_str = self._build_device_string(device)
+        freq_hz = int(frequency_mhz * 1e6)
+
+        cmd = [
+            'rx_sdr',
+            '-d', device_str,
+            '-f', str(freq_hz),
+            '-s', str(sample_rate),
+            '-F', 'CU8',
+        ]
+
+        if gain is not None and gain > 0:
+            cmd.extend(['-g', self._format_gain(gain)])
+
+        if bias_t:
+            cmd.append('-T')
+
+        # Output to stdout
+        cmd.append('-')
+
+        return cmd
+
     def get_capabilities(self) -> SDRCapabilities:
         """Return Airspy capabilities."""
         return self.CAPABILITIES

utils/sdr/base.py

@@ -186,6 +186,41 @@ class CommandBuilder(ABC):
         """Return hardware capabilities for this SDR type."""
         pass
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build raw I/Q capture command for streaming samples to stdout.
+
+        Used for real-time waterfall/spectrum display. Output is unsigned
+        8-bit I/Q pairs (cu8) written continuously to stdout.
+
+        Args:
+            device: The SDR device to use
+            frequency_mhz: Center frequency in MHz
+            sample_rate: Sample rate in Hz (default 2048000)
+            gain: Gain in dB (None for auto)
+            ppm: PPM frequency correction
+            bias_t: Enable bias-T power (for active antennas)
+            output_format: Output sample format (default 'cu8')
+
+        Returns:
+            Command as list of strings for subprocess
+
+        Raises:
+            NotImplementedError: If the SDR type does not support I/Q capture.
+        """
+        raise NotImplementedError(
+            f"{self.__class__.__name__} does not support raw I/Q capture"
+        )
+
     @classmethod
     @abstractmethod
     def get_sdr_type(cls) -> SDRType:

utils/sdr/hackrf.py

@@ -185,6 +185,44 @@ class HackRFCommandBuilder(CommandBuilder):
 
         return cmd
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build rx_sdr command for raw I/Q capture with HackRF.
+
+        Outputs unsigned 8-bit I/Q pairs to stdout for waterfall display.
+        """
+        device_str = self._build_device_string(device)
+        freq_hz = int(frequency_mhz * 1e6)
+
+        cmd = [
+            'rx_sdr',
+            '-d', device_str,
+            '-f', str(freq_hz),
+            '-s', str(sample_rate),
+            '-F', 'CU8',
+        ]
+
+        if gain is not None and gain > 0:
+            lna, vga = self._split_gain(gain)
+            cmd.extend(['-g', f'LNA={lna},VGA={vga}'])
+
+        if bias_t:
+            cmd.append('-T')
+
+        # Output to stdout
+        cmd.append('-')
+
+        return cmd
+
     def get_capabilities(self) -> SDRCapabilities:
         """Return HackRF capabilities."""
         return self.CAPABILITIES

utils/sdr/limesdr.py

@@ -162,6 +162,41 @@ class LimeSDRCommandBuilder(CommandBuilder):
 
         return cmd
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build rx_sdr command for raw I/Q capture with LimeSDR.
+
+        Outputs unsigned 8-bit I/Q pairs to stdout for waterfall display.
+        Note: LimeSDR does not support bias-T, parameter is ignored.
+        """
+        device_str = self._build_device_string(device)
+        freq_hz = int(frequency_mhz * 1e6)
+
+        cmd = [
+            'rx_sdr',
+            '-d', device_str,
+            '-f', str(freq_hz),
+            '-s', str(sample_rate),
+            '-F', 'CU8',
+        ]
+
+        if gain is not None and gain > 0:
+            cmd.extend(['-g', f'LNAH={int(gain)}'])
+
+        # Output to stdout
+        cmd.append('-')
+
+        return cmd
+
     def get_capabilities(self) -> SDRCapabilities:
         """Return LimeSDR capabilities."""
         return self.CAPABILITIES

utils/sdr/rtlsdr.py

@@ -231,6 +231,45 @@ class RTLSDRCommandBuilder(CommandBuilder):
 
         return cmd
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build rtl_sdr command for raw I/Q capture.
+
+        Outputs unsigned 8-bit I/Q pairs to stdout for waterfall display.
+        """
+        rtl_sdr_path = get_tool_path('rtl_sdr') or 'rtl_sdr'
+        freq_hz = int(frequency_mhz * 1e6)
+
+        cmd = [
+            rtl_sdr_path,
+            '-d', self._get_device_arg(device),
+            '-f', str(freq_hz),
+            '-s', str(sample_rate),
+        ]
+
+        if gain is not None and gain > 0:
+            cmd.extend(['-g', str(gain)])
+
+        if ppm is not None and ppm != 0:
+            cmd.extend(['-p', str(ppm)])
+
+        if bias_t:
+            cmd.append('-T')
+
+        # Output to stdout
+        cmd.append('-')
+
+        return cmd
+
     def get_capabilities(self) -> SDRCapabilities:
         """Return RTL-SDR capabilities."""
         return self.CAPABILITIES

utils/sdr/sdrplay.py

@@ -163,6 +163,43 @@ class SDRPlayCommandBuilder(CommandBuilder):
 
         return cmd
 
+    def build_iq_capture_command(
+        self,
+        device: SDRDevice,
+        frequency_mhz: float,
+        sample_rate: int = 2048000,
+        gain: Optional[float] = None,
+        ppm: Optional[int] = None,
+        bias_t: bool = False,
+        output_format: str = 'cu8',
+    ) -> list[str]:
+        """
+        Build rx_sdr command for raw I/Q capture with SDRPlay.
+
+        Outputs unsigned 8-bit I/Q pairs to stdout for waterfall display.
+        """
+        device_str = self._build_device_string(device)
+        freq_hz = int(frequency_mhz * 1e6)
+
+        cmd = [
+            'rx_sdr',
+            '-d', device_str,
+            '-f', str(freq_hz),
+            '-s', str(sample_rate),
+            '-F', 'CU8',
+        ]
+
+        if gain is not None and gain > 0:
+            cmd.extend(['-g', f'IFGR={int(gain)}'])
+
+        if bias_t:
+            cmd.append('-T')
+
+        # Output to stdout
+        cmd.append('-')
+
+        return cmd
+
     def get_capabilities(self) -> SDRCapabilities:
         """Return SDRPlay capabilities."""
         return self.CAPABILITIES

utils/waterfall_fft.py

@@ -0,0 +1,122 @@
+"""FFT pipeline for real-time waterfall display.
+
+Converts raw I/Q samples from SDR hardware into quantized power spectrum
+frames suitable for binary WebSocket transmission.
+"""
+
+from __future__ import annotations
+
+import struct
+
+import numpy as np
+
+
+def cu8_to_complex(raw: bytes) -> np.ndarray:
+    """Convert unsigned 8-bit I/Q bytes to complex64.
+
+    RTL-SDR (and rx_sdr with -F cu8) outputs interleaved unsigned 8-bit
+    I/Q pairs where 128 is the zero point.
+
+    Args:
+        raw: Raw bytes, length must be even (I/Q pairs).
+
+    Returns:
+        Complex64 array of length len(raw) // 2.
+    """
+    iq = np.frombuffer(raw, dtype=np.uint8).astype(np.float32)
+    # Normalize: 0 -> -1.0, 128 -> ~0.0, 255 -> +1.0
+    iq = (iq - 127.5) / 127.5
+    return iq[0::2] + 1j * iq[1::2]
+
+
+def compute_power_spectrum(
+    samples: np.ndarray,
+    fft_size: int = 1024,
+    avg_count: int = 4,
+) -> np.ndarray:
+    """Compute averaged power spectrum in dBm.
+
+    Applies a Hann window, computes FFT, converts to power (dB),
+    and averages over multiple segments.
+
+    Args:
+        samples: Complex64 array, length >= fft_size * avg_count.
+        fft_size: Number of FFT bins.
+        avg_count: Number of segments to average.
+
+    Returns:
+        Float32 array of length fft_size with power in dB (fftshift'd).
+    """
+    window = np.hanning(fft_size).astype(np.float32)
+    accum = np.zeros(fft_size, dtype=np.float32)
+    actual_avg = 0
+
+    for i in range(avg_count):
+        offset = i * fft_size
+        if offset + fft_size > len(samples):
+            break
+        segment = samples[offset : offset + fft_size] * window
+        spectrum = np.fft.fft(segment)
+        power = np.real(spectrum * np.conj(spectrum))
+        # Avoid log10(0)
+        power = np.maximum(power, 1e-20)
+        accum += 10.0 * np.log10(power)
+        actual_avg += 1
+
+    if actual_avg == 0:
+        return np.full(fft_size, -100.0, dtype=np.float32)
+
+    accum /= actual_avg
+    return np.fft.fftshift(accum).astype(np.float32)
+
+
+def quantize_to_uint8(
+    power_db: np.ndarray,
+    db_min: float = -90.0,
+    db_max: float = -20.0,
+) -> bytes:
+    """Clamp and scale dB values to 0-255.
+
+    Args:
+        power_db: Float32 array of power values in dB.
+        db_min: Value mapped to 0.
+        db_max: Value mapped to 255.
+
+    Returns:
+        Bytes of length len(power_db), each in [0, 255].
+    """
+    db_range = db_max - db_min
+    if db_range <= 0:
+        db_range = 1.0
+    scaled = (power_db - db_min) / db_range * 255.0
+    clamped = np.clip(scaled, 0, 255).astype(np.uint8)
+    return clamped.tobytes()
+
+
+def build_binary_frame(
+    start_freq: float,
+    end_freq: float,
+    quantized_bins: bytes,
+) -> bytes:
+    """Pack a binary waterfall frame for WebSocket transmission.
+
+    Wire format (little-endian):
+        [uint8 msg_type=0x01]
+        [float32 start_freq]
+        [float32 end_freq]
+        [uint16 bin_count]
+        [uint8[] bins]
+
+    Total size = 11 + bin_count bytes.
+
+    Args:
+        start_freq: Start frequency in MHz.
+        end_freq: End frequency in MHz.
+        quantized_bins: Pre-quantized uint8 bin data.
+
+    Returns:
+        Binary frame bytes.
+    """
+    bin_count = len(quantized_bins)
+    header = struct.pack('<BffH', 0x01, start_freq, end_freq, bin_count)
+    return header + quantized_bins