Private
Public Access
134 lines
4.8 KiB
Python
134 lines
4.8 KiB
Python
"""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
|
|
|
|
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.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.started_at = None
|
|
self.zone_clear_seen = False
|
|
|
|
def reset(self):
|
|
with self._lock:
|
|
self.stopped = False
|
|
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",
|
|
}
|