Spaces:
Sleeping
Sleeping
File size: 2,072 Bytes
eb8c5e1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
# 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()
|