plugin_state.py

"""
Serial Studio plugin state helper.

Wraps the extensions.saveState / extensions.loadState commands and adds
project-change detection by polling project.getStatus, since neither the
TCP nor gRPC API broadcasts a project-changed event.

Usage:
    state = StateManager(
        client, "my-plugin", root,
        on_state_loaded=app._restore_from_dict,
        on_project_changed=app._reset_for_new_project,
    )
    state.restore()
    state.start_polling()

    # later, on any user change:
    state.save({"windows": [...], "settings": {...}})

    # on quit:
    state.save_now({"windows": [...]})
"""

import threading
import time


class StateManager:
    POLL_INTERVAL_MS = 2000
    SAVE_DEBOUNCE_S = 0.25

    def __init__(
        self,
        client,
        plugin_id,
        root,
        on_state_loaded=None,
        on_project_changed=None,
    ):
        self._client = client
        self._plugin_id = plugin_id
        self._root = root
        self._on_state_loaded = on_state_loaded or (lambda state: None)
        self._on_project_changed = on_project_changed or (lambda path, title: None)

        self._save_lock = threading.Lock()
        self._save_pending = False
        self._pending_state = None

        self._project_path = None
        self._project_title = None
        self._project_seen = False

    # ── Save ────────────────────────────────────────────────────────────────

    def save(self, state):
        """Schedule a debounced background save. Safe to call from any thread."""
        with self._save_lock:
            self._pending_state = state
            if self._save_pending:
                return
            self._save_pending = True

        threading.Thread(target=self._do_save, daemon=True).start()

    def save_now(self, state):
        """Save synchronously (use on quit). Returns True on success."""
        if not self._client.connected:
            return False
        try:
            ok, _ = self._client.execute(
                "extensions.saveState",
                {"pluginId": self._plugin_id, "state": state},
            )
            return bool(ok)
        except Exception:
            return False

    def _do_save(self):
        time.sleep(self.SAVE_DEBOUNCE_S)
        with self._save_lock:
            self._save_pending = False
            state = self._pending_state
            self._pending_state = None
        if state is not None:
            self.save_now(state)

    # ── Restore ─────────────────────────────────────────────────────────────

    def restore(self):
        """Load state from the active project; fire on_state_loaded on UI thread."""
        threading.Thread(target=self._do_restore, daemon=True).start()

    def _do_restore(self):
        if not self._client.connected:
            return
        try:
            ok, result = self._client.execute(
                "extensions.loadState", {"pluginId": self._plugin_id}
            )
        except Exception:
            return
        if not ok or not isinstance(result, dict):
            return
        state = result.get("state", result)
        if state:
            self._root.after(0, lambda: self._on_state_loaded(state))

    # ── Project change polling ──────────────────────────────────────────────

    def start_polling(self):
        """Begin periodic project.getStatus polling."""
        self._poll()

    def _poll(self):
        if not getattr(self._client, "running", True):
            return
        if self._client.connected:
            threading.Thread(target=self._do_poll, daemon=True).start()
        self._root.after(self.POLL_INTERVAL_MS, self._poll)

    def _do_poll(self):
        try:
            ok, result = self._client.execute("project.getStatus", {})
        except Exception:
            return
        if not ok or not isinstance(result, dict):
            return
        path = result.get("filePath", "") or ""
        title = result.get("title", "") or ""
        self._root.after(0, lambda: self._apply_project(path, title))

    def _apply_project(self, path, title):
        if not self._project_seen:
            self._project_seen = True
            self._project_path = path
            self._project_title = title
            return
        if path == self._project_path and title == self._project_title:
            return
        self._project_path = path
        self._project_title = title
        self._on_project_changed(path, title)