#!/usr/bin/env python3 """ Shelly TRV Schedule Manager — BLE GATT GUI Tool Connects to a Shelly TRV device over BLE to read/write schedule rules via the 'schedules' GATT characteristic (UUID b77abf7f-66b5-4899-821d-efb177843783). Binary protocol (matches sl_ble_event_handler.c): READ response: For each active rule, packed sequentially: uint8_t rule_id uint8_t enable (0 or 1) float target (IEEE 754 LE, temperature in °C) uint8_t timespec_len char[] timespec (cron expression, timespec_len bytes) Terminated by a single 0xFF byte. WRITE commands: 0x01 — Add rule: [0x01, enable, float target (4B LE), ts_len, timespec...] 0x02 — Update rule: [0x02, rule_id, enable, float target (4B LE), ts_len, timespec...] 0x03 — Remove rule: [0x03, rule_id] 0xFF — Remove all: [0xFF] Connection is opened per request and closed after each operation. """ import asyncio import math import struct import sys import tkinter as tk import tkinter.font as tkfont from tkinter import ttk, messagebox from typing import List, Optional try: from bleak import BleakClient, BleakError except ImportError: print("ERROR: 'bleak' package is required. Install with: pip install bleak") sys.exit(1) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- SCHEDULES_CHAR_UUID = "b77abf7f-66b5-4899-821d-efb177843783" BLE_CONNECT_TIMEOUT = 30 # seconds MAX_RULES = 10 # Write command bytes CMD_ADD = 0x01 CMD_UPDATE = 0x02 CMD_REMOVE = 0x03 CMD_REMOVE_ALL = 0xFF # --------------------------------------------------------------------------- # Data model # --------------------------------------------------------------------------- class ScheduleRule: """Represents a single schedule rule as stored on the device.""" def __init__(self, rule_id: int, enable: bool, target: float, timespec: str): self.rule_id = rule_id self.enable = enable self.target = target # °C self.timespec = timespec # cron expression def __repr__(self): state = "ON" if self.enable else "OFF" return f"Rule {self.rule_id}: {state} {self.target:.1f}°C '{self.timespec}'" # --------------------------------------------------------------------------- # Binary codec — mirrors firmware ble_read_schedules / ble_write_schedules # --------------------------------------------------------------------------- def decode_schedules(raw: bytes) -> List[ScheduleRule]: """Parse the binary blob returned by a GATT read on the schedules characteristic.""" rules: List[ScheduleRule] = [] off = 0 while off < len(raw): rule_id = raw[off]; off += 1 if rule_id == 0xFF: break # terminator if off + 5 > len(raw): break # truncated enable = bool(raw[off]); off += 1 target = struct.unpack_from(" len(raw): break # truncated timespec = raw[off:off + ts_len].decode("utf-8", errors="replace") off += ts_len rules.append(ScheduleRule(rule_id, enable, target, timespec)) return rules def encode_add(enable: bool, target: float, timespec: str) -> bytes: """Build the payload for CMD_ADD (0x01).""" ts_bytes = timespec.encode("utf-8") return struct.pack(" bytes: """Build the payload for CMD_UPDATE (0x02).""" ts_bytes = timespec.encode("utf-8") return struct.pack(" bytes: """Build the payload for CMD_REMOVE (0x03).""" return struct.pack(" bytes: """Build the payload for CMD_REMOVE_ALL (0xFF).""" return struct.pack(" List[ScheduleRule]: async with BleakClient(mac, timeout=BLE_CONNECT_TIMEOUT) as client: raw = await client.read_gatt_char(SCHEDULES_CHAR_UUID) return decode_schedules(raw) async def ble_write_schedules(mac: str, payload: bytes) -> None: async with BleakClient(mac, timeout=BLE_CONNECT_TIMEOUT) as client: await client.write_gatt_char(SCHEDULES_CHAR_UUID, payload, response=True) # --------------------------------------------------------------------------- # GUI # --------------------------------------------------------------------------- class ScheduleManagerApp(tk.Tk): """Main application window.""" def __init__(self): super().__init__() self.title("Shelly TRV Schedule Manager") self.resizable(True, True) self.minsize(780, 520) self._rules: List[ScheduleRule] = [] self._loop = asyncio.new_event_loop() self._configure_styles() self._build_ui() # ---- Style configuration ----------------------------------------------- def _configure_styles(self): style = ttk.Style(self) self._tree_font = tkfont.nametofont("TkDefaultFont").copy() self._tree_font.configure(size=11) self._heading_font = self._tree_font.copy() self._heading_font.configure(weight="bold") row_height = self._tree_font.metrics("linespace") + 8 style.configure("Schedule.Treeview", font=self._tree_font, rowheight=row_height) style.configure("Schedule.Treeview.Heading", font=self._heading_font) # ---- UI construction --------------------------------------------------- def _build_ui(self): # --- Top frame: MAC address & Read/Write buttons --- top = ttk.Frame(self, padding=8) top.pack(fill=tk.X) ttk.Label(top, text="MAC Address:").pack(side=tk.LEFT) self._mac_var = tk.StringVar(value="") mac_entry = ttk.Entry(top, textvariable=self._mac_var, width=22, font=("Courier", 12)) mac_entry.pack(side=tk.LEFT, padx=(4, 12)) ttk.Button(top, text="Read Schedules", command=self._on_read).pack(side=tk.LEFT, padx=2) ttk.Button(top, text="Remove All", command=self._on_remove_all).pack(side=tk.RIGHT, padx=2) # --- Status bar --- self._status_var = tk.StringVar(value="Enter MAC address and press Read Schedules.") status_bar = ttk.Label(self, textvariable=self._status_var, relief=tk.SUNKEN, anchor=tk.W, padding=(6, 3)) status_bar.pack(fill=tk.X, side=tk.BOTTOM) # --- Main paned area: rule list (left) + editor (right) --- pane = ttk.PanedWindow(self, orient=tk.HORIZONTAL) pane.pack(fill=tk.BOTH, expand=True, padx=6, pady=4) # -- Left: rule list -- left = ttk.LabelFrame(pane, text="Rules on Device", padding=6) pane.add(left, weight=1) columns = ("id", "en", "target", "timespec") self._tree = ttk.Treeview(left, columns=columns, show="headings", selectmode="browse", height=12, style="Schedule.Treeview") # Column definitions: (key, heading_text, anchor, stretch) col_defs = [ ("id", "ID", tk.CENTER, False), ("en", "Enabled", tk.CENTER, False), ("target", "Target °C", tk.CENTER, False), ("timespec", "Timespec (cron)", tk.W, True), ] hfont = self._heading_font pad = 24 # extra pixels for sort arrow / internal padding for key, text, anchor, stretch in col_defs: self._tree.heading(key, text=text) min_w = hfont.measure(text) + pad self._tree.column(key, width=min_w, minwidth=min_w, anchor=anchor, stretch=stretch) scroll = ttk.Scrollbar(left, orient=tk.VERTICAL, command=self._tree.yview) self._tree.configure(yscrollcommand=scroll.set) self._tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scroll.pack(side=tk.RIGHT, fill=tk.Y) self._tree.bind("<>", self._on_select) btn_row = ttk.Frame(left) btn_row.pack(fill=tk.X, pady=(4, 0)) ttk.Button(btn_row, text="Remove Selected", command=self._on_remove).pack(side=tk.LEFT, padx=2) ttk.Button(btn_row, text="Refresh", command=self._on_read).pack(side=tk.RIGHT, padx=2) # -- Right: editor -- right = ttk.LabelFrame(pane, text="Add / Update Rule", padding=10) pane.add(right, weight=1) row = 0 ttk.Label(right, text="Rule ID (for update):").grid(row=row, column=0, sticky=tk.W, pady=3) self._id_var = tk.StringVar(value="") ttk.Entry(right, textvariable=self._id_var, width=6).grid(row=row, column=1, sticky=tk.W, pady=3) ttk.Label(right, text="(leave empty to add new)", font=("TkDefaultFont", 9, "italic") ).grid(row=row, column=2, sticky=tk.W, padx=6) row += 1 ttk.Label(right, text="Enabled:").grid(row=row, column=0, sticky=tk.W, pady=3) self._enable_var = tk.BooleanVar(value=True) ttk.Checkbutton(right, variable=self._enable_var).grid(row=row, column=1, sticky=tk.W, pady=3) row += 1 ttk.Label(right, text="Target (°C):").grid(row=row, column=0, sticky=tk.W, pady=3) self._target_var = tk.StringVar(value="21.0") ttk.Entry(right, textvariable=self._target_var, width=8).grid(row=row, column=1, sticky=tk.W, pady=3) row += 1 ttk.Label(right, text="Timespec (cron):").grid(row=row, column=0, sticky=tk.W, pady=3) self._ts_var = tk.StringVar(value="0 0 7 * * MON-FRI") ts_entry = ttk.Entry(right, textvariable=self._ts_var, width=30) ts_entry.grid(row=row, column=1, columnspan=2, sticky=tk.EW, pady=3) row += 1 ttk.Separator(right, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=3, sticky=tk.EW, pady=8) row += 1 btn_frame = ttk.Frame(right) btn_frame.grid(row=row, column=0, columnspan=3, sticky=tk.EW) ttk.Button(btn_frame, text="Add Rule", command=self._on_add).pack(side=tk.LEFT, padx=4) ttk.Button(btn_frame, text="Update Rule", command=self._on_update).pack(side=tk.LEFT, padx=4) row += 1 ttk.Separator(right, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=3, sticky=tk.EW, pady=8) # --- Cron help --- row += 1 help_text = ( "Cron format: SEC MIN HOUR DOM MON DOW\n" "Examples:\n" " 0 0 7 * * MON-FRI — weekdays at 07:00\n" " 0 30 22 * * * — every day at 22:30\n" " 0 0 8 * * SAT,SUN — weekends at 08:00" ) ttk.Label(right, text=help_text, justify=tk.LEFT, font=("Courier", 9), foreground="gray").grid( row=row, column=0, columnspan=3, sticky=tk.W, pady=(4, 0)) right.columnconfigure(2, weight=1) # ---- helpers ----------------------------------------------------------- def _mac(self) -> Optional[str]: mac = self._mac_var.get().strip() if not mac: messagebox.showwarning("MAC required", "Please enter the device MAC address.") return None return mac def _set_status(self, msg: str, error: bool = False): prefix = "ERROR: " if error else "" self._status_var.set(prefix + msg) def _run_async(self, coro): """Run an async coroutine, blocking the GUI thread (simple approach).""" return self._loop.run_until_complete(coro) def _refresh_tree(self): self._tree.delete(*self._tree.get_children()) for r in self._rules: self._tree.insert("", tk.END, iid=str(r.rule_id), values=(r.rule_id, "Yes" if r.enable else "No", f"{r.target:.1f}", r.timespec)) # ---- event handlers ---------------------------------------------------- def _on_select(self, _event): sel = self._tree.selection() if not sel: return iid = sel[0] for r in self._rules: if str(r.rule_id) == iid: self._id_var.set(str(r.rule_id)) self._enable_var.set(r.enable) self._target_var.set(f"{r.target:.1f}") self._ts_var.set(r.timespec) break def _on_read(self): mac = self._mac() if not mac: return self._set_status(f"Connecting to {mac} …") self.update_idletasks() try: self._rules = self._run_async(ble_read_schedules(mac)) self._refresh_tree() n = len(self._rules) self._set_status(f"Read OK — {n} rule{'s' if n != 1 else ''} found.") except Exception as exc: self._rules = [] self._refresh_tree() self._set_status(str(exc), error=True) def _on_add(self): mac = self._mac() if not mac: return try: target = float(self._target_var.get()) except ValueError: messagebox.showerror("Invalid target", "Target must be a number (°C).") return ts = self._ts_var.get().strip() if not ts: messagebox.showerror("Empty timespec", "Timespec (cron expression) cannot be empty.") return enable = self._enable_var.get() payload = encode_add(enable, target, ts) self._set_status(f"Adding rule → {mac} …") self.update_idletasks() try: self._run_async(ble_write_schedules(mac, payload)) self._set_status("Add rule OK. Reading back…") self.update_idletasks() self._rules = self._run_async(ble_read_schedules(mac)) self._refresh_tree() self._set_status(f"Add rule OK — {len(self._rules)} rule(s) on device.") except Exception as exc: self._set_status(str(exc), error=True) def _on_update(self): mac = self._mac() if not mac: return rid_str = self._id_var.get().strip() if not rid_str: messagebox.showerror("Rule ID required", "Enter the Rule ID to update (or select a row).") return try: rule_id = int(rid_str) except ValueError: messagebox.showerror("Invalid Rule ID", "Rule ID must be an integer.") return try: target = float(self._target_var.get()) except ValueError: messagebox.showerror("Invalid target", "Target must be a number (°C).") return ts = self._ts_var.get().strip() if not ts: messagebox.showerror("Empty timespec", "Timespec (cron expression) cannot be empty.") return enable = self._enable_var.get() payload = encode_update(rule_id, enable, target, ts) self._set_status(f"Updating rule {rule_id} → {mac} …") self.update_idletasks() try: self._run_async(ble_write_schedules(mac, payload)) self._set_status("Update OK. Reading back…") self.update_idletasks() self._rules = self._run_async(ble_read_schedules(mac)) self._refresh_tree() self._set_status(f"Update rule {rule_id} OK — {len(self._rules)} rule(s) on device.") except Exception as exc: self._set_status(str(exc), error=True) def _on_remove(self): mac = self._mac() if not mac: return sel = self._tree.selection() if not sel: messagebox.showinfo("No selection", "Select a rule in the list to remove.") return rule_id = int(sel[0]) if not messagebox.askyesno("Confirm remove", f"Remove rule {rule_id} from {mac}?"): return payload = encode_remove(rule_id) self._set_status(f"Removing rule {rule_id} → {mac} …") self.update_idletasks() try: self._run_async(ble_write_schedules(mac, payload)) self._set_status("Remove OK. Reading back…") self.update_idletasks() self._rules = self._run_async(ble_read_schedules(mac)) self._refresh_tree() self._set_status(f"Remove rule {rule_id} OK — {len(self._rules)} rule(s) on device.") except Exception as exc: self._set_status(str(exc), error=True) def _on_remove_all(self): mac = self._mac() if not mac: return if not messagebox.askyesno("Confirm remove ALL", f"Remove ALL schedule rules from {mac}?\n\nThis cannot be undone."): return payload = encode_remove_all() self._set_status(f"Removing all rules → {mac} …") self.update_idletasks() try: self._run_async(ble_write_schedules(mac, payload)) self._set_status("Remove-all OK. Reading back…") self.update_idletasks() self._rules = self._run_async(ble_read_schedules(mac)) self._refresh_tree() self._set_status("Remove all OK — device has no schedule rules.") except Exception as exc: self._set_status(str(exc), error=True) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(): app = ScheduleManagerApp() app.mainloop() if __name__ == "__main__": main()