hackrf_scanner.py

#!/usr/bin/env python3
"""
HackRF Frequency Scanner

This script uses a HackRF device to scan a range of frequencies and stops when
it detects a signal strength above a specified threshold.
"""

import argparse
import os
import select
import sys
import termios
import time
import tty
import yaml
import numpy as np
from scipy import signal
import matplotlib.pyplot as plt
from threading import Event, Thread
from subprocess import Popen, PIPE, STDOUT

def load_config(config_file):
    """Load configuration from YAML file."""
    try:
        with open(config_file, 'r') as f:
            config = yaml.safe_load(f)
        return config
    except Exception as e:
        print(f"Error loading config: {e}")
        sys.exit(1)

def parse_hackrf_sweep_output(line):
    """Parse a line of output from hackrf_sweep and extract frequency and power data."""
    try:
        parts = line.strip().split(', ')
        if len(parts) >= 7:  # Typical format from hackrf_sweep
            # Make sure this is a data line with frequency information
            # Skip lines that don't have valid frequency information
            if not parts[2].isdigit() or not parts[3].isdigit():
                return None, None
                
            date = parts[0]
            time_str = parts[1]
            hz_low = int(parts[2])
            hz_high = int(parts[3])
            hz_bin_width = float(parts[4])
            
            # Extract dBm values - they start from the 6th element
            try:
                dbm_values = [float(x) for x in parts[6:]]
                # Calculate frequencies for each bin
                freqs = np.linspace(hz_low, hz_high, len(dbm_values))
                return freqs, dbm_values
            except ValueError:
                # If conversion to float fails, this isn't a data line
                pass
    except Exception as e:
        # More detailed error information for debugging
        # Don't print for every line to avoid flooding output
        if line and any(x in line for x in ['error', 'fail', 'sweep', 'hackrf']):
            print(f"Parse error: {str(e)[:100]} for line: {line.strip()[:50]}...")
    
    return None, None

