core/seed_manager.py

# core/seed_manager.py
import json
import random
import threading
from pathlib import Path

DATA_DIR = Path(__file__).resolve().parent.parent / "data"
SEED_FILE = DATA_DIR / "seeds.json"
SEED_FILE.parent.mkdir(parents=True, exist_ok=True)

_lock = threading.Lock()

class SeedManager:
    """
    Persistent and thread-safe seed manager for reproducible image/video generation.
    Maps: scene_id (str/int) -> seed (int)
    """

    def __init__(self, path: Path = SEED_FILE):
        self.path = path
        self._load()

    def _load(self):
        """Load seeds from disk safely."""
        with _lock:
            if self.path.exists():
                try:
                    with open(self.path, "r", encoding="utf-8") as f:
                        self._store = json.load(f)
                except Exception:
                    self._store = {}
            else:
                self._store = {}

    def _save(self):
        """Save seeds to disk safely."""
        with _lock:
            with open(self.path, "w", encoding="utf-8") as f:
                json.dump(self._store, f, indent=2)

    def get_seed(self, scene_id: str | int) -> int | None:
        """Return existing seed for a scene_id or None if not set."""
        return self._store.get(str(scene_id))

    def store_seed(self, scene_id: str | int, seed: int):
        """Store a seed persistently for a given scene_id."""
        self._store[str(scene_id)] = int(seed)
        self._save()

    def ensure_seed(self, scene_id: str | int) -> int:
        """
        Return existing seed or generate/store a new one.
        Guarantees deterministic reproducible seed for the same scene_id.
        """
        s = self.get_seed(scene_id)
        if s is None:
            s = random.randint(0, 2**31 - 1)
            self.store_seed(scene_id, s)
        return s

    def reset(self):
        """Clear all seeds (useful for testing or new projects)."""
        with _lock:
            self._store = {}
            self._save()