""" Multi-hardware SDR device detection. Detects RTL-SDR devices via rtl_test and other SDR hardware via SoapySDR. """ from __future__ import annotations import contextlib import logging import re import subprocess import time from utils.dependencies import get_tool_path from .base import SDRCapabilities, SDRDevice, SDRType logger = logging.getLogger(__name__) # Cache HackRF detection results so polling endpoints don't repeatedly run # hackrf_info while the device is actively streaming in SubGHz mode. _hackrf_cache: list[SDRDevice] = [] _hackrf_cache_ts: float = 0.0 _HACKRF_CACHE_TTL_SECONDS = 3.0 # Cache all-device detection results. Multiple endpoints call # detect_all_devices() on the same page load (e.g. /devices and /adsb/tools # both trigger it from DOMContentLoaded). On a Pi the subprocess calls # (rtl_test, SoapySDRUtil, hackrf_info) each take seconds and block the # single gevent worker, serialising every other request behind them. # A short TTL cache avoids duplicate subprocess storms. _all_devices_cache: list[SDRDevice] = [] _all_devices_cache_ts: float = 0.0 _ALL_DEVICES_CACHE_TTL_SECONDS = 5.0 def _hackrf_probe_blocked() -> bool: """Return True when probing HackRF would interfere with an active stream.""" try: from utils.subghz import get_subghz_manager return get_subghz_manager().active_mode in {"rx", "decode", "tx", "sweep"} except Exception: return False def _get_capabilities_for_type(sdr_type: SDRType) -> SDRCapabilities: """Get default capabilities for an SDR type.""" # Import here to avoid circular imports from .airspy import AirspyCommandBuilder from .hackrf import HackRFCommandBuilder from .limesdr import LimeSDRCommandBuilder from .rtlsdr import RTLSDRCommandBuilder from .sdrplay import SDRPlayCommandBuilder from .usrp import USRPCommandBuilder builders = { SDRType.RTL_SDR: RTLSDRCommandBuilder, SDRType.LIME_SDR: LimeSDRCommandBuilder, SDRType.HACKRF: HackRFCommandBuilder, SDRType.AIRSPY: AirspyCommandBuilder, SDRType.SDRPLAY: SDRPlayCommandBuilder, SDRType.USRP: USRPCommandBuilder, } 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) -> SDRType | None: """Map SoapySDR driver name to SDRType.""" mapping = { "rtlsdr": SDRType.RTL_SDR, "lime": SDRType.LIME_SDR, "limesdr": SDRType.LIME_SDR, "hackrf": SDRType.HACKRF, "airspy": SDRType.AIRSPY, "airspyhf": SDRType.AIRSPY, # Airspy HF+ uses same builder "sdrplay": SDRType.SDRPLAY, "uhd": SDRType.USRP, # Future support # '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] = [] rtl_test_path = get_tool_path("rtl_test") if not rtl_test_path: logger.debug("rtl_test not found, skipping RTL-SDR detection") return devices try: import os import platform env = os.environ.copy() if platform.system() == "Darwin": lib_paths = ["/usr/local/lib", "/opt/homebrew/lib"] current_ld = env.get("DYLD_LIBRARY_PATH", "") env["DYLD_LIBRARY_PATH"] = ":".join(lib_paths + [current_ld] if current_ld else lib_paths) try: result = subprocess.run( [rtl_test_path, "-t"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, env=env, ) except subprocess.TimeoutExpired: logger.warning("rtl_test timed out after 5s") return [] output = result.stderr + result.stdout # Parse device info from rtl_test output # Format: "0: Realtek, RTL2838UHIDIR, SN: 00000001" # Require a non-empty serial to avoid matching malformed lines like "SN:". device_pattern = r"(\d+):\s+(.+?),\s*SN:\s*(\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), 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 _find_soapy_util() -> str | None: """Find SoapySDR utility command (name varies by distribution).""" # Try different command names used across distributions for cmd in ["SoapySDRUtil", "soapy_sdr_util", "soapysdr-util"]: tool_path = get_tool_path(cmd) if tool_path: return tool_path return None def _get_soapy_env() -> dict: """Get environment variables needed for SoapySDR on macOS. On macOS with Homebrew, SoapySDR modules are installed in paths that require SOAPY_SDR_ROOT or DYLD_LIBRARY_PATH to be set. This fixes detection issues where modules like SoapyHackRF are installed but not found by SoapySDRUtil. See: https://github.com/smittix/intercept/issues/77 """ import os import platform env = os.environ.copy() if platform.system() == "Darwin": # Homebrew paths for Apple Silicon and Intel Macs homebrew_paths = ["/opt/homebrew", "/usr/local"] lib_paths = [] for base in homebrew_paths: lib_path = f"{base}/lib" if os.path.isdir(lib_path): lib_paths.append(lib_path) if lib_paths: current_dyld = env.get("DYLD_LIBRARY_PATH", "") env["DYLD_LIBRARY_PATH"] = ":".join(lib_paths + ([current_dyld] if current_dyld else [])) # Set SOAPY_SDR_ROOT if we found Homebrew installation for base in homebrew_paths: if os.path.isdir(f"{base}/lib/SoapySDR"): env["SOAPY_SDR_ROOT"] = base break return env def detect_soapy_devices(skip_types: set[SDRType] | None = None) -> list[SDRDevice]: """ Detect SDR devices via SoapySDR. This detects LimeSDR, HackRF, Airspy, and other SoapySDR-compatible devices. Args: skip_types: Set of SDRType values to skip (e.g., if already found via native detection) """ devices: list[SDRDevice] = [] skip_types = skip_types or set() soapy_cmd = _find_soapy_util() if not soapy_cmd: logger.debug("SoapySDR utility not found, skipping SoapySDR detection") return devices try: # Use macOS-aware environment to find Homebrew-installed modules env = _get_soapy_env() result = subprocess.run([soapy_cmd, "--find"], capture_output=True, text=True, timeout=10, env=env) # 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, skip_types) 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, skip_types) 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], skip_types: set[SDRType] ) -> 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 device types that were already found via native detection if sdr_type in skip_types: logger.debug(f"Skipping {driver} from SoapySDR (already found via native detection)") 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. """ global _hackrf_cache, _hackrf_cache_ts now = time.time() # While HackRF is actively streaming in SubGHz mode, skip probe calls. # Re-running hackrf_info during active RX/TX can disrupt the USB stream. if _hackrf_probe_blocked(): return list(_hackrf_cache) if _hackrf_cache and (now - _hackrf_cache_ts) < _HACKRF_CACHE_TTL_SECONDS: return list(_hackrf_cache) devices: list[SDRDevice] = [] hackrf_info_path = get_tool_path("hackrf_info") if not hackrf_info_path: _hackrf_cache = devices _hackrf_cache_ts = now return devices try: result = subprocess.run([hackrf_info_path], capture_output=True, text=True, timeout=5) # Combine stdout + stderr: newer firmware may print to stderr, # and hackrf_info may exit non-zero when device is briefly busy # but still output valid info. output = f"{result.stdout or ''}\n{result.stderr or ''}" # Parse hackrf_info output # Extract board name from "Board ID Number: X (Name)" and serial from .hackrf import HackRFCommandBuilder serial_pattern = re.compile( r"^\s*Serial\s+number:\s*(.+)$", re.IGNORECASE | re.MULTILINE, ) board_pattern = re.compile( r"Board\s+ID\s+Number:\s*\d+\s*\(([^)]+)\)", re.IGNORECASE, ) serials_found = [] for raw in serial_pattern.findall(output): # Normalise legacy formats like "0x1234 5678" to plain hex. serial = re.sub(r"0x", "", raw, flags=re.IGNORECASE) serial = re.sub(r"[^0-9A-Fa-f]", "", serial) if serial: serials_found.append(serial) boards_found = board_pattern.findall(output) for i, serial in enumerate(serials_found): board_name = boards_found[i] if i < len(boards_found) else "HackRF" devices.append( SDRDevice( sdr_type=SDRType.HACKRF, index=i, name=board_name, serial=serial, driver="hackrf", capabilities=HackRFCommandBuilder.CAPABILITIES, ) ) # Fallback: check if any HackRF found without serial if not devices and re.search(r"Found\s+HackRF", output, re.IGNORECASE): board_match = board_pattern.search(output) board_name = board_match.group(1) if board_match else "HackRF" devices.append( SDRDevice( sdr_type=SDRType.HACKRF, index=0, name=board_name, serial="Unknown", driver="hackrf", capabilities=HackRFCommandBuilder.CAPABILITIES, ) ) except Exception as e: logger.debug(f"HackRF detection error: {e}") _hackrf_cache = list(devices) _hackrf_cache_ts = now return devices def probe_rtlsdr_device(device_index: int) -> str | None: """Probe whether an RTL-SDR device is available at the USB level. Runs a quick ``rtl_test`` invocation targeting a single device to check for USB claim errors that indicate the device is held by an external process (or a stale handle from a previous crash). Args: device_index: The RTL-SDR device index to probe. Returns: An error message string if the device cannot be opened, or ``None`` if the device is available. """ rtl_test_path = get_tool_path("rtl_test") if not rtl_test_path: # Can't probe without rtl_test — let the caller proceed and # surface errors from the actual decoder process instead. return None try: import os import platform env = os.environ.copy() if platform.system() == "Darwin": lib_paths = ["/usr/local/lib", "/opt/homebrew/lib"] current_ld = env.get("DYLD_LIBRARY_PATH", "") env["DYLD_LIBRARY_PATH"] = ":".join(lib_paths + [current_ld] if current_ld else lib_paths) # Use Popen with early termination instead of run() with full timeout. # rtl_test prints device info to stderr quickly, then keeps running # its test loop. We kill it as soon as we see success or failure. proc = subprocess.Popen( [rtl_test_path, "-d", str(device_index), "-t"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, ) import select error_found = False device_found = False deadline = time.monotonic() + 3.0 try: while time.monotonic() < deadline: remaining = deadline - time.monotonic() if remaining <= 0: break # Wait for stderr output with timeout ready, _, _ = select.select([proc.stderr], [], [], min(remaining, 0.1)) if ready: line = proc.stderr.readline() if not line: break # EOF — process closed stderr # Check for no-device messages first (before success check, # since "No supported devices found" also contains "Found" + "device") if "no supported devices" in line.lower() or "no matching devices" in line.lower(): error_found = True break if "usb_claim_interface" in line or "Failed to open" in line: error_found = True break if "Found" in line and "device" in line.lower(): # Device opened successfully — no need to wait longer device_found = True break if proc.poll() is not None: break # Process exited if not device_found and not error_found and proc.poll() is not None and proc.returncode != 0: # rtl_test exited with error and we never saw a success message error_found = True finally: with contextlib.suppress(OSError): proc.kill() proc.wait() if device_found: # Allow the kernel to fully release the USB interface # before the caller opens the device with dump1090/rtl_fm/etc. time.sleep(0.5) if error_found: logger.warning(f"RTL-SDR device {device_index} USB probe failed: device busy or unavailable") return ( f"SDR device {device_index} is not available — " f"check that the SDR device is connected and not in use by another process." ) except Exception as e: logger.debug(f"RTL-SDR probe error for device {device_index}: {e}") return None def detect_all_devices(force: bool = False) -> list[SDRDevice]: """ Detect all connected SDR devices across all supported hardware types. Results are cached for a few seconds so that multiple callers hitting this within the same page-load cycle (e.g. /devices + /adsb/tools) do not each spawn a full set of blocking subprocess probes. Args: force: Bypass the cache and re-probe hardware. Returns a unified list of SDRDevice objects sorted by type and index. """ global _all_devices_cache, _all_devices_cache_ts now = time.time() if not force and _all_devices_cache_ts and (now - _all_devices_cache_ts) < _ALL_DEVICES_CACHE_TTL_SECONDS: logger.debug("Returning cached device list (%d device(s))", len(_all_devices_cache)) return list(_all_devices_cache) devices: list[SDRDevice] = [] skip_in_soapy: set[SDRType] = set() # RTL-SDR via native tool (primary method) rtlsdr_devices = detect_rtlsdr_devices() devices.extend(rtlsdr_devices) if rtlsdr_devices: skip_in_soapy.add(SDRType.RTL_SDR) # Native HackRF detection (primary method) hackrf_devices = detect_hackrf_devices() devices.extend(hackrf_devices) if hackrf_devices: skip_in_soapy.add(SDRType.HACKRF) # SoapySDR devices (LimeSDR, Airspy, and fallback for HackRF/RTL-SDR if native failed) soapy_devices = detect_soapy_devices(skip_types=skip_in_soapy) devices.extend(soapy_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})") # Update cache _all_devices_cache = list(devices) _all_devices_cache_ts = time.time() return devices def get_cached_devices() -> list[SDRDevice] | None: """Return the cached device list without probing hardware. Returns None if no cached data is available (never probed). """ if _all_devices_cache_ts == 0.0: return None return list(_all_devices_cache) def invalidate_device_cache() -> None: """Clear the all-devices cache so the next call re-probes hardware.""" global _all_devices_cache, _all_devices_cache_ts _all_devices_cache = [] _all_devices_cache_ts = 0.0