dune-weaver/dune-weaver-touch/backend.py

from PySide6.QtCore import QObject, Signal, Property, Slot, QTimer
from PySide6.QtQml import QmlElement
import asyncio
import base64
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,
                             friendly_error, posix_tz)
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")


def _clamp_int(value, default, lo, hi):
    """Coerce ``value`` to an int within [lo, hi], falling back to ``default``."""
    try:
        return max(lo, min(hi, int(value)))
    except (TypeError, ValueError):
        return default


@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
    # Consecutive poll failures before the UI flips to "disconnected". The
    # board's single-threaded web server stalls status reads for seconds while
    # it serves file transfers, so one slow/failed poll means "busy", not
    # "gone". A hard-down board fails fast (connection refused), so real
    # disconnects are still detected within a few seconds.
    STATUS_FAIL_THRESHOLD = 3

    # 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
    patternPreviewReady = Signal(str, str)  # patternName, preview rendered late
    executionStopped = Signal()
    errorOccurred = Signal(str)
    serialPortsUpdated = Signal(list)          # now: list of discovered table URLs
    discoveredTablesUpdated = Signal(list)     # [{name, url}] from mDNS discovery
    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)

        # 'ball' tracker state (firmware-native effect id 38; the blob follows
        # the sand ball). Mirrors the board's NVS; written live via /sand_led.
        self._led_color2 = "#000000"      # background colour when bg == "static"
        self._led_ball_fgbright = 255     # blob brightness (0..255)
        self._led_ball_bgbright = 255     # background brightness (0..255)
        self._led_ball_size = 3           # glow size in LEDs (1..30 in the UI)
        self._led_ball_bg = "static"      # "static" | "off" | any effect name
        self._led_ball_direction = "cw"   # "cw" | "ccw"
        self._led_ball_align = 0          # rotate the blob onto the ball (0..359)
        self._led_last_non_ball_effect = 2  # restored when the ball toggle is off

        # 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()

        # Apply the saved table password ($Sand/Password key) to the client.
        self.client.set_api_key(self._decode_password(self._table_password_b64))

        # 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
        self._poll_inflight = False   # never stack polls on the busy board
        self._poll_failures = 0       # consecutive failures (see threshold)
        self._discovered_tables = []  # [{name, url}] from the last mDNS browse
        self._table_name = ""         # firmware hostname from /sand_status
        self._table_password_b64 = "" # $Sand/Password, base64 like the backend

        # Playlist run state from /sand_status's playlist object (firmware
        # sequences playlists; this is read-only telemetry, like the
        # backend's translate_status).
        self._playlist_active = False
        self._playlist_index = 0      # 0-based position in the playlist
        self._playlist_total = 0
        self._playlist_name = ""
        self._next_pattern = ""       # next pattern (display name, no path/.thr)
        self._playlist_clearing = False
        self._pause_remaining = -1    # seconds until next pattern; -1 = not pausing
        self._pause_total = -1

        # 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._discover(auto_connect=True))

    async def _discover(self, auto_connect=False):
        """mDNS-browse for tables; auto_connect picks one on startup.

        Preference order: the last-connected table (saved URL) if it's on the
        network, otherwise the first table found — the touch panel should
        come up connected without being asked.
        """
        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._discovered_tables = [{"name": t.name, "url": t.base_url} for t in tables]
        self.serialPortsUpdated.emit(self._serial_ports)
        self.discoveredTablesUpdated.emit(self._discovered_tables)
        if not auto_connect:
            return
        if not tables:
            self._set_reconnect_status("No table found. Enter the table address.")
            return
        target = next((t for t in tables if t.base_url == self._saved_table_url),
                      tables[0])
        which = "last-connected" if target.base_url == self._saved_table_url else "first discovered"
        logger.info(f"Auto-connecting to the {which} table: {target.base_url}")
        self._connect_to(target.base_url)

    # ==================== Status polling ====================
    @Slot()
    def _tick_status(self):
        _run(self._poll_status())

    async def _poll_status(self):
        if not self.client.base_url or self._poll_inflight:
            return
        self._poll_inflight = True
        try:
            data = await self.client.status()
        except Exception as exc:
            self._poll_failures += 1
            reason = str(exc) or type(exc).__name__
            # 401 = board is password-locked; deterministic, so say so
            # instead of the generic "retrying" message.
            if getattr(exc, "status", None) == 401:
                self._on_unreachable(reason)
                self._set_reconnect_status(
                    "Table requires a password — set it below.")
            elif self._poll_failures >= self.STATUS_FAIL_THRESHOLD:
                self._on_unreachable(reason)
            else:
                logger.debug(f"Status poll failed ({self._poll_failures}/"
                             f"{self.STATUS_FAIL_THRESHOLD}): {reason}")
            return
        finally:
            self._poll_inflight = False
        self._poll_failures = 0
        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())

        # GRBL states can carry a substate suffix ("Hold:0") — strip it, like
        # the backend's execution._state() does, or pause is never detected.
        state = (status.get("state") or "").split(":", 1)[0]
        self._table_name = status.get("hostname", "") or self._table_name

        # 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}'")
            preview = self._preview_path(new_file)
            self.executionStarted.emit(new_file, preview)
            if not preview:
                _run(self._render_preview_late(new_file))
        self._current_file = new_file

        self._is_running = bool(status.get("running", False))

        # Playlist telemetry (between-patterns pause countdown + position)
        pl = status.get("playlist") or {}
        self._playlist_active = bool(pl.get("active"))
        self._playlist_index = int(pl.get("index", 0) or 0)
        self._playlist_total = int(pl.get("total", 0) or 0)
        self._playlist_name = str(pl.get("name") or "")
        nxt = str(pl.get("next") or "").rsplit("/", 1)[-1]
        self._next_pattern = nxt[:-4] if nxt.endswith(".thr") else nxt
        self._playlist_clearing = bool(pl.get("clearing"))

        def _secs(v):
            return int(v) if isinstance(v, (int, float)) and v >= 0 else -1
        self._pause_remaining = _secs(pl.get("pause_remaining", -1))
        self._pause_total = _secs(pl.get("pause_total", -1))

        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:
            # epoch + POSIX tz, like the backend — board-side quiet-hours and
            # autostart schedules run on the board's local time.
            await self.client.sync_time(int(time.time()), tz=posix_tz())
            self._time_synced = True
        except Exception as exc:
            logger.debug(f"time sync failed: {exc}")

    # ==================== Table password ($Sand/Password) ====================
    @staticmethod
    def _decode_password(b64: str):
        if not b64:
            return None
        try:
            return base64.b64decode(b64).decode("utf-8")
        except Exception:
            return None

    @Slot(str)
    def setTablePassword(self, password):
        """Store the table's API password (empty string clears it)."""
        password = (password or "").strip()
        self._table_password_b64 = (
            base64.b64encode(password.encode("utf-8")).decode("ascii")
            if password else "")
        self.client.set_api_key(password or None)
        self._save_local_settings()
        logger.info("Table password %s", "set" if password else "cleared")
        _run(self._poll_status())

    @Property(bool, notify=settingsLoaded)
    def hasTablePassword(self):
        return bool(self._table_password_b64)

    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 ""

    async def _render_preview_late(self, rel_name):
        """Render an uncached preview for the executing pattern, then notify.

        The Execution page otherwise only ever sees whatever was cached at
        the moment the pattern started."""
        try:
            import thr_preview
            path = await thr_preview.render_preview(
                self.client, self.client.base_url, rel_name)
        except Exception as exc:
            logger.debug(f"late preview render failed for {rel_name}: {exc}")
            return
        if path:
            self.patternPreviewReady.emit(rel_name, path)

    # ==================== 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(str, notify=statusChanged)
    def tableName(self):
        return self._table_name

    @Property(bool, notify=statusChanged)
    def playlistActive(self):
        return self._playlist_active

    @Property(int, notify=statusChanged)
    def playlistIndex(self):
        return self._playlist_index

    @Property(int, notify=statusChanged)
    def playlistTotal(self):
        return self._playlist_total

    @Property(str, notify=statusChanged)
    def playlistName(self):
        return self._playlist_name

    @Property(str, notify=statusChanged)
    def nextPattern(self):
        return self._next_pattern

    @Property(bool, notify=statusChanged)
    def playlistClearing(self):
        return self._playlist_clearing

    @Property(int, notify=statusChanged)
    def pauseRemaining(self):
        return self._pause_remaining

    @Property(int, notify=statusChanged)
    def pauseTotal(self):
        return self._pause_total

    @Property(list, notify=discoveredTablesUpdated)
    def discoveredTables(self):
        return self._discovered_tables

    @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._discover(auto_connect=True))

    @Slot()
    def refreshSerialPorts(self):
        """Re-run mDNS discovery to populate the table picker (list only —
        never switches tables; connecting is an explicit user tap)."""
        logger.info("Discovering tables...")
        _run(self._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._poll_failures = 0  # fresh table, fresh streak
        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._table_name = ""
        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)
            preview = self._preview_path(fileName)
            self.executionStarted.emit(fileName, preview)
            if not preview:
                _run(self._render_preview_late(fileName))
        except Exception as exc:
            logger.error(f"executePattern failed: {exc}")
            self.errorOccurred.emit("Couldn't start the pattern. " + friendly_error(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"Couldn't {label}. " + friendly_error(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("Couldn't start the playlist. " + friendly_error(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("Couldn't change the speed. " + friendly_error(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', "")
                self._table_password_b64 = settings.get('table_password', "")
            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,
                'table_password': self._table_password_b64,
                '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

    # -- 'ball' tracker properties (firmware effect id 38) -----------------
    @Property(str, notify=ledStatusChanged)
    def ledColor2(self):
        return self._led_color2

    @Property(int, notify=ledStatusChanged)
    def ledBallFgBright(self):
        return self._led_ball_fgbright

    @Property(int, notify=ledStatusChanged)
    def ledBallBgBright(self):
        return self._led_ball_bgbright

    @Property(int, notify=ledStatusChanged)
    def ledBallSize(self):
        return self._led_ball_size

    @Property(str, notify=ledStatusChanged)
    def ledBallBg(self):
        return self._led_ball_bg

    @Property(str, notify=ledStatusChanged)
    def ledBallDirection(self):
        return self._led_ball_direction

    @Property(int, notify=ledStatusChanged)
    def ledBallAlign(self):
        return self._led_ball_align

    @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('#')}"

            # 'ball' tracker params (read NVS names, written live via /sand_led).
            color2 = settings.get("LED/Color2", "000000")
            self._led_color2 = f"#{color2.lstrip('#')}"
            self._led_ball_fgbright = _clamp_int(settings.get("LED/BallBright"), 255, 0, 255)
            self._led_ball_bgbright = _clamp_int(settings.get("LED/BallBgBright"), 255, 0, 255)
            self._led_ball_size = _clamp_int(settings.get("LED/BallSize"), 3, 1, 200)
            self._led_ball_align = _clamp_int(settings.get("LED/Align"), 0, 0, 359)
            self._led_ball_bg = (settings.get("LED/BallBg") or "static").lower()
            direction = (settings.get("LED/Direction") or "cw").lower()
            self._led_ball_direction = direction if direction in ("cw", "ccw") else "cw"
            if effect != "ball":
                self._led_last_non_ball_effect = self._led_current_effect

        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
            if effect not in ("ball", "off"):
                self._led_last_non_ball_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("Couldn't update the light. " + friendly_error(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
            if name not in ("ball", "off"):
                self._led_last_non_ball_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()

    # -- 'ball' tracker control (firmware-native effect id 38) -------------
    BALL_EFFECT_ID = LED_EFFECTS.index("ball")

    @Slot(bool)
    def setBallTracker(self, on):
        """Enable/disable the ball tracker by toggling the 'ball' effect.
        Mirrors the mobile app's ball switch: remembers the previous non-ball
        effect and restores it when turned off."""
        if on:
            if self._led_current_effect != self.BALL_EFFECT_ID:
                self._led_last_non_ball_effect = self._led_current_effect
            self.setLedEffect(self.BALL_EFFECT_ID)
        else:
            restore = self._led_last_non_ball_effect or LED_EFFECTS.index("rainbow")
            self.setLedEffect(restore)

    @Slot(str)
    def setLedColor2Hex(self, hexColor):
        """Background colour for the ball tracker (used when bg == 'static')."""
        hexColor = hexColor.lstrip('#')
        if len(hexColor) != 6:
            logger.warning(f"Invalid hex color2: {hexColor}")
            return
        self._led_color2 = f"#{hexColor}"
        _run(self._apply_led(color2=hexColor))
        self.ledStatusChanged.emit()

    @Slot(int)
    def setLedBallFgBright(self, value):
        self._led_ball_fgbright = _clamp_int(value, 255, 0, 255)
        _run(self._apply_led(fgbright=self._led_ball_fgbright))
        self.ledStatusChanged.emit()

    @Slot(int)
    def setLedBallBgBright(self, value):
        self._led_ball_bgbright = _clamp_int(value, 255, 0, 255)
        _run(self._apply_led(bgbright=self._led_ball_bgbright))
        self.ledStatusChanged.emit()

    @Slot(int)
    def setLedBallSize(self, value):
        self._led_ball_size = _clamp_int(value, 3, 1, 200)
        _run(self._apply_led(size=self._led_ball_size))
        self.ledStatusChanged.emit()

    @Slot(int)
    def setLedBallAlign(self, value):
        self._led_ball_align = _clamp_int(value, 0, 0, 359)
        _run(self._apply_led(align=self._led_ball_align))
        self.ledStatusChanged.emit()

    @Slot(str)
    def setLedBallDirection(self, direction):
        if direction not in ("cw", "ccw"):
            return
        self._led_ball_direction = direction
        _run(self._apply_led(direction=direction))
        self.ledStatusChanged.emit()

    @Slot(str)
    def setLedBallBg(self, bg):
        if not bg:
            return
        self._led_ball_bg = str(bg)
        _run(self._apply_led(bg=self._led_ball_bg))
        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("Couldn't shut down the Pi. " + friendly_error(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 []