File size: 3,778 Bytes
151124f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """xros_server.py — WebSocket server exposing the shared world-state API.
Runs a single WorldState and lets any number of clients read and mutate it
over a JSON-over-WebSocket protocol. This is the "world as a system service"
primitive from DiOS: apps consume one shared world model instead of each
reconstructing their own.
Protocol (JSON messages, one object per line):
client -> server:
{"op": "snapshot"} -> full current world
{"op": "spawn", "type": "...", "pose": {...}, "properties": {...}}
{"op": "move", "id": "...", "pose": {...}}
{"op": "remove", "id": "..."}
server -> client (after any mutation, to every connected client):
{"op": "spawn", "object": {...}}
{"op": "move", "object": {...}}
{"op": "remove", "id": "..."}
Run: python xros_server.py --host 127.0.0.1 --port 8765
"""
from __future__ import annotations
import argparse
import asyncio
import json
import websockets
from xros_world import Pose, WorldState
from xros_shell import SpatialShell
async def handle(ws, world: WorldState, shell: SpatialShell) -> None:
"""One connection = one app. Subscribe it to the event bus, then serve it."""
queue = world.subscribe()
# Forward world events to this client.
forward = asyncio.create_task(_forward(queue, ws))
try:
async for raw in ws:
try:
msg = json.loads(raw)
except json.JSONDecodeError:
await ws.send(json.dumps({"op": "error", "error": "bad json"}))
continue
await _dispatch(ws, world, shell, msg)
finally:
forward.cancel()
world.unsubscribe(queue)
async def _forward(queue: asyncio.Queue, ws) -> None:
while True:
event = await queue.get()
await ws.send(json.dumps(event))
async def _dispatch(ws, world: WorldState, shell: SpatialShell, msg: dict) -> None:
op = msg.get("op")
try:
if op == "command":
output = await shell.execute(msg.get("text", ""))
await ws.send(json.dumps({"op": "output", "text": output}))
elif op == "snapshot":
await ws.send(json.dumps({"op": "snapshot", "objects": await world.snapshot()}))
elif op == "spawn":
pose = Pose(**msg.get("pose", {})) if msg.get("pose") else Pose()
obj = await world.spawn(
type=msg.get("type", "entity"),
pose=pose,
properties=msg.get("properties"),
owner=msg.get("owner"),
)
await ws.send(json.dumps({"op": "ack", "id": obj.id}))
elif op == "move":
obj = await world.move(msg["id"], Pose(**msg["pose"]))
await ws.send(json.dumps({"op": "ack", "id": obj.id}))
elif op == "remove":
await world.remove(msg["id"])
await ws.send(json.dumps({"op": "ack", "id": msg["id"]}))
else:
await ws.send(json.dumps({"op": "error", "error": f"unknown op {op!r}"}))
except KeyError as e:
await ws.send(json.dumps({"op": "error", "error": f"unknown object {e}"}))
except (TypeError, ValueError) as e:
await ws.send(json.dumps({"op": "error", "error": str(e)}))
async def main(host: str, port: int) -> None:
world = WorldState()
shell = SpatialShell(world)
async with websockets.serve(lambda ws: handle(ws, world, shell), host, port):
print(f"XROS spatial shell listening on ws://{host}:{port}")
await asyncio.Future() # run forever
if __name__ == "__main__":
p = argparse.ArgumentParser()
p.add_argument("--host", default="127.0.0.1")
p.add_argument("--port", type=int, default=8765)
args = p.parse_args()
asyncio.run(main(args.host, args.port))
|