"""Sichere Zustandslogik für den simulierten Effektablauf.""" import json import os import tempfile import threading import time from pathlib import Path MODES = ("aus", "atmosphaere", "sanft", "voll") SEQUENCE_SECONDS = 3.0 COOLDOWN_SECONDS = 10.0 class Controller: def __init__(self, state_dir: Path, clock=time.monotonic): self._lock = threading.RLock() self._clock = clock self._path = state_dir / "settings.json" self._path.parent.mkdir(parents=True, exist_ok=True) self.mode, self.config_error = self._load_mode() self.stopped = False self.zone_clear_seen = False self.started_at = None self.cooldown_until = 0.0 self.last_event = None self.generation = 0 def _load_mode(self): if not self._path.exists(): return "voll", None try: mode = json.loads(self._path.read_text(encoding="utf-8"))["mode"] if mode not in MODES: raise ValueError("ungültiger Modus") return mode, None except (OSError, ValueError, KeyError, TypeError) as exc: return "aus", f"Einstellungen fehlerhaft: {exc}" def _save_mode(self, mode: str): temp_path = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=self._path.parent, prefix="settings-", suffix=".tmp", delete=False, ) as stream: temp_path = Path(stream.name) json.dump({"mode": mode}, stream) stream.flush() os.fsync(stream.fileno()) os.replace(temp_path, self._path) finally: if temp_path is not None: temp_path.unlink(missing_ok=True) def set_mode(self, mode: str): if mode not in MODES: raise ValueError("unbekannter Modus") with self._lock: self._save_mode(mode) if mode != self.mode: self.generation += 1 self.started_at = None self.zone_clear_seen = False self.mode = mode self.config_error = None def zone_clear(self): with self._lock: if not self.stopped: self.zone_clear_seen = True def trigger(self): """Nur neuer Trigger nach freiem Bereich; bislang ohne Aktorausgabe.""" with self._lock: now = self._clock() if self.stopped: return False, "Stopp aktiv" if self.config_error: return False, "Einstellungen fehlerhaft" if self.mode not in ("sanft", "voll"): return False, "Modus löst keinen Schreck aus" if not self.zone_clear_seen: return False, "Auslösebereich zuerst freigeben" if self.started_at is not None and now - self.started_at < SEQUENCE_SECONDS: return False, "Sequenz läuft" if now < self.cooldown_until: return False, "Sperrzeit läuft" self.started_at = now self.cooldown_until = now + COOLDOWN_SECONDS self.zone_clear_seen = False self.last_event = {"mode": self.mode, "time": time.strftime("%Y-%m-%d %H:%M:%S")} return True, "Simulation gestartet" def stop(self): with self._lock: self.stopped = True self.generation += 1 self.started_at = None self.zone_clear_seen = False def reset(self): with self._lock: self.stopped = False self.generation += 1 self.started_at = None self.zone_clear_seen = False def status(self): with self._lock: now = self._clock() elapsed = None if self.started_at is None else now - self.started_at if self.stopped: phase = "Stopp" elif elapsed is not None and elapsed < 0.8: phase = "Stille" elif elapsed is not None and elapsed < 1.7: phase = "Kratzen (simuliert)" elif elapsed is not None and elapsed < 2.4: phase = "Schreck (simuliert)" if self.mode == "voll" else "Sanft (simuliert)" elif elapsed is not None and elapsed < SEQUENCE_SECONDS: phase = "Rückkehr (simuliert)" elif now < self.cooldown_until: phase = "Sperrzeit" else: phase = "Bereit" if self.zone_clear_seen else "Bereich frei abwarten" return { "mode": self.mode, "phase": phase, "stopped": self.stopped, "zone_clear_seen": self.zone_clear_seen, "cooldown_seconds": round(max(0, self.cooldown_until - now), 1), "last_event": self.last_event, "config_error": self.config_error, "actuators": "nur Simulation", }