def scan_frequencies(config, stop_event=None, continuous_mode=False, collect_data=False, max_samples=100):
    """
    Scan frequencies using hackrf_sweep and stop when signal exceeds threshold.
    
    Args:
        config: Configuration dictionary
        stop_event: Threading event to signal when to stop
        continuous_mode: If True, continues scanning until manually stopped
        collect_data: If True, collect data for plotting even below threshold
        max_samples: Maximum number of frequency samples to collect
    
    Returns:
        If collect_data is True, returns a tuple of (freq_data, power_data, max_freq, max_power)
        Otherwise, returns a tuple of (max_freq, max_power) if signal above threshold is found
    """
    start_freq = config['start_frequency'] / 1e6  # Convert to MHz for hackrf_sweep
    end_freq = config['end_frequency'] / 1e6  # Convert to MHz for hackrf_sweep
    threshold = config['dbm_threshold']
    gain = config.get('gain', 20)
    
    # For 300 MHz range (2.2 GHz to 2.5 GHz), we need a larger bin width
    # The HackRF has a limit of 8184 FFT bins for the entire range
    # Using a value of 5000 kHz (5 MHz) which should be safe
    bin_size_khz = 5000  # Fixed at 5 MHz
    
    # Format frequency parameters as integers for hackrf_sweep
    start_freq_int = int(start_freq)
    end_freq_int = int(end_freq)
    
    cmd = [
        'hackrf_sweep',
        '-f', f"{start_freq_int}:{end_freq_int}",
        '-w', f"{bin_size_khz}",
        '-g', f"{gain}"
    ]
    
    # Only add the one-shot flag if not in continuous mode
    if not continuous_mode:
        cmd.append('-1')
    
    print(f"Starting HackRF sweep from {start_freq} MHz to {end_freq} MHz")
    print(f"Signal threshold: {threshold} dBm")
    print(f"Running command: {' '.join(cmd)}")
    
    process = Popen(cmd, stdout=PIPE, stderr=STDOUT, text=True, bufsize=1, universal_newlines=True)
    
    try:
        max_freq = None
        max_power = float('-inf')
        empty_count = 0  # Track empty lines for error detection
        
        # For data collection and plotting
        all_freqs = []
        all_powers = []
        
        line_count = 0
        print("Waiting for HackRF data...")
        
        for line in iter(process.stdout.readline, ''):
            line_count += 1
            
            # Print raw data occasionally for debugging
            if line_count <= 5 or line_count % 20 == 0:
                print(f"DEBUG: Raw line[{line_count}]: {line.strip()[:100]}")
            
            # Reset empty line counter when we get data
            if line.strip():
                empty_count = 0
            else:
                empty_count += 1
                if empty_count > 10:
                    print("No data received for several iterations. Is the HackRF connected?")
                    print("Note: hackrf_sweep might not be working properly or returning expected data format.")
                    if not continuous_mode:
                        break
                    empty_count = 0  # Reset counter and continue trying
                continue
                
            if stop_event and stop_event.is_set():
                break
                
            # Debug the raw line contents if it matches specific keywords
            if line.strip() and ('2200000000' in line or '2500000000' in line or 'MHz' in line):
                print(f"FREQUENCY DATA: {line.strip()[:150]}")
                
            freqs, powers = parse_hackrf_sweep_output(line)
            if freqs is not None and powers is not None:
                # Data collection for plotting
                if collect_data and len(freqs) > 0 and len(powers) > 0:
                    all_freqs.extend(freqs)
                    all_powers.extend(powers)
                
                print(f"PARSED DATA: Got {len(powers)} power values from {freqs[0]/1e6:.2f} to {freqs[-1]/1e6:.2f} MHz")
                # Find the maximum power in this sweep
                if len(powers) > 0:
                    sweep_max_idx = np.argmax(powers)
                    sweep_max_power = powers[sweep_max_idx]
                    sweep_max_freq = freqs[sweep_max_idx]
                    
                    # Update overall maximum if this is higher
                    if sweep_max_power > max_power:
                        max_power = sweep_max_power
                        max_freq = sweep_max_freq
                    
                    # Print more detailed information about the scan
                    print(f"Scanning: {freqs[0]/1e6:.2f}-{freqs[-1]/1e6:.2f} MHz | "
                          f"Max power: {sweep_max_power:.2f} dBm at {sweep_max_freq/1e6:.2f} MHz")
                    
                    # Print a simplified spectrum visualization
                    if len(powers) > 10:
                        # Create a simple ASCII spectrum visualization
                        print("Spectrum: ", end="")
                        for p in powers[::len(powers)//10][:10]:  # Sample 10 points
                            bars = int((p + 100) // 5)  # Normalize to positive range
                            print("█" * min(bars, 10), end=" ")
                        print(f" | Peak: {sweep_max_power:.1f} dBm")
                    
                    # Check if we've found a signal above threshold
                    if sweep_max_power > threshold:
                        print(f"\n*** SIGNAL DETECTED ***")
                        print(f"Frequency: {sweep_max_freq/1e6:.4f} MHz")
                        print(f"Power: {sweep_max_power:.2f} dBm")
                        
                        # If we're collecting data, return all data plus the peak
                        if collect_data:
                            return all_freqs, all_powers, sweep_max_freq, sweep_max_power
                        else:
                            return sweep_max_freq, sweep_max_power
    
    except KeyboardInterrupt:
        print("\nScan interrupted by user")
    except Exception as e:
        print(f"\nError during scanning: {e}")
        if continuous_mode:
            print("Attempting to continue...")
            return None, None
    finally:
        try:
            process.terminate()
            process.wait(timeout=5)
        except:
            try:
                process.kill()
            except:
                pass  # Process might already be gone
    
    # Return collected data if requested
    if collect_data:
        if all_freqs and all_powers:
            print(f"\nCollected {len(all_freqs)} data points across {start_freq}-{end_freq} MHz range")
        else:
            print("\nNo data points collected for plotting")
            # Return empty lists instead of None to avoid unpacking errors
            all_freqs = []
            all_powers = []
        return all_freqs, all_powers, max_freq, max_power
    
    # Standard operation mode
    if max_freq is not None:
        print(f"\nMaximum signal found: {max_power:.2f} dBm at {max_freq/1e6:.4f} MHz")
        print(f"(Did not exceed threshold of {threshold} dBm)")
        return max_freq, max_power  # Return the max values even if below threshold
    else:
        print("\nNo valid signals detected")
    
    return None, None

def plot_console(freqs, powers, config, width=80, height=10):
    """Create an ASCII plot of the spectrum for console output with simple bars."""
    if not freqs or not powers or len(freqs) == 0 or len(powers) == 0:
        print("No data available for console plot")
        return
    
    # Convert frequencies to MHz
    freq_mhz = [f/1e6 for f in freqs]
    
    # Create frequency bins across the range
    min_freq = config['start_frequency']/1e6
    max_freq = config['end_frequency']/1e6
    num_bins = width - 10  # Leave space for labels
    
    # Create bins
    freq_bins = np.linspace(min_freq, max_freq, num_bins)
    power_bins = [float('-inf')] * num_bins  # Initialize with very low values
    
    # Assign powers to bins (taking the maximum in each bin)
    for f, p in zip(freq_mhz, powers):
        bin_idx = int((f - min_freq) / (max_freq - min_freq) * (num_bins-1))
        if 0 <= bin_idx < num_bins and p > power_bins[bin_idx]:
            power_bins[bin_idx] = p
    
    # Fill in gaps with interpolation to make a continuous display
    for i in range(num_bins):
        if power_bins[i] == float('-inf'):
            # Find closest non-empty bins
            left_idx = right_idx = i
            left_val = right_val = float('-inf')
            
            # Find closest value to the left
            for j in range(i-1, -1, -1):
                if power_bins[j] != float('-inf'):
                    left_idx = j
                    left_val = power_bins[j]
                    break
            
            # Find closest value to the right
            for j in range(i+1, num_bins):
                if power_bins[j] != float('-inf'):
                    right_idx = j
                    right_val = power_bins[j]
                    break
            
            # Interpolate if we have values on both sides
            if left_val != float('-inf') and right_val != float('-inf'):
                power_bins[i] = left_val + (right_val - left_val) * (i - left_idx) / (right_idx - left_idx)
            # Or use nearest non-empty value
            elif left_val != float('-inf'):
                power_bins[i] = left_val
            elif right_val != float('-inf'):
                power_bins[i] = right_val
            else:
                power_bins[i] = -100  # Default value if no neighbors found
    
    # Find min/max for scaling
    min_power = min(power_bins)
    max_power = max(power_bins)
    
    # Ensure min_power and max_power are different
    if max_power - min_power < 10:
        min_power = max_power - 10
    
    # Create the ASCII plot
    result = []
    result.append(f"Signal Strength vs Frequency ({min_freq:.0f}-{max_freq:.0f} MHz)")
    result.append(f"Power range: {min_power:.1f} to {max_power:.1f} dBm (threshold: {config['dbm_threshold']} dBm)")
    result.append("-" * width)
    
    # Draw the plot
    for h in range(height, 0, -1):
        # Calculate the power level for this row
        power_level = min_power + (max_power - min_power) * h / height
        
        # Print the power label on the y-axis
        if h == height:
            row = f"{max_power:6.1f} |" 
        elif h == 1:
            row = f"{min_power:6.1f} |"
        elif h == height // 2:
            mid_power = min_power + (max_power - min_power) / 2
            row = f"{mid_power:6.1f} |"
        else:
            row = "       |"
        
        # Draw the data points - simplified to just use full blocks for continuous bars
        for power in power_bins:
            if power >= power_level:
                row += "█"  # Full block for points above this level
            else:
                row += " "  # Empty for points below this level
                
        result.append(row)
    
    # Draw the x-axis
    result.append("       " + "-" * num_bins)
    
    # Draw the frequency labels
    x_labels = ""
    for i in range(5):
        pos = i * (num_bins - 1) // 4
        freq_val = min_freq + (max_freq - min_freq) * i / 4
        label = f"{freq_val:.0f}"
        x_labels += label + " " * (pos - len(x_labels))
    
    # Print x-axis labels with some spacing
    result.append("       " + x_labels + f"{max_freq:.0f}")
    result.append(" " * 7 + "Frequency (MHz)")
    
    return "\n".join(result)

def plot_results(freq, power, config, all_freqs=None, all_powers=None, filename="frequency_scan.png", block=False):
    """Plot the detected signal and save to file."""
    # Note: We no longer need to print the ASCII plot here as it's done in the main loop every second
    
    # Then create the matplotlib plot
    plt.figure(figsize=(12, 8))
    plt.axhline(y=config['dbm_threshold'], color='r', linestyle='--', label=f"Threshold ({config['dbm_threshold']} dBm)")
    
    # If we have a full frequency sweep, plot it
    if all_freqs and len(all_freqs) > 0 and all_powers and len(all_powers) > 0:
        # Convert to MHz for plotting
        freq_mhz = [f/1e6 for f in all_freqs]
        
        # Create heatmap-style plot by binning data
        freq_bins = np.linspace(config['start_frequency']/1e6, config['end_frequency']/1e6, 300)
        power_bins = {}
        
        # Group by frequency bin
        for f, p in zip(freq_mhz, all_powers):
            # Find nearest bin
            bin_idx = int((f - freq_bins[0]) / (freq_bins[-1] - freq_bins[0]) * (len(freq_bins)-1))
            if 0 <= bin_idx < len(freq_bins):
                bin_freq = freq_bins[bin_idx]
                if bin_freq not in power_bins:
                    power_bins[bin_freq] = []
                power_bins[bin_freq].append(p)
        
        # Calculate max power for each bin
        bin_freqs = []
        bin_powers = []
        for bin_freq, powers in power_bins.items():
            if powers:  # Skip empty bins
                bin_freqs.append(bin_freq)
                bin_powers.append(max(powers))  # Take max power in each bin
        
        # Sort by frequency
        sorted_data = sorted(zip(bin_freqs, bin_powers))
        if sorted_data:
            bin_freqs, bin_powers = zip(*sorted_data)
            
            # Plot the frequency sweep
            plt.plot(bin_freqs, bin_powers, 'b-', alpha=0.7, linewidth=1)
            plt.scatter(bin_freqs, bin_powers, color='b', marker='.', s=5, alpha=0.5, label="Frequency Sweep")
    
    # If we have a specific strong signal, highlight it
    if freq is not None and power is not None:
        plt.scatter(freq/1e6, power, color='r', marker='o', s=100, label="Peak Signal")
        plt.annotate(f"{power:.2f} dBm", (freq/1e6, power), 
                     xytext=(10, 10), textcoords='offset points', 
                     color='red', fontweight='bold')
    
    # Plot styling
    plt.title(f"HackRF Frequency Scan: {config['start_frequency']/1e6:.1f}-{config['end_frequency']/1e6:.1f} MHz")
    plt.xlabel('Frequency (MHz)')
    plt.ylabel('Power (dBm)')
    plt.grid(True, alpha=0.3)
    plt.legend()
    
    # Y-axis limits to focus on relevant signal range
    if all_powers and len(all_powers) > 0:
        plt.ylim([min(min(all_powers), -100), max(max(all_powers), config['dbm_threshold']) + 5])
    else:
        # Default range if no data
        plt.ylim([-100, config['dbm_threshold'] + 5])
    
    # Save the plot
    plt.savefig(filename, dpi=300)
    print(f"Plot saved to {filename}")
    
    if block:
        plt.show()
    else:
        # Non-blocking plot display
        plt.draw()
        plt.pause(0.001)  # Small pause to update the figure
        plt.close('all')  # Close it after drawing to avoid blocking

def is_data_available():
    """Check if there is data available on stdin without blocking."""
    return select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], [])

def get_keypress():
    """Get a single keypress without requiring Enter."""
    if is_data_available():
        return sys.stdin.read(1)
    return None

def start_hackrf_sweep(config):
    """Start the hackrf_sweep process and return the process object."""
    start_freq = config['start_frequency'] / 1e6  # Convert to MHz for hackrf_sweep
    end_freq = config['end_frequency'] / 1e6  # Convert to MHz for hackrf_sweep
    gain = config.get('gain', 20)
    
    # Format frequency parameters as integers for hackrf_sweep
    start_freq_int = int(start_freq)
    end_freq_int = int(end_freq)
    
    # Fixed bin width to avoid exceeding FFT limits
    bin_size_khz = 5000  # 5 MHz
    
    cmd = [
        'hackrf_sweep',
        '-f', f"{start_freq_int}:{end_freq_int}",
        '-w', f"{bin_size_khz}",
        '-g', f"{gain}"
    ]
    
    print(f"Starting HackRF sweep from {start_freq} MHz to {end_freq} MHz")
    print(f"Running command: {' '.join(cmd)}")
    
    try:
        # Process with text mode enabled (no need to decode manually)
        process = Popen(cmd, stdout=PIPE, stderr=STDOUT, text=True, bufsize=1, universal_newlines=True)
        return process
    except Exception as e:
        print(f"Error starting HackRF sweep: {e}")
        return None

def main():
    parser = argparse.ArgumentParser(description='HackRF Frequency Scanner')
    parser.add_argument('--config', default='config.yaml', help='Path to configuration file')
    parser.add_argument('--plot', action='store_true', help='Plot now and also enable on-demand plotting')
    parser.add_argument('--continuous', action='store_true', 
                      help='Run in continuous mode until manually interrupted')
    parser.add_argument('--collect-time', type=int, default=30,
                      help='Time in seconds to collect data before plotting')
    args = parser.parse_args()
    
    # Load configuration
    config = load_config(args.config)
    
    print("\n=== HackRF Frequency Scanner ===")
    print(f"Scanning range: {config['start_frequency']/1e6:.1f} MHz to {config['end_frequency']/1e6:.1f} MHz")
    print(f"Threshold: {config['dbm_threshold']} dBm | Gain: {config.get('gain', 20)}")
    print("Press 'p' to plot current data | Press 'q' to quit")
    
    # Set terminal to raw mode to read keystrokes without requiring Enter
    old_settings = termios.tcgetattr(sys.stdin)
    
    # Variables to track scan data
    all_freqs = []
    all_powers = []
    top_signals = []  # List of (freq, power) tuples for top 5 signals
    last_display_time = 0
    plot_count = 0
    
    try:
        # Set terminal to raw mode (single character input without Enter)
        tty.setcbreak(sys.stdin.fileno())
        
        # Start the HackRF sweep process
        process = start_hackrf_sweep(config)
        if not process:
            print("Error starting HackRF sweep. Exiting.")
            return
        
        # Clear the screen initially
        os.system('clear')
        print("=== HackRF Frequency Scanner ===")
        print(f"Scanning {config['start_frequency']/1e6:.1f} MHz to {config['end_frequency']/1e6:.1f} MHz")
        print("Waiting for data...")
        print("\nPress 'p' to plot current data | Press 'q' to quit")
        
        # Main processing loop
        while True:
            # Check for keypresses
            if is_data_available():
                key = sys.stdin.read(1)
                if key == 'q':
                    print("\nQuitting by user request")
                    break
                elif key == 'p':
                    print("\nGenerating plot file...")
                    plot_count += 1
                    if len(all_freqs) > 0:
                        # Get the strongest signal for the plot title
                        if top_signals:
                            max_freq, max_power = top_signals[0]
                        else:
                            max_freq, max_power = None, None
                            
                        # Generate plot with a unique filename - save to file only, don't display
                        timestamp = time.strftime("%Y%m%d-%H%M%S")
                        filename = f"frequency_scan_{timestamp}.png"
                        
                        # Create a matplotlib plot without displaying it
                        plt.figure(figsize=(12, 8))
                        plt.axhline(y=config['dbm_threshold'], color='r', linestyle='--', 
                                   label=f"Threshold ({config['dbm_threshold']} dBm)")
                        
                        # If we have a max signal to highlight
                        if max_freq is not None and max_power is not None:
                            plt.scatter([max_freq], [max_power], color='r', s=100, zorder=10, 
                                       label=f"Peak: {max_freq/1e6:.1f} MHz @ {max_power:.1f} dBm")
                        
                        # Plot all collected frequency data
                        plt.scatter(all_freqs, all_powers, s=2, alpha=0.5, color='b', label="Collected Data")
                        
                        # Add a trendline if possible
                        if len(all_freqs) > 100:
                            try:
                                # Sort the data by frequency
                                sorted_indices = np.argsort(all_freqs)
                                sorted_freqs = np.array(all_freqs)[sorted_indices]
                                sorted_powers = np.array(all_powers)[sorted_indices]
                                
                                # Apply smoothing
                                window_len = min(1001, len(all_freqs) // 10 * 2 + 1)  # Ensure window length is odd
                                if window_len > 3:
                                    smoothed = np.convolve(sorted_powers, np.ones(window_len)/window_len, mode='valid')
                                    smoothed_freqs = sorted_freqs[window_len//2:-(window_len//2)]
                                    plt.plot(smoothed_freqs, smoothed, 'g-', linewidth=2, alpha=0.7, label="Trend")
                            except Exception as e:
                                print(f"Could not create trend line: {e}")
                        
                        # Formatting
                        plt.xlabel('Frequency (Hz)')
                        plt.ylabel('Signal Strength (dBm)')
                        plt.title(f"HackRF Frequency Scan: {config['start_frequency']/1e6:.1f}-{config['end_frequency']/1e6:.1f} MHz")
                        plt.grid(True, alpha=0.3)
                        plt.legend()
                        
                        # Set vertical range
                        plt.ylim([min(min(all_powers), -100), max(max(all_powers), config['dbm_threshold']) + 5])
                        
                        # Save to file and close plot (without displaying)
                        plt.savefig(filename, dpi=300)
                        plt.close('all')
                        print(f"\nPlot saved to {filename}")
                    else:
                        print("No data collected yet for plotting")
            
            # Read a line from the HackRF process
            try:
                # Check if we need to decode (depends on how we set up the process)
                line = process.stdout.readline()
                if isinstance(line, bytes):
                    line = line.decode('utf-8')
                    
                if not line:
                    if process.poll() is not None:
                        print("\nHackRF process exited. Restarting...")
                        process = start_hackrf_sweep(config)
                        if not process:
                            print("Error restarting HackRF sweep. Exiting.")
                            break
                    continue
            except Exception as e:
                print(f"Error reading from HackRF process: {e}")
                continue
            
            # Parse the data
            freqs, powers = parse_hackrf_sweep_output(line)
            if freqs is not None and powers is not None and len(freqs) > 0 and len(powers) > 0:
                # Add to our collected data
                all_freqs.extend(freqs)
                all_powers.extend(powers)
                
                # Find the peak power in this sweep
                if len(powers) > 0:
                    max_idx = np.argmax(powers)
                    sweep_max_power = powers[max_idx]
                    sweep_max_freq = freqs[max_idx] if max_idx < len(freqs) else 0
                    
                    # Update top signals
                    found = False
                    for i, (f, p) in enumerate(top_signals):
                        # Update if we found a stronger signal at a similar frequency
                        if abs(f - sweep_max_freq) < 1e6 and sweep_max_power > p:  # Within 1 MHz
                            top_signals[i] = (sweep_max_freq, sweep_max_power)
                            found = True
                            break
                    if not found:
                        top_signals.append((sweep_max_freq, sweep_max_power))
                    
                    # Sort by power (descending) and keep only top 5
                    top_signals.sort(key=lambda x: x[1], reverse=True)
                    top_signals = top_signals[:5]
                
                # Update the display every second
                current_time = time.time()
                if current_time - last_display_time >= 1:
                    # Clear the console 
                    os.system('clear')
                    print(f"=== HackRF Frequency Scanner ===\nScanning {config['start_frequency']/1e6:.1f}-{config['end_frequency']/1e6:.1f} MHz")
                    
                    # Create the ASCII spectrum plot (compact version)
                    if len(all_freqs) > 100:  # Only plot if we have enough data points
                        console_plot = plot_console(all_freqs, all_powers, config, height=8)
                        print(console_plot)
                    
                    # Display the top 5 signals
                    print("\nTop 5 Signals:")
                    if top_signals:
                        for freq, power in top_signals:
                            print(f"{freq/1e6:.1f} MHz => {power:.1f} dBm")
                    else:
                        print("No signals detected yet")
                    
                    print("\nPress 'p' to plot current data | Press 'q' to quit")
                    
                    # Limit the amount of data we store
                    if len(all_freqs) > 150000:  # Prevent memory issues
                        # Keep the most recent data
                        all_freqs = all_freqs[-100000:]
                        all_powers = all_powers[-100000:]
                    
                    last_display_time = current_time
        
        # Clean up the process
        if process and process.poll() is None:
            process.terminate()
            process.wait(timeout=2)
            
    except Exception as e:
        print(f"\nError in main loop: {e}")
    except Exception as e:
        print(f"\nError: {e}")
    finally:
        # Restore terminal settings
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
        print("\nScanner stopped.")

if __name__ == "__main__":
    main()