Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-24 14:50:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add multi-SDR hardware support (LimeSDR, HackRF) and setup script

Browse Source 
- Add SDR hardware abstraction layer (utils/sdr/) with support for:
  - RTL-SDR (existing, using native rtl_* tools)
  - LimeSDR (via SoapySDR)
  - HackRF (via SoapySDR)
- Add hardware type selector to UI with capabilities display
- Add automatic device detection across all supported hardware
- Add hardware-specific parameter validation (frequency/gain ranges)
- Add setup.sh script for automated dependency installation
- Update README with multi-SDR docs, installation guide, troubleshooting
- Add SoapySDR/LimeSDR/HackRF to dependency definitions
- Fix dump1090 detection for Homebrew on Apple Silicon Macs
- Remove defunct NOAA-15/18/19 satellites, add NOAA-21

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
James Smith
2026-01-02 14:23:51 +00:00
parent 3437a2fc0a
commit 5ed9674e1f
17 changed files with 1957 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
utils/sdr/__init__.py Normal file
View File

@@ -0,0 +1,196 @@
"""
SDR Hardware Abstraction Layer.
This module provides a unified interface for multiple SDR hardware types
including RTL-SDR, LimeSDR, and HackRF. Use SDRFactory to detect devices
and get appropriate command builders.
Example usage:
from utils.sdr import SDRFactory, SDRType
# Detect all connected devices
devices = SDRFactory.detect_devices()
# Get a command builder for a specific device
builder = SDRFactory.get_builder_for_device(devices[0])
# Or get a builder by type
builder = SDRFactory.get_builder(SDRType.RTL_SDR)
# Build commands
cmd = builder.build_fm_demod_command(device, frequency_mhz=153.35)
"""
from typing import Optional
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
from .detection import detect_all_devices
from .rtlsdr import RTLSDRCommandBuilder
from .limesdr import LimeSDRCommandBuilder
from .hackrf import HackRFCommandBuilder
from .validation import (
SDRValidationError,
validate_frequency,
validate_gain,
validate_sample_rate,
validate_ppm,
validate_device_index,
validate_squelch,
get_capabilities_for_type,
)
class SDRFactory:
"""Factory for creating SDR command builders and detecting devices."""
_builders: dict[SDRType, type[CommandBuilder]] = {
SDRType.RTL_SDR: RTLSDRCommandBuilder,
SDRType.LIME_SDR: LimeSDRCommandBuilder,
SDRType.HACKRF: HackRFCommandBuilder,
}
@classmethod
def get_builder(cls, sdr_type: SDRType) -> CommandBuilder:
"""
Get a command builder for the specified SDR type.
Args:
sdr_type: The SDR hardware type
Returns:
CommandBuilder instance for the specified type
Raises:
ValueError: If the SDR type is not supported
"""
builder_class = cls._builders.get(sdr_type)
if not builder_class:
raise ValueError(f"Unsupported SDR type: {sdr_type}")
return builder_class()
@classmethod
def get_builder_for_device(cls, device: SDRDevice) -> CommandBuilder:
"""
Get a command builder for a specific device.
Args:
device: The SDR device
Returns:
CommandBuilder instance for the device's type
"""
return cls.get_builder(device.sdr_type)
@classmethod
def detect_devices(cls) -> list[SDRDevice]:
"""
Detect all available SDR devices.
Returns:
List of detected SDR devices
"""
return detect_all_devices()
@classmethod
def get_supported_types(cls) -> list[SDRType]:
"""
Get list of supported SDR types.
Returns:
List of supported SDRType values
"""
return list(cls._builders.keys())
@classmethod
def get_capabilities(cls, sdr_type: SDRType) -> SDRCapabilities:
"""
Get capabilities for an SDR type.
Args:
sdr_type: The SDR hardware type
Returns:
SDRCapabilities for the specified type
"""
builder = cls.get_builder(sdr_type)
return builder.get_capabilities()
@classmethod
def get_all_capabilities(cls) -> dict[str, dict]:
"""
Get capabilities for all supported SDR types.
Returns:
Dictionary mapping SDR type names to capability dicts
"""
capabilities = {}
for sdr_type in cls._builders:
caps = cls.get_capabilities(sdr_type)
capabilities[sdr_type.value] = {
'name': sdr_type.name.replace('_', ' '),
'freq_min_mhz': caps.freq_min_mhz,
'freq_max_mhz': caps.freq_max_mhz,
'gain_min': caps.gain_min,
'gain_max': caps.gain_max,
'sample_rates': caps.sample_rates,
'supports_bias_t': caps.supports_bias_t,
'supports_ppm': caps.supports_ppm,
'tx_capable': caps.tx_capable,
}
return capabilities
@classmethod
def create_default_device(
cls,
sdr_type: SDRType,
index: int = 0,
serial: str = 'N/A'
) -> SDRDevice:
"""
Create a default device object for a given SDR type.
Useful when device detection didn't provide full details but
you know the hardware type.
Args:
sdr_type: The SDR hardware type
index: Device index (default 0)
serial: Device serial (default 'N/A')
Returns:
SDRDevice with default capabilities for the type
"""
caps = cls.get_capabilities(sdr_type)
return SDRDevice(
sdr_type=sdr_type,
index=index,
name=f'{sdr_type.name.replace("_", " ")} Device {index}',
serial=serial,
driver=sdr_type.value,
capabilities=caps
)
# Export commonly used items at package level
__all__ = [
# Factory
'SDRFactory',
# Types and classes
'SDRType',
'SDRDevice',
'SDRCapabilities',
'CommandBuilder',
# Builders
'RTLSDRCommandBuilder',
'LimeSDRCommandBuilder',
'HackRFCommandBuilder',
# Validation
'SDRValidationError',
'validate_frequency',
'validate_gain',
'validate_sample_rate',
'validate_ppm',
'validate_device_index',
'validate_squelch',
'get_capabilities_for_type',
]

149
utils/sdr/base.py Normal file
View File

@@ -0,0 +1,149 @@
"""
Base classes and types for SDR hardware abstraction.
This module provides the core abstractions for supporting multiple SDR hardware
types (RTL-SDR, LimeSDR, HackRF, etc.) through a unified interface.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class SDRType(Enum):
"""Supported SDR hardware types."""
RTL_SDR = "rtlsdr"
LIME_SDR = "limesdr"
HACKRF = "hackrf"
# Future support
# USRP = "usrp"
# BLADE_RF = "bladerf"
@dataclass
class SDRCapabilities:
"""Hardware capabilities for an SDR device."""
sdr_type: SDRType
freq_min_mhz: float # Minimum frequency in MHz
freq_max_mhz: float # Maximum frequency in MHz
gain_min: float # Minimum gain in dB
gain_max: float # Maximum gain in dB
sample_rates: list[int] = field(default_factory=list) # Supported sample rates
supports_bias_t: bool = False # Bias-T support
supports_ppm: bool = True # PPM correction support
tx_capable: bool = False # Can transmit
@dataclass
class SDRDevice:
"""Detected SDR device."""
sdr_type: SDRType
index: int
name: str
serial: str
driver: str # e.g., "rtlsdr", "lime", "hackrf"
capabilities: SDRCapabilities
def to_dict(self) -> dict:
"""Convert to dictionary for JSON serialization."""
return {
'index': self.index,
'name': self.name,
'serial': self.serial,
'sdr_type': self.sdr_type.value,
'driver': self.driver,
'capabilities': {
'freq_min_mhz': self.capabilities.freq_min_mhz,
'freq_max_mhz': self.capabilities.freq_max_mhz,
'gain_min': self.capabilities.gain_min,
'gain_max': self.capabilities.gain_max,
'sample_rates': self.capabilities.sample_rates,
'supports_bias_t': self.capabilities.supports_bias_t,
'supports_ppm': self.capabilities.supports_ppm,
'tx_capable': self.capabilities.tx_capable,
}
}
class CommandBuilder(ABC):
"""Abstract base class for building SDR commands."""
@abstractmethod
def build_fm_demod_command(
self,
device: SDRDevice,
frequency_mhz: float,
sample_rate: int = 22050,
gain: Optional[float] = None,
ppm: Optional[int] = None,
modulation: str = "fm",
squelch: Optional[int] = None
) -> list[str]:
"""
Build FM demodulation command (for pager, iridium).
Args:
device: The SDR device to use
frequency_mhz: Center frequency in MHz
sample_rate: Audio sample rate (default 22050 for pager)
gain: Gain in dB (None for auto)
ppm: PPM frequency correction
modulation: Modulation type (fm, am, etc.)
squelch: Squelch level
Returns:
Command as list of strings for subprocess
"""
pass
@abstractmethod
def build_adsb_command(
self,
device: SDRDevice,
gain: Optional[float] = None
) -> list[str]:
"""
Build ADS-B decoder command.
Args:
device: The SDR device to use
gain: Gain in dB (None for auto)
Returns:
Command as list of strings for subprocess
"""
pass
@abstractmethod
def build_ism_command(
self,
device: SDRDevice,
frequency_mhz: float = 433.92,
gain: Optional[float] = None,
ppm: Optional[int] = None
) -> list[str]:
"""
Build ISM band decoder command (433MHz sensors).
Args:
device: The SDR device to use
frequency_mhz: Center frequency in MHz (default 433.92)
gain: Gain in dB (None for auto)
ppm: PPM frequency correction
Returns:
Command as list of strings for subprocess
"""
pass
@abstractmethod
def get_capabilities(self) -> SDRCapabilities:
"""Return hardware capabilities for this SDR type."""
pass
@classmethod
@abstractmethod
def get_sdr_type(cls) -> SDRType:
"""Return the SDR type this builder handles."""
pass

306
utils/sdr/detection.py Normal file
View File

@@ -0,0 +1,306 @@
"""
Multi-hardware SDR device detection.
Detects RTL-SDR devices via rtl_test and other SDR hardware via SoapySDR.
"""
import logging
import re
import shutil
import subprocess
from typing import Optional
from .base import SDRCapabilities, SDRDevice, SDRType
logger = logging.getLogger(__name__)
def _check_tool(name: str) -> bool:
"""Check if a tool is available in PATH."""
return shutil.which(name) is not None
def _get_capabilities_for_type(sdr_type: SDRType) -> SDRCapabilities:
"""Get default capabilities for an SDR type."""
# Import here to avoid circular imports
from .rtlsdr import RTLSDRCommandBuilder
from .limesdr import LimeSDRCommandBuilder
from .hackrf import HackRFCommandBuilder
builders = {
SDRType.RTL_SDR: RTLSDRCommandBuilder,
SDRType.LIME_SDR: LimeSDRCommandBuilder,
SDRType.HACKRF: HackRFCommandBuilder,
}
builder_class = builders.get(sdr_type)
if builder_class:
return builder_class.CAPABILITIES
# Fallback generic capabilities
return SDRCapabilities(
sdr_type=sdr_type,
freq_min_mhz=1.0,
freq_max_mhz=6000.0,
gain_min=0.0,
gain_max=50.0,
sample_rates=[2048000],
supports_bias_t=False,
supports_ppm=False,
tx_capable=False
)
def _driver_to_sdr_type(driver: str) -> Optional[SDRType]:
"""Map SoapySDR driver name to SDRType."""
mapping = {
'rtlsdr': SDRType.RTL_SDR,
'lime': SDRType.LIME_SDR,
'limesdr': SDRType.LIME_SDR,
'hackrf': SDRType.HACKRF,
# Future support
# 'uhd': SDRType.USRP,
# 'bladerf': SDRType.BLADE_RF,
}
return mapping.get(driver.lower())
def detect_rtlsdr_devices() -> list[SDRDevice]:
"""
Detect RTL-SDR devices using rtl_test.
This uses the native rtl_test tool for best compatibility with
existing RTL-SDR installations.
"""
devices: list[SDRDevice] = []
if not _check_tool('rtl_test'):
logger.debug("rtl_test not found, skipping RTL-SDR detection")
return devices
try:
result = subprocess.run(
['rtl_test', '-t'],
capture_output=True,
text=True,
timeout=5
)
output = result.stderr + result.stdout
# Parse device info from rtl_test output
# Format: "0: Realtek, RTL2838UHIDIR, SN: 00000001"
device_pattern = r'(\d+):\s+(.+?)(?:,\s*SN:\s*(\S+))?$'
from .rtlsdr import RTLSDRCommandBuilder
for line in output.split('\n'):
line = line.strip()
match = re.match(device_pattern, line)
if match:
devices.append(SDRDevice(
sdr_type=SDRType.RTL_SDR,
index=int(match.group(1)),
name=match.group(2).strip().rstrip(','),
serial=match.group(3) or 'N/A',
driver='rtlsdr',
capabilities=RTLSDRCommandBuilder.CAPABILITIES
))
# Fallback: if we found devices but couldn't parse details
if not devices:
found_match = re.search(r'Found (\d+) device', output)
if found_match:
count = int(found_match.group(1))
for i in range(count):
devices.append(SDRDevice(
sdr_type=SDRType.RTL_SDR,
index=i,
name=f'RTL-SDR Device {i}',
serial='Unknown',
driver='rtlsdr',
capabilities=RTLSDRCommandBuilder.CAPABILITIES
))
except subprocess.TimeoutExpired:
logger.warning("rtl_test timed out")
except Exception as e:
logger.debug(f"RTL-SDR detection error: {e}")
return devices
def detect_soapy_devices() -> list[SDRDevice]:
"""
Detect SDR devices via SoapySDR.
This detects LimeSDR, HackRF, USRP, BladeRF, and other SoapySDR-compatible
devices. RTL-SDR devices may also appear here but we prefer the native
detection for those.
"""
devices: list[SDRDevice] = []
if not _check_tool('SoapySDRUtil'):
logger.debug("SoapySDRUtil not found, skipping SoapySDR detection")
return devices
try:
result = subprocess.run(
['SoapySDRUtil', '--find'],
capture_output=True,
text=True,
timeout=10
)
# Parse SoapySDR output
# Format varies but typically includes lines like:
# " driver = lime"
# " serial = 0009060B00123456"
# " label = LimeSDR Mini [USB 3.0] 0009060B00123456"
current_device: dict = {}
device_counts: dict[SDRType, int] = {}
for line in result.stdout.split('\n'):
line = line.strip()
# Start of new device block
if line.startswith('Found device'):
if current_device.get('driver'):
_add_soapy_device(devices, current_device, device_counts)
current_device = {}
continue
# Parse key = value pairs
if ' = ' in line:
key, value = line.split(' = ', 1)
key = key.strip()
value = value.strip()
current_device[key] = value
# Don't forget the last device
if current_device.get('driver'):
_add_soapy_device(devices, current_device, device_counts)
except subprocess.TimeoutExpired:
logger.warning("SoapySDRUtil timed out")
except Exception as e:
logger.debug(f"SoapySDR detection error: {e}")
return devices
def _add_soapy_device(
devices: list[SDRDevice],
device_info: dict,
device_counts: dict[SDRType, int]
) -> None:
"""Add a device from SoapySDR detection to the list."""
driver = device_info.get('driver', '').lower()
sdr_type = _driver_to_sdr_type(driver)
if not sdr_type:
logger.debug(f"Unknown SoapySDR driver: {driver}")
return
# Skip RTL-SDR devices from SoapySDR (we use native detection)
if sdr_type == SDRType.RTL_SDR:
return
# Track device index per type
if sdr_type not in device_counts:
device_counts[sdr_type] = 0
index = device_counts[sdr_type]
device_counts[sdr_type] += 1
devices.append(SDRDevice(
sdr_type=sdr_type,
index=index,
name=device_info.get('label', device_info.get('driver', 'Unknown')),
serial=device_info.get('serial', 'N/A'),
driver=driver,
capabilities=_get_capabilities_for_type(sdr_type)
))
def detect_hackrf_devices() -> list[SDRDevice]:
"""
Detect HackRF devices using native hackrf_info tool.
Fallback for when SoapySDR is not available.
"""
devices: list[SDRDevice] = []
if not _check_tool('hackrf_info'):
return devices
try:
result = subprocess.run(
['hackrf_info'],
capture_output=True,
text=True,
timeout=5
)
# Parse hackrf_info output
# Look for "Serial number:" lines
serial_pattern = r'Serial number:\s*(\S+)'
from .hackrf import HackRFCommandBuilder
serials_found = re.findall(serial_pattern, result.stdout)
for i, serial in enumerate(serials_found):
devices.append(SDRDevice(
sdr_type=SDRType.HACKRF,
index=i,
name=f'HackRF One',
serial=serial,
driver='hackrf',
capabilities=HackRFCommandBuilder.CAPABILITIES
))
# Fallback: check if any HackRF found without serial
if not devices and 'Found HackRF' in result.stdout:
devices.append(SDRDevice(
sdr_type=SDRType.HACKRF,
index=0,
name='HackRF One',
serial='Unknown',
driver='hackrf',
capabilities=HackRFCommandBuilder.CAPABILITIES
))
except Exception as e:
logger.debug(f"HackRF detection error: {e}")
return devices
def detect_all_devices() -> list[SDRDevice]:
"""
Detect all connected SDR devices across all supported hardware types.
Returns a unified list of SDRDevice objects sorted by type and index.
"""
devices: list[SDRDevice] = []
# RTL-SDR via native tool (primary method)
devices.extend(detect_rtlsdr_devices())
# SoapySDR devices (LimeSDR, HackRF, etc.)
soapy_devices = detect_soapy_devices()
devices.extend(soapy_devices)
# Native HackRF detection (fallback if SoapySDR didn't find it)
hackrf_from_soapy = any(d.sdr_type == SDRType.HACKRF for d in soapy_devices)
if not hackrf_from_soapy:
devices.extend(detect_hackrf_devices())
# Sort by type name, then index
devices.sort(key=lambda d: (d.sdr_type.value, d.index))
logger.info(f"Detected {len(devices)} SDR device(s)")
for d in devices:
logger.debug(f" {d.sdr_type.value}:{d.index} - {d.name} (serial: {d.serial})")
return devices

148
utils/sdr/hackrf.py Normal file
View File

@@ -0,0 +1,148 @@
"""
HackRF command builder implementation.
Uses SoapySDR-based tools for FM demodulation and signal capture.
HackRF supports 1 MHz to 6 GHz frequency range.
"""
from typing import Optional
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
class HackRFCommandBuilder(CommandBuilder):
"""HackRF command builder using SoapySDR tools."""
CAPABILITIES = SDRCapabilities(
sdr_type=SDRType.HACKRF,
freq_min_mhz=1.0, # 1 MHz
freq_max_mhz=6000.0, # 6 GHz
gain_min=0.0,
gain_max=62.0, # LNA (0-40) + VGA (0-62)
sample_rates=[2000000, 4000000, 8000000, 10000000, 20000000],
supports_bias_t=True,
supports_ppm=False,
tx_capable=True
)
def _build_device_string(self, device: SDRDevice) -> str:
"""Build SoapySDR device string for HackRF."""
if device.serial and device.serial != 'N/A':
return f'driver=hackrf,serial={device.serial}'
return f'driver=hackrf'
def _split_gain(self, gain: float) -> tuple[int, int]:
"""
Split total gain into LNA and VGA components.
HackRF has two gain stages:
- LNA: 0-40 dB (RF amplifier)
- VGA: 0-62 dB (IF amplifier)
This function distributes the requested gain across both stages.
"""
if gain <= 40:
# All to LNA first
return int(gain), 0
else:
# Max out LNA, rest to VGA
lna = 40
vga = min(62, int(gain - 40))
return lna, vga
def build_fm_demod_command(
self,
device: SDRDevice,
frequency_mhz: float,
sample_rate: int = 22050,
gain: Optional[float] = None,
ppm: Optional[int] = None,
modulation: str = "fm",
squelch: Optional[int] = None
) -> list[str]:
"""
Build SoapySDR rx_fm command for FM demodulation.
For pager decoding and iridium capture with HackRF.
"""
device_str = self._build_device_string(device)
cmd = [
'rx_fm',
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
'-s', str(sample_rate),
]
if gain is not None and gain > 0:
lna, vga = self._split_gain(gain)
cmd.extend(['-g', f'LNA={lna},VGA={vga}'])
if squelch is not None and squelch > 0:
cmd.extend(['-l', str(squelch)])
# Output to stdout
cmd.append('-')
return cmd
def build_adsb_command(
self,
device: SDRDevice,
gain: Optional[float] = None
) -> list[str]:
"""
Build dump1090/readsb command with SoapySDR support for ADS-B decoding.
Uses readsb which has better SoapySDR support.
"""
device_str = self._build_device_string(device)
cmd = [
'readsb',
'--net',
'--device-type', 'soapysdr',
'--device', device_str,
'--quiet'
]
if gain is not None:
cmd.extend(['--gain', str(int(gain))])
return cmd
def build_ism_command(
self,
device: SDRDevice,
frequency_mhz: float = 433.92,
gain: Optional[float] = None,
ppm: Optional[int] = None
) -> list[str]:
"""
Build rtl_433 command with SoapySDR support for ISM band decoding.
rtl_433 has native SoapySDR support via -d flag.
"""
device_str = self._build_device_string(device)
cmd = [
'rtl_433',
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-F', 'json'
]
if gain is not None and gain > 0:
cmd.extend(['-g', str(int(gain))])
return cmd
def get_capabilities(self) -> SDRCapabilities:
"""Return HackRF capabilities."""
return self.CAPABILITIES
@classmethod
def get_sdr_type(cls) -> SDRType:
"""Return SDR type."""
return SDRType.HACKRF

136
utils/sdr/limesdr.py Normal file
View File

@@ -0,0 +1,136 @@
"""
LimeSDR command builder implementation.
Uses SoapySDR-based tools for FM demodulation and signal capture.
LimeSDR supports 100 kHz to 3.8 GHz frequency range.
"""
from typing import Optional
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
class LimeSDRCommandBuilder(CommandBuilder):
"""LimeSDR command builder using SoapySDR tools."""
CAPABILITIES = SDRCapabilities(
sdr_type=SDRType.LIME_SDR,
freq_min_mhz=0.1, # 100 kHz
freq_max_mhz=3800.0, # 3.8 GHz
gain_min=0.0,
gain_max=73.0, # Combined LNA + TIA + PGA
sample_rates=[1000000, 2000000, 4000000, 8000000, 10000000, 20000000],
supports_bias_t=False,
supports_ppm=False, # Uses TCXO, no PPM correction needed
tx_capable=True
)
def _build_device_string(self, device: SDRDevice) -> str:
"""Build SoapySDR device string for LimeSDR."""
if device.serial and device.serial != 'N/A':
return f'driver=lime,serial={device.serial}'
return f'driver=lime'
def build_fm_demod_command(
self,
device: SDRDevice,
frequency_mhz: float,
sample_rate: int = 22050,
gain: Optional[float] = None,
ppm: Optional[int] = None,
modulation: str = "fm",
squelch: Optional[int] = None
) -> list[str]:
"""
Build SoapySDR rx_fm command for FM demodulation.
For pager decoding and iridium capture with LimeSDR.
"""
device_str = self._build_device_string(device)
cmd = [
'rx_fm',
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
'-s', str(sample_rate),
]
if gain is not None and gain > 0:
# LimeSDR gain is applied to LNAH element
cmd.extend(['-g', f'LNAH={int(gain)}'])
if squelch is not None and squelch > 0:
cmd.extend(['-l', str(squelch)])
# Output to stdout
cmd.append('-')
return cmd
def build_adsb_command(
self,
device: SDRDevice,
gain: Optional[float] = None
) -> list[str]:
"""
Build dump1090 command with SoapySDR support for ADS-B decoding.
Uses dump1090 compiled with SoapySDR support, or readsb as alternative.
Note: Requires dump1090 with SoapySDR support or readsb.
"""
device_str = self._build_device_string(device)
# Try readsb first (better SoapySDR support), fallback to dump1090
cmd = [
'readsb',
'--net',
'--device-type', 'soapysdr',
'--device', device_str,
'--quiet'
]
if gain is not None:
cmd.extend(['--gain', str(int(gain))])
return cmd
def build_ism_command(
self,
device: SDRDevice,
frequency_mhz: float = 433.92,
gain: Optional[float] = None,
ppm: Optional[int] = None
) -> list[str]:
"""
Build rtl_433 command with SoapySDR support for ISM band decoding.
rtl_433 has native SoapySDR support via -d flag.
"""
device_str = self._build_device_string(device)
cmd = [
'rtl_433',
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-F', 'json'
]
if gain is not None and gain > 0:
cmd.extend(['-g', str(int(gain))])
# PPM not typically needed for LimeSDR (TCXO)
# but include if specified
if ppm is not None and ppm != 0:
cmd.extend(['-p', str(ppm)])
return cmd
def get_capabilities(self) -> SDRCapabilities:
"""Return LimeSDR capabilities."""
return self.CAPABILITIES
@classmethod
def get_sdr_type(cls) -> SDRType:
"""Return SDR type."""
return SDRType.LIME_SDR

121
utils/sdr/rtlsdr.py Normal file
View File

@@ -0,0 +1,121 @@
"""
RTL-SDR command builder implementation.
Uses native rtl_* tools (rtl_fm, rtl_433) and dump1090 for maximum compatibility
with existing RTL-SDR installations. No SoapySDR dependency required.
"""
from typing import Optional
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
class RTLSDRCommandBuilder(CommandBuilder):
"""RTL-SDR command builder using native rtl_* tools."""
CAPABILITIES = SDRCapabilities(
sdr_type=SDRType.RTL_SDR,
freq_min_mhz=24.0,
freq_max_mhz=1766.0,
gain_min=0.0,
gain_max=49.6,
sample_rates=[250000, 1024000, 1800000, 2048000, 2400000],
supports_bias_t=True,
supports_ppm=True,
tx_capable=False
)
def build_fm_demod_command(
self,
device: SDRDevice,
frequency_mhz: float,
sample_rate: int = 22050,
gain: Optional[float] = None,
ppm: Optional[int] = None,
modulation: str = "fm",
squelch: Optional[int] = None
) -> list[str]:
"""
Build rtl_fm command for FM demodulation.
Used for pager decoding and iridium capture.
"""
cmd = [
'rtl_fm',
'-d', str(device.index),
'-f', f'{frequency_mhz}M',
'-M', modulation,
'-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 squelch is not None and squelch > 0:
cmd.extend(['-l', str(squelch)])
# Output to stdout for piping
cmd.append('-')
return cmd
def build_adsb_command(
self,
device: SDRDevice,
gain: Optional[float] = None
) -> list[str]:
"""
Build dump1090 command for ADS-B decoding.
Uses dump1090 with network output for SBS data streaming.
"""
cmd = [
'dump1090',
'--net',
'--device-index', str(device.index),
'--quiet'
]
if gain is not None:
cmd.extend(['--gain', str(int(gain))])
return cmd
def build_ism_command(
self,
device: SDRDevice,
frequency_mhz: float = 433.92,
gain: Optional[float] = None,
ppm: Optional[int] = None
) -> list[str]:
"""
Build rtl_433 command for ISM band sensor decoding.
Outputs JSON for easy parsing.
"""
cmd = [
'rtl_433',
'-d', str(device.index),
'-f', f'{frequency_mhz}M',
'-F', 'json'
]
if gain is not None and gain > 0:
cmd.extend(['-g', str(int(gain))])
if ppm is not None and ppm != 0:
cmd.extend(['-p', str(ppm)])
return cmd
def get_capabilities(self) -> SDRCapabilities:
"""Return RTL-SDR capabilities."""
return self.CAPABILITIES
@classmethod
def get_sdr_type(cls) -> SDRType:
"""Return SDR type."""
return SDRType.RTL_SDR

257
utils/sdr/validation.py Normal file
View File

@@ -0,0 +1,257 @@
"""
Hardware-specific parameter validation for SDR devices.
Validates frequency, gain, sample rate, and other parameters against
the capabilities of specific SDR hardware.
"""
from typing import Optional
from .base import SDRCapabilities, SDRDevice, SDRType
class SDRValidationError(ValueError):
"""Raised when SDR parameter validation fails."""
pass
def validate_frequency(
freq_mhz: float,
device: Optional[SDRDevice] = None,
capabilities: Optional[SDRCapabilities] = None
) -> float:
"""
Validate frequency against device capabilities.
Args:
freq_mhz: Frequency in MHz
device: SDR device (optional, takes precedence)
capabilities: SDR capabilities (used if device not provided)
Returns:
Validated frequency in MHz
Raises:
SDRValidationError: If frequency is out of range
"""
if device:
caps = device.capabilities
elif capabilities:
caps = capabilities
else:
# Default RTL-SDR range for backwards compatibility
caps = SDRCapabilities(
sdr_type=SDRType.RTL_SDR,
freq_min_mhz=24.0,
freq_max_mhz=1766.0,
gain_min=0.0,
gain_max=50.0
)
if not caps.freq_min_mhz <= freq_mhz <= caps.freq_max_mhz:
raise SDRValidationError(
f"Frequency {freq_mhz} MHz out of range for {caps.sdr_type.value}. "
f"Valid range: {caps.freq_min_mhz}-{caps.freq_max_mhz} MHz"
)
return freq_mhz
def validate_gain(
gain: float,
device: Optional[SDRDevice] = None,
capabilities: Optional[SDRCapabilities] = None
) -> float:
"""
Validate gain against device capabilities.
Args:
gain: Gain in dB
device: SDR device (optional, takes precedence)
capabilities: SDR capabilities (used if device not provided)
Returns:
Validated gain in dB
Raises:
SDRValidationError: If gain is out of range
"""
if device:
caps = device.capabilities
elif capabilities:
caps = capabilities
else:
# Default range for backwards compatibility
caps = SDRCapabilities(
sdr_type=SDRType.RTL_SDR,
freq_min_mhz=24.0,
freq_max_mhz=1766.0,
gain_min=0.0,
gain_max=50.0
)
# Allow 0 for auto gain
if gain == 0:
return gain
if not caps.gain_min <= gain <= caps.gain_max:
raise SDRValidationError(
f"Gain {gain} dB out of range for {caps.sdr_type.value}. "
f"Valid range: {caps.gain_min}-{caps.gain_max} dB"
)
return gain
def validate_sample_rate(
rate: int,
device: Optional[SDRDevice] = None,
capabilities: Optional[SDRCapabilities] = None,
snap_to_nearest: bool = True
) -> int:
"""
Validate sample rate against device capabilities.
Args:
rate: Sample rate in Hz
device: SDR device (optional, takes precedence)
capabilities: SDR capabilities (used if device not provided)
snap_to_nearest: If True, return nearest valid rate instead of raising
Returns:
Validated sample rate in Hz
Raises:
SDRValidationError: If rate is invalid and snap_to_nearest is False
"""
if device:
caps = device.capabilities
elif capabilities:
caps = capabilities
else:
return rate # No validation without capabilities
if not caps.sample_rates:
return rate # No restrictions
if rate in caps.sample_rates:
return rate
if snap_to_nearest:
# Find closest valid rate
closest = min(caps.sample_rates, key=lambda x: abs(x - rate))
return closest
raise SDRValidationError(
f"Sample rate {rate} Hz not supported by {caps.sdr_type.value}. "
f"Valid rates: {caps.sample_rates}"
)
def validate_ppm(
ppm: int,
device: Optional[SDRDevice] = None,
capabilities: Optional[SDRCapabilities] = None
) -> int:
"""
Validate PPM frequency correction.
Args:
ppm: PPM correction value
device: SDR device (optional, takes precedence)
capabilities: SDR capabilities (used if device not provided)
Returns:
Validated PPM value
Raises:
SDRValidationError: If PPM is out of range or not supported
"""
if device:
caps = device.capabilities
elif capabilities:
caps = capabilities
else:
caps = None
# Check if device supports PPM
if caps and not caps.supports_ppm:
if ppm != 0:
# Warn but don't fail - some hardware just ignores PPM
pass
return 0 # Return 0 to indicate no correction
# Standard PPM range
if not -1000 <= ppm <= 1000:
raise SDRValidationError(
f"PPM correction {ppm} out of range. Valid range: -1000 to 1000"
)
return ppm
def validate_device_index(index: int) -> int:
"""
Validate device index.
Args:
index: Device index (0-255)
Returns:
Validated device index
Raises:
SDRValidationError: If index is out of range
"""
if not 0 <= index <= 255:
raise SDRValidationError(
f"Device index {index} out of range. Valid range: 0-255"
)
return index
def validate_squelch(squelch: int) -> int:
"""
Validate squelch level.
Args:
squelch: Squelch level (0-1000, 0 = off)
Returns:
Validated squelch level
Raises:
SDRValidationError: If squelch is out of range
"""
if not 0 <= squelch <= 1000:
raise SDRValidationError(
f"Squelch {squelch} out of range. Valid range: 0-1000"
)
return squelch
def get_capabilities_for_type(sdr_type: SDRType) -> SDRCapabilities:
"""
Get default capabilities for an SDR type.
Args:
sdr_type: The SDR type
Returns:
SDRCapabilities for the specified type
"""
from .rtlsdr import RTLSDRCommandBuilder
from .limesdr import LimeSDRCommandBuilder
from .hackrf import HackRFCommandBuilder
builders = {
SDRType.RTL_SDR: RTLSDRCommandBuilder,
SDRType.LIME_SDR: LimeSDRCommandBuilder,
SDRType.HACKRF: HackRFCommandBuilder,
}
builder_class = builders.get(sdr_type)
if builder_class:
return builder_class.CAPABILITIES
raise SDRValidationError(f"Unknown SDR type: {sdr_type}")