Spaces:
Paused
Paused
File size: 6,608 Bytes
3865888 | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """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 ----
@property
def day_fraction(self) -> float:
return (self.clock_sec % DAY_LENGTH_SEC) / DAY_LENGTH_SEC
@property
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()
|