Add USB-level device probe to prevent cryptic rtl_fm crashes

When an external process (or stale handle from a crash) holds an SDR
device, claim_sdr_device() registry check passes but rtl_fm fails with
usb_claim_interface error -6. This adds a quick rtl_test probe inside
claim_sdr_device() so all modes get a clear error message before the
decoder pipeline is launched.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -244,6 +244,10 @@ sdr_device_registry_lock = threading.Lock()
 def claim_sdr_device(device_index: int, mode_name: str) -> str | None:
     """Claim an SDR device for a mode.
 
+    Checks the in-app registry first, then probes the USB device to
+    catch stale handles held by external processes (e.g. a leftover
+    rtl_fm from a previous crash).
+
     Args:
         device_index: The SDR device index to claim
         mode_name: Name of the mode claiming the device (e.g., 'sensor', 'rtlamr')
@@ -255,6 +259,16 @@ def claim_sdr_device(device_index: int, mode_name: str) -> str | None:
         if device_index in sdr_device_registry:
             in_use_by = sdr_device_registry[device_index]
             return f'SDR device {device_index} is in use by {in_use_by}. Stop {in_use_by} first or use a different device.'
+
+        # Probe the USB device to catch external processes holding the handle
+        try:
+            from utils.sdr.detection import probe_rtlsdr_device
+            usb_error = probe_rtlsdr_device(device_index)
+            if usb_error:
+                return usb_error
+        except Exception:
+            pass  # If probe fails, let the caller proceed normally
+
         sdr_device_registry[device_index] = mode_name
         return None
 

utils/sdr/__init__.py

@@ -26,7 +26,7 @@ from __future__ import annotations
 from typing import Optional
 
 from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
-from .detection import detect_all_devices
+from .detection import detect_all_devices, probe_rtlsdr_device
 from .rtlsdr import RTLSDRCommandBuilder
 from .limesdr import LimeSDRCommandBuilder
 from .hackrf import HackRFCommandBuilder
@@ -229,4 +229,6 @@ __all__ = [
     'validate_device_index',
     'validate_squelch',
     'get_capabilities_for_type',
+    # Device probing
+    'probe_rtlsdr_device',
 ]

utils/sdr/detection.py

@@ -348,6 +348,68 @@ def detect_hackrf_devices() -> list[SDRDevice]:
     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.
+    """
+    if not _check_tool('rtl_test'):
+        # 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
+            )
+
+        result = subprocess.run(
+            ['rtl_test', '-d', str(device_index), '-t'],
+            capture_output=True,
+            text=True,
+            timeout=3,
+            env=env,
+        )
+        output = result.stderr + result.stdout
+
+        if 'usb_claim_interface' in output or 'Failed to open' in output:
+            logger.warning(
+                f"RTL-SDR device {device_index} USB probe failed: "
+                f"device busy or unavailable"
+            )
+            return (
+                f'SDR device {device_index} is busy at the USB level — '
+                f'another process outside INTERCEPT may be using it. '
+                f'Check for stale rtl_fm/rtl_433/dump1090 processes, '
+                f'or try a different device.'
+            )
+
+    except subprocess.TimeoutExpired:
+        # rtl_test opened the device successfully and is running the
+        # test — that means the device *is* available.
+        pass
+    except Exception as e:
+        logger.debug(f"RTL-SDR probe error for device {device_index}: {e}")
+
+    return None
+
+
 def detect_all_devices() -> list[SDRDevice]:
     """
     Detect all connected SDR devices across all supported hardware types.