pre-push commit

battleflow/state.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+import threading, time
+from dataclasses import asdict
+from typing import Any, Dict, List
+from .models import Actor, KNOWN_CONDITIONS, CONDITION_EMOJI, CONDITION_ICON_FILES
+
+def now_ms() -> int:
+    return int(time.time() * 1000)
+
+class CombatState:
+    def __init__(self) -> None:
+        self.actors: List[Actor] = []
+        self.turn_idx: int = 0
+        self.round: int = 1
+        self.visible: bool = True
+        self.dead_mode: str = 'normal'  # 'normal' | 'shrink' | 'hide'
+        self.updated_at: int = now_ms()
+        self.version: int = 1
+        self._cv = threading.Condition()
+
+    def touch(self) -> None:
+        self.updated_at = now_ms()
+        self.version += 1
+
+    def sorted_actors(self) -> List[Actor]:
+        return sorted(self.actors, key=lambda a: a.init, reverse=True)
+
+    def normalize(self) -> None:
+        self.actors = self.sorted_actors()
+        if self.turn_idx >= len(self.actors):
+            self.turn_idx = 0
+
+    def to_public(self) -> Dict[str, Any]:
+        return {
+            'actors': [asdict(a) for a in self.sorted_actors()],
+            'turn_idx': self.turn_idx,
+            'round': self.round,
+            'visible': self.visible,
+            'dead_mode': self.dead_mode,
+            'updated_at': self.updated_at,
+            'version': self.version,
+            'condition_icons': CONDITION_ICON_FILES,
+            'condition_emoji': CONDITION_EMOJI,
+            'known_conditions': KNOWN_CONDITIONS,
+        }
+
+    def broadcast(self) -> None:
+        with self._cv:
+            self.touch()
+            self._cv.notify_all()
+
+    def wait_for(self, since: int, timeout: float = 25.0) -> Dict[str, Any]:
+        end = time.time() + timeout
+        with self._cv:
+            while self.version <= since:
+                remaining = end - time.time()
+                if remaining <= 0:
+                    break
+                self._cv.wait(timeout=remaining)
+            return self.to_public()
+
+# Singleton
+STATE = CombatState()