296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""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()
|