""" Board-owned settings sync — the host side of the "board NVS is canonical" rule. The FluidNC firmware persists table behavior settings in NVS ($Playlist/*, $Sands/*, $Sand/*) and the native mobile apps edit them directly on the board. This module keeps the web backend in agreement: - Auto-play on boot ($Playlist/Autostart*) lives ONLY on the board — it fires when the *table* powers on, whether or not this backend is running. The backend proxies reads/writes for the web UI and mirrors host playlists to the board SD so autostart has something to run. - Still Sands quiet hours ($Sands/*) are stored on the board, but the firmware only enforces them for its own playlist sequencing — an explicit $Sand/Run (how this backend plays each pattern) deliberately bypasses them. So the host keeps enforcing quiet hours itself via state.scheduled_pause_* in pattern_manager; this module pushes UI edits to the board and adopts board values (mobile app edits) back into host state. - The board clock must be set for quiet hours / autostart schedules; the host pushes its epoch + POSIX timezone on connect and whenever the tz changes. All pushes are best-effort: the board being unreachable never blocks a host-side settings save (host enforcement still works without the board's copy). """ import logging import os import threading import time from modules.core.state import state logger = logging.getLogger(__name__) # Host slot day names (state.scheduled_pause_time_slots) <-> firmware 3-letter codes. _DAY_TO_BOARD = { "sunday": "sun", "monday": "mon", "tuesday": "tue", "wednesday": "wed", "thursday": "thu", "friday": "fri", "saturday": "sat", } _BOARD_TO_DAY = {v: k for k, v in _DAY_TO_BOARD.items()} _BOARD_DAY_ORDER = ["sun", "mon", "tue", "wed", "thu", "fri", "sat"] CLEAR_MODES = ("none", "adaptive", "in", "out", "sideway", "random") def _is_on(value) -> bool: return str(value or "").upper() in ("ON", "1", "TRUE") # --------------------------------------------------------------------------- # Still Sands slot conversion: host dicts <-> "$Sands/Slots" spec string # --------------------------------------------------------------------------- def slots_to_board(time_slots: list) -> str: """Host slot dicts -> 'HH:MM-HH:MM@days,...' ($Sands/Slots syntax).""" parts = [] for slot in time_slots or []: start = slot.get("start_time") end = slot.get("end_time") if not start or not end: continue days = slot.get("days", "daily") if days == "custom": codes = sorted( {_DAY_TO_BOARD[d] for d in slot.get("custom_days", []) if d in _DAY_TO_BOARD}, key=_BOARD_DAY_ORDER.index, ) days = "+".join(codes) if codes else "daily" parts.append(f"{start}-{end}@{days}") return ",".join(parts) def _normalize_time(value: str) -> str | None: """'9:5' -> '09:05'; None when not a valid HH:MM.""" hours, sep, minutes = value.strip().partition(":") if not sep or not hours.isdigit() or not minutes.isdigit(): return None h, m = int(hours), int(minutes) if h > 23 or m > 59: return None return f"{h:02d}:{m:02d}" def board_to_slots(spec: str) -> list: """'$Sands/Slots' spec string -> host slot dicts (invalid entries dropped).""" slots = [] for part in (spec or "").split(","): part = part.strip() if not part: continue times, _, days_spec = part.partition("@") start, dash, end = times.strip().partition("-") start, end = _normalize_time(start), _normalize_time(end or "") if not dash or not start or not end: continue days_spec = (days_spec or "daily").strip().lower() if days_spec in ("", "daily", "weekdays", "weekends"): days, custom_days = (days_spec or "daily"), [] else: custom_days = [_BOARD_TO_DAY[c.strip()] for c in days_spec.split("+") if c.strip() in _BOARD_TO_DAY] days = "custom" if custom_days else "daily" slots.append({ "start_time": start, "end_time": end, "days": days, "custom_days": custom_days, }) return slots # --------------------------------------------------------------------------- # Clock sync # --------------------------------------------------------------------------- def posix_tz(iana_name: str | None = None) -> str | None: """POSIX TZ rule string for an IANA zone (or the system zone when None). Modern TZif files (v2+) end with a footer line holding exactly this string (e.g. 'EST5EDT,M3.2.0,M11.1.0'), which is what the firmware's $Time/Zone wants. Returns None when the zone file can't be read. """ path = f"/usr/share/zoneinfo/{iana_name}" if iana_name else "/etc/localtime" try: with open(path, "rb") as f: data = f.read() if not data.startswith(b"TZif"): return None # Footer = text between the last two newlines of the file. end = data.rfind(b"\n") if end <= 0: return None begin = data.rfind(b"\n", 0, end) footer = data[begin + 1:end].decode("ascii").strip() return footer or None except Exception as e: logger.debug(f"Could not derive POSIX tz for {iana_name or 'system'}: {e}") return None def sync_board_time(conn=None) -> dict: """Push the host's clock (and effective quiet-hours timezone) to the board. Quiet hours are computed host-side in state.scheduled_pause_timezone (or the system zone); pushing the same zone keeps board-side schedules (autostart, firmware playlists) aligned with what the user sees in the UI. """ conn = conn or state.conn if not conn: raise RuntimeError("No board connection") tz = posix_tz(state.scheduled_pause_timezone) result = conn.set_time(epoch=int(time.time()), tz=tz) logger.info(f"Synced board clock (tz={tz or 'unchanged'}): {result}") return result # --------------------------------------------------------------------------- # Still Sands push / adopt # --------------------------------------------------------------------------- def push_still_sands(conn=None) -> None: """Write the host's Still Sands settings to the board's $Sands/* NVS keys.""" conn = conn or state.conn if not conn: return conn.set_setting("Sands/Enabled", "ON" if state.scheduled_pause_enabled else "OFF") slots = slots_to_board(state.scheduled_pause_time_slots) if slots: conn.set_setting("Sands/Slots", slots) conn.set_setting("Sands/FinishPattern", "ON" if state.scheduled_pause_finish_pattern else "OFF") # One UI toggle drives both LED systems: host WLED and the board's own ring. conn.set_setting("Sands/LedOff", "ON" if state.scheduled_pause_control_wled else "OFF") logger.info("Pushed Still Sands settings to board") def adopt_still_sands(settings_map: dict) -> bool: """Adopt the board's $Sands/* values into host state (mobile-app edits win). Returns True when anything changed. The timezone is not adopted: the board stores a POSIX rule derived from the host's IANA zone, which isn't reversible — the host zone selection stays authoritative. """ enabled = _is_on(settings_map.get("Sands/Enabled")) finish = _is_on(settings_map.get("Sands/FinishPattern", "ON")) led_off = _is_on(settings_map.get("Sands/LedOff")) slots = board_to_slots(settings_map.get("Sands/Slots", "")) changed = ( enabled != state.scheduled_pause_enabled or finish != state.scheduled_pause_finish_pattern or led_off != state.scheduled_pause_control_wled or slots != state.scheduled_pause_time_slots ) if changed: state.scheduled_pause_enabled = enabled state.scheduled_pause_finish_pattern = finish state.scheduled_pause_control_wled = led_off state.scheduled_pause_time_slots = slots state.save() logger.info("Adopted Still Sands settings from board") return changed # --------------------------------------------------------------------------- # Auto-home cadence ($Playlist/AutoHome) — mirrors the host auto_home settings # so firmware-sequenced playlists (autostart) drift-correct the same way. # --------------------------------------------------------------------------- def push_auto_home(conn=None) -> None: conn = conn or state.conn if not conn: return every = state.auto_home_after_patterns if state.auto_home_enabled else 0 conn.set_setting("Playlist/AutoHome", max(0, int(every or 0))) # --------------------------------------------------------------------------- # Auto-play on boot ($Playlist/Autostart*) — board-only, proxied for the web UI # --------------------------------------------------------------------------- def get_board_settings(conn=None) -> dict: """Shape the board's /sand_settings + clock into the web UI's structure.""" conn = conn or state.conn if not conn: raise RuntimeError("No board connection") s = conn.get_settings() status = conn.get_status() return { "reachable": True, "firmware_version": status.get("fw"), "state": status.get("state"), "time": status.get("time") or conn.get_time(), "autostart": { "playlist": s.get("Playlist/Autostart", ""), "run_mode": "single" if (s.get("Playlist/AutostartMode", "loop").lower() == "single") else "loop", "shuffle": _is_on(s.get("Playlist/AutostartShuffle")), "pause_seconds": int(float(s.get("Playlist/AutostartPause", 0) or 0)), "pause_from_start": _is_on(s.get("Playlist/AutostartPauseFromStart")), "clear_pattern": s.get("Playlist/AutostartClear", "none") or "none", }, "homing_mode": (s.get("Sand/HomingMode") or "sensor").lower(), "theta_offset": float(s.get("Sand/ThetaOffset", 0) or 0), "auto_home_every": int(float(s.get("Playlist/AutoHome", 0) or 0)), } def apply_autostart(update: dict, conn=None) -> None: """Write autostart fields to the board. `update` uses the UI field names.""" conn = conn or state.conn if not conn: raise RuntimeError("No board connection") if "playlist" in update: # Empty value disables auto-play on boot. conn.set_setting("Playlist/Autostart", update["playlist"] or "") if "run_mode" in update: mode = "single" if update["run_mode"] == "single" else "loop" conn.set_setting("Playlist/AutostartMode", mode) if "shuffle" in update: conn.set_setting("Playlist/AutostartShuffle", "ON" if update["shuffle"] else "OFF") if "pause_seconds" in update: conn.set_setting("Playlist/AutostartPause", max(0, int(update["pause_seconds"] or 0))) if "pause_from_start" in update: conn.set_setting("Playlist/AutostartPauseFromStart", "ON" if update["pause_from_start"] else "OFF") if "clear_pattern" in update: clear = update["clear_pattern"] if update["clear_pattern"] in CLEAR_MODES else "none" conn.set_setting("Playlist/AutostartClear", clear) # --------------------------------------------------------------------------- # Playlist mirroring — firmware playlists are /playlists/.txt on the SD, # one SD pattern path per line. Autostart runs these, so host playlist CRUD is # mirrored (and the selected playlist's patterns are ensured on the board). # --------------------------------------------------------------------------- def _playlist_sd_content(files: list, resolve) -> str: lines = [resolve(f) for f in files or []] return "\n".join(lines) + "\n" def mirror_playlist(name: str, files: list, conn=None, ensure_patterns: bool = False, resolve=None) -> None: """Write a host playlist to the board as /playlists/.txt (best-effort). `resolve` maps host entries to SD paths (make_sd_path_resolver); pass the caller's resolver when patterns are ensured separately so the playlist lines and the uploads land on the same paths.""" from modules.core.pattern_manager import _ensure_on_board, make_sd_path_resolver conn = conn or state.conn if not conn: return resolve = resolve or make_sd_path_resolver(conn) try: sd_path = f"/playlists/{name}.txt" data = _playlist_sd_content(files, resolve).encode("utf-8") conn.upload_file(sd_path, data, "/playlists") logger.info(f"Mirrored playlist '{name}' to board ({len(files or [])} patterns)") except Exception as e: logger.warning(f"Could not mirror playlist '{name}' to board: {e}") return if ensure_patterns: for f in files or []: _ensure_on_board(f, resolve(f)) def mirror_playlist_async(name: str, files: list) -> None: """Fire-and-forget mirror from sync code paths (never blocks the caller).""" threading.Thread(target=mirror_playlist, args=(name, files), daemon=True).start() def unmirror_playlist_async(name: str) -> None: threading.Thread(target=unmirror_playlist, args=(name,), daemon=True).start() def unmirror_playlist(name: str, conn=None) -> None: """Delete a playlist's mirror from the board SD (best-effort).""" conn = conn or state.conn if not conn: return try: conn.delete_file("/playlists", f"{name}.txt") logger.info(f"Removed playlist '{name}' from board") except Exception as e: logger.debug(f"Could not remove playlist '{name}' from board: {e}") def _from_playlist_sd_line(line: str) -> str: """Invert _to_sd_path for playlist lines: '/patterns/x.thr' -> 'x.thr' (host playlist entries are relative to the patterns/ dir).""" p = line.replace("\\", "/").strip() if p.startswith("/sd/"): p = p[3:] if p.startswith("/patterns/"): return p[len("/patterns/"):] return p.rsplit("/", 1)[-1] def _make_host_path_resolver(): """Board SD paths don't always mirror the host patterns/ tree: files can reach the board via the mobile app or a copied SD card while the host holds the same pattern elsewhere (e.g. under custom_patterns/). Adopted playlist entries must point at real host files or previews and play both 404. Returns a resolve(rel) -> rel function: exact path if it exists, else a unique catalog suffix match, else a unique basename match, else the raw path unchanged (the pattern genuinely isn't on the host). The catalog is scanned lazily once per adoption pass and results are memoized, so the resolution is deterministic and repeat adoptions stay stable.""" from modules.core import pattern_manager cache: dict = {} catalog: list = [] scanned = False def resolve(rel: str) -> str: nonlocal catalog, scanned if rel in cache: return cache[rel] result = rel if not os.path.exists(os.path.join(pattern_manager.THETA_RHO_DIR, rel)): if not scanned: scanned = True try: catalog = pattern_manager.list_theta_rho_files() except Exception as e: logger.warning(f"Could not scan pattern catalog: {e}") suffix = "/" + rel matches = [f for f in catalog if f.endswith(suffix)] if not matches: basename = rel.rsplit("/", 1)[-1] matches = [f for f in catalog if f.rsplit("/", 1)[-1] == basename] if len(matches) == 1: result = matches[0] cache[rel] = result return result return resolve def adopt_board_playlists(conn=None) -> None: """Read the board's /playlists/*.txt into the host catalog. The board is the source of truth for playlists (mobile apps edit it directly); the host only WRITES on deliberate user actions (playlist CRUD, pressing play, selecting an autostart playlist) — never automatically. Board copies win for names that exist on the board; host-only playlists are kept (they reach the board the next time the user edits or plays them). Reads only — no SD writes, no flash wear. """ from modules.core import playlist_manager conn = conn or state.conn if not conn: return try: names = conn.list_playlists() except Exception as e: logger.warning(f"Could not list board playlists: {e}") return if not isinstance(names, list): return try: playlists = playlist_manager.load_playlists() except Exception: playlists = {} changed = [] resolve_host_path = _make_host_path_resolver() for board_name in names: fname = board_name if board_name.endswith(".txt") else f"{board_name}.txt" name = fname[:-4] if not name: continue try: data = conn.fetch_file(f"/playlists/{fname}") except Exception: continue # listed but unreadable (e.g. deleted mid-scan) — skip files = [resolve_host_path(_from_playlist_sd_line(l)) for l in data.decode("utf-8", "replace").splitlines() if l.strip()] entry = playlists.get(name) current = entry.get("files", entry) if isinstance(entry, dict) else entry if current != files: playlists[name] = {**entry, "files": files} if isinstance(entry, dict) else files changed.append(name) if changed: playlist_manager.save_playlists(playlists) logger.info(f"Adopted board playlists: {', '.join(changed)}") def adopt_auto_home(settings_map: dict) -> None: """Adopt the board's $Playlist/AutoHome cadence (0 = disabled). The host pushes it only when the user edits the setting in the web UI.""" raw = settings_map.get("Playlist/AutoHome") if raw is None: return try: every = int(float(raw)) except (TypeError, ValueError): return enabled = every > 0 if enabled == state.auto_home_enabled and ( not enabled or every == state.auto_home_after_patterns): return state.auto_home_enabled = enabled if enabled: state.auto_home_after_patterns = every state.save_debounced() logger.info(f"Adopted board auto-home cadence: {every or 'disabled'}") # --------------------------------------------------------------------------- # Custom clear patterns — the firmware picks and runs its own clear files # (/patterns/clear_from_in.thr, clear_from_out.thr per its playlist: config). # A "custom" clear is implemented by uploading the chosen pattern's content # over those fixed paths (and restoring the stock file when cleared). # --------------------------------------------------------------------------- def push_custom_clears(conn=None) -> None: """Upload the effective clear files to the board (best effort).""" conn = conn or state.conn if not conn: return from modules.core.pattern_manager import THETA_RHO_DIR import os for board_name, custom in ( ("clear_from_in.thr", state.custom_clear_from_in), ("clear_from_out.thr", state.custom_clear_from_out), ): source = os.path.join(THETA_RHO_DIR, custom) if custom \ else os.path.join(THETA_RHO_DIR, board_name) if not os.path.exists(source): logger.warning(f"Clear source missing, not mirrored: {source}") continue try: with open(source, "rb") as f: conn.upload_file(f"/patterns/{board_name}", f.read(), "/patterns") logger.info(f"Board clear file {board_name} <- {os.path.basename(source)}") except Exception as e: logger.warning(f"Could not push clear file {board_name}: {e}") def push_custom_clears_async() -> None: threading.Thread(target=push_custom_clears, daemon=True).start() # --------------------------------------------------------------------------- # Connect-time reconciliation, called after the board connection is up. # --------------------------------------------------------------------------- def sync_on_connect(conn=None) -> None: """Connect-time reconciliation: clock push + adopt board-owned state. Read-only toward the board's SD/NVS (except the clock, which is RAM/RTC): Still Sands, auto-home cadence and playlists are ADOPTED from the board. The host never auto-pushes content — uploads happen only on deliberate user actions (play, playlist CRUD, autostart selection, settings edits). """ conn = conn or state.conn if not conn: return try: sync_board_time(conn) except Exception as e: logger.warning(f"Board clock sync failed: {e}") try: settings_map = conn.get_settings() adopt_still_sands(settings_map) adopt_auto_home(settings_map) except Exception as e: logger.warning(f"Could not adopt board settings: {e}") adopt_board_playlists(conn)