Merge pull request #1026 from cartpauj/feat/faster-balloon-detection-with-ka9q

feat: faster balloon detection with ka9q using async

Author: darksidelemm

auto_rx/auto_rx.py

@@ -27,7 +27,7 @@
 import traceback
 import os
 from dateutil.parser import parse
-from queue import Queue
+from queue import Queue, Empty
 
 if sys.version_info < (3, 6):
     print("CRITICAL - radiosonde_auto_rx requires Python 3.6 or newer!")
@@ -183,6 +183,7 @@ def start_scanner():
             wideband_sondes=config["wideband_sondes"],
             temporary_block_list=temporary_block_list,
             temporary_block_time=config["temporary_block_time"],
+            max_async_scan_workers=config["max_async_scan_workers"],
         )
 
         # Add a reference into the sdr_list entry
@@ -1103,14 +1104,28 @@ def main():
     if args.type != None:
         handle_scan_results()
 
-    # Loop.
+    # Loop - Event-driven processing with periodic maintenance
     while True:
-        # Check for finished tasks.
-        clean_task_list()
-        # Handle any new scan results.
+        # Wait for new scan results or timeout after 2 seconds
+        # This allows immediate response when results arrive while still doing periodic cleanup
+        try:
+            # Block until result arrives or timeout
+            result = autorx.scan_results.get(timeout=2.0)
+            # Put it back for handle_scan_results() to process
+            # Note: This get/put pattern allows us to block-wait for queue activity while
+            # maintaining handle_scan_results()'s existing logic (which checks qsize and processes all items).
+            # Alternative approach would be to refactor handle_scan_results to accept items directly.
+            autorx.scan_results.put(result)
+        except Empty:
+            # Timeout - normal condition when no scan results for 2s
+            pass
+
+        # Process any scan results in the queue (including the one we just got, if any)
+        # handle_scan_results() checks qsize and processes all available results
         handle_scan_results()
-        # Sleep a little bit.
-        time.sleep(2)
+
+        # Check for finished tasks
+        clean_task_list()
 
         if len(autorx.sdr_list) == 0:
             # No Functioning SDRs!

auto_rx/autorx/config.py

@@ -139,6 +139,7 @@ def read_auto_rx_config(filename, no_sdr_test=False):
         "temporary_block_time": 60,
         "rs41_drift_tweak": False,
         "decoder_stats": False,
+        "max_async_scan_workers": 4,
         "ngp_tweak": False,
         # Rotator Settings
         "enable_rotator": False,
@@ -370,6 +371,16 @@ def read_auto_rx_config(filename, no_sdr_test=False):
             "advanced", "synchronous_upload"
         )
 
+        # Max async scan workers - validate and cap to reasonable limits
+        _max_workers = config.getint("advanced", "max_async_scan_workers")
+        if _max_workers < 1:
+            logging.warning(f"Config - max_async_scan_workers must be at least 1, setting to 1")
+            _max_workers = 1
+        elif _max_workers > 32:
+            logging.warning(f"Config - max_async_scan_workers capped at 32 (was {_max_workers})")
+            _max_workers = 32
+        auto_rx_config["max_async_scan_workers"] = _max_workers
+
         # Rotator Settings
         auto_rx_config["rotator_enabled"] = config.getboolean(
             "rotator", "rotator_enabled"

auto_rx/autorx/scan.py

@@ -25,6 +25,14 @@
 )
 from .sdr_wrappers import test_sdr, reset_sdr, get_sdr_name, get_sdr_iq_cmd, get_sdr_fm_cmd, get_power_spectrum, shutdown_sdr
 
+# Import async scanning for concurrent peak detection
+try:
+    from .scan_async import run_async_scan
+    ASYNC_SCAN_AVAILABLE = True
+except ImportError:
+    ASYNC_SCAN_AVAILABLE = False
+    logging.warning("Async scanning not available - falling back to sequential scanning")
+
 
 try:
     from .web import flask_emit_event
@@ -680,7 +688,8 @@ def __init__(
         temporary_block_list={},
         temporary_block_time=60,
         ngp_tweak=False,
-        wideband_sondes=False
+        wideband_sondes=False,
+        max_async_scan_workers=4
     ):
         """Initialise a Sonde Scanner Object.
 
@@ -770,6 +779,7 @@ def __init__(
         self.callback = callback
         self.save_detection_audio = save_detection_audio
         self.wideband_sondes = wideband_sondes
+        self.max_async_scan_workers = max_async_scan_workers
 
         # Temporary block list.
         self.temporary_block_list = temporary_block_list.copy()
@@ -1118,44 +1128,98 @@ def sonde_search(self, first_only=False):
             )
 
         # Run rs_detect on each peak frequency, to determine if there is a sonde there.
-        for freq in peak_frequencies:
+        # OPTIMIZATION: Use concurrent async scanning ONLY with KA9Q-radio
+        # KA9Q provides virtual SDR channels that can actually scan concurrently
+        if ASYNC_SCAN_AVAILABLE and self.sdr_type == "KA9Q" and len(peak_frequencies) > 1:
+            try:
+                import os
+                cpu_count = os.cpu_count() or 1
+
+                # With KA9Q, we can use multiple virtual channels concurrently
+                # The scanner creates temporary KA9Q channels that don't consume task slots
+                # Cap concurrency to min of: configured max, CPU cores, and number of peaks
+                max_concurrent = min(
+                    self.max_async_scan_workers,
+                    cpu_count,
+                    len(peak_frequencies)
+                )
 
-            _freq = float(freq)
+                self.log_info(f"KA9Q: Using concurrent peak detection with {max_concurrent} workers")
+
+                detections = run_async_scan(
+                    peak_frequencies=peak_frequencies,
+                    max_concurrent=max_concurrent,
+                    rs_path=self.rs_path,
+                    dwell_time=self.detect_dwell_time,
+                    sdr_type=self.sdr_type,
+                    sdr_hostname=self.sdr_hostname,
+                    sdr_port=self.sdr_port,
+                    ss_iq_path=self.ss_iq_path,
+                    rtl_fm_path=self.rtl_fm_path,
+                    rtl_device_idx=self.rtl_device_idx,
+                    ppm=self.ppm,
+                    gain=self.gain,
+                    bias=self.bias,
+                    save_detection_audio=self.save_detection_audio,
+                    wideband_sondes=self.wideband_sondes
+                )
 
-            # Exit opportunity.
-            if self.sonde_scanner_running == False:
-                return []
+                # Process results
+                for _freq, detected in detections:
+                    if self.sonde_scanner_running == False:
+                        return []
 
-            (detected, offset_est) = detect_sonde(
-                _freq,
-                sdr_type=self.sdr_type,
-                sdr_hostname=self.sdr_hostname,
-                sdr_port=self.sdr_port,
-                ss_iq_path = self.ss_iq_path,
-                rtl_fm_path=self.rtl_fm_path,
-                rtl_device_idx=self.rtl_device_idx,
-                ppm=self.ppm,
-                gain=self.gain,
-                bias=self.bias,
-                dwell_time=self.detect_dwell_time,
-                save_detection_audio=self.save_detection_audio,
-                wideband_sondes=self.wideband_sondes
-            )
+                    _search_results.append([_freq, detected])
+                    self.send_to_callback([[_freq, detected]])
+
+                    if first_only:
+                        return _search_results
+
+            except Exception as e:
+                import traceback
+                self.log_error(f"Async scanning failed: {e}, falling back to sequential")
+                self.log_debug(f"Async scan traceback: {traceback.format_exc()}")
+
+        # Standard sequential scanning (for RTLSDR, SpyServer, or single peaks)
+        else:
+            for freq in peak_frequencies:
+
+                _freq = float(freq)
+
+                # Exit opportunity.
+                if self.sonde_scanner_running == False:
+                    return []
+
+                (detected, offset_est) = detect_sonde(
+                    _freq,
+                    sdr_type=self.sdr_type,
+                    sdr_hostname=self.sdr_hostname,
+                    sdr_port=self.sdr_port,
+                    ss_iq_path = self.ss_iq_path,
+                    rtl_fm_path=self.rtl_fm_path,
+                    rtl_device_idx=self.rtl_device_idx,
+                    ppm=self.ppm,
+                    gain=self.gain,
+                    bias=self.bias,
+                    dwell_time=self.detect_dwell_time,
+                    save_detection_audio=self.save_detection_audio,
+                    wideband_sondes=self.wideband_sondes
+                )
 
-            if detected != None:
-                # Quantize the detected frequency (with offset) to 1 kHz
-                _freq = round((_freq + offset_est) / 1000.0) * 1000.0
+                if detected != None:
+                    # Quantize the detected frequency (with offset) to 1 kHz
+                    _freq = round((_freq + offset_est) / 1000.0) * 1000.0
 
-                # Add a detected sonde to the output array
-                _search_results.append([_freq, detected])
+                    # Add a detected sonde to the output array
+                    _search_results.append([_freq, detected])
 
-                # Immediately send this result to the callback.
-                self.send_to_callback([[_freq, detected]])
-                # If we only want the first detected sonde, then return now.
-                if first_only:
-                    return _search_results
+                    # Immediately send this result to the callback.
+                    self.send_to_callback([[_freq, detected]])
+                    # If we only want the first detected sonde, then return now.
+                    if first_only:
+                        return _search_results
 
-                # Otherwise, we continue....
+                    # Otherwise, we continue....
 
         if len(_search_results) == 0:
             self.log_debug("No sondes detected.")