93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
"""Find Sound Tower advertisements without connecting or loading Tk.
|
|
|
|
Some MX-T40 advertisements have no device name. Model UUIDs are therefore
|
|
important: the observed FE40 advertisement must still produce a selectable
|
|
speaker. These UUIDs identify discovery records, not verified control support.
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import threading
|
|
from dataclasses import dataclass
|
|
|
|
MODEL_UUIDS = {
|
|
f"0000{short}-0000-1000-8000-00805f9b34fb" for short in ("ff40", "fe40", "ff41", "fe41")
|
|
}
|
|
|
|
|
|
def is_candidate(device, advertisement):
|
|
names = f"{device.name or ''} {advertisement.local_name or ''}".upper()
|
|
return any(name in names for name in ("T40", "MX-T", "MX-ST", "SOUND TOWER")) or bool(
|
|
MODEL_UUIDS.intersection(value.lower() for value in advertisement.service_uuids)
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Speaker:
|
|
address: str
|
|
name: str
|
|
|
|
@property
|
|
def label(self):
|
|
return f"{self.name} — {self.address}"
|
|
|
|
|
|
def selected_address(selection, speakers):
|
|
"""Resolve a discovered label while preserving manually entered addresses."""
|
|
selection = selection.strip()
|
|
return next((speaker.address for speaker in speakers if speaker.label == selection), selection)
|
|
|
|
|
|
async def scan_speakers(timeout=8):
|
|
from bleak import BleakScanner
|
|
|
|
found = await BleakScanner.discover(timeout=timeout, return_adv=True)
|
|
speakers = {}
|
|
for device, advertisement in found.values():
|
|
if not is_candidate(device, advertisement):
|
|
continue
|
|
name = advertisement.local_name or device.name
|
|
if not name:
|
|
name = "MX-T40 (model advertisement)"
|
|
# Keep addresses in the choice label to distinguish identical names;
|
|
# deduplicate case-insensitively without changing the backend's address.
|
|
speakers[device.address.casefold()] = Speaker(device.address, name)
|
|
return sorted(speakers.values(), key=lambda speaker: (speaker.name.casefold(), speaker.address))
|
|
|
|
|
|
class ScanWorker(threading.Thread):
|
|
"""Run the bounded BLE scan off Tk's thread; return results through its queue."""
|
|
|
|
def __init__(self, events, timeout=8):
|
|
super().__init__(daemon=True)
|
|
self.events = events
|
|
self.timeout = timeout
|
|
|
|
def run(self):
|
|
try:
|
|
self.events.put(("scan_results", asyncio.run(scan_speakers(self.timeout))))
|
|
except Exception as exc:
|
|
self.events.put(("scan_error", str(exc)))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="List nearby Sound Tower names and Bluetooth addresses"
|
|
)
|
|
parser.add_argument("--timeout", type=float, default=8)
|
|
args = parser.parse_args()
|
|
if args.timeout <= 0:
|
|
parser.error("timeout must be positive")
|
|
try:
|
|
speakers = asyncio.run(scan_speakers(args.timeout))
|
|
except Exception as exc:
|
|
parser.exit(1, f"Bluetooth scan failed: {exc}\n")
|
|
if not speakers:
|
|
print("No Sound Tower found. Turn it on, close the phone app, and scan again.")
|
|
for speaker in speakers:
|
|
print(speaker.label)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|