Initial

assembly_player/state.py

@@ -0,0 +1,81 @@
+"""Per-assembly playback progress, persisted so playback can resume exactly
+where the user left off, across app restarts and reboots."""
+import datetime
+import json
+import os
+
+XDG_STATE_HOME = os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state"))
+STATE_DIR = os.path.join(XDG_STATE_HOME, "assembly-player")
+STATE_PATH = os.path.join(STATE_DIR, "state.json")
+
+
+def _now():
+    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
+
+
+def load_all():
+    if not os.path.exists(STATE_PATH):
+        return {}
+    try:
+        with open(STATE_PATH, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError):
+        return {}
+
+
+def save_all(data):
+    os.makedirs(STATE_DIR, exist_ok=True)
+    tmp_path = STATE_PATH + ".tmp"
+    with open(tmp_path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+    os.replace(tmp_path, STATE_PATH)
+
+
+def _blank_entry():
+    return {"sessions": [], "current": None, "position": 0.0, "completed": [], "updated": _now()}
+
+
+class AssemblyState:
+    """Convenience wrapper around the state entry for a single assembly code."""
+
+    def __init__(self, code):
+        self.code = code
+        self._all = load_all()
+        self._entry = self._all.setdefault(code, _blank_entry())
+
+    def set_sessions(self, ordered_titles):
+        self._entry["sessions"] = ordered_titles
+        self._persist()
+
+    def update_progress(self, title, position):
+        self._entry["current"] = title
+        self._entry["position"] = position
+        self._entry["updated"] = _now()
+        self._persist()
+
+    def mark_completed(self, title):
+        if title not in self._entry["completed"]:
+            self._entry["completed"].append(title)
+        if self._entry.get("current") == title:
+            self._entry["current"] = None
+            self._entry["position"] = 0.0
+        self._persist()
+
+    def resume_target(self):
+        """Return (title, position) to resume at, or (None, 0) if nothing to resume."""
+        current = self._entry.get("current")
+        sessions = self._entry.get("sessions") or []
+        completed = set(self._entry.get("completed") or [])
+
+        if current and current not in completed:
+            return current, self._entry.get("position", 0.0)
+
+        for title in sessions:
+            if title not in completed:
+                return title, 0.0
+
+        return None, 0.0
+
+    def _persist(self):
+        self._all[self.code] = self._entry
+        save_all(self._all)