Spaces:
Paused
Paused
| """Persistent 3D world runtime (V6 phase_04/19/22). | |
| Owns: spec, physics, world clock, object states (doors/food/water), character | |
| roster. Snapshot/restore covers physics + clock + consumed + roster config. | |
| TRUE_WORLD lives here; organism knowledge lives in spatial memory (never the | |
| reverse). | |
| """ | |
| import hashlib | |
| import json | |
| import math | |
| import time | |
| from typing import Any, Dict, List, Optional | |
| from src.world3d.spec import build_world_spec | |
| from src.world3d.physics import PhysicsWorld | |
| DAY_LENGTH_SEC = 600.0 # 10-minute full day cycle | |
| class World3D: | |
| def __init__(self, seed: int = 7, characters: Optional[List[Dict[str, Any]]] = None): | |
| self.spec = build_world_spec(seed) | |
| self.seed = seed | |
| self.physics = PhysicsWorld(self.spec, characters) | |
| self.clock_sec = 0.0 | |
| self.tick = 0 | |
| self.consumed: List[str] = [] # food ids eaten (bodies parked) | |
| self.door_state: Dict[str, str] = {d["id"]: "closed" for d in self.spec["doors"]} | |
| self.events: List[Dict[str, Any]] = [] | |
| # ---- clock ---- | |
| def day_fraction(self) -> float: | |
| return (self.clock_sec % DAY_LENGTH_SEC) / DAY_LENGTH_SEC | |
| def is_night(self) -> bool: | |
| return self.day_fraction < 0.25 or self.day_fraction > 0.8 | |
| # ---- stepping ---- | |
| def step(self, dt: float = 0.05, substeps: int = 5) -> None: | |
| for _ in range(substeps): | |
| self.physics.step(dt / substeps) | |
| # door torques are single impulses: clear after integration | |
| for d in self.spec["doors"]: | |
| try: | |
| da = int(self.physics.model.jnt_dofadr[self._hinge(d["id"])]) | |
| self.physics.data.qfrc_applied[da] = 0.0 | |
| except Exception: | |
| pass | |
| self.clock_sec += dt | |
| self.tick += 1 | |
| self._update_doors() | |
| def _update_doors(self) -> None: | |
| for d in self.spec["doors"]: | |
| try: | |
| jid = self._hinge(d["id"]) | |
| ang = float(self.physics.data.qpos[int(self.physics.model.jnt_qposadr[jid])]) | |
| state = "open" if abs(ang) > 0.35 else "closed" | |
| if state != self.door_state[d["id"]]: | |
| self.door_state[d["id"]] = state | |
| self.log("door", {"door": d["id"], "state": state, "angle": round(ang, 3)}) | |
| except Exception: | |
| pass | |
| def _hinge(self, door_id: str) -> int: | |
| import mujoco | |
| jid = mujoco.mj_name2id(self.physics.model, mujoco.mjtObj.mjOBJ_JOINT, | |
| f"hinge_{door_id}") | |
| if jid < 0: | |
| raise KeyError(door_id) | |
| return jid | |
| # ---- queries ---- | |
| def log(self, kind: str, payload: Dict[str, Any]) -> None: | |
| self.events.append({"tick": self.tick, "t": round(self.clock_sec, 2), | |
| "kind": kind, **payload}) | |
| def in_water(self, x: float, y: float) -> bool: | |
| w = self.spec["water"] | |
| return abs(x - w["cx"]) <= w["sx"] / 2 and abs(y - w["cy"]) <= w["sy"] / 2 | |
| def nearest_food(self, x: float, y: float, max_d: float = 1.2) -> Optional[Dict[str, Any]]: | |
| best, bd = None, max_d | |
| for f in self.spec["foods"]: | |
| if f["id"] in self.consumed: | |
| continue | |
| d = math.hypot(f["x"] - x, f["y"] - y) | |
| if d < bd: | |
| best, bd = f, d | |
| return best | |
| # ---- interactions (real state transitions) ---- | |
| def push_door(self, door_id: str, torque: float = 30.0) -> Dict[str, Any]: | |
| try: | |
| jid = self._hinge(door_id) | |
| da = int(self.physics.model.jnt_dofadr[jid]) | |
| self.physics.data.qfrc_applied[da] += float(torque) | |
| return {"status": "EXECUTED", "door": door_id, "torque": torque, | |
| "state": self.door_state.get(door_id, "unknown")} | |
| except KeyError: | |
| return {"status": "ERROR", "reason": f"unknown door {door_id!r}"} | |
| def eat_food(self, char_name: str) -> Dict[str, Any]: | |
| st = self.physics.char_state(char_name) | |
| f = self.nearest_food(st["pos"][0], st["pos"][1]) | |
| if f is None: | |
| return {"status": "NOTHING_NEARBY"} | |
| self.consumed.append(f["id"]) | |
| # park the body far below the world (deterministic, no rebuild) | |
| self.physics.teleport(f"bod_{f['id']}", 0.0, 0.0, -50.0) | |
| self.log("consumed", {"food": f["id"], "by": char_name, "energy": f["energy"]}) | |
| return {"status": "CONSUMED", "food": f["id"], "energy": f["energy"]} | |
| def object_states(self) -> Dict[str, Any]: | |
| return {"doors": dict(self.door_state), | |
| "foods_remaining": [f["id"] for f in self.spec["foods"] | |
| if f["id"] not in self.consumed], | |
| "foods_consumed": list(self.consumed)} | |
| # ---- persistence ---- | |
| def state_hash(self) -> str: | |
| h = hashlib.sha256() | |
| h.update(self.physics.data.qpos.tobytes()) | |
| h.update(self.physics.data.qvel.tobytes()) | |
| h.update(json.dumps({"tick": self.tick, "clock": round(self.clock_sec, 4), | |
| "consumed": sorted(self.consumed), | |
| "doors": self.door_state}, sort_keys=True).encode()) | |
| return h.hexdigest() | |
| def snapshot(self) -> Dict[str, Any]: | |
| from src.version import VERSION | |
| return {"version": "world3d_v1", "app_version": VERSION, "seed": self.seed, | |
| "spec_hash": self.spec.get("spec_hash", ""), | |
| "physics": self.physics.snapshot(), "clock_sec": self.clock_sec, | |
| "tick": self.tick, "consumed": list(self.consumed), | |
| "door_state": dict(self.door_state), | |
| "events": self.events[-200:]} | |
| def restore(self, snap: Dict[str, Any]) -> None: | |
| if snap.get("spec_hash", "") != self.spec.get("spec_hash", ""): | |
| raise ValueError("world snapshot belongs to a different spec") | |
| # re-park consumed bodies BEFORE physics restore (teleport is part of state) | |
| for fid in snap.get("consumed", []): | |
| try: | |
| self.physics.teleport(f"bod_{fid}", 0.0, 0.0, -50.0) | |
| except KeyError: | |
| pass | |
| self.consumed = list(snap.get("consumed", [])) | |
| if snap.get("door_state"): | |
| self.door_state = dict(snap["door_state"]) | |
| self.physics.restore(snap["physics"]) | |
| self.clock_sec = float(snap.get("clock_sec", 0.0)) | |
| self.tick = int(snap.get("tick", 0)) | |
| self.events = list(snap.get("events", [])) | |
| self._update_doors() | |