"""Tk interface; Bluetooth workers and protocol code never import this module.""" import queue import tkinter as tk 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 class App: def __init__(self, root, address="", channel=1, prepare=False, transport="ble"): self.root = root self.link = None self.events = queue.Queue() self.ready = False self.channel = channel self.prepare = prepare self.transport = transport self.activity = ActivityLog() self.discovered = [] self.scanning = False root.title("MX-T40 · Sound Tower") root.minsize(520, 640) root.protocol("WM_DELETE_WINDOW", self.close) style = ttk.Style() if "vista" in style.theme_names(): style.theme_use("vista") style.configure("TButton", padding=(10, 6)) outer = ttk.Frame(root, padding=20) outer.pack(fill="both", expand=True) ttk.Label(outer, text="MX-T40", font=("Segoe UI", 22, "bold")).pack(anchor="w") ttk.Label(outer, text="Sound Tower controller", font=("Segoe UI", 11)).pack( anchor="w", pady=(0, 18) ) connection = ttk.Frame(outer) connection.pack(fill="x") 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.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.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="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}." ) ttk.Label(outer, textvariable=self.status, wraplength=440).pack(anchor="w", pady=(10, 16)) self.controls = ttk.Frame(outer) self.controls.pack(fill="x") self.selection("EQ preset", EQ, commands.set_eq) self.selection("Bass boost", {"On": True, "Off": False}, commands.set_bass) self.selection("Lighting mode", LIGHTS, commands.set_lighting) self.selection("Source", commands.SOURCES, commands.set_source) row = ttk.Frame(self.controls) row.pack(fill="x", pady=5) ttk.Label(row, text="Lighting color", width=17).pack(side="left") ttk.Button(row, text="Choose color…", command=self.color).pack(side="left") row = ttk.Frame(self.controls) row.pack(fill="x", pady=(12, 5)) for label, action in [ ("Volume −", "volume_down"), ("Volume +", "volume_up"), ("Mute", "mute"), ("Play / pause", "play_pause"), ]: ttk.Button( row, text=label, command=lambda a=action: self.send(commands.remote_key(a)) ).pack(side="left", padx=(0, 5)) ttk.Button( self.controls, text="Read EQ and lighting status", command=self.read_status ).pack(anchor="w", pady=8) ttk.Label(outer, text="Speaker status", font=("Segoe UI", 11, "bold")).pack( anchor="w", pady=(16, 4) ) self.report = tk.StringVar(value="Waiting for a connection.") ttk.Label(outer, textvariable=self.report, wraplength=440).pack(anchor="w") activity_header = ttk.Frame(outer) activity_header.pack(fill="x", pady=(12, 0)) ttk.Label(activity_header, text="Activity", font=("Segoe UI", 11, "bold")).pack(side="left") self.raw_packets = tk.BooleanVar(value=False) ttk.Checkbutton( activity_header, text="Show raw packets", variable=self.raw_packets, command=self.render_log, ).pack(side="right") self.log = tk.Text( outer, height=7, width=58, state="disabled", font=("Segoe UI", 10), wrap="word" ) self.log.pack(fill="both", expand=True, pady=(12, 0)) self.enable(False) if transport != "ble": self.refresh() root.after(60, self.poll) def selection(self, label, values, build): row = ttk.Frame(self.controls) row.pack(fill="x", pady=5) ttk.Label(row, text=label, width=17).pack(side="left") choice = ttk.Combobox(row, values=list(values), state="readonly", width=17) choice.set(next(iter(values))) choice.pack(side="left", padx=(0, 8)) ttk.Button(row, text="Apply", command=lambda: self.send(build(values[choice.get()]))).pack( side="right" ) def enable(self, enabled): def visit(widget): for child in widget.winfo_children(): if isinstance(child, ttk.Combobox): child.configure(state="readonly" if enabled else "disabled") elif isinstance(child, ttk.Button): child.configure(state="normal" if enabled else "disabled") visit(child) 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 ports = [p.device for p in list_ports.comports()] self.port.configure(values=ports) except ImportError: # Direct RFCOMM does not need pyserial. Missing it only removes # optional COM-port enumeration from the address dropdown. self.port.configure(values=[]) def toggle(self): if self.link: self.link.stop() self.connect.configure(state="disabled") self.enable(False) self.ready = False self.status.set("Disconnecting…") return port = self.port.get().strip() port = selected_address(port, self.discovered) if not 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) else: 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() def send(self, payload): if self.ready and self.link: try: self.link.commands.put_nowait(payload) self.status.set("Command queued. Speaker reports appear below when received.") except queue.Full: self.status.set("Command queue is full; wait a moment and try again.") def color(self): color, _ = colorchooser.askcolor(parent=self.root, title="Lighting color") if color: self.send(commands.set_rgb(*(round(v) for v in color))) # The owner observes the speaker switch to COLOR on RGB writes. # Ask for the actual mode rather than treating its ACK as proof. self.send(commands.lighting_status()) def read_status(self): for payload in commands.status_queries(): self.send(payload) def poll(self): log_changed = False try: while True: 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) self.status.set("Speaker handshake complete.") elif kind == "closed": self.link = None self.ready = False 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": self.report.set(describe_reply(value)) elif kind in ("log", "frame"): self.activity.record(kind, value) log_changed = True except queue.Empty: pass if log_changed: self.render_log() self.root.after(60, self.poll) def render_log(self): """Render the same bounded history in English or English plus raw bytes.""" self.log.configure( state="normal", font=("Consolas", 9) if self.raw_packets.get() else ("Segoe UI", 10) ) self.log.delete("1.0", "end") self.log.insert("end", self.activity.text(raw=self.raw_packets.get())) self.log.see("end") self.log.configure(state="disabled") def close(self): if self.link: self.link.stop() self.root.destroy() def launch(address="", channel=1, prepare=False, transport="ble"): """Create and run the Tk window on the calling (main) thread.""" root = tk.Tk() App(root, address, channel, prepare, transport) root.mainloop()