"""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.")