"""Visitor state tracking for the Matrix frontend. Tracks active visitors as they connect and move around the 3D world, and provides serialization for Matrix protocol broadcast messages. """ import time from dataclasses import dataclass, field from datetime import UTC, datetime @dataclass class VisitorState: """State for a single visitor in the Matrix. Attributes ---------- visitor_id: Unique identifier for the visitor (client ID). display_name: Human-readable name shown above the visitor. position: 3D coordinates (x, y, z) in the world. rotation: Rotation angle in degrees (0-360). connected_at: ISO timestamp when the visitor connected. """ visitor_id: str display_name: str = "" position: dict[str, float] = field(default_factory=lambda: {"x": 0.0, "y": 0.0, "z": 0.0}) rotation: float = 0.0 connected_at: str = field( default_factory=lambda: datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") ) def __post_init__(self): """Set display_name to visitor_id if not provided; copy position dict.""" if not self.display_name: self.display_name = self.visitor_id # Copy position to avoid shared mutable state self.position = dict(self.position) class VisitorRegistry: """Registry of active visitors in the Matrix. Thread-safe singleton pattern (Python GIL protects dict operations). Used by the WebSocket layer to track and broadcast visitor positions. """ _instance: "VisitorRegistry | None" = None def __new__(cls) -> "VisitorRegistry": """Singleton constructor.""" if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._visitors: dict[str, VisitorState] = {} return cls._instance def add( self, visitor_id: str, display_name: str = "", position: dict | None = None ) -> VisitorState: """Add a new visitor to the registry. Parameters ---------- visitor_id: Unique identifier for the visitor. display_name: Optional display name (defaults to visitor_id). position: Optional initial position (defaults to origin). Returns ------- The newly created VisitorState. """ visitor = VisitorState( visitor_id=visitor_id, display_name=display_name, position=position if position else {"x": 0.0, "y": 0.0, "z": 0.0}, ) self._visitors[visitor_id] = visitor return visitor def remove(self, visitor_id: str) -> bool: """Remove a visitor from the registry. Parameters ---------- visitor_id: The visitor to remove. Returns ------- True if the visitor was found and removed, False otherwise. """ if visitor_id in self._visitors: del self._visitors[visitor_id] return True return False def update_position( self, visitor_id: str, position: dict[str, float], rotation: float | None = None, ) -> bool: """Update a visitor's position and rotation. Parameters ---------- visitor_id: The visitor to update. position: New 3D coordinates (x, y, z). rotation: Optional new rotation angle. Returns ------- True if the visitor was found and updated, False otherwise. """ if visitor_id not in self._visitors: return False self._visitors[visitor_id].position = position if rotation is not None: self._visitors[visitor_id].rotation = rotation return True def get(self, visitor_id: str) -> VisitorState | None: """Get a single visitor's state. Parameters ---------- visitor_id: The visitor to retrieve. Returns ------- The VisitorState if found, None otherwise. """ return self._visitors.get(visitor_id) def get_all(self) -> list[dict]: """Get all active visitors as Matrix protocol message dicts. Returns ------- List of visitor_state dicts ready for WebSocket broadcast. Each dict has: type, visitor_id, data (with display_name, position, rotation, connected_at), and ts. """ now = int(time.time()) return [ { "type": "visitor_state", "visitor_id": v.visitor_id, "data": { "display_name": v.display_name, "position": v.position, "rotation": v.rotation, "connected_at": v.connected_at, }, "ts": now, } for v in self._visitors.values() ] def clear(self) -> None: """Remove all visitors (useful for testing).""" self._visitors.clear() def __len__(self) -> int: """Return the number of active visitors.""" return len(self._visitors)