username/rf-mapper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: skip Bluetooth scanning on Termux/Android

Browse Source 
Bleak requires D-Bus which isn't available on Android. Detect Termux
environment and skip both Classic BT and BLE scanning gracefully.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
User
2026-02-01 13:52:23 +01:00
parent b0e6d1107c
commit 3ff43de5ea
1 changed files with 89 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
src/rf_mapper/scanner.py
View File

@@ -4,12 +4,19 @@ import subprocess
import re import re
import json import json
import asyncio import asyncio
import os
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from dataclasses import dataclass, asdict from dataclasses import dataclass, asdict
from bleak import BleakScanner from bleak import BleakScanner
def is_termux() -> bool:
"""Detect if running in Termux (Android)."""
return os.environ.get('TERMUX_VERSION') is not None or \
os.path.exists('/data/data/com.termux')
from .oui import OUILookup from .oui import OUILookup
from .bluetooth_class import BluetoothClassDecoder from .bluetooth_class import BluetoothClassDecoder
from .distance import estimate_distance, rssi_to_quality, rssi_bar from .distance import estimate_distance, rssi_to_quality, rssi_bar
@@ -184,91 +191,97 @@ class RFScanner:
""" """
devices = [] devices = []
# Classic Bluetooth scan # Classic Bluetooth scan (not supported on Termux/Android)
try: if is_termux():
print(f"Scanning Classic Bluetooth ({timeout} seconds)...") print("Skipping Classic BT scan (not supported on Termux)")
result = subprocess.run( else:
['sudo', 'hcitool', 'inq', '--flush'], try:
capture_output=True, print(f"Scanning Classic Bluetooth ({timeout} seconds)...")
text=True, result = subprocess.run(
timeout=timeout + 10 ['sudo', 'hcitool', 'inq', '--flush'],
) capture_output=True,
text=True,
for line in result.stdout.split('\n'): timeout=timeout + 10
match = re.match(
r'\s*([0-9A-Fa-f:]+)\s+clock offset:\s*\S+\s+class:\s*(\S+)',
line
)
if match:
addr = match.group(1)
device_class = match.group(2)
name = self._get_bt_name(addr)
rssi = self._get_bt_rssi(addr)
dev_type, dev_subtype = self.bt_decoder.decode(device_class)
devices.append(BluetoothDevice(
address=addr,
name=name,
rssi=rssi,
device_class=device_class,
device_type=f"{dev_type}" + (f" ({dev_subtype})" if dev_subtype else ""),
manufacturer=self.oui_lookup.lookup(addr)
))
except Exception as e:
print(f"Classic BT scan error: {e}")
# BLE scan using bleak
try:
print(f"Scanning BLE devices ({timeout} seconds)...")
async def _ble_scan():
return await BleakScanner.discover(
timeout=timeout,
return_adv=True
) )
ble_results = asyncio.run(_ble_scan()) for line in result.stdout.split('\n'):
match = re.match(
r'\s*([0-9A-Fa-f:]+)\s+clock offset:\s*\S+\s+class:\s*(\S+)',
line
)
if match:
addr = match.group(1)
device_class = match.group(2)
name = self._get_bt_name(addr)
rssi = self._get_bt_rssi(addr)
dev_type, dev_subtype = self.bt_decoder.decode(device_class)
seen_addrs = {d.address for d in devices} devices.append(BluetoothDevice(
for device, adv_data in ble_results.values(): address=addr,
addr = device.address name=name,
if addr not in seen_addrs and addr != 'LE': rssi=rssi,
seen_addrs.add(addr) device_class=device_class,
device_type=f"{dev_type}" + (f" ({dev_subtype})" if dev_subtype else ""),
manufacturer=self.oui_lookup.lookup(addr)
))
except Exception as e:
print(f"Classic BT scan error: {e}")
name = device.name or adv_data.local_name or '<unknown>' # BLE scan using bleak (not supported on Termux/Android)
rssi = adv_data.rssi # Real RSSI from advertisement if is_termux():
manufacturer = self.oui_lookup.lookup(addr) print("Skipping BLE scan (not supported on Termux)")
else:
try:
print(f"Scanning BLE devices ({timeout} seconds)...")
# Extract manufacturer data if available async def _ble_scan():
if adv_data.manufacturer_data and not manufacturer: return await BleakScanner.discover(
# First 2 bytes are company ID timeout=timeout,
for company_id in adv_data.manufacturer_data.keys(): return_adv=True
manufacturer = f"Company ID: {company_id}" )
break
# Infer device type ble_results = asyncio.run(_ble_scan())
inferred_type = infer_device_type_from_name(name)
if not inferred_type:
inferred_type = infer_device_type_from_manufacturer(manufacturer)
if not inferred_type: seen_addrs = {d.address for d in devices}
if is_random_mac(addr): for device, adv_data in ble_results.values():
device_type = "BLE Device (Random MAC)" addr = device.address
if addr not in seen_addrs and addr != 'LE':
seen_addrs.add(addr)
name = device.name or adv_data.local_name or '<unknown>'
rssi = adv_data.rssi # Real RSSI from advertisement
manufacturer = self.oui_lookup.lookup(addr)
# Extract manufacturer data if available
if adv_data.manufacturer_data and not manufacturer:
# First 2 bytes are company ID
for company_id in adv_data.manufacturer_data.keys():
manufacturer = f"Company ID: {company_id}"
break
# Infer device type
inferred_type = infer_device_type_from_name(name)
if not inferred_type:
inferred_type = infer_device_type_from_manufacturer(manufacturer)
if not inferred_type:
if is_random_mac(addr):
device_type = "BLE Device (Random MAC)"
else:
device_type = "Low Energy Device"
else: else:
device_type = "Low Energy Device" device_type = inferred_type
else:
device_type = inferred_type
devices.append(BluetoothDevice( devices.append(BluetoothDevice(
address=addr, address=addr,
name=name, name=name,
rssi=rssi, # Real RSSI instead of -70 rssi=rssi, # Real RSSI instead of -70
device_class="BLE", device_class="BLE",
device_type=device_type, device_type=device_type,
manufacturer=manufacturer manufacturer=manufacturer
)) ))
except Exception as e: except Exception as e:
print(f"BLE scan error: {e}") print(f"BLE scan error: {e}")
# Auto-identify unknown devices # Auto-identify unknown devices
if auto_identify and devices: if auto_identify and devices: