import tkinter as tk import customtkinter as ctk import threading import queue import time import logging from datetime import datetime from broadlink_core import (create_device, learn_code, load_codes, save_codes, send_command, discover_devices, DEFAULT_CONFIG) ctk.set_appearance_mode("System") ctk.set_default_color_theme("blue") class BroadlinkApp(ctk.CTk): def __init__(self): super().__init__() self.title("Broadlink Manager Pro") self.geometry("880x680") self.minsize(750, 550) self.queue = queue.Queue() self.device = None self.codes = load_codes() self.setup_ui() self.bind_shortcuts() self.log("🚀 Приложение запущено. Загружено команд: " + str(len(self.codes))) self.after(100, self.process_queue) def setup_ui(self): # Настройки устройства cfg = ctk.CTkFrame(self) cfg.pack(fill="x", padx=10, pady=10) ctk.CTkLabel(cfg, text="🌐 IP:").pack(side="left", padx=(10,0)) self.ip_var = tk.StringVar(value=DEFAULT_CONFIG["ip"]) ctk.CTkEntry(cfg, textvariable=self.ip_var, width=130).pack(side="left", padx=5) ctk.CTkLabel(cfg, text="🔗 MAC:").pack(side="left", padx=5) self.mac_var = tk.StringVar(value=DEFAULT_CONFIG["mac"]) ctk.CTkEntry(cfg, textvariable=self.mac_var, width=140).pack(side="left", padx=5) ctk.CTkButton(cfg, text="🔌 Подключить", command=self.connect_device).pack(side="right", padx=10) # Управление ctrl = ctk.CTkFrame(self) ctrl.pack(fill="x", padx=10, pady=10) ctk.CTkLabel(ctrl, text="📜 Команда:").pack(side="left", padx=(10,5)) self.cmd_var = tk.StringVar() self.cmd_combo = ctk.CTkComboBox(ctrl, variable=self.cmd_var, values=sorted(self.codes.keys()), width=200) self.cmd_combo.pack(side="left", padx=5) if self.codes: self.cmd_combo.set(next(iter(sorted(self.codes.keys())))) ctk.CTkButton(ctrl, text="📡 Обучение", command=self.learn_selected).pack(side="left", padx=5) ctk.CTkButton(ctrl, text="🚀 Отправить", command=self.send_selected).pack(side="left", padx=5) ctk.CTkButton(ctrl, text="⏯ Все подряд", command=self.send_all).pack(side="left", padx=5) ctk.CTkButton(ctrl, text="🔍 Поиск", command=self.discover).pack(side="left", padx=5) ctk.CTkButton(ctrl, text="🔄 Обновить", command=self.refresh_codes).pack(side="left", padx=5) # Лог self.log_frame = ctk.CTkTextbox(self, height=220) self.log_frame.pack(fill="both", expand=True, padx=10, pady=10) self.status_bar = ctk.CTkLabel(self, text="Готово", anchor="w") self.status_bar.pack(fill="x", padx=10, pady=(0,5)) def bind_shortcuts(self): self.bind("", lambda e: self.discover()) self.bind("", lambda e: self.learn_selected()) self.bind("", lambda e: self.send_selected()) self.bind("", lambda e: self.send_all()) self.bind("", lambda e: self.refresh_codes()) self.bind("", lambda e: self.quit()) def log(self, msg: str): self.queue.put(("log", f"[{datetime.now().strftime('%H:%M:%S')}] {msg}\n")) def process_queue(self): try: while True: action, data = self.queue.get_nowait() if action == "log": self.log_frame.insert("end", data) self.log_frame.see("end") elif action == "status": self.status_bar.configure(text=data) elif action == "update_combo": self.cmd_combo.configure(values=sorted(data.keys())) if data: self.cmd_combo.set(next(iter(sorted(data.keys())))) except queue.Empty: pass self.after(100, self.process_queue) def get_dev(self): if not self.device: try: self.device = create_device(self.ip_var.get(), self.mac_var.get()) except Exception as e: self.queue.put(("status", f"❌ {e}")) return None return self.device def connect_device(self): threading.Thread(target=self._connect_worker, daemon=True).start() def _connect_worker(self): self.queue.put(("status", "🔌 Подключение...")) self.device = create_device(self.ip_var.get(), self.mac_var.get()) self.queue.put(("status", "✅ Подключено" if self.device else "❌ Ошибка подключения")) def learn_selected(self): threading.Thread(target=self._learn_worker, daemon=True).start() def _learn_worker(self): dev = self.get_dev() if not dev: return name = self.cmd_var.get().strip() self.queue.put(("status", f"📡 Обучение '{name}'... Нажмите кнопку на пульте")) code = learn_code(dev, name) if code: self.codes[name] = code save_codes(self.codes) self.queue.put(("update_combo", self.codes)) self.queue.put(("status", f"✅ '{name}' сохранено")) else: self.queue.put(("status", "❌ Обучение отменено")) def send_selected(self): threading.Thread(target=self._send_worker, daemon=True).start() def _send_worker(self): dev = self.get_dev() if not dev: return name = self.cmd_var.get().strip() self.queue.put(("status", f"🚀 Отправка '{name}'...")) if send_command(dev, name, self.codes): self.queue.put(("status", "✅ Отправлено")) else: self.queue.put(("status", "❌ Ошибка отправки")) def send_all(self): threading.Thread(target=self._send_all_worker, daemon=True).start() def _send_all_worker(self): dev = self.get_dev() if not dev: return self.queue.put(("status", "⏯ Отправка всех команд...")) for name in sorted(self.codes.keys()): self.queue.put(("status", f"📤 {name}")) send_command(dev, name, self.codes) time.sleep(1.5) self.queue.put(("status", "🏁 Все команды отправлены")) def refresh_codes(self): self.codes = load_codes() self.queue.put(("update_combo", self.codes)) self.queue.put(("log", "📜 Список команд обновлён\n")) def discover(self): threading.Thread(target=self._discover_worker, daemon=True).start() def _discover_worker(self): self.queue.put(("status", "🔍 Поиск устройств...")) devs = discover_devices() for d in devs: self.queue.put(("log", f"Найдено: {d['host']} | MAC: {d['mac']} | Тип: 0x{d['type']:04x}\n")) self.queue.put(("status", f"🔍 Найдено: {len(devs)}"))