Method _linux_interactive_scan

actions/ble.py:1285–1450

Perform an interactive Bluetooth scan on Linux that captures devices in real-time This is more reliable than relying on 'bluetoothctl devices' which may not update

(self, duration: int)

        return devices
    
    def _linux_interactive_scan(self, duration: int) -> Dict[str, Dict[str, Any]]:
        """
        Perform an interactive Bluetooth scan on Linux that captures devices in real-time
        This is more reliable than relying on 'bluetoothctl devices' which may not update
        """
        devices = {}
        
        try:
            import threading
            import queue
            
            # Create a queue to collect output
            output_queue = queue.Queue()
            
            # Start bluetoothctl process
            proc = subprocess.Popen(
                ['bluetoothctl'],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1
            )
            
            # Thread to read output
            def read_output():
                try:
                    for line in proc.stdout:
                        output_queue.put(line)
                except:
                    pass
            
            reader_thread = threading.Thread(target=read_output, daemon=True)
            reader_thread.start()
            
            # Send commands
            self.logger.info("Sending 'power on' to bluetoothctl")
            proc.stdin.write('power on\n')
            proc.stdin.flush()
            time.sleep(1)
            
            self.logger.info("Sending 'scan on' to bluetoothctl")
            proc.stdin.write('scan on\n')
            proc.stdin.flush()
            
            # Monitor output for device discoveries
            start_time = time.time()
            self.logger.info(f"Monitoring for {duration} seconds...")
            
            while time.time() - start_time < duration:
                try:
                    line = output_queue.get(timeout=1)
                    line = line.strip()
                    
                    # Log interesting lines
                    if any(x in line for x in ['Device', 'CHG', 'NEW']):
                        self.logger.debug(f"BT Output: {line}")