video_decoder.py

import sys
import os
import SoapySDR
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CF32
import numpy as np
from scipy import signal
import cv2
import logging
import time
import argparse

# Setup logging
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

def fm_demodulate(iq_samples, smooth=False):
    """Perform FM demodulation on complex I/Q samples with optional smoothing."""
    phase = np.angle(iq_samples)
    demod = np.diff(np.unwrap(phase))
    demod = np.pad(demod, (0, 1), mode='edge')
    if smooth:
        demod = signal.savgol_filter(demod, 11, 3)  # Smoothing filter
    return demod

def process_video_signal(demod, sample_rate, freq_mhz, band_name, target_spl=1018, lines_per_frame=525, active_fraction=0.85, line_offset=10, h_sync_adjust=0, active_lines=485, frame_rate_factor=1.0, frame_width=640, frame_height=480, fps=0):
    """Convert demodulated signal to video frames with OSD."""
    samples_per_line = int(target_spl * frame_rate_factor)
    samples_per_frame = samples_per_line * lines_per_frame
    logging.debug(f"Frame params: {samples_per_line} samples/line, {lines_per_frame} lines, {samples_per_frame} samples/frame, active_fraction={active_fraction}, line_offset={line_offset}, h_sync_adjust={h_sync_adjust}, active_lines={active_lines}, frame_rate_factor={frame_rate_factor}")

    if len(demod) < samples_per_frame:
        logging.debug(f"Not enough samples: got {len(demod)}, need {samples_per_frame}")
        return None

    # Normalize demodulated signal
    demod = (demod - np.min(demod)) / (np.max(demod) - np.min(demod))

    frame = np.zeros((active_lines, frame_width))

    for i in range(active_lines):
        line_start = i * samples_per_line + line_offset + h_sync_adjust
        if line_start < 0:
            line_start = 0
        line_end = line_start + int(samples_per_line * active_fraction)
        if line_end > len(demod):
            break
        line_samples = demod[line_start:line_end]
        if len(line_samples) > 0:
            x = np.linspace(0, 1, len(line_samples))
            x_new = np.linspace(0, 1, frame_width)
            line_resampled = np.interp(x_new, x, line_samples)
            frame[i, :] = line_resampled

    if active_lines < 100:
        logging.debug("Insufficient lines for frame")
        return None

    frame = cv2.resize(frame, (frame_width, frame_height), interpolation=cv2.INTER_LINEAR)
    frame = (frame * 255).astype(np.uint8)
    frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)

    # Draw OSD: FPS, Frequency, Band Name (top-left)
    osd_text = [
        f"FPS: {fps:.1f}",
        f"Freq: {freq_mhz} MHz",
        f"Band: {band_name}"
    ]
    for i, text in enumerate(osd_text):
        cv2.putText(frame, text, (10, 30 + i * 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 2, cv2.LINE_AA)
        cv2.putText(frame, text, (10, 30 + i * 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1, cv2.LINE_AA)

    # Draw "Press Q to stop" (bottom-center)
    quit_text = "Press Q to stop"
    text_size = cv2.getTextSize(quit_text, cv2.FONT_HERSHEY_SIMPLEX, 0.7, 1)[0]
    text_x = (frame_width - text_size[0]) // 2
    cv2.putText(frame, quit_text, (text_x, frame_height - 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 2, cv2.LINE_AA)
    cv2.putText(frame, quit_text, (text_x, frame_height - 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1, cv2.LINE_AA)

    # Draw parameters (bottom-right)
    param_text = [
        f"SPL: {target_spl}",
        f"Frac: {active_fraction:.2f}",
        f"Offset: {line_offset}",
        f"HSync: {h_sync_adjust}",
        f"Lines: {active_lines}",
        f"FRate: {frame_rate_factor:.2f}"
    ]
    for i, text in enumerate(param_text):
        text_size = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, 0.7, 1)[0]
        text_x = frame_width - text_size[0] - 10
        text_y = frame_height - 30 - i * 30
        cv2.putText(frame, text, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 2, cv2.LINE_AA)
        cv2.putText(frame, text, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1, cv2.LINE_AA)

    return frame

def main():
    parser = argparse.ArgumentParser(description="Capture and display FM video from HackRF")
    parser.add_argument("frequency", type=float, help="Frequency in MHz (e.g., 5362)")
    parser.add_argument("band_name", type=str, help="Band name for OSD (e.g., L1)")
    args = parser.parse_args()

    freq_mhz = args.frequency
    band_name = args.band_name

    # Ensure library path
    os.environ["DYLD_LIBRARY_PATH"] = "/opt/homebrew/lib:" + os.environ.get("DYLD_LIBRARY_PATH", "")
    logging.info(f"DYLD_LIBRARY_PATH: {os.environ['DYLD_LIBRARY_PATH']}")

    # List available devices
    logging.info("Probing available SoapySDR devices...")
    devices = SoapySDR.Device.enumerate()
    for i, dev in enumerate(devices):
        logging.info(f"Device {i}: {dev}")
    if not devices:
        logging.error("No SoapySDR devices found. Ensure HackRF is connected and drivers are installed.")
        sys.exit(1)

    # Initialize HackRF
    logging.info(f"Attempting to initialize device: {devices[0]}")
    try:
        sdr = SoapySDR.Device(devices[0])
    except Exception as e:
        logging.error(f"Failed to initialize HackRF: {str(e)}")
        logging.error("Ensure HackRF is connected and accessible. Try running without sudo:")
        logging.error("  SoapySDRUtil --probe=\"driver=hackrf\"")
        logging.error("If permission denied, check USB access or reinstall hackrf:")
        logging.error("  brew reinstall hackrf")
        sys.exit(1)

    # Configure device
    sample_rate = 16e6
    bandwidth = 6e6
    target_spl = 1018
    lines_per_frame = 525
    active_fraction = 0.85
    line_offset = 10
    h_sync_adjust = 0
    active_lines = 485  # NTSC default after blanking
    frame_rate_factor = 1.0
    lna_gain = 40
    vga_gain = 30
    smooth_demod = False
    is_pal = False
    try:
        sdr.setSampleRate(SOAPY_SDR_RX, 0, sample_rate)
        sdr.setBandwidth(SOAPY_SDR_RX, 0, bandwidth)
        sdr.setFrequency(SOAPY_SDR_RX, 0, freq_mhz * 1e6)
        sdr.setGainMode(SOAPY_SDR_RX, 0, False)
        sdr.setGain(SOAPY_SDR_RX, 0, "LNA", lna_gain)
        sdr.setGain(SOAPY_SDR_RX, 0, "VGA", vga_gain)
        logging.info(f"Configured SDR: sample_rate={sample_rate} Hz, bandwidth={bandwidth} Hz, freq={freq_mhz} MHz, LNA={lna_gain} dB, VGA={vga_gain} dB")
    except Exception as e:
        logging.error(f"Failed to configure SDR: {str(e)}")
        sdr = None
        sys.exit(1)

    # Setup stream
    try:
        stream = sdr.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32, [0], {"buffers": "32"})
        sdr.activateStream(stream)
        logging.info("Stream activated with 32 buffers")
    except Exception as e:
        logging.error(f"Failed to setup stream: {str(e)}")
        sdr = None
        sys.exit(1)

    # Buffers
    buff_size = 131072
    buff = np.zeros(buff_size, dtype=np.complex64)
    sample_buffer = np.array([], dtype=np.complex64)
    samples_needed = target_spl * lines_per_frame

    # Display window
    cv2.namedWindow("Analog Video", cv2.WINDOW_NORMAL)

    # FPS calculation
    frame_count = 0
    fps_start_time = time.time()
    fps = 0.0
    sample_rates = [8e6, 10e6, 16e6]
    sample_rate_index = 2  # Start at 16 Msps

    try:
        timeout_count = 0
        max_timeouts = 10
        no_signal_count = 0
        max_no_signal = 100

        while True:
            sr = sdr.readStream(stream, [buff], buff_size, timeoutUs=100000)
            if sr.ret == SoapySDR.SOAPY_SDR_TIMEOUT:
                timeout_count += 1
                logging.warning(f"Stream read timeout ({timeout_count}/{max_timeouts}): {sr.ret}")
                if timeout_count >= max_timeouts:
                    logging.info("Switching to lower sample rate (10 Msps) due to timeouts")
                    sdr.deactivateStream(stream)
                    sdr.setSampleRate(SOAPY_SDR_RX, 0, 10e6)
                    sample_rate = 10e6
                    sample_rate_index = 1
                    target_spl = int(target_spl * 10e6 / sample_rates[sample_rate_index + 1])
                    samples_needed = target_spl * lines_per_frame
                    sdr.activateStream(stream)
                    timeout_count = 0
                time.sleep(0.01)
                continue
            elif sr.ret <= 0:
                logging.warning(f"Stream read error: {sr.ret}")
                continue

            timeout_count = 0
            iq_samples = buff[:sr.ret]
            signal_power = np.mean(np.abs(iq_samples) ** 2)
            logging.debug(f"Signal power: {signal_power:.4f}")

            if signal_power < 1e-6:
                no_signal_count += 1
                logging.debug("No significant signal detected")
                if no_signal_count >= max_no_signal:
                    logging.error(f"No signal detected for too long. Verify frequency ({freq_mhz} MHz) and antenna.")
                    break
                continue

            no_signal_count = 0
            sample_buffer = np.concatenate((sample_buffer, iq_samples))
            logging.debug(f"Accumulated samples: {len(sample_buffer)}/{samples_needed}")

            if len(sample_buffer) >= samples_needed:
                logging.debug("Attempting to process frame")
                demod = fm_demodulate(sample_buffer[:samples_needed], smooth=smooth_demod)
                logging.debug(f"Demod stats: min={np.min(demod):.4f}, max={np.max(demod):.4f}, mean={np.mean(demod):.4f}")
                frame = process_video_signal(
                    demod, sample_rate, freq_mhz, band_name, target_spl=target_spl,
                    lines_per_frame=lines_per_frame, active_fraction=active_fraction,
                    line_offset=line_offset, h_sync_adjust=h_sync_adjust, active_lines=active_lines,
                    frame_rate_factor=frame_rate_factor, fps=fps
                )
                if frame is not None:
                    logging.debug("Frame processed successfully")
                    cv2.imshow("Analog Video", frame)
                    frame_count += 1
                else:
                    logging.debug("Frame processing failed")
                sample_buffer = sample_buffer[samples_needed:]

            # Update FPS every second
            current_time = time.time()
            if current_time - fps_start_time >= 1.0:
                fps = frame_count / max(current_time - fps_start_time, 1e-6)
                logging.debug(f"FPS updated: {fps:.1f}")
                frame_count = 0
                fps_start_time = current_time

            # Handle key presses
            key = cv2.waitKey(1)
            if key in [ord('q'), ord('Q')]:
                logging.info("Quitting due to 'q' or 'Q' key press")
                break
            elif key in [ord('a'), ord('A')]:  # A key
                target_spl = max(1, target_spl - 1)
                samples_needed = target_spl * lines_per_frame
                logging.info(f"Decreased target_spl to {target_spl}, samples_needed={samples_needed}")
            elif key in [ord('d'), ord('D')]:  # D key
                target_spl += 1
                samples_needed = target_spl * lines_per_frame
                logging.info(f"Increased target_spl to {target_spl}, samples_needed={samples_needed}")
            elif key in [ord('s'), ord('S')]:  # S key: toggle PAL/NTSC
                is_pal = not is_pal
                if is_pal:
                    target_spl = int(64e-6 * sample_rate)  # ~1224 at 16 Msps
                    lines_per_frame = 625
                    active_lines = 575  # PAL default after blanking
                    standard = "PAL"
                else:
                    target_spl = int(63.5e-6 * sample_rate)  # ~1017 at 16 Msps
                    lines_per_frame = 525
                    active_lines = 485  # NTSC default after blanking
                    standard = "NTSC"
                samples_needed = target_spl * lines_per_frame
                logging.info(f"Switched to {standard}: target_spl={target_spl}, lines_per_frame={lines_per_frame}, active_lines={active_lines}, samples_needed={samples_needed}")
            elif key in [ord('w'), ord('W')]:  # W key: cycle active_fraction
                active_fractions = [0.80, 0.85, 0.90]
                current_idx = active_fractions.index(active_fraction) if active_fraction in active_fractions else 1
                active_fraction = active_fractions[(current_idx + 1) % len(active_fractions)]
                logging.info(f"Set active_fraction to {active_fraction}")
            elif key in [ord('z'), ord('Z')]:  # Z key: decrease line_offset
                line_offset = max(-50, line_offset - 10)
                logging.info(f"Decreased line_offset to {line_offset}")
            elif key in [ord('x'), ord('X')]:  # X key: increase line_offset
                line_offset = min(50, line_offset + 10)
                logging.info(f"Increased line_offset to {line_offset}")
            elif key in [ord('c'), ord('C')]:  # C key: cycle sample_rate
                sample_rate_index = (sample_rate_index + 1) % len(sample_rates)
                new_sample_rate = sample_rates[sample_rate_index]
                logging.info(f"Switching to sample_rate={new_sample_rate} Hz")
                sdr.deactivateStream(stream)
                sdr.setSampleRate(SOAPY_SDR_RX, 0, new_sample_rate)
                target_spl = int(target_spl * new_sample_rate / sample_rate)
                sample_rate = new_sample_rate
                samples_needed = target_spl * lines_per_frame
                sdr.activateStream(stream)
                logging.info(f"Updated target_spl={target_spl}, samples_needed={samples_needed}")
            elif key in [ord('q'), ord('Q')]:  # Q key: decrease h_sync_adjust
                h_sync_adjust = max(-20, h_sync_adjust - 1)
                logging.info(f"Decreased h_sync_adjust to {h_sync_adjust}")
            elif key in [ord('e'), ord('E')]:  # E key: increase h_sync_adjust
                h_sync_adjust = min(20, h_sync_adjust + 1)
                logging.info(f"Increased h_sync_adjust to {h_sync_adjust}")
            elif key in [ord('r'), ord('R')]:  # R key: increase active_lines
                active_lines = min(600, active_lines + 10)
                logging.info(f"Increased active_lines to {active_lines}")
            elif key in [ord('f'), ord('F')]:  # F key: decrease active_lines
                active_lines = max(300, active_lines - 10)
                logging.info(f"Decreased active_lines to {active_lines}")
            elif key in [ord('t'), ord('T')]:  # T key: increase frame_rate_factor
                frame_rate_factor = min(1.1, frame_rate_factor + 0.01)
                samples_needed = int(target_spl * frame_rate_factor) * lines_per_frame
                logging.info(f"Increased frame_rate_factor to {frame_rate_factor}, samples_needed={samples_needed}")
            elif key in [ord('g'), ord('G')]:  # G key: decrease frame_rate_factor
                frame_rate_factor = max(0.9, frame_rate_factor - 0.01)
                samples_needed = int(target_spl * frame_rate_factor) * lines_per_frame
                logging.info(f"Decreased frame_rate_factor to {frame_rate_factor}, samples_needed={samples_needed}")
            elif key in [ord('y'), ord('Y')]:  # Y key: decrease LNA gain
                lna_gain = max(0, lna_gain - 2)
                sdr.setGain(SOAPY_SDR_RX, 0, "LNA", lna_gain)
                logging.info(f"Decreased LNA gain to {lna_gain} dB")
            elif key in [ord('u'), ord('U')]:  # U key: increase LNA gain
                lna_gain = min(48, lna_gain + 2)
                sdr.setGain(SOAPY_SDR_RX, 0, "LNA", lna_gain)
                logging.info(f"Increased LNA gain to {lna_gain} dB")
            elif key in [ord('i'), ord('I')]:  # I key: decrease VGA gain
                vga_gain = max(0, vga_gain - 2)
                sdr.setGain(SOAPY_SDR_RX, 0, "VGA", vga_gain)
                logging.info(f"Decreased VGA gain to {vga_gain} dB")
            elif key in [ord('o'), ord('O')]:  # O key: increase VGA gain
                vga_gain = min(62, vga_gain + 2)
                sdr.setGain(SOAPY_SDR_RX, 0, "VGA", vga_gain)
                logging.info(f"Increased VGA gain to {vga_gain} dB")
            elif key in [ord('v'), ord('V')]:  # V key: toggle demod smoothing
                smooth_demod = not smooth_demod
                logging.info(f"Demod smoothing {'enabled' if smooth_demod else 'disabled'}")

    except KeyboardInterrupt:
        logging.info("Interrupted by user")
    except Exception as e:
        logging.error(f"Error during streaming: {str(e)}")
    finally:
        if sdr is not None:
            try:
                sdr.deactivateStream(stream)
                sdr.closeStream(stream)
                logging.info("Stream closed")
            except:
                pass
            sdr = None
        cv2.destroyAllWindows()
        logging.info("Display window closed")

if __name__ == "__main__":
    main()