memory.py

"""Persistent memory for Nova — facts, reminders, and command history.

Stored as JSON at ``~/.nova/memory.json``. Thread-safe via an internal
lock so the FastAPI server, push-to-talk loop, and wake-word loop can
all share the same Memory instance.
"""

from __future__ import annotations

import json
import os
import threading
import uuid
from datetime import datetime, timezone


def _now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
from pathlib import Path
from typing import Any, Dict, List, Optional

_DEFAULT_PATH = "~/.nova/memory.json"
_HISTORY_CAP = 1000


class Memory:
    """Tiny JSON-backed key/value + log store."""

    def __init__(self, path: str = _DEFAULT_PATH) -> None:
        self.path = Path(os.path.expanduser(path))
        self._lock = threading.RLock()
        self._data: Dict[str, Any] = {
            "facts": {},
            "reminders": [],
            "history": [],
        }
        self._load()

    # ---------- IO ---------- #

    def _load(self) -> None:
        try:
            if self.path.exists():
                with self.path.open("r", encoding="utf-8") as f:
                    raw = json.load(f)
                if isinstance(raw, dict):
                    self._data["facts"] = dict(raw.get("facts") or {})
                    self._data["reminders"] = list(raw.get("reminders") or [])
                    self._data["history"] = list(raw.get("history") or [])
        except Exception as e:
            print(f"⚠️  Memory load failed ({e}); starting fresh.")

    def _save(self) -> None:
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            tmp = self.path.with_suffix(".json.tmp")
            with tmp.open("w", encoding="utf-8") as f:
                json.dump(self._data, f, indent=2, ensure_ascii=False)
            tmp.replace(self.path)
        except Exception as e:
            print(f"⚠️  Memory save failed: {e}")

    # ---------- facts ---------- #

    def remember(self, key: str, value: str) -> None:
        if not key:
            return
        with self._lock:
            self._data["facts"][str(key).strip().lower()] = str(value)
            self._save()

    def recall(self, key: str) -> Optional[str]:
        if not key:
            return None
        with self._lock:
            return self._data["facts"].get(str(key).strip().lower())

    def list_facts(self) -> Dict[str, str]:
        with self._lock:
            return dict(self._data["facts"])

    def forget(self, key: str) -> bool:
        with self._lock:
            k = str(key).strip().lower()
            if k in self._data["facts"]:
                del self._data["facts"][k]
                self._save()
                return True
            return False

    # ---------- history ---------- #

    def add_history(self, command: str, response: str) -> None:
        entry = {
            "timestamp": _now_iso(),
            "command": str(command or ""),
            "response": str(response or ""),
        }
        with self._lock:
            hist = self._data["history"]
            hist.append(entry)
            if len(hist) > _HISTORY_CAP:
                # FIFO trim
                del hist[: len(hist) - _HISTORY_CAP]
            self._save()

    def get_history(self, limit: int = 50) -> List[Dict[str, Any]]:
        with self._lock:
            hist = self._data["history"]
            if limit <= 0:
                return list(hist)
            return list(hist[-limit:])

    def clear_history(self) -> None:
        with self._lock:
            self._data["history"] = []
            self._save()

    # ---------- reminders ---------- #

    def add_reminder(self, text: str, when: str) -> Dict[str, Any]:
        entry = {
            "id": uuid.uuid4().hex[:8],
            "text": str(text or ""),
            "when": str(when or ""),
            "created": _now_iso(),
        }
        with self._lock:
            self._data["reminders"].append(entry)
            self._save()
        return entry

    def get_reminders(self) -> List[Dict[str, Any]]:
        with self._lock:
            return list(self._data["reminders"])

    def remove_reminder(self, reminder_id: str) -> bool:
        with self._lock:
            before = len(self._data["reminders"])
            self._data["reminders"] = [
                r for r in self._data["reminders"] if r.get("id") != reminder_id
            ]
            if len(self._data["reminders"]) != before:
                self._save()
                return True
            return False


# Module-level singleton convenience
_default_memory: Optional[Memory] = None
_default_lock = threading.Lock()


def get_default_memory() -> Memory:
    global _default_memory
    with _default_lock:
        if _default_memory is None:
            _default_memory = Memory()
        return _default_memory


if __name__ == "__main__":
    m = Memory()
    m.remember("my name", "Kanishk")
    m.add_history("hello", "hi there")
    m.add_reminder("buy milk", "tomorrow 9am")
    print("Facts:", m.list_facts())
    print("History:", m.get_history(5))
    print("Reminders:", m.get_reminders())
# note 6
# note 14
# note 22
# note 30
# note 38
# note 46
# note 54
# note 62
# note 70
# note 78
# note 86
# note 94
# note 102
# note 110
# note 118