spatial shell

xros_shell.py

@@ -0,0 +1,169 @@
+"""xros_shell.py — the spatial shell: a command interpreter over the shared world.
+
+This is the "terminal" half of the spatial OS. The terminal (in the Quest app)
+sends a command string; the shell parses it, mutates the shared WorldState, and
+returns text output. World mutations are broadcast to every connected client by
+the server, so a `spawn panel` typed in the terminal makes a panel appear in the
+spatial scene for everyone.
+
+Commands:
+  help                          list commands
+  ls | list                     list every object in the world
+  spawn <type> [x y z] [opts]   create an object (panel, cube, ...)
+  move <id> x y z               move an object
+  rm <id> | remove <id>         remove an object
+  clear                         remove every object
+  gen <description>             generative UI (stub: rule-based panel generator)
+
+The `gen` command is the stub for the generative-UI layer: it maps a natural-
+language description to a panel type via keyword rules, so the concept works
+end to end before a real LLM is wired in.
+"""
+
+from __future__ import annotations
+
+import shlex
+from typing import List, Optional, Tuple
+
+from xros_world import Pose, WorldState
+
+HELP_TEXT = """spatial shell commands:
+  help
+  ls | list
+  spawn <type> [x y z] [--title T] [--color C]
+  move <id> x y z
+  rm <id> | remove <id>
+  clear
+  gen <description>   (generative UI — stub)
+"""
+
+# Keyword -> panel type rules for the generative-UI stub.
+GEN_RULES: List[Tuple[List[str], str]] = [
+    (["clock", "time", "watch"], "clock"),
+    (["note", "memo", "remind", "remember"], "note"),
+    (["todo", "task", "checklist", "list"], "todo"),
+    (["chart", "graph", "plot", "bar"], "chart"),
+    (["image", "photo", "picture", "pic"], "image"),
+    (["video", "player", "movie"], "video"),
+    (["map", "location", "where"], "map"),
+    (["browser", "web", "search"], "browser"),
+]
+
+
+class SpatialShell:
+    def __init__(self, world: WorldState):
+        self.world = world
+
+    async def execute(self, command: str) -> str:
+        """Run one command line and return its text output."""
+        if not command.strip():
+            return ""
+        try:
+            parts = shlex.split(command)
+        except ValueError as e:
+            return f"error: {e}"
+        if not parts:
+            return ""
+        cmd, args = parts[0].lower(), parts[1:]
+        handler = getattr(self, f"cmd_{cmd}", None)
+        if handler is None:
+            return f"error: unknown command '{cmd}' (try 'help')"
+        try:
+            return await handler(args)
+        except Exception as e:  # surface any error as terminal text, never crash
+            return f"error: {e}"
+
+    # ---- commands ------------------------------------------------------
+
+    async def cmd_help(self, args: List[str]) -> str:
+        return HELP_TEXT.strip()
+
+    async def cmd_ls(self, args: List[str]) -> str:
+        objs = await self.world.snapshot()
+        if not objs:
+            return "(empty world)"
+        lines = []
+        for o in objs:
+            x, y, z = o["pose"]["position"]
+            title = o["properties"].get("title", "")
+            lines.append(f"{o['id'][:8]}  {o['type']:<8}  ({x:.1f}, {y:.1f}, {z:.1f})  {title}")
+        return "\n".join(lines)
+
+    async def cmd_spawn(self, args: List[str]) -> str:
+        type_, pos, opts = self._parse_spawn(args)
+        obj = await self.world.spawn(
+            type=type_,
+            pose=Pose(position=pos),
+            properties=opts,
+        )
+        title = opts.get("title", "")
+        return f"spawned {type_} {obj.id[:8]}" + (f" \"{title}\"" if title else "")
+
+    async def cmd_move(self, args: List[str]) -> str:
+        if len(args) < 4:
+            return "usage: move <id> x y z"
+        obj_id, xyz = args[0], [float(a) for a in args[1:4]]
+        obj = await self.world.move(obj_id, Pose(position=xyz))
+        return f"moved {obj_id[:8]} to ({xyz[0]:.1f}, {xyz[1]:.1f}, {xyz[2]:.1f})"
+
+    async def cmd_rm(self, args: List[str]) -> str:
+        if not args:
+            return "usage: rm <id>"
+        obj = await self.world.remove(args[0])
+        return f"removed {args[0][:8]} ({obj.type})"
+
+    async def cmd_clear(self, args: List[str]) -> str:
+        objs = await self.world.snapshot()
+        for o in objs:
+            await self.world.remove(o["id"])
+        return f"cleared {len(objs)} object(s)"
+
+    async def cmd_gen(self, args: List[str]) -> str:
+        """Generative-UI stub: map a description to a panel via keyword rules."""
+        if not args:
+            return "usage: gen <description>"
+        description = " ".join(args)
+        panel_type = self._classify(description)
+        obj = await self.world.spawn(
+            type="panel",
+            pose=Pose(position=[0.0, 0.0, -1.5]),
+            properties={"title": description, "panel_type": panel_type},
+        )
+        return f"generated {panel_type} panel {obj.id[:8]} for \"{description}\""
+
+    # ---- helpers -------------------------------------------------------
+
+    def _parse_spawn(self, args: List[str]) -> Tuple[str, List[float], dict]:
+        """Parse `spawn <type> [x y z] [--title T] [--color C]`."""
+        if not args:
+            raise ValueError("usage: spawn <type> [x y z] [--title T] [--color C]")
+        type_ = args[0]
+        rest = args[1:]
+        pos: List[float] = [0.0, 0.0, -1.5]  # default: 1.5m in front of the user
+        opts: dict = {}
+        i = 0
+        # leading three numbers = position
+        nums: List[float] = []
+        while i < len(rest) and len(nums) < 3:
+            try:
+                nums.append(float(rest[i]))
+                i += 1
+            except ValueError:
+                break
+        if nums:
+            pos = nums
+        # remaining --key value pairs
+        while i < len(rest):
+            if rest[i].startswith("--") and i + 1 < len(rest):
+                opts[rest[i][2:]] = rest[i + 1]
+                i += 2
+            else:
+                i += 1
+        return type_, pos, opts
+
+    def _classify(self, description: str) -> str:
+        low = description.lower()
+        for keywords, panel_type in GEN_RULES:
+            if any(k in low for k in keywords):
+                return panel_type
+        return "generic"