From f75d3fb558f0b8b43439a4f97d419e38b65492d3 Mon Sep 17 00:00:00 2001 From: lda Date: Sun, 25 Oct 2026 04:59:07 +0700 Subject: [PATCH] Add in-app speaker discovery and connection guide --- README.md | 45 +++++++++++++++++--- controller/ble_probe.py | 13 +----- controller/discovery.py | 92 +++++++++++++++++++++++++++++++++++++++++ controller/gui.py | 88 +++++++++++++++++++++++++++++++++++---- docs/ARCHITECTURE.md | 6 +++ tests/test_discovery.py | 69 +++++++++++++++++++++++++++++++ tests/test_imports.py | 2 +- 7 files changed, 287 insertions(+), 28 deletions(-) create mode 100644 controller/discovery.py create mode 100644 tests/test_discovery.py diff --git a/README.md b/README.md index b228f4d..5dbb863 100644 --- a/README.md +++ b/README.md @@ -14,19 +14,52 @@ Use Windows Python for a Windows Bluetooth adapter. From WSL, invoke `uv.exe` so Bluetooth runs on Windows rather than inside WSL. ```powershell -uv run soundtower --address YOUR_SPEAKER_MAC +uv run soundtower ``` For WSL: ```bash -uv.exe run soundtower --address YOUR_SPEAKER_MAC +uv.exe run soundtower ``` -Replace `YOUR_SPEAKER_MAC` with the Bluetooth address of your speaker. Keep it -powered on, close the phone app, and click **Connect**. The address can also be -entered in the window. `uv run python -m controller` remains supported, as does -the earlier `--with-requirements requirements.txt` launch command. +1. Turn on Windows Bluetooth and power on the speaker. Close the phone's Sound + Tower app if it is holding the control connection. +2. Click **Scan**. The window stays responsive during the eight-second search. +3. Choose your speaker in the **Speaker** dropdown, then click **Connect**. + A single result is selected automatically when the field is empty; the app + never connects automatically. Addresses distinguish speakers with the same name. +4. Wait for **Speaker handshake complete**, then use the controls. + +The MX-T40 can advertise without a name. The selector recognizes its known +model advertisement and shows **MX-T40 (model advertisement)** in that case. +Discovery of another Sound Tower model is not a compatibility guarantee. + +`uv run python -m controller` remains supported, as does the earlier +`--with-requirements requirements.txt` launch command. + +### Finding or entering the Bluetooth address manually + +The selector includes the Bluetooth address after each speaker name. You can +also list candidate names and addresses from the terminal without opening Tk +or connecting to a speaker: + +```powershell +uv run python -m controller.discovery +``` + +An address looks like `12:34:56:78:9A:BC`. Copy your speaker's actual address, +not that example, and either type it in the editable **Speaker** field or use: + +```powershell +uv run soundtower --address YOUR_SPEAKER_MAC +``` + +Use the address observed in the BLE scan rather than assuming an audio-device +display name is an address. If no speaker appears, ensure Windows Bluetooth +and the speaker are on, close the phone app, and scan again. An already open +phone connection can affect advertising. Scanning lists advertisements only; +it does not send control commands or pair/remove devices. Python 3.11 or newer is required; physical device testing used Windows Python 3.14. The interpreter needs Tcl/Tk. If the interpreter provided by uv lacks Tk, diff --git a/controller/ble_probe.py b/controller/ble_probe.py index 987cda0..2fdd01f 100644 --- a/controller/ble_probe.py +++ b/controller/ble_probe.py @@ -3,18 +3,7 @@ import argparse import asyncio -# MX-T40 model advertisements from the Android APK's strings.xml. These -# identify discovery records; they are not assumed to be writable services. -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 "T40" in names or bool( - MODEL_UUIDS.intersection(value.lower() for value in advertisement.service_uuids) - ) +from controller.discovery import is_candidate async def inspect(timeout): diff --git a/controller/discovery.py b/controller/discovery.py new file mode 100644 index 0000000..f4d82bb --- /dev/null +++ b/controller/discovery.py @@ -0,0 +1,92 @@ +"""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() diff --git a/controller/gui.py b/controller/gui.py index 7e64751..c1bbbd1 100644 --- a/controller/gui.py +++ b/controller/gui.py @@ -6,6 +6,7 @@ from tkinter import colorchooser, ttk from controller import commands from controller.ble_link import BLELink +from controller.discovery import ScanWorker, selected_address from controller.link import Link from controller.messages import ActivityLog, describe_reply from controller.protocol import EQ, LIGHTS @@ -21,8 +22,10 @@ class App: self.prepare = prepare self.transport = transport self.activity = ActivityLog() + self.discovered = [] + self.scanning = False root.title("MX-T40 · Sound Tower") - root.minsize(480, 640) + root.minsize(520, 640) root.protocol("WM_DELETE_WINDOW", self.close) style = ttk.Style() if "vista" in style.theme_names(): @@ -36,15 +39,27 @@ class App: ) connection = ttk.Frame(outer) connection.pack(fill="x") - ttk.Label(connection, text="Address / port").pack(side="left") - self.port = ttk.Combobox(connection, width=20) + connection.columnconfigure(0, weight=1) + ttk.Label(connection, text="Speaker" if transport == "ble" else "Address / port").grid( + row=0, column=0, sticky="w" + ) + self.port = ttk.Combobox(connection, width=30) self.port.set(address) - self.port.pack(side="left", padx=8) - ttk.Button(connection, text="Refresh", command=self.refresh).pack(side="left") + self.port.grid(row=1, column=0, sticky="ew", padx=(0, 8)) + self.scan_button = ttk.Button( + connection, text="Scan" if transport == "ble" else "Refresh", command=self.refresh + ) + self.scan_button.grid(row=1, column=1, padx=(0, 8)) self.connect = ttk.Button(connection, text="Connect", command=self.toggle) - self.connect.pack(side="right") + self.connect.grid(row=1, column=2) + if transport == "ble": + ttk.Label( + connection, + text="Turn the speaker on → Scan → choose it → Connect. Or enter its Bluetooth address.", + wraplength=440, + ).grid(row=2, column=0, columnspan=3, sticky="w", pady=(8, 0)) self.status = tk.StringVar( - value="Enter the speaker’s Bluetooth address. Using Samsung BLE control." + value="Click Scan to find nearby speakers. Close the phone app if the speaker does not appear." if transport == "ble" else f"Enter the speaker’s Bluetooth address. Direct RFCOMM channel: {channel}." ) @@ -94,7 +109,8 @@ class App: ) self.log.pack(fill="both", expand=True, pady=(12, 0)) self.enable(False) - self.refresh() + if transport != "ble": + self.refresh() root.after(60, self.poll) def selection(self, label, values, build): @@ -120,6 +136,16 @@ class App: visit(self.controls) def refresh(self): + if self.link or self.scanning: + return + if self.transport == "ble": + self.scanning = True + self.scan_button.configure(text="Scanning…", state="disabled") + self.connect.configure(state="disabled") + self.port.configure(state="disabled") + self.status.set("Searching for nearby speakers…") + ScanWorker(self.events).start() + return try: from serial.tools import list_ports @@ -139,8 +165,13 @@ class App: self.status.set("Disconnecting…") return port = self.port.get().strip() + port = selected_address(port, self.discovered) if not port: - self.status.set("Enter the MX-T40 Bluetooth address, or select its outgoing COM port.") + self.status.set( + "Click Scan and choose your speaker, or enter its Bluetooth address." + if self.transport == "ble" + else "Enter the Bluetooth address or outgoing COM port." + ) return if self.transport == "ble": self.link = BLELink(port, self.events) @@ -148,6 +179,7 @@ class App: self.link = Link(port, self.events, self.channel, self.prepare) self.activity.pending.clear() self.port.configure(state="disabled") + self.scan_button.configure(state="disabled") self.connect.configure(text="Disconnect") self.status.set("Connecting to speaker…") self.link.start() @@ -179,6 +211,43 @@ class App: kind, value = self.events.get_nowait() if kind in ("status", "error"): self.status.set(value) + elif kind in ("scan_results", "scan_error"): + self.scanning = False + self.scan_button.configure(text="Scan", state="normal") + self.connect.configure(state="normal") + self.port.configure(state="normal") + if kind == "scan_error": + self.status.set( + f"Scan failed: {value}. Check Windows Bluetooth is on and try again." + ) + continue + previous = selected_address(self.port.get(), self.discovered) + self.discovered = value + self.port.configure(values=[speaker.label for speaker in value]) + matching = next( + ( + speaker + for speaker in value + if speaker.address.casefold() == previous.casefold() + ), + None, + ) + if matching: + self.port.set(matching.label) + elif len(value) == 1 and not previous: + self.port.set(value[0].label) + elif previous: + # Preserve a manual address or a prior device that is + # temporarily absent; never silently select another. + self.port.set(previous) + if value: + self.status.set( + f"Found {len(value)} speaker(s). Choose yours, then Connect." + ) + else: + self.status.set( + "No speaker found. Turn it on, close the phone app, and scan again. A manual address also works." + ) elif kind == "ready": self.ready = True self.enable(True) @@ -189,6 +258,7 @@ class App: self.enable(False) self.connect.configure(text="Connect", state="normal") self.port.configure(state="normal") + self.scan_button.configure(state="normal") if not self.status.get().startswith(("No ", "Malformed", "could", "Install")): self.status.set("Disconnected. " + self.status.get()) elif kind == "payload": diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 171bbee..a5026b9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -14,6 +14,7 @@ the package or running `--help` does not load Tk or Bleak. | `ble_link.py` | Bleak connection, notifications, characteristic writes | | `link.py`, `transport.py` | Experimental RFCOMM and serial adapters | | `gui.py` | Tk controls and event rendering | +| `discovery.py` | Read-only BLE discovery, unnamed model matching, and selectable labels | | `cli.py`, `__main__.py` | Installed command and backwards-compatible module entry | | `ble_probe.py`, `probe.py` | Endpoint diagnostics | | `windows_bluetooth.py` | Experimental RFCOMM initialization | @@ -41,6 +42,11 @@ async event loop lives entirely within the worker thread. The GUI submits payloads through a bounded command queue and consumes events through another queue; workers never call Tk methods. +Discovery uses a separate bounded-duration worker and the same event queue. +It returns plain speaker records without opening a GATT connection. Tk maps a +selected label to its address; manual addresses remain supported. Scanning is +disabled while a connection is active so scans do not compete with BLE traffic. + BLE notifications enter a bounded async queue and can contain partial or combined packets. The decoder retains framing state between notifications. Queue overflow terminates the connection rather than silently losing framing. diff --git a/tests/test_discovery.py b/tests/test_discovery.py new file mode 100644 index 0000000..60e3600 --- /dev/null +++ b/tests/test_discovery.py @@ -0,0 +1,69 @@ +"""Discovery fixtures include the owner's actual unnamed FE40 advertisement.""" + +import asyncio +import queue +import sys +import types +import unittest +from unittest.mock import AsyncMock, patch + +from controller.discovery import ScanWorker, Speaker, scan_speakers, selected_address + + +def advertisement(name, address, uuids=()): + return ( + types.SimpleNamespace(name=name, address=address), + types.SimpleNamespace(local_name=name, service_uuids=list(uuids)), + ) + + +class DiscoveryTests(unittest.TestCase): + def test_unnamed_model_advertisement_and_named_speaker_are_selectable(self): + found = { + "unnamed": advertisement( + None, "12:34:56:78:9A:BC", ["0000fe40-0000-1000-8000-00805f9b34fb"] + ), + "named": advertisement("[AV] MX-T40", "12:34:56:78:9A:BD"), + "unrelated": advertisement( + "Headphones", "12:34:56:78:9A:BE", ["0000ffa0-0000-1000-8000-00805f9b34fb"] + ), + } + discover = AsyncMock(return_value=found) + with patch.dict( + sys.modules, + {"bleak": types.SimpleNamespace(BleakScanner=types.SimpleNamespace(discover=discover))}, + ): + speakers = asyncio.run(scan_speakers(timeout=3)) + self.assertEqual(len(speakers), 2) + self.assertEqual( + {speaker.address for speaker in speakers}, {"12:34:56:78:9A:BC", "12:34:56:78:9A:BD"} + ) + unnamed = next(speaker for speaker in speakers if speaker.address.endswith("BC")) + self.assertEqual(unnamed.name, "MX-T40 (model advertisement)") + self.assertIn(unnamed.address, unnamed.label) + discover.assert_awaited_once_with(timeout=3, return_adv=True) + + def test_choice_resolves_to_address_and_preserves_manual_input(self): + speaker = Speaker("12:34:56:78:9A:BC", "MX-T40") + self.assertEqual(selected_address(speaker.label, [speaker]), speaker.address) + self.assertEqual(selected_address(" 12:34:56:78:9A:BD ", [speaker]), "12:34:56:78:9A:BD") + + def test_identical_names_remain_distinguishable(self): + speakers = [Speaker("12:34:56:78:9A:BC", "MX-T40"), Speaker("12:34:56:78:9A:BD", "MX-T40")] + self.assertNotEqual(speakers[0].label, speakers[1].label) + self.assertEqual(selected_address(speakers[1].label, speakers), speakers[1].address) + + def test_scan_error_is_delivered_to_gui_queue(self): + events = queue.Queue() + with patch( + "controller.discovery.scan_speakers", + AsyncMock(side_effect=RuntimeError("Bluetooth is off")), + ): + ScanWorker(events).run() + self.assertEqual(events.get_nowait(), ("scan_error", "Bluetooth is off")) + + def test_empty_scan_has_a_result_event(self): + events = queue.Queue() + with patch("controller.discovery.scan_speakers", AsyncMock(return_value=[])): + ScanWorker(events).run() + self.assertEqual(events.get_nowait(), ("scan_results", [])) diff --git a/tests/test_imports.py b/tests/test_imports.py index af7439c..50cc01a 100644 --- a/tests/test_imports.py +++ b/tests/test_imports.py @@ -9,7 +9,7 @@ class ImportTests(unittest.TestCase): def test_headless_import_and_help(self): script = """ import sys -from controller import protocol, session, messages, cli +from controller import protocol, session, messages, discovery, cli assert 'tkinter' not in sys.modules assert 'bleak' not in sys.modules try: