Spaces:
Running
Running
File size: 5,013 Bytes
d660fa9 f4542ce d660fa9 f4542ce d660fa9 f4542ce | 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 | """perceive — turns a frozen WorldSnapshot into what one agent can observe.
Pure functions, no LLM, no side effects: builds nearest-first Observation
lists in both tile (Chebyshev) and pixel (Euclidean) spaces.
Architecture: used by the engine's perceive phase via snapshot
agents_within_px; the novelty fingerprint that gates LLM decisions is
computed from these observations.
Design: strict purity keeps perception reproducible and cheap.
"""
from __future__ import annotations
from typing import List, Optional
from pydantic import BaseModel
from src.core.log import get_logger
from src.core.snapshot import WorldSnapshot
from src.core.world_state import AgentState
from src.config import DEFAULT_PERCEPTION_RADIUS
logger = get_logger(__name__)
class Observation(BaseModel):
"""One thing an agent noticed this tick."""
tick: int
observer_id: str
subject_agent_id: str
description: str # human-readable, e.g. "Gurnoor is eating breakfast"
location_id: Optional[str] = None
distance: int = 0 # chebyshev tiles from observer; lower = more salient
def perceive(
snapshot: WorldSnapshot,
agent_id: str,
radius: int = DEFAULT_PERCEPTION_RADIUS,
) -> List[Observation]:
"""
Return everything `agent_id` can currently observe in `snapshot`,
nearest first.
Called from tick_graph.py's perceive node with the tick's single shared
snapshot -- never call this once the underlying WorldState might have
moved on; always pass the snapshot the whole tick is using.
"""
if agent_id not in snapshot.agents:
raise KeyError(f"'{agent_id}' not present in this snapshot")
me = snapshot.get_agent(agent_id)
nearby = snapshot.agents_near(agent_id, radius)
observations: List[Observation] = []
for other in nearby:
distance = max(
abs(other.position.x - me.position.x),
abs(other.position.y - me.position.y),
)
observations.append(
Observation(
tick=snapshot.tick,
observer_id=agent_id,
subject_agent_id=other.agent_id,
description=_describe(other),
location_id=other.position.location_id,
distance=distance,
)
)
observations.sort(key=lambda o: o.distance)
if observations:
logger.debug(
"Agent '%s' perceived %d nearby agent(s) at tick %d",
agent_id, len(observations), snapshot.tick,
)
return observations
def perceive_px(
snapshot: WorldSnapshot,
agent_id: str,
radius_px: float,
) -> List[Observation]:
"""
Pixel-space (Euclidean circle) perception -- everything `agent_id` can see
within `radius_px` pixels, nearest first. This is the 0-LLM sensory input
behind the 50px proximity circle. Distance is rounded pixel distance.
"""
if agent_id not in snapshot.agents:
raise KeyError(f"'{agent_id}' not present in this snapshot")
me = snapshot.get_agent(agent_id)
nearby = snapshot.agents_within_px(agent_id, radius_px)
observations: List[Observation] = []
for other in nearby:
dx = other.position.x - me.position.x
dy = other.position.y - me.position.y
dist = int(round((dx * dx + dy * dy) ** 0.5))
observations.append(
Observation(
tick=snapshot.tick,
observer_id=agent_id,
subject_agent_id=other.agent_id,
description=_describe(other),
location_id=other.position.location_id,
distance=dist,
)
)
# agents_within_px already returns nearest-first; keep that order.
if observations:
logger.debug(
"Agent '%s' perceived %d agent(s) within %spx at tick %d",
agent_id, len(observations), radius_px, snapshot.tick,
)
return observations
def _describe(agent: AgentState) -> str:
if agent.current_action is not None:
return f"{agent.agent_id} is {agent.current_action.description}"
return f"{agent.agent_id} is idle"
# Standalone sanity check
if __name__ == "__main__":
from src.core.world_state import CurrentAction, Position, WorldState
from src.core.snapshot import take_snapshot
world = WorldState()
world.register_agent("a", Position(x=0, y=0, location_id="quad"))
world.register_agent("b", Position(x=2, y=1, location_id="quad"))
world.set_agent_action(
"b", CurrentAction(description="reading a book", start_tick=0, end_tick=30)
)
snap = take_snapshot(world)
obs_a = perceive(snap, "a", radius=5)
print(obs_a)
assert len(obs_a) == 1
assert obs_a[0].subject_agent_id == "b"
assert "reading a book" in obs_a[0].description
obs_a_tight = perceive(snap, "a", radius=1)
assert len(obs_a_tight) == 0, "radius 1 should not see an agent at chebyshev distance 2"
print("perceive.py sanity check passed.") |