feat: feed Evennia world events into Nexus websocket bridge

nexus/evennia_ws_bridge.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Publish Evennia telemetry logs into the Nexus websocket bridge."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import re
+from pathlib import Path
+from typing import Iterable
+
+import websockets
+
+from nexus.evennia_event_adapter import actor_located, command_issued, command_result, room_snapshot, session_bound
+
+ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+
+
+def strip_ansi(text: str) -> str:
+    return ANSI_RE.sub("", text or "")
+
+
+def clean_lines(text: str) -> list[str]:
+    text = strip_ansi(text).replace("\r", "")
+    return [line.strip() for line in text.split("\n") if line.strip()]
+
+
+def parse_room_output(text: str):
+    lines = clean_lines(text)
+    if len(lines) < 2:
+        return None
+    title = lines[0]
+    desc = lines[1]
+    exits = []
+    objects = []
+    for line in lines[2:]:
+        if line.startswith("Exits:"):
+            raw = line.split(":", 1)[1].strip()
+            raw = raw.replace(" and ", ", ")
+            exits = [{"key": token.strip(), "destination_id": token.strip().title(), "destination_key": token.strip().title()} for token in raw.split(",") if token.strip()]
+        elif line.startswith("You see:"):
+            raw = line.split(":", 1)[1].strip()
+            raw = raw.replace(" and ", ", ")
+            parts = [token.strip() for token in raw.split(",") if token.strip()]
+            objects = [{"id": p.removeprefix('a ').removeprefix('an '), "key": p.removeprefix('a ').removeprefix('an '), "short_desc": p} for p in parts]
+    return {"title": title, "desc": desc, "exits": exits, "objects": objects}
+
+
+def normalize_event(raw: dict, hermes_session_id: str) -> list[dict]:
+    out: list[dict] = []
+    event = raw.get("event")
+    actor = raw.get("actor", "Timmy")
+    timestamp = raw.get("timestamp")
+
+    if event == "connect":
+        out.append(session_bound(hermes_session_id, evennia_account=actor, evennia_character=actor, timestamp=timestamp))
+        parsed = parse_room_output(raw.get("output", ""))
+        if parsed:
+            out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
+            out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
+        return out
+
+    if event == "command":
+        cmd = raw.get("command", "")
+        output = raw.get("output", "")
+        out.append(command_issued(hermes_session_id, actor, cmd, timestamp=timestamp))
+        success = not output.startswith("Command '") and not output.startswith("Could not find")
+        out.append(command_result(hermes_session_id, actor, cmd, strip_ansi(output), success=success, timestamp=timestamp))
+        parsed = parse_room_output(output)
+        if parsed:
+            out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
+            out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
+        return out
+
+    return out
+
+
+async def playback(log_path: Path, ws_url: str):
+    hermes_session_id = log_path.stem
+    async with websockets.connect(ws_url) as ws:
+        for line in log_path.read_text(encoding="utf-8").splitlines():
+            if not line.strip():
+                continue
+            raw = json.loads(line)
+            for event in normalize_event(raw, hermes_session_id):
+                await ws.send(json.dumps(event))
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Publish Evennia telemetry into the Nexus websocket bridge")
+    parser.add_argument("log_path", help="Path to Evennia telemetry JSONL")
+    parser.add_argument("--ws", default="ws://127.0.0.1:8765", help="Nexus websocket bridge URL")
+    args = parser.parse_args()
+    asyncio.run(playback(Path(args.log_path).expanduser(), args.ws))
+
+
+if __name__ == "__main__":
+    main()