from PySide6.QtCore import QObject, Signal, Property, Slot, QTimer from PySide6.QtQml import QmlElement import asyncio import json import logging import subprocess import threading import time from pathlib import Path import os from firmware_client import FirmwareClient, LED_EFFECTS, LED_PALETTES import discovery # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%H:%M:%S' ) logger = logging.getLogger("DuneWeaver") QML_IMPORT_NAME = "DuneWeaver" QML_IMPORT_MAJOR_VERSION = 1 # Firmware clear-mode -> touch UI pre-execution option (reverse of CLEAR_MODE_MAP) _FW_TO_UI_CLEAR = { "in": "clear_center", "out": "clear_perimeter", "adaptive": "adaptive", "none": "none", "sideway": "adaptive", "random": "adaptive", } def _run(coro): """Schedule a coroutine on the running (qasync) event loop, if any.""" try: asyncio.get_event_loop().create_task(coro) except RuntimeError: logger.warning("No running event loop to schedule task") @QmlElement class Backend(QObject): """Backend controller: drives a headless FluidNC sand table over HTTP.""" # Constants SETTINGS_FILE = "touch_settings.json" DEFAULT_SCREEN_TIMEOUT = 300 # 5 minutes in seconds STATUS_POLL_MS = 1000 # /sand_status poll interval # Predefined timeout options (in seconds) TIMEOUT_OPTIONS = { "30 seconds": 30, "1 minute": 60, "5 minutes": 300, "10 minutes": 600, "Never": 0 # 0 means never timeout } # Predefined speed options (motor feed, mm/min) SPEED_OPTIONS = { "50": 50, "100": 100, "150": 150, "200": 200, "300": 300, "500": 500 } # Predefined pause between patterns options (in seconds) PAUSE_OPTIONS = { "0s": 0, "1 min": 60, "5 min": 300, "15 min": 900, "30 min": 1800, "1 hour": 3600, "2 hour": 7200, "3 hour": 10800, "4 hour": 14400, "5 hour": 18000, "6 hour": 21600, "12 hour": 43200 } # Signals statusChanged = Signal() progressChanged = Signal() connectionChanged = Signal() executionStarted = Signal(str, str) # patternName, patternPreview executionStopped = Signal() errorOccurred = Signal(str) serialPortsUpdated = Signal(list) # now: list of discovered table URLs serialConnectionChanged = Signal(bool) # now: table reachable currentPortChanged = Signal(str) # now: current table URL speedChanged = Signal(int) settingsLoaded = Signal() screenStateChanged = Signal(bool) screenTimeoutChanged = Signal(int) pauseBetweenPatternsChanged = Signal(int) pausedChanged = Signal(bool) playlistSettingsChanged = Signal() patternsRefreshCompleted = Signal(bool, str) # Playlist management signals playlistCreated = Signal(bool, str) playlistDeleted = Signal(bool, str) patternAddedToPlaylist = Signal(bool, str) playlistModified = Signal(bool, str) # Backend/table connection status signals backendConnectionChanged = Signal(bool) reconnectStatusChanged = Signal(str) # LED control signals ledStatusChanged = Signal() ledEffectsLoaded = Signal(list) ledPalettesLoaded = Signal(list) # LCD brightness signals lcdBrightnessChanged = Signal() def __init__(self): super().__init__() self.client = FirmwareClient.instance() # Status properties self._current_file = "" self._progress = 0 self._is_running = False self._is_paused = False self._is_connected = False self._serial_ports = [] self._serial_connected = False self._current_port = "" self._current_speed = 130 self._auto_play_on_boot = False self._pause_between_patterns = 10800 # Playlist settings (mirror the table's NVS; also persisted locally) self._playlist_shuffle = True self._playlist_run_mode = "loop" self._playlist_clear_pattern = "adaptive" # Connection status self._backend_connected = False self._reconnect_status = "Looking for table..." self._saved_table_url = "" # LCD brightness state self._lcd_brightness = 255 self._lcd_brightness_path = "" self._lcd_max_brightness = 0 # LED control state self._led_provider = "none" # "none" or "dw_leds" self._led_connected = False self._led_power_on = False self._led_brightness = 100 # 0..100 for QML (firmware is 0..255) self._led_effects = [] self._led_palettes = [] self._led_current_effect = 0 self._led_current_palette = 0 self._led_color = "#ffffff" self._led_last_effect = 2 # remembered on power-off (default rainbow) # Screen management self._screen_on = True self._screen_timeout = self.DEFAULT_SCREEN_TIMEOUT self._last_activity = time.time() self._screen_transition_lock = threading.Lock() self._last_screen_change = 0 self._screen_timer = QTimer() self._screen_timer.timeout.connect(self._check_screen_timeout) self._screen_timer.start(1000) # Load local settings first self._load_local_settings() self._detect_backlight() # Point the shared client at the saved / env-configured table. env_url = os.environ.get("DUNE_WEAVER_URL", "") initial_url = env_url or self._saved_table_url if initial_url: self.client.set_base_url(initial_url) self._current_port = self.client.base_url # Status polling replaces the old WebSocket status stream. self._status_timer = QTimer() self._status_timer.timeout.connect(self._tick_status) self._time_synced = False # Kick everything off once the event loop is running. QTimer.singleShot(200, self._start) @Slot() def _start(self): self._status_timer.start(self.STATUS_POLL_MS) if not self.client.base_url: # No configured table -> try to discover one automatically. _run(self._auto_discover()) async def _auto_discover(self): self._set_reconnect_status("Searching for tables (mDNS)...") try: tables = await discovery.discover_tables(timeout=3.0) except Exception as exc: logger.warning(f"Discovery failed: {exc}") tables = [] self._serial_ports = [t.base_url for t in tables] self.serialPortsUpdated.emit(self._serial_ports) if len(tables) == 1: logger.info(f"Auto-connecting to the only discovered table: {tables[0].base_url}") self._connect_to(tables[0].base_url) elif not tables: self._set_reconnect_status("No table found. Enter the table address.") else: self._set_reconnect_status(f"{len(tables)} tables found. Pick one.") # ==================== Status polling ==================== @Slot() def _tick_status(self): _run(self._poll_status()) async def _poll_status(self): if not self.client.base_url: return try: data = await self.client.status() except Exception as exc: self._on_unreachable(str(exc)) return self._on_status(data) def _on_status(self, status): # Reachability if not self._backend_connected: self._backend_connected = True self._is_connected = True self._serial_connected = True self._current_port = self.client.base_url self._reconnect_status = "Connected" self.connectionChanged.emit() self.backendConnectionChanged.emit(True) self.serialConnectionChanged.emit(True) self.currentPortChanged.emit(self._current_port) self.reconnectStatusChanged.emit("Connected") # Load settings + LED config on (re)connect. self.loadControlSettings() self.loadLedConfig() _run(self._sync_time_once()) state = status.get("state", "") # Current pattern / execution start detection raw_file = status.get("file", "") or "" new_file = raw_file for prefix in ("/sd/patterns/", "/patterns/", "/sd/", "/"): if new_file.startswith(prefix): new_file = new_file[len(prefix):] break if new_file and new_file != self._current_file: logger.info(f"Pattern changed to '{new_file}'") self.executionStarted.emit(new_file, self._preview_path(new_file)) self._current_file = new_file self._is_running = bool(status.get("running", False)) new_paused = (state == "Hold") if new_paused != self._is_paused: self._is_paused = new_paused self.pausedChanged.emit(new_paused) feed = status.get("feed") if feed and int(feed) != self._current_speed: self._current_speed = int(feed) self.speedChanged.emit(self._current_speed) prog = status.get("progress") if prog is not None: self._progress = 0 if prog < 0 else float(prog) * 100.0 # Live LED snapshot (effect/brightness) from status led = status.get("led") if isinstance(led, dict): self._ingest_led_status(led) self.statusChanged.emit() self.progressChanged.emit() def _on_unreachable(self, reason): if self._backend_connected or self._is_connected: logger.warning(f"Table unreachable: {reason}") self._backend_connected = False self._is_connected = False if self._serial_connected: self._serial_connected = False self.serialConnectionChanged.emit(False) self._reconnect_status = "Table connection lost, retrying..." self.connectionChanged.emit() self.backendConnectionChanged.emit(False) self.reconnectStatusChanged.emit(self._reconnect_status) def _set_reconnect_status(self, msg): self._reconnect_status = msg self.reconnectStatusChanged.emit(msg) async def _sync_time_once(self): if self._time_synced: return try: await self.client.sync_time(int(time.time())) self._time_synced = True except Exception as exc: logger.debug(f"time sync failed: {exc}") def _preview_path(self, rel_name): """Best-effort cached preview path for a pattern (may be empty).""" try: import thr_preview return thr_preview.cached_preview(self.client.base_url, rel_name) except Exception: return "" # ==================== Properties ==================== @Property(str, notify=statusChanged) def currentFile(self): return self._current_file @Property(float, notify=progressChanged) def progress(self): return self._progress @Property(bool, notify=statusChanged) def isRunning(self): return self._is_running @Property(bool, notify=pausedChanged) def isPaused(self): return self._is_paused @Property(bool, notify=connectionChanged) def isConnected(self): return self._is_connected @Property(list, notify=serialPortsUpdated) def serialPorts(self): return self._serial_ports @Property(bool, notify=serialConnectionChanged) def serialConnected(self): return self._serial_connected @Property(str, notify=currentPortChanged) def currentPort(self): return self._current_port @Property(int, notify=speedChanged) def currentSpeed(self): return self._current_speed @Property(bool, notify=settingsLoaded) def autoPlayOnBoot(self): return self._auto_play_on_boot @Property(bool, notify=backendConnectionChanged) def backendConnected(self): return self._backend_connected @Property(str, notify=reconnectStatusChanged) def reconnectStatus(self): return self._reconnect_status # ==================== Connection management ==================== @Slot() def retryConnection(self): logger.debug("Manual connection retry requested") if self.client.base_url: _run(self._poll_status()) else: _run(self._auto_discover()) @Slot() def refreshSerialPorts(self): """Re-run mDNS discovery to populate the table picker.""" logger.info("Discovering tables...") _run(self._auto_discover()) @Slot(str) def connectSerial(self, port): """Point the app at a table (URL or host).""" logger.info(f"Connecting to table: {port}") self._connect_to(port) def _connect_to(self, url): self.client.set_base_url(url) self._current_port = self.client.base_url self._saved_table_url = self.client.base_url self._time_synced = False self._save_local_settings() self.currentPortChanged.emit(self._current_port) self._set_reconnect_status(f"Connecting to {self._current_port}...") _run(self._poll_status()) @Slot() def disconnectSerial(self): logger.info("Disconnecting from table...") self.client.set_base_url("") self._current_port = "" self._serial_connected = False self._backend_connected = False self._is_connected = False self.serialConnectionChanged.emit(False) self.currentPortChanged.emit("") self.backendConnectionChanged.emit(False) self.connectionChanged.emit() # ==================== Pattern execution ==================== @Slot(str, str) def executePattern(self, fileName, preExecution="adaptive"): logger.info(f"ExecutePattern: '{fileName}' (clear={preExecution})") _run(self._execute_pattern(fileName, preExecution)) async def _execute_pattern(self, fileName, preExecution): try: await self.client.run_pattern(fileName, preExecution) self.executionStarted.emit(fileName, self._preview_path(fileName)) except Exception as exc: logger.error(f"executePattern failed: {exc}") self.errorOccurred.emit(str(exc)) @Slot() def stopExecution(self): _run(self._simple_action(self.client.stop, on_ok=self.executionStopped.emit, label="stop")) @Slot() def pauseExecution(self): logger.info("Pausing execution...") _run(self._simple_action(self.client.pause, label="pause")) @Slot() def resumeExecution(self): logger.info("Resuming execution...") _run(self._simple_action(self.client.resume, label="resume")) @Slot() def skipPattern(self): logger.info("Skipping pattern...") _run(self._simple_action(self.client.playlist_skip, label="skip")) async def _simple_action(self, coro_fn, *, on_ok=None, label="action"): try: await coro_fn() if on_ok: on_ok() except Exception as exc: logger.error(f"{label} failed: {exc}") self.errorOccurred.emit(f"{label} failed: {exc}") @Slot(str, float, str, str, bool) def executePlaylist(self, playlistName, pauseTime=0.0, clearPattern="adaptive", runMode="single", shuffle=False): logger.info(f"ExecutePlaylist: '{playlistName}' pause={pauseTime} " f"clear={clearPattern} mode={runMode} shuffle={shuffle}") _run(self._execute_playlist(playlistName, pauseTime, clearPattern, runMode, shuffle)) async def _execute_playlist(self, name, pauseTime, clearPattern, runMode, shuffle): try: await self.client.run_playlist( name, pause_time=pauseTime, clear_pattern=clearPattern, run_mode=runMode, shuffle=shuffle) except Exception as exc: logger.error(f"executePlaylist failed: {exc}") self.errorOccurred.emit(str(exc)) # ==================== Hardware movement ==================== @Slot() def sendHome(self): logger.debug("Sending home command...") _run(self._simple_action(self.client.home, label="home")) @Slot() def moveToCenter(self): logger.info("Moving to center...") _run(self._simple_action(lambda: self.client.goto(rho=0), label="move to center")) @Slot() def moveToPerimeter(self): logger.info("Moving to perimeter...") _run(self._simple_action(lambda: self.client.goto(rho=1), label="move to perimeter")) # ==================== Speed ==================== @Slot(int) def setSpeed(self, speed): logger.debug(f"Setting speed (feed) to: {speed}") _run(self._set_speed(speed)) async def _set_speed(self, speed): try: await self.client.set_feed_mm(speed) self._current_speed = speed self.speedChanged.emit(speed) except Exception as exc: logger.error(f"set speed failed: {exc}") self.errorOccurred.emit(str(exc)) @Slot(result='QStringList') def getSpeedOptions(self): return list(self.SPEED_OPTIONS.keys()) @Slot(result=str) def getCurrentSpeedOption(self): for option, value in self.SPEED_OPTIONS.items(): if value == self._current_speed: return option return str(self._current_speed) @Slot(str) def setSpeedByOption(self, option): if option in self.SPEED_OPTIONS: self.setSpeed(self.SPEED_OPTIONS[option]) else: logger.warning(f"Unknown speed option: {option}") # ==================== Auto play on boot ==================== @Slot(bool) def setAutoPlayOnBoot(self, enabled): logger.info(f"Setting auto play on boot: {enabled}") # The firmware auto-plays a *named* playlist ($Playlist/Autostart). We # can reliably disable it here; enabling requires choosing a playlist # (done from the playlist page), so True is best-effort. if not enabled: _run(self._simple_action(lambda: self.client.set_autostart(""), label="clear autostart")) else: logger.info("Enable auto-play: select a playlist to autostart on the playlist page") self._auto_play_on_boot = enabled self.settingsLoaded.emit() # ==================== Load settings from the table ==================== @Slot() def loadControlSettings(self): logger.debug("Loading control settings from table...") _run(self._load_settings()) async def _load_settings(self): try: settings = await self.client.settings() except Exception as exc: logger.debug(f"Could not load settings: {exc}") return def as_int(key, default): try: return int(float(settings.get(key, default))) except (TypeError, ValueError): return default # Speed / feed feed = as_int("THR/Feed", self._current_speed) if feed and feed != self._current_speed: self._current_speed = feed self.speedChanged.emit(feed) # Playlist settings (mirror the table) mode = settings.get("Playlist/Mode", self._playlist_run_mode) if mode in ("single", "loop"): self._playlist_run_mode = mode self._playlist_shuffle = str(settings.get("Playlist/Shuffle", "ON")).upper() == "ON" self._pause_between_patterns = as_int("Playlist/PauseTime", self._pause_between_patterns) fw_clear = settings.get("Playlist/ClearPattern", "adaptive") self._playlist_clear_pattern = _FW_TO_UI_CLEAR.get(fw_clear, "adaptive") self._auto_play_on_boot = bool(settings.get("Playlist/Autostart", "").strip()) self.playlistSettingsChanged.emit() self.pauseBetweenPatternsChanged.emit(self._pause_between_patterns) self.settingsLoaded.emit() logger.info("Settings loaded from table") # ==================== Screen management ==================== @Property(bool, notify=screenStateChanged) def screenOn(self): return self._screen_on @Property(int, notify=screenTimeoutChanged) def screenTimeout(self): return self._screen_timeout @screenTimeout.setter def setScreenTimeout(self, timeout): if self._screen_timeout != timeout: self._screen_timeout = timeout self._save_local_settings() self.screenTimeoutChanged.emit(timeout) @Slot(result='QStringList') def getScreenTimeoutOptions(self): return list(self.TIMEOUT_OPTIONS.keys()) @Slot(result=str) def getCurrentScreenTimeoutOption(self): current_timeout = self._screen_timeout for option, value in self.TIMEOUT_OPTIONS.items(): if value == current_timeout: return option if current_timeout == 0: return "Never" elif current_timeout < 60: return f"{current_timeout} seconds" elif current_timeout < 3600: minutes = current_timeout // 60 return f"{minutes} minute{'s' if minutes != 1 else ''}" else: hours = current_timeout // 3600 return f"{hours} hour{'s' if hours != 1 else ''}" @Slot(str) def setScreenTimeoutByOption(self, option): if option in self.TIMEOUT_OPTIONS: timeout_value = self.TIMEOUT_OPTIONS[option] if self._screen_timeout != timeout_value: self._screen_timeout = timeout_value self._save_local_settings() self.screenTimeoutChanged.emit(timeout_value) else: logger.warning(f"Unknown timeout option: {option}") @Slot() def turnScreenOn(self): if not self._screen_on: self._turn_screen_on() self._reset_activity_timer() @Slot() def turnScreenOff(self): self._turn_screen_off() @Slot() def resetActivityTimer(self): self._reset_activity_timer() if not self._screen_on: self._turn_screen_on() def _turn_screen_on(self): with self._screen_transition_lock: time_since_change = time.time() - self._last_screen_change if time_since_change < 2.0: return if self._screen_on: return try: restore_brightness = self._lcd_brightness if self._lcd_brightness > 0 else self._lcd_max_brightness or 255 screen_on_script = Path('/usr/local/bin/screen-on') if screen_on_script.exists(): result = subprocess.run(['sudo', '/usr/local/bin/screen-on'], capture_output=True, text=True, timeout=5) if result.returncode != 0: logger.warning(f"screen-on script failed: {result.stderr}") if self._lcd_brightness_path and restore_brightness < (self._lcd_max_brightness or 255): subprocess.run( ['sudo', 'sh', '-c', f'echo {restore_brightness} > {self._lcd_brightness_path}'], check=False, timeout=5) else: if self._lcd_brightness_path: subprocess.run(['sudo', 'sh', '-c', f'echo 0 > /sys/class/graphics/fb0/blank && echo {restore_brightness} > {self._lcd_brightness_path}'], check=False, timeout=5) else: subprocess.run(['sudo', 'sh', '-c', f'echo 0 > /sys/class/graphics/fb0/blank && echo {restore_brightness} > /sys/class/backlight/*/brightness'], check=False, timeout=5) self._screen_on = True self._last_screen_change = time.time() self.screenStateChanged.emit(True) except Exception as e: logger.error(f"Failed to turn screen on: {e}") def _turn_screen_off(self): with self._screen_transition_lock: time_since_change = time.time() - self._last_screen_change if time_since_change < 2.0: return if not self._screen_on: return try: screen_off_script = Path('/usr/local/bin/screen-off') if screen_off_script.exists(): result = subprocess.run(['sudo', '/usr/local/bin/screen-off'], capture_output=True, text=True, timeout=10) if result.returncode == 0: logger.info("Screen turned OFF (screen-off script)") else: logger.warning(f"screen-off script failed: return code {result.returncode}") else: subprocess.run(['sudo', 'sh', '-c', 'echo 0 > /sys/class/backlight/*/brightness && echo 1 > /sys/class/graphics/fb0/blank'], check=False, timeout=5) self._screen_on = False self._last_screen_change = time.time() self.screenStateChanged.emit(False) except Exception as e: logger.error(f"Failed to turn screen off: {e}") def _reset_activity_timer(self): self._last_activity = time.time() def _check_screen_timeout(self): if self._screen_on and self._screen_timeout > 0: idle_time = time.time() - self._last_activity if idle_time > self._screen_timeout: self._turn_screen_off() # ==================== Local settings persistence ==================== def _load_local_settings(self): try: if os.path.exists(self.SETTINGS_FILE): with open(self.SETTINGS_FILE, 'r') as f: settings = json.load(f) screen_timeout = settings.get('screen_timeout', self.DEFAULT_SCREEN_TIMEOUT) if isinstance(screen_timeout, (int, float)) and screen_timeout >= 0: self._screen_timeout = int(screen_timeout) self._lcd_brightness = settings.get('lcd_brightness', 255) self._lcd_brightness_path = settings.get('lcd_brightness_path', "") self._lcd_max_brightness = settings.get('lcd_max_brightness', 0) self._pause_between_patterns = settings.get('pause_between_patterns', 10800) self._playlist_shuffle = settings.get('playlist_shuffle', True) self._playlist_run_mode = settings.get('playlist_run_mode', "loop") self._playlist_clear_pattern = settings.get('playlist_clear_pattern', "adaptive") self._saved_table_url = settings.get('table_url', "") else: self._save_local_settings() except Exception as e: logger.error(f"Error loading local settings: {e}, using defaults") self._screen_timeout = self.DEFAULT_SCREEN_TIMEOUT def _save_local_settings(self): try: settings = { 'screen_timeout': self._screen_timeout, 'lcd_brightness': self._lcd_brightness, 'lcd_brightness_path': self._lcd_brightness_path, 'lcd_max_brightness': self._lcd_max_brightness, 'pause_between_patterns': self._pause_between_patterns, 'playlist_shuffle': self._playlist_shuffle, 'playlist_run_mode': self._playlist_run_mode, 'playlist_clear_pattern': self._playlist_clear_pattern, 'table_url': self._saved_table_url, 'version': '2.0' } with open(self.SETTINGS_FILE, 'w') as f: json.dump(settings, f, indent=2) except Exception as e: logger.error(f"Error saving local settings: {e}") # ==================== Pause between patterns ==================== @Slot(result='QStringList') def getPauseOptions(self): return list(self.PAUSE_OPTIONS.keys()) @Slot(result=str) def getCurrentPauseOption(self): current_pause = self._pause_between_patterns for option, value in self.PAUSE_OPTIONS.items(): if value == current_pause: return option if current_pause == 0: return "0s" elif current_pause < 60: return f"{current_pause}s" elif current_pause < 3600: return f"{current_pause // 60} min" else: return f"{current_pause // 3600} hour" @Slot(str) def setPauseByOption(self, option): if option in self.PAUSE_OPTIONS: pause_value = self.PAUSE_OPTIONS[option] if self._pause_between_patterns != pause_value: self._pause_between_patterns = pause_value self._save_local_settings() _run(self._simple_action( lambda: self.client.command(f"$Playlist/PauseTime={pause_value}"), label="set pause")) self.pauseBetweenPatternsChanged.emit(pause_value) else: logger.warning(f"Unknown pause option: {option}") @Property(int, notify=pauseBetweenPatternsChanged) def pauseBetweenPatterns(self): return self._pause_between_patterns # ==================== Playlist settings ==================== @Property(bool, notify=playlistSettingsChanged) def playlistShuffle(self): return self._playlist_shuffle @Slot(bool) def setPlaylistShuffle(self, enabled): if self._playlist_shuffle != enabled: self._playlist_shuffle = enabled self._save_local_settings() _run(self._simple_action( lambda: self.client.command(f"$Playlist/Shuffle={'ON' if enabled else 'OFF'}"), label="set shuffle")) self.playlistSettingsChanged.emit() @Property(str, notify=playlistSettingsChanged) def playlistRunMode(self): return self._playlist_run_mode @Slot(str) def setPlaylistRunMode(self, mode): if mode in ("single", "loop") and self._playlist_run_mode != mode: self._playlist_run_mode = mode self._save_local_settings() _run(self._simple_action( lambda: self.client.command(f"$Playlist/Mode={mode}"), label="set run mode")) self.playlistSettingsChanged.emit() @Property(str, notify=playlistSettingsChanged) def playlistClearPattern(self): return self._playlist_clear_pattern @Slot(str) def setPlaylistClearPattern(self, pattern): valid = ["adaptive", "clear_center", "clear_perimeter", "none"] if pattern in valid and self._playlist_clear_pattern != pattern: self._playlist_clear_pattern = pattern self._save_local_settings() from firmware_client import CLEAR_MODE_MAP fw = CLEAR_MODE_MAP.get(pattern, "adaptive") _run(self._simple_action( lambda: self.client.command(f"$Playlist/ClearPattern={fw}"), label="set clear pattern")) self.playlistSettingsChanged.emit() # ==================== LED control ==================== @Property(str, notify=ledStatusChanged) def ledProvider(self): return self._led_provider @Property(bool, notify=ledStatusChanged) def ledConnected(self): return self._led_connected @Property(bool, notify=ledStatusChanged) def ledPowerOn(self): return self._led_power_on @Property(int, notify=ledStatusChanged) def ledBrightness(self): return self._led_brightness @Property(list, notify=ledEffectsLoaded) def ledEffects(self): return self._led_effects @Property(list, notify=ledPalettesLoaded) def ledPalettes(self): return self._led_palettes @Property(int, notify=ledStatusChanged) def ledCurrentEffect(self): return self._led_current_effect @Property(int, notify=ledStatusChanged) def ledCurrentPalette(self): return self._led_current_palette @Property(str, notify=ledStatusChanged) def ledColor(self): return self._led_color @Slot() def loadLedConfig(self): logger.debug("Loading LED configuration...") _run(self._load_led_config()) async def _load_led_config(self): try: settings = await self.client.settings() except Exception as exc: logger.debug(f"LED config load failed: {exc}") return has_leds = any(k.startswith("LED/") for k in settings) self._led_provider = "dw_leds" if has_leds else "none" self._led_connected = has_leds if has_leds: # Expose the fixed catalogues as {id, name} lists for the QML page. self._led_effects = [{"id": i, "name": n} for i, n in enumerate(LED_EFFECTS)] self._led_palettes = [{"id": i, "name": n} for i, n in enumerate(LED_PALETTES)] self.ledEffectsLoaded.emit(self._led_effects) self.ledPalettesLoaded.emit(self._led_palettes) effect = settings.get("LED/Effect", "off") if effect in LED_EFFECTS: self._led_current_effect = LED_EFFECTS.index(effect) palette = settings.get("LED/Palette", "rainbow") if palette in LED_PALETTES: self._led_current_palette = LED_PALETTES.index(palette) self._led_power_on = (effect != "off") if self._led_power_on: self._led_last_effect = self._led_current_effect try: self._led_brightness = round(int(settings.get("LED/Brightness", 255)) * 100 / 255) except (TypeError, ValueError): self._led_brightness = 100 color = settings.get("LED/Color", "ffffff") self._led_color = f"#{color.lstrip('#')}" self.ledStatusChanged.emit() def _ingest_led_status(self, led): """Update live LED effect/brightness from a /sand_status snapshot.""" changed = False effect = led.get("effect") if effect in LED_EFFECTS: idx = LED_EFFECTS.index(effect) if idx != self._led_current_effect: self._led_current_effect = idx changed = True power = (effect != "off") if power != self._led_power_on: self._led_power_on = power changed = True if power: self._led_last_effect = idx b = led.get("brightness") if b is not None: scaled = round(int(b) * 100 / 255) if scaled != self._led_brightness: self._led_brightness = scaled changed = True if changed: self.ledStatusChanged.emit() @Slot() def refreshLedStatus(self): logger.debug("Refreshing LED status...") _run(self._load_led_config()) @Slot() def toggleLedPower(self): self.setLedPower(not self._led_power_on) @Slot(bool) def setLedPower(self, on): logger.debug(f"Setting LED power: {on}") if on: effect = LED_EFFECTS[self._led_last_effect] if self._led_last_effect else "rainbow" if effect == "off": effect = "rainbow" _run(self._apply_led(effect=effect)) self._led_power_on = True self._led_current_effect = LED_EFFECTS.index(effect) else: if self._led_current_effect: self._led_last_effect = self._led_current_effect _run(self._apply_led(effect="off")) self._led_power_on = False self.ledStatusChanged.emit() async def _apply_led(self, **kwargs): try: await self.client.set_led(**kwargs) except Exception as exc: logger.error(f"LED update failed: {exc}") self.errorOccurred.emit(str(exc)) @Slot(int) def setLedBrightness(self, value): logger.debug(f"Setting LED brightness: {value}") value = max(0, min(100, value)) self._led_brightness = value _run(self._apply_led(brightness=round(value * 255 / 100))) self.ledStatusChanged.emit() @Slot(int, int, int) def setLedColor(self, r, g, b): self.setLedColorHex(f"#{r:02x}{g:02x}{b:02x}") @Slot(str) def setLedColorHex(self, hexColor): hexColor = hexColor.lstrip('#') if len(hexColor) != 6: logger.warning(f"Invalid hex color: {hexColor}") return self._led_color = f"#{hexColor}" _run(self._apply_led(color=hexColor)) self.ledStatusChanged.emit() @Slot(int) def setLedEffect(self, effectId): logger.debug(f"Setting LED effect: {effectId}") if 0 <= effectId < len(LED_EFFECTS): name = LED_EFFECTS[effectId] self._led_current_effect = effectId self._led_power_on = (name != "off") if self._led_power_on: self._led_last_effect = effectId _run(self._apply_led(effect=name)) self.ledStatusChanged.emit() @Slot(int) def setLedPalette(self, paletteId): logger.debug(f"Setting LED palette: {paletteId}") if 0 <= paletteId < len(LED_PALETTES): self._led_current_palette = paletteId _run(self._apply_led(palette=LED_PALETTES[paletteId])) self.ledStatusChanged.emit() # ==================== LCD brightness ==================== def _detect_backlight(self): if not self._lcd_brightness_path: try: result = subprocess.run(['ls', '/sys/class/backlight'], capture_output=True, text=True, timeout=2) if result.returncode == 0 and result.stdout.strip(): device = result.stdout.strip().split('\n')[0] self._lcd_brightness_path = f"/sys/class/backlight/{device}/brightness" logger.info(f"Auto-detected backlight path: {self._lcd_brightness_path}") else: logger.warning("No backlight device found") return except Exception as e: logger.warning(f"Failed to detect backlight path: {e}") return backlight_dir = str(Path(self._lcd_brightness_path).parent) if self._lcd_max_brightness == 0: try: result = subprocess.run(['cat', f"{backlight_dir}/max_brightness"], capture_output=True, text=True, timeout=2) if result.returncode == 0 and result.stdout.strip(): self._lcd_max_brightness = int(result.stdout.strip()) else: self._lcd_max_brightness = 255 except Exception: self._lcd_max_brightness = 255 try: result = subprocess.run(['cat', self._lcd_brightness_path], capture_output=True, text=True, timeout=2) if result.returncode == 0 and result.stdout.strip(): self._lcd_brightness = int(result.stdout.strip()) except Exception as e: logger.warning(f"Failed to read current brightness: {e}") @Property(int, notify=lcdBrightnessChanged) def lcdBrightness(self): return self._lcd_brightness @Property(int, notify=lcdBrightnessChanged) def lcdMaxBrightness(self): return self._lcd_max_brightness @Slot(int) def setLcdBrightness(self, value): value = max(0, min(value, self._lcd_max_brightness)) if not self._lcd_brightness_path: logger.warning("No backlight path configured") return try: subprocess.run(['sudo', 'sh', '-c', f'echo {value} > {self._lcd_brightness_path}'], check=False, timeout=5) self._lcd_brightness = value self._save_local_settings() self.lcdBrightnessChanged.emit() except Exception as e: logger.error(f"Failed to set LCD brightness: {e}") # ==================== Pattern refresh ==================== @Slot() def refreshPatterns(self): """Re-fetch the pattern catalogue from the table (models reload).""" logger.debug("Refreshing patterns...") # onPatternsRefreshCompleted in QML triggers patternModel.refresh(). self.patternsRefreshCompleted.emit(True, "Patterns refreshed") # ==================== System control ==================== @Slot() def restartBackend(self): """Reboot the table controller (closest analog to the old backend restart).""" logger.info("Rebooting the table...") _run(self._simple_action(self.client.reboot, label="reboot table")) @Slot() def shutdownPi(self): """Shut down the local touch host (Raspberry Pi).""" logger.info("Shutting down the Pi...") try: subprocess.run(['sudo', 'shutdown', '-h', 'now'], check=False, timeout=5) except Exception as e: logger.error(f"Shutdown failed: {e}") self.errorOccurred.emit(str(e)) # ==================== Playlist management ==================== @Slot(str) def createPlaylist(self, playlistName): logger.debug(f"Creating playlist: {playlistName}") _run(self._create_playlist(playlistName)) async def _create_playlist(self, name): try: content = f"# {name}\n".encode("utf-8") await self.client.upload_file(f"{name}.txt", content, path="/playlists") self.playlistCreated.emit(True, f"Created: {name}") except Exception as exc: logger.error(f"create playlist failed: {exc}") self.playlistCreated.emit(False, f"Failed: {exc}") @Slot(str) def deletePlaylist(self, playlistName): logger.info(f"Deleting playlist: {playlistName}") _run(self._delete_playlist(playlistName)) async def _delete_playlist(self, name): try: await self.client.delete_file(f"{name}.txt", path="/playlists") self.playlistDeleted.emit(True, f"Deleted: {name}") except Exception as exc: logger.error(f"delete playlist failed: {exc}") self.playlistDeleted.emit(False, f"Failed: {exc}") @Slot(str, str) def addPatternToPlaylist(self, playlistName, patternPath): logger.info(f"Adding pattern to playlist: {patternPath} -> {playlistName}") _run(self._add_pattern_to_playlist(playlistName, patternPath)) async def _add_pattern_to_playlist(self, name, patternPath): try: # Fetch current contents, append the SD path, re-upload. try: raw = await self.client.fetch_sd_file(f"/playlists/{name}.txt") lines = raw.decode("utf-8", errors="ignore").splitlines() except Exception: lines = [f"# {name}"] sd_path = self._to_sd_pattern_path(patternPath) lines.append(sd_path) content = ("\n".join(lines) + "\n").encode("utf-8") await self.client.upload_file(f"{name}.txt", content, path="/playlists") self.patternAddedToPlaylist.emit(True, f"Added to {name}") except Exception as exc: logger.error(f"add pattern failed: {exc}") self.patternAddedToPlaylist.emit(False, f"Failed: {exc}") @Slot(str, list) def updatePlaylistPatterns(self, playlistName, patterns): logger.debug(f"Updating playlist {playlistName} -> {len(patterns)} patterns") _run(self._update_playlist_patterns(playlistName, patterns)) async def _update_playlist_patterns(self, name, patterns): try: lines = [f"# {name}"] lines += [self._to_sd_pattern_path(p) for p in patterns] content = ("\n".join(lines) + "\n").encode("utf-8") await self.client.upload_file(f"{name}.txt", content, path="/playlists") self.playlistModified.emit(True, f"Updated: {name}") except Exception as exc: logger.error(f"update playlist failed: {exc}") self.playlistModified.emit(False, f"Failed: {exc}") @staticmethod def _to_sd_pattern_path(pattern): """Normalize a pattern reference to an SD-absolute /patterns path.""" p = str(pattern).strip() if p.startswith("/patterns/") or p.startswith("/sd/"): return p.replace("/sd/", "/", 1) if p.startswith("/sd/") else p return "/patterns/" + p.lstrip("/") @Slot(result=list) def getPlaylistNames(self): """Synchronous accessor kept for QML compatibility (best-effort).""" # Playlists now live on the table; the PlaylistModel is the source of # truth. This returns an empty list rather than reading a local file. return []