feat: Cache device names and listen for property updates

Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
This commit is contained in:
2025-12-31 00:05:56 +00:00
parent 61eb680695
commit 58397e6b4c

25
scan.py
View File

@@ -8,12 +8,15 @@ logging.basicConfig(stream=sys.stdout,
import asyncio import asyncio
from dbus_next.aio import MessageBus from dbus_next.aio import MessageBus
from dbus_next.constants import BusType from dbus_next.constants import BusType
from dbus_next import Message
BLUEZ_SERVICE = 'org.bluez' BLUEZ_SERVICE = 'org.bluez'
ADAPTER_IFACE = 'org.bluez.Adapter1' ADAPTER_IFACE = 'org.bluez.Adapter1'
DEVICE_IFACE = 'org.bluez.Device1' DEVICE_IFACE = 'org.bluez.Device1'
bus = None
seen_devices = set() seen_devices = set()
name_cache = {}
async def get_adapter(bus): async def get_adapter(bus):
"""Gets the first Bluetooth adapter found.""" """Gets the first Bluetooth adapter found."""
@@ -32,15 +35,25 @@ def on_interfaces_added(path, interfaces):
"""Callback for when a new D-Bus interface is added.""" """Callback for when a new D-Bus interface is added."""
if DEVICE_IFACE in interfaces: if DEVICE_IFACE in interfaces:
device_properties = interfaces[DEVICE_IFACE] device_properties = interfaces[DEVICE_IFACE]
address = device_properties.get('Address') address_variant = device_properties.get('Address')
alias = device_properties.get('Alias') if not address_variant:
if address and alias and alias.value: return
addr_str = address.value
addr_str = address_variant.value
alias_variant = device_properties.get('Alias')
alias = alias_variant.value if alias_variant else name_cache.get(addr_str)
if alias:
# Update cache if we found a new alias
if alias_variant and alias_variant.value:
name_cache[addr_str] = alias_variant.value
if addr_str not in seen_devices: if addr_str not in seen_devices:
seen_devices.add(addr_str) seen_devices.add(addr_str)
logging.info(f"Found: {alias.value} ({addr_str}) (new)") logging.info(f"Found: {alias} ({addr_str}) (new)")
else: else:
logging.info(f"Found: {alias.value} ({addr_str})") # Log repeat discoveries only if they have a name
logging.info(f"Found: {alias} ({addr_str})")
async def main(): async def main():
"""Main function to run the scanner.""" """Main function to run the scanner."""