world state

xros_world.py

@@ -0,0 +1,123 @@
+"""xros_world.py — the core of the DiOS/XROS world-state service.
+
+This implements the central claim of the DiOS paper (arXiv:2201.03256):
+the physical-digital world is a *single shared resource* among applications.
+Instead of every app reconstructing its own view of the world, apps read and
+mutate one shared world model.
+
+The module is dependency-free (stdlib only) so the store can be unit-tested
+and reused anywhere. Networking lives in xros_server.py.
+
+Concepts:
+  - WorldObject: a node in the scene graph (an entity in the shared world).
+  - WorldState:  the registry of objects + an event bus that publishes
+                 spawn/move/remove events to every subscriber.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+from dataclasses import dataclass, field, asdict
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class Pose:
+    """Position (meters), rotation (quaternion x,y,z,w), scale (unitless)."""
+    position: List[float] = field(default_factory=lambda: [0.0, 0.0, 0.0])
+    rotation: List[float] = field(default_factory=lambda: [0.0, 0.0, 0.0, 1.0])
+    scale: List[float] = field(default_factory=lambda: [1.0, 1.0, 1.0])
+
+
+@dataclass
+class WorldObject:
+    """A single entity in the shared world (a scene-graph node)."""
+    id: str
+    type: str
+    pose: Pose
+    properties: Dict[str, Any] = field(default_factory=dict)
+    owner: Optional[str] = None
+    created_at: float = field(default_factory=time.time)
+    updated_at: float = field(default_factory=time.time)
+
+    def to_dict(self) -> Dict[str, Any]:
+        d = asdict(self)
+        d["pose"] = asdict(self.pose)
+        return d
+
+
+class WorldState:
+    """The shared world: an object registry plus a fan-out event bus.
+
+    Thread-safety: all mutations are serialized through an asyncio.Lock, and
+    events are delivered to per-subscriber asyncio.Queues so a slow client
+    never blocks a fast one.
+    """
+
+    def __init__(self) -> None:
+        self._objects: Dict[str, WorldObject] = {}
+        self._subscribers: List[asyncio.Queue] = []
+        self._lock = asyncio.Lock()
+
+    # ---- mutations -----------------------------------------------------
+
+    async def spawn(
+        self,
+        type: str,
+        pose: Optional[Pose] = None,
+        properties: Optional[Dict[str, Any]] = None,
+        owner: Optional[str] = None,
+    ) -> WorldObject:
+        obj = WorldObject(
+            id=uuid.uuid4().hex,
+            type=type,
+            pose=pose or Pose(),
+            properties=properties or {},
+            owner=owner,
+        )
+        async with self._lock:
+            self._objects[obj.id] = obj
+        await self._publish({"op": "spawn", "object": obj.to_dict()})
+        return obj
+
+    async def move(self, obj_id: str, pose: Pose) -> WorldObject:
+        async with self._lock:
+            obj = self._objects[obj_id]  # KeyError if unknown -> caller handles
+            obj.pose = pose
+            obj.updated_at = time.time()
+        await self._publish({"op": "move", "object": obj.to_dict()})
+        return obj
+
+    async def remove(self, obj_id: str) -> WorldObject:
+        async with self._lock:
+            obj = self._objects.pop(obj_id)  # KeyError if unknown
+        await self._publish({"op": "remove", "id": obj_id})
+        return obj
+
+    # ---- reads ---------------------------------------------------------
+
+    async def snapshot(self) -> List[Dict[str, Any]]:
+        async with self._lock:
+            return [o.to_dict() for o in self._objects.values()]
+
+    async def get(self, obj_id: str) -> Optional[WorldObject]:
+        async with self._lock:
+            return self._objects.get(obj_id)
+
+    # ---- pub/sub -------------------------------------------------------
+
+    def subscribe(self) -> asyncio.Queue:
+        """Return a queue that will receive every future world event."""
+        q: asyncio.Queue = asyncio.Queue()
+        self._subscribers.append(q)
+        return q
+
+    def unsubscribe(self, q: asyncio.Queue) -> None:
+        if q in self._subscribers:
+            self._subscribers.remove(q)
+
+    async def _publish(self, event: Dict[str, Any]) -> None:
+        for q in list(self._subscribers):
+            q.put_nowait(event)