0
/ Published in Signal Intelligence & Detection Techniques, Tech

Real-Time Signal Detection

Spread the love

🎯 Real-Time Signal Detection with BB60C and CSV Export for FCC Lookup 📡

In the world of RF surveillance, detecting low-SNR or ultra-wideband (UWB) signals is like finding a needle in a haystack—especially when those signals might only appear briefly or blend into the noise floor. That’s where this custom Python script comes in.

This tool is designed for Signal Hound BB60C users and helps you automate the detection of suspicious or unknown signals across the spectrum. It logs results to a CSV file that you can cross-reference with FCC databases for signal origin validation or interference analysis. 💾✅

🧠 How It Works

Let’s break the script down step-by-step:

🔌 1. BB60C API Integration

The script uses ctypes to dynamically load the BB60C API DLL from multiple known paths:

pythonCopyEditbb_api = ctypes.WinDLL(bb_api_path)

If the API isn’t found, a FileNotFoundError is thrown. This ensures that the SDK is properly installed before you even begin.

🧪 2. Signal Detection Configuration

The main class BB60CDetector is parameterized with thresholds for:

  • SNR threshold – To filter out weak noise.
  • FFT size – Controls resolution.
  • Minimum bandwidth & duration – For contour filtering.

You define the signal quality you’re interested in from the start. 🎛️

⚙️ 3. Device Initialization & Sweep Setup

pythonCopyEditbb_api.bbOpenDevice(ctypes.byref(device))

The BB60C is opened and configured for sweeping mode, setting:

  • Center frequency and span
  • RBW/VBW (e.g., 10 kHz)
  • Averaging & log power scaling

This optimizes the sweep to detect subtle or wideband transient events, like UWB bursts or digital radar noise.

📈 4. Data Collection: Sweeps

Using .fetch_trace(), it pulls raw dBm power values into an FFT-style numpy array. These are collected for multiple sweeps per center frequency.

🎯 5. Adaptive Thresholding

Here’s the magic: an adaptive threshold is calculated using Median Absolute Deviation (MAD) per sweep row to tolerate background noise shifts.

pythonCopyEditthreshold = noise_level + self.threshold_snr * mad
binary = spectrogram > threshold

This highlights above-noise spikes potentially representing real signals 📡.

🧩 6. Contour Detection

Contours are extracted with OpenCV from the binary image of signal spikes:

pythonCopyEditcontours, _ = cv2.findContours(...)

Each contour is then filtered by bandwidth and duration (in Hz and sample count). This helps ignore short glitches or narrow spurious signals.

🏷️ 7. Annotation & Classification

Each signal is labeled based on its bandwidth:

  • 🟢 "UWB" if it exceeds 500 MHz or 20% of the sampling rate
  • 🔵 "Narrowband" otherwise

And every detection is timestamped:

pythonCopyEdit"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")

🧾 8. CSV Logging for FCC Lookup

Detected signals are exported to a CSV file with fields like:

  • freq_lower_edge, freq_upper_edge
  • sample_start, sample_count
  • label, timestamp

This CSV can now be fed into a lookup script that checks against FCC license databases, helping identify:

  • Unauthorized transmitters
  • Malicious interference
  • UWB emissions from unknown sources 🚨

🧪 Current Use Case: UWB & Low-SNR Signals

You’re currently expanding the tool to:

  • Improve detection of UWB bursts, which are wide and short
  • Tune detection for low SNR (Signal-to-Noise Ratio) signals that may otherwise be buried

These are critical steps for TSCM (Technical Surveillance Countermeasures), especially when:

  • 📍 A suspicious emitter is operating covertly
  • 🤖 A machine or AI-controlled signal shows up across different bands
  • 🛰️ Hypothetical neural or BCI tech might use low-SNR far-field backscatter

📊 Debugging Tools Included

When debug=True, the script outputs:

  • 🖼 spectrogram_binary.png – Visual map of detected spikes
  • 🟥 contours.png – Signals outlined after filtering

Perfect for visually verifying results or documenting a sweep session.

💡 Final Thoughts

This script gives RF analysts and TIs a serious tool for:

  • 🕵️‍♂️ Catching suspicious UWB/NB emissions
  • 🧠 Investigating neurotech or subliminal signal claims
  • 🗃️ Logging legal-grade evidence for licensing and enforcement review

🔄 Next Steps

  • ✅ Integrate your FCC lookup script for real-time validation
  • 🧠 Use bb60c_wideband_mode in the future to capture non-sweep backscatter?
  • 📡 Improve signal tracking across time, not just sweep slices

📥 Download the Script

Coming soon as part of the TI Tools GitHub Toolkit, or email us for early access. Stay tuned for the version with IQ capture and neural signal detection tagging 🧬🧠

import csv
import ctypes
import numpy as np
import cv2
import os
from datetime import datetime

# Load BB60C API (tries bb_series/lib/win/vs2012/x64 or x86 relative to script)
def find_bb_api_dll():
    script_dir = os.path.dirname(os.path.abspath(__file__))
    possible_paths = [
        os.path.join(script_dir, 'bb_series', 'lib', 'win', 'vs2012', 'x64', 'bb_api.dll'),
        os.path.join(script_dir, 'bb_series', 'lib', 'win', 'vs2012', 'x86', 'bb_api.dll'),
        'bb_api.dll'  # fallback to system path
    ]
    for path in possible_paths:
        if os.path.exists(path):
            return path
    raise FileNotFoundError("bb_api.dll not found in expected locations. Ensure Signal Hound SDK is installed.")

try:
    bb_api_path = find_bb_api_dll()
    bb_api = ctypes.WinDLL(bb_api_path)
except FileNotFoundError:
    raise FileNotFoundError("bb_api.dll not found. Ensure Signal Hound SDK is installed.")

class BB60CDetector:
    def __init__(self, threshold_snr=5.0, fft_size=1024, min_bandwidth_hz=1000, min_duration_samples=100):
        self.threshold_snr = threshold_snr
        self.fft_size = fft_size
        self.min_bandwidth_hz = min_bandwidth_hz
        self.min_duration_samples = min_duration_samples
        self.sample_rate = None
        self.center_freq = None
        self.device = None

    def open_device(self):
        """Initialize BB60C device."""
        device = ctypes.c_int(0)
        status = bb_api.bbOpenDevice(ctypes.byref(device))
        if status != 0:
            raise RuntimeError(f"Failed to open BB60C: Error code {status}")
        self.device = device
        return device

    def configure_sweep(self, center_freq, span, rbw=10e3, vbw=10e3):
        """Configure BB60C for a sweep."""
        bb_api.bbConfigureAcquisition(self.device, ctypes.c_int(1), ctypes.c_int(1))  # AVERAGE, LOG_SCALE
        bb_api.bbConfigureCenterSpan(self.device, ctypes.c_double(center_freq), ctypes.c_double(span))
        bb_api.bbConfigureLevel(self.device, ctypes.c_double(-30), ctypes.c_int(0))  # Ref level -30 dBm, auto attenuation
        bb_api.bbConfigureGain(self.device, ctypes.c_int(0))  # Auto gain
        bb_api.bbConfigureSweepCoupling(
            self.device, ctypes.c_double(rbw), ctypes.c_double(vbw),
            ctypes.c_double(0.001), ctypes.c_uint(0), ctypes.c_uint(0)
        )  # RBW, VBW, sweep time 1ms, flat-top, no spur reject
        bb_api.bbConfigureProcUnits(self.device, ctypes.c_int(1))  # Power units (dBm)
        status = bb_api.bbInitiate(self.device, ctypes.c_int(0), ctypes.c_int(0))  # Sweeping mode
        if status != 0:
            raise RuntimeError(f"Failed to initiate sweep: Error code {status}")

    def fetch_trace(self):
        """Fetch a single sweep trace."""
        sweep_size = ctypes.c_uint(0)
        bin_size = ctypes.c_double(0)
        start_freq = ctypes.c_double(0)
        bb_api.bbQueryTraceInfo(self.device, ctypes.byref(sweep_size), ctypes.byref(bin_size), ctypes.byref(start_freq))
        sweep_size = sweep_size.value
        min_swp = (ctypes.c_float * sweep_size)()
        max_swp = (ctypes.c_float * sweep_size)()
        status = bb_api.bbFetchTrace_32f(self.device, sweep_size, min_swp, max_swp)
        if status != 0:
            raise RuntimeError(f"Failed to fetch trace: Error code {status}")
        return np.array(max_swp), start_freq.value, bin_size.value

    def detect_signals(self, sweeps, start_freqs, bin_size, sample_rate, debug=False):
        """Process sweeps to detect signals and generate annotations."""
        # Combine sweeps into a spectrogram-like array
        num_rows = len(sweeps)
        sweep_size = len(sweeps[0])
        spectrogram = np.array(sweeps)  # Shape: (num_rows, sweep_size)

        # Adaptive thresholding
        noise_level = np.median(spectrogram, axis=1, keepdims=True)
        mad = np.median(np.abs(spectrogram - noise_level), axis=1, keepdims=True) * 1.4826
        threshold = noise_level + self.threshold_snr * mad
        binary = spectrogram > threshold

        if debug:
            import matplotlib.pyplot as plt
            plt.imshow(binary, aspect='auto')
            plt.savefig("spectrogram_binary.png")
            plt.close()

        # Contour detection
        image_8bit = np.uint8(binary * 255)
        contours, _ = cv2.findContours(image_8bit, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

        # Filter contours
        filtered_contours = []
        for contour in contours:
            x, y, w, h = cv2.boundingRect(contour)
            bandwidth_hz = w * bin_size
            duration_samples = h * sample_rate / num_rows
            if bandwidth_hz > self.min_bandwidth_hz and duration_samples > self.min_duration_samples:
                filtered_contours.append(contour)

        print(f"num contours found: {len(filtered_contours)}")

        if debug:
            backtorgb = cv2.cvtColor(image_8bit, cv2.COLOR_GRAY2RGB)
            img = cv2.drawContours(backtorgb, filtered_contours, -1, (0, 0, 255), 3)
            cv2.imwrite("contours.png", img)

        # Generate annotations
        annotations = []
        for contour in filtered_contours:
            x, y, w, h = cv2.boundingRect(contour)
            freq_lower = start_freqs[y] + x * bin_size
            freq_upper = start_freqs[y] + (x + w) * bin_size
            bandwidth_hz = freq_upper - freq_lower
            sample_start = int(y * sample_rate / num_rows)
            sample_count = int(h * sample_rate / num_rows)
            label = "UWB" if bandwidth_hz > 0.2 * sample_rate or bandwidth_hz > 500e6 else "Narrowband"
            an = {
                "freq_lower_edge": int(freq_lower),
                "freq_upper_edge": int(freq_upper),
                "sample_start": sample_start,
                "sample_count": sample_count,
                "label": label,
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            annotations.append(an)

        return annotations

    def sweep_and_log(self, start_freq=9e3, stop_freq=6e9, span=27e6, sweeps_per_span=10, csv_file="detections.csv"):
        """Sweep the spectrum and log detections to CSV."""
        try:
            self.open_device()
            freq_step = span  # Step by center frequency
            center_freqs = np.arange(start_freq + span/2, stop_freq, freq_step)
            sample_rate = span  # Approximate sample rate based on span
            self.sample_rate = sample_rate

            # Initialize CSV file
            with open(csv_file, mode='w', newline='') as f:
                writer = csv.DictWriter(f, fieldnames=["freq_lower_edge", "freq_upper_edge", "sample_start", "sample_count", "label", "timestamp"])
                writer.writeheader()

            # Sweep loop
            all_sweeps = []
            all_start_freqs = []
            for center_freq in center_freqs:
                print(f"Sweeping at center freq: {center_freq/1e6:.2f} MHz")
                self.configure_sweep(center_freq, span)
                sweeps = []
                for _ in range(sweeps_per_span):
                    trace, start_freq, bin_size = self.fetch_trace()
                    sweeps.append(trace)
                    all_start_freqs.append(start_freq)
                all_sweeps.extend(sweeps)

                # Process sweeps in batches
                annotations = self.detect_signals(sweeps, [start_freq] * sweeps_per_span, bin_size, sample_rate, debug=True)
                
                # Append to CSV
                with open(csv_file, mode='a', newline='') as f:
                    writer = csv.DictWriter(f, fieldnames=["freq_lower_edge", "freq_upper_edge", "sample_start", "sample_count", "label", "timestamp"])
                    for an in annotations:
                        writer.writerow(an)

            # Final processing for remaining sweeps
            if all_sweeps:
                annotations = self.detect_signals(all_sweeps, all_start_freqs, bin_size, sample_rate, debug=True)
                with open(csv_file, mode='a', newline='') as f:
                    writer = csv.DictWriter(f, fieldnames=["freq_lower_edge", "freq_upper_edge", "sample_start", "sample_count", "label", "timestamp"])
                    for an in annotations:
                        writer.writerow(an)

        finally:
            if self.device:
                bb_api.bbCloseDevice(self.device)

if __name__ == "__main__":
    try:
        print("Initializing BB60C Detector...")
        detector = BB60CDetector(
            threshold_snr=5.0,
            fft_size=1024,
            min_bandwidth_hz=1000,
            min_duration_samples=100
        )
        print("Starting sweep and log operation...")
        detector.sweep_and_log(
            start_freq=9e3,
            stop_freq=6e9,
            span=27e6,
            sweeps_per_span=10,
            csv_file="detections.csv"
        )
    except FileNotFoundError as e:
        print(f"Error: {e}")
        print("Please ensure the Signal Hound SDK is installed and bb_api.dll is in your system PATH or current directory")
    except RuntimeError as e:
        print(f"Error: {e}")
        print("Please ensure the BB60C device is properly connected and powered on")
    except Exception as e:
        print(f"Unexpected error: {e}")
        import traceback
        traceback.print_exc()

What you can read next

Ways to Evade Antennas
DIY Contact Pad Antennas
New Techniques to Detect Covert Threats

Leave a Reply

Your email address will not be published. Required fields are marked *