"""Actions — the agent's executor: turns a validated day plan into movement and activity on the map, tick by tick. Owns the AgentActionManager (last/current/next action state machine), location resolution (place id -> pixel position), route computation via pathfinder, and conversation freeze/resume. Architecture: the only module that advances agent position; called by the brain/body each tick and checkpointed whole by checkpoint_manager. Design: travel is schedule-aware (arrives by the plan's end time); the agent keeps its origin location_id while in transit. """ from __future__ import annotations import sys from pathlib import Path BACKEND_ROOT = Path(__file__).resolve().parents[2] if str(BACKEND_ROOT) not in sys.path: sys.path.append(str(BACKEND_ROOT)) import json import math import random from enum import Enum from typing import Any, Dict, List, Optional, Tuple from pydantic import BaseModel, Field from src.core.log import get_logger from src.core.world_state import Position logger = get_logger(__name__) def _interpolate_line(a: Tuple[int, int], b: Tuple[int, int], steps: int = 40) -> List[Tuple[int, int]]: """A straight line of `steps` points from a to b (inclusive). Used as a fallback route so agents still animate when no walkable path is found.""" (x0, y0), (x1, y1) = a, b steps = max(2, steps) return [ (round(x0 + (x1 - x0) * i / (steps - 1)), round(y0 + (y1 - y0) * i / (steps - 1))) for i in range(steps) ] # --------------------------------------------------------------------------- # Action types # --------------------------------------------------------------------------- class ActionType(str, Enum): MOVE = "move" MISC = "misc" CONVERSATION = "conversation" # --------------------------------------------------------------------------- # Action state # --------------------------------------------------------------------------- class ActionState(BaseModel): """One action in the agent's queue.""" action_type: ActionType description: str start_time: str # "HH:MM" end_time: str # "HH:MM" location_id: str # places.json ID sub_area: Optional[str] = None position: Optional[Position] = None # pixel coords (filled by resolver) path: Optional[List[Tuple[int, int]]] = None # if MOVE, the pixel path path_index: int = 0 # current position along path energy_change: float = 0.0 # total change over entire action emotion_change: float = 0.0 # total change over entire action energy_target: Optional[float] = None # declared cumulative energy at action end (0-1); None = delta-based is_final_plan_action: bool = False event_id: Optional[str] = None # data-driven world event, when applicable # --------------------------------------------------------------------------- # Location resolver -- converts location_id -> pixel Position # --------------------------------------------------------------------------- # places.json ID -> entrypoint.json building name _PLACES_TO_ENTRYPOINT: Dict[str, str] = { "Brahmaputra_Boys 1": "Brahmputra Boys 1", "Brahmaputra_Boys 2": "Brahmputra Boys 2", "mess": "Mess", "LHC": "Lecture Hall complex", "admin_block": "Admin Block", "sac_utility_block": "Utility/ SAC", "main_gate": "Campus Main gate", "central_lawn": "Main fest Ground", "auditorium": "Auditorium + Library", "library": "Auditorium + Library", "sports_complex": "Volleyball Court", "health_centre": "Admin Block", "Chenab": "Chenab", "Beas": "Beas", "Satluj": "Satluj", "Jhelum": "Jhelum", "Ravi": "Ravi", "SAB": "SAB", "computer_science_department": "CS Department", "electrical_department": "Electrical Department", "mechanical_department": "Mechanical Department", "chemical_department": "Chemical Department", } class LocationResolver: """Maps places.json location_id strings to pixel Position objects.""" def __init__( self, places_path: Optional[Path] = None, entrypoint_path: Optional[Path] = None, ): from src.config import ENVIRONMENT_DIR places_path = places_path or ENVIRONMENT_DIR / "places.json" entrypoint_path = entrypoint_path or ENVIRONMENT_DIR / "entrypoint.json" self._entrypoints: Dict[str, Dict[str, Any]] = {} self._places_inline: Dict[str, Tuple[int, int]] = {} self._name_to_polykey: Dict[str, str] = {} # entrypoint name -> "PolygonBuilding_N" self._bbox_by_name: Dict[str, Tuple[int, int, int, int]] = {} # name -> (xmin,ymin,xmax,ymax) # Load entrypoints if entrypoint_path.exists(): raw = json.loads(entrypoint_path.read_text()) for key, data in raw.items(): name = data.get("name", "") self._entrypoints[name] = {"x": data["x"], "y": data["y"]} self._name_to_polykey[name] = key # Load inline coordinates from places.json (3 locations have this) if places_path.exists(): raw = json.loads(places_path.read_text(encoding="utf-8")) for loc in raw.get("locations", []): inline = loc.get("Entry_point Coordinates", "") if inline and "x" in inline and "y" in inline: # Parse "x = 712, y = 414" parts = inline.split(",") x = int(parts[0].split("=")[1].strip()) y = int(parts[1].split("=")[1].strip()) self._places_inline[loc["id"]] = (x, y) # Load building bounding boxes: union all decomposed parts per polygon. bbox_path = ENVIRONMENT_DIR / "buildings_polygon_decomposed.json" polykey_bbox: Dict[str, Tuple[int, int, int, int]] = {} if bbox_path.exists(): try: parts = json.loads(bbox_path.read_text()) for p in parts: base = p.get("building_name", "").rsplit("_part", 1)[0] corners = [p.get("top_left"), p.get("top_right"), p.get("bottom_left"), p.get("bottom_right")] xs = [c[0] for c in corners if c] ys = [c[1] for c in corners if c] if not xs or not ys: continue x0, y0, x1, y1 = min(xs), min(ys), max(xs), max(ys) if base in polykey_bbox: px0, py0, px1, py1 = polykey_bbox[base] polykey_bbox[base] = (min(px0, x0), min(py0, y0), max(px1, x1), max(py1, y1)) else: polykey_bbox[base] = (x0, y0, x1, y1) except Exception as e: logger.warning("[LocationResolver] failed to load building bboxes: %s", e) # Link entrypoint name -> bbox via the shared polygon key. for name, key in self._name_to_polykey.items(): if key in polykey_bbox: self._bbox_by_name[name] = polykey_bbox[key] logger.info( "[LocationResolver] loaded %d entrypoints, %d inline coords, %d building bboxes", len(self._entrypoints), len(self._places_inline), len(self._bbox_by_name), ) def resolve(self, location_id: str) -> Optional[Position]: """Convert a places.json ID to a Position with pixel coordinates.""" if not isinstance(location_id, str) or not location_id.strip(): logger.warning("[LocationResolver] invalid empty location_id: %r", location_id) return None # Strategy 1: inline coordinates from places.json if location_id in self._places_inline: x, y = self._places_inline[location_id] return Position(x=x, y=y, location_id=location_id) # Strategy 2: explicit override dictionary entrypoint_name = _PLACES_TO_ENTRYPOINT.get(location_id) if entrypoint_name and entrypoint_name in self._entrypoints: data = self._entrypoints[entrypoint_name] return Position(x=data["x"], y=data["y"], location_id=location_id) # Strategy 3: fuzzy match (lowercase, strip underscores/spaces) normalized = location_id.lower().replace("_", "").replace(" ", "") for name, data in self._entrypoints.items(): if name.lower().replace("_", "").replace(" ", "") == normalized: return Position(x=data["x"], y=data["y"], location_id=location_id) logger.warning("[LocationResolver] could not resolve '%s'", location_id) return None def has_location(self, location_id: str) -> bool: """Check if we can resolve this ID.""" return self.resolve(location_id) is not None def _entrypoint_name_for(self, location_id: str) -> Optional[str]: """Resolve a location_id to its entrypoint building name (for bbox lookup).""" name = _PLACES_TO_ENTRYPOINT.get(location_id) if name and name in self._entrypoints: return name normalized = location_id.lower().replace("_", "").replace(" ", "") for nm in self._entrypoints: if nm.lower().replace("_", "").replace(" ", "") == normalized: return nm return None def bbox_for(self, location_id: str) -> Optional[Tuple[int, int, int, int]]: """Return (xmin, ymin, xmax, ymax) for a location's building, or None.""" name = self._entrypoint_name_for(location_id) if name and name in self._bbox_by_name: return self._bbox_by_name[name] return None def random_interior_point( self, location_id: str, occupied: Optional[List[Tuple[int, int]]] = None, min_gap: int = 6, max_tries: int = 40, ) -> Position: """A random pixel INSIDE the building's bounding box, kept at least `min_gap` px away from any `occupied` point. Falls back to a small jitter around the entrypoint when no bbox is known.""" occupied = occupied or [] bbox = self.bbox_for(location_id) ep = self.resolve(location_id) if bbox is not None: x0, y0, x1, y1 = bbox # small inset so points aren't exactly on the wall if x1 - x0 > 4: x0, x1 = x0 + 2, x1 - 2 if y1 - y0 > 4: y0, y1 = y0 + 2, y1 - 2 elif ep is not None: x0, y0, x1, y1 = ep.x - 14, ep.y - 14, ep.x + 14, ep.y + 14 else: return Position(x=0, y=0, location_id=location_id) best = None for _ in range(max_tries): x = random.randint(int(x0), int(x1)) y = random.randint(int(y0), int(y1)) if all((x - ox) ** 2 + (y - oy) ** 2 >= min_gap * min_gap for ox, oy in occupied): return Position(x=x, y=y, location_id=location_id) best = (x, y) x, y = best if best else (int((x0 + x1) / 2), int((y0 + y1) / 2)) return Position(x=x, y=y, location_id=location_id) # --------------------------------------------------------------------------- # Agent action manager -- per-agent state machine # --------------------------------------------------------------------------- class AgentActionManager: """ Manages the action lifecycle for one agent. Reads the day plan, tracks last/current/next action, handles movement between locations, and updates WorldState each tick. """ def __init__( self, agent_id: str, day_plan: List[Dict[str, Any]], initial_position: Position, resolver: LocationResolver, ): self.agent_id = agent_id self.day_plan = sorted(day_plan, key=lambda a: a.get("start", "00:00")) self.resolver = resolver self.position = initial_position self.last_action: Optional[ActionState] = None self.current_action: Optional[ActionState] = None self.next_action: Optional[ActionState] = None # Conversation support self._conversation_mode: bool = False self._pending_plan_action: Optional[ActionState] = None # Last-action detection (end-of-day transition) self._entered_last_action: bool = False @property def is_last_action(self) -> bool: """True if the current action is the final entry in the day plan. Uses description match only — start_time may differ when rescheduled after a conversation or a long move that shifted the timeline.""" return bool(self.current_action and self.current_action.is_final_plan_action) @property def is_idle(self) -> bool: """ True when the day plan is fully consumed — no action covers the current time and there are no more plan entries. """ return self.current_action is None def _hhmm_to_minutes(self, hhmm: str) -> int: """Convert 'HH:MM' to minutes since midnight.""" h, m = hhmm.split(":") return int(h) * 60 + int(m) def _minutes_to_hhmm(self, minutes: int) -> str: """Convert minutes since midnight to 'HH:MM'.""" h, m = divmod(minutes, 60) return f"{h:02d}:{m:02d}" def _find_current_plan_action(self, hhmm: str) -> Optional[Dict[str, Any]]: """Find the day plan action that covers this time.""" minute = self._hhmm_to_minutes(hhmm) for action in self.day_plan: start = self._hhmm_to_minutes(action.get("start", "00:00")) end = self._hhmm_to_minutes(action.get("end", "24:00")) if start <= minute < end: return action return None def _create_action_state(self, plan_action: Dict[str, Any]) -> ActionState: """Create an ActionState from a day plan entry.""" location_id = plan_action.get("location_id", "") position = self.resolver.resolve(location_id) return ActionState( action_type=ActionType.MISC, description=plan_action.get("action", "unknown"), start_time=plan_action.get("start", "00:00"), end_time=plan_action.get("end", "24:00"), location_id=location_id, sub_area=plan_action.get("sub_area"), position=position, energy_change=plan_action.get("energy_change", 0.0), emotion_change=plan_action.get("emotion_change", 0.0), energy_target=plan_action.get("energy_target"), is_final_plan_action=bool(self.day_plan and plan_action is self.day_plan[-1]), event_id=plan_action.get("world_event_id"), ) def _create_move_action( self, from_pos: Position, to_pos: Position, start_time: str, end_time: str, ) -> ActionState: """Create a MOVE action that walks the agent DOOR to DOOR. The route is: current spot -> current building's entry door -> (pathfinder route along the walkable grid) -> destination building's entry door. Arrival then settles the agent inside the destination (handled in tick()). If the pathfinder can't produce a route we fall back to a densely interpolated straight line so motion is still visible. """ # Resolve the two front doors (walkable entrypoints). from_door = self.resolver.resolve(from_pos.location_id) if from_pos.location_id else None to_door = self.resolver.resolve(to_pos.location_id) if to_pos.location_id else None d_from = (from_door.x, from_door.y) if from_door else (from_pos.x, from_pos.y) d_to = (to_door.x, to_door.y) if to_door else (to_pos.x, to_pos.y) road: List[Tuple[int, int]] = [] try: from pathfinder import shortest_path road = shortest_path(d_from, d_to) or [] except Exception as e: logger.warning("[Actions] pathfinder failed for %s: %s", self.agent_id, e) road = [] if len(road) < 2: road = _interpolate_line(d_from, d_to, steps=40) # Prepend the agent's current spot so it visibly steps out to its door. path: List[Tuple[int, int]] = [(from_pos.x, from_pos.y)] + [p for p in road] return ActionState( action_type=ActionType.MOVE, description=f"Walk from {from_pos.location_id or 'here'} to {to_pos.location_id or 'there'}", start_time=start_time, end_time=end_time, location_id=to_pos.location_id or "", position=to_pos, path=path, path_index=0, ) def _advance_to_next_plan_action(self, current_hhmm: str) -> None: """Pick the next action from the day plan and set up current/next.""" plan_action = self._find_current_plan_action(current_hhmm) if plan_action is None: # This should be unreachable for validated plans. It can still # occur when resuming a checkpoint created before stricter replan # validation existed, or after a damaged external plan edit. Keep # the agent embodied instead of surfacing a misleading permanent # ``Idle`` state while the next calendar-day plan is prepared. now = self._hhmm_to_minutes(current_hhmm) later_starts = [ self._hhmm_to_minutes(action.get("start", "24:00")) for action in self.day_plan if self._hhmm_to_minutes(action.get("start", "24:00")) > now ] end_time = self._minutes_to_hhmm(min(later_starts) if later_starts else 24 * 60) self.current_action = ActionState( action_type=ActionType.MISC, description="Unscheduled downtime", start_time=current_hhmm, end_time=end_time, location_id=self.position.location_id or "", position=self.position, ) logger.warning( "[Actions] %s: no plan action at %s; using bounded recovery activity until %s", self.agent_id, current_hhmm, end_time, ) return target_location_id = plan_action.get("location_id") if not isinstance(target_location_id, str) or not target_location_id.strip(): logger.error( "[Actions] %s: plan action '%s' has no valid location; using bounded recovery activity", self.agent_id, plan_action.get("action", "unknown"), ) now = self._hhmm_to_minutes(current_hhmm) end_time = plan_action.get("end", self._minutes_to_hhmm(min(now + 30, 24 * 60))) self.current_action = ActionState( action_type=ActionType.MISC, description="Recovering from invalid planned location", start_time=current_hhmm, end_time=end_time, location_id=self.position.location_id or "", position=self.position, ) return target_position = self.resolver.resolve(target_location_id) if target_position is None: logger.warning( "[Actions] %s: cannot resolve location '%s' — skipping action", self.agent_id, target_location_id, ) return # Check if agent is already at the target location already_at_location = ( self.position.location_id == target_location_id or ( self.position.x == target_position.x and self.position.y == target_position.y ) ) if already_at_location: # Agent is already here — start the activity immediately self.current_action = self._create_action_state(plan_action) self.current_action.start_time = current_hhmm else: # Agent needs to move first — create move action with BFS path, # then compute travel time from the schedule's allocated duration # so the agent arrives exactly when the plan expects. self.current_action = self._create_move_action( self.position, target_position, current_hhmm, "23:59" ) path = self.current_action.path path_len = len(path) if path else 100 # Schedule-aware speed: arrive by the plan's end time. plan_end_min = self._hhmm_to_minutes(plan_action.get("end", "24:00")) start_minute = self._hhmm_to_minutes(current_hhmm) activity_minutes = max(1, plan_end_min - start_minute) min_on_site_minutes = min(10, max(1, activity_minutes // 2)) max_travel_minutes = max(1, activity_minutes - min_on_site_minutes) natural_travel_minutes = max(1, math.ceil(path_len / 50)) # Transit must leave time for the scheduled on-site activity. travel_minutes = min(natural_travel_minutes, max_travel_minutes) end_minute = start_minute + travel_minutes self.current_action.end_time = self._minutes_to_hhmm(min(end_minute, 24 * 60)) # The actual activity becomes the next action self.next_action = self._create_action_state(plan_action) def set_conversation_action(self, other_agent_id: str, world_tick: Optional[int] = None) -> ActionState: """ Override the current action with a conversation (no pre-set duration). Agent stays in conversation mode until resume_from_conversation() is called by the background LLM task. """ hhmm = self._minutes_to_hhmm((world_tick or 0) % (24 * 60)) if self.current_action: self._pending_plan_action = self.current_action conv_action = ActionState( action_type=ActionType.CONVERSATION, description=f"Chatting with {other_agent_id}", start_time=hhmm, end_time="23:59", # placeholder — overridden by resume_from_conversation location_id=self.position.location_id or "", position=self.position, ) self.current_action = conv_action self._conversation_mode = True return conv_action def resume_from_conversation( self, new_day_plan: Optional[List[Dict[str, Any]]] = None, world_tick: Optional[int] = None, ) -> None: """ Leave conversation mode and resume the interrupted action when the plan has not changed. A paused route resumes from its actual point. """ if new_day_plan is not None: self.day_plan = sorted(new_day_plan, key=lambda a: a.get("start", "00:00")) pending = self._pending_plan_action self._conversation_mode = False self._pending_plan_action = None self._entered_last_action = False if pending is None or new_day_plan is not None: self.current_action = None self.next_action = None return if pending.action_type == ActionType.MOVE and pending.path: old_path = pending.path old_index = min(pending.path_index, len(old_path) - 1) fraction_done = old_index / max(1, len(old_path) - 1) pending.path = [(self.position.x, self.position.y)] + old_path[old_index + 1:] pending.path_index = 0 if world_tick is not None: duration = max(1, self._hhmm_to_minutes(pending.end_time) - self._hhmm_to_minutes(pending.start_time)) remaining = max(1, math.ceil(duration * (1.0 - fraction_done))) now = world_tick % (24 * 60) pending.start_time = self._minutes_to_hhmm(now) pending.end_time = self._minutes_to_hhmm(min(24 * 60, now + remaining)) if len(pending.path) < 2: # The conversation happened at the final waypoint. Do not # leave a one-point MOVE that cannot interpolate; commit the # arrival and continue its queued on-site activity instead. if pending.location_id: self.position = self.resolver.random_interior_point(pending.location_id) if self.next_action is not None: self.current_action = self.next_action self.next_action = None else: self.current_action = None return self.current_action = pending def replace_day_plan(self, new_day_plan: List[Dict[str, Any]]) -> None: """ Replace the day plan mid-stream (e.g. at end-of-day transition). Keeps the current action if it's still valid, otherwise advances. """ self.day_plan = sorted(new_day_plan, key=lambda a: a.get("start", "00:00")) self._pending_plan_action = None self._entered_last_action = False def begin_new_day(self, new_day_plan: List[Dict[str, Any]]) -> None: """Install a new calendar day's plan without moving the agent. Position is intentionally retained: the first action of the new plan must account for where the previous day actually ended. Schedule bookkeeping is reset so a previous day's ``24:00`` action cannot wrap around and continue into the next day. """ self.day_plan = sorted(new_day_plan, key=lambda a: a.get("start", "00:00")) self.last_action = self.current_action self.current_action = None self.next_action = None self._pending_plan_action = None self._conversation_mode = False self._entered_last_action = False def tick(self, world_tick: int, snapshot: Any = None) -> Optional[ActionState]: """ Advance one tick. Returns the current action state (or None). This is the main entry point called each simulation tick. """ hhmm = self._minutes_to_hhmm(world_tick % (24 * 60)) # In conversation mode: don't advance, just return current action. # The agent stays frozen until resume_from_conversation() is called. if self._conversation_mode: return self.current_action # First tick or after action completed — pick next action if self.current_action is None: self._advance_to_next_plan_action(hhmm) if self.current_action is None: return None # Check if current action is finished current_end = self._hhmm_to_minutes(self.current_action.end_time) current_minute = self._hhmm_to_minutes(hhmm) # Let a MOVE render its terminal path coordinate for one tick before # committing the arrival and changing its action label. is_finished = ( current_minute > current_end if self.current_action.action_type == ActionType.MOVE else current_minute >= current_end ) if is_finished: # Action finished — advance was_last = self.is_last_action self.last_action = self.current_action self.current_action = None # If we were moving and have a next action, start it if self.last_action.action_type == ActionType.MOVE and self.next_action is not None: self.current_action = self.next_action self.next_action = None # Arrived at the destination building — settle at a random spot # INSIDE its footprint (not stacked on the entrypoint). if self.current_action.location_id: self.position = self.resolver.random_interior_point( self.current_action.location_id ) else: self.position = self.current_action.position or self.position else: # Movement completed — settle inside the destination building if self.last_action.action_type == ActionType.MOVE and self.last_action.location_id: self.position = self.resolver.random_interior_point( self.last_action.location_id ) elif self.last_action.action_type == ActionType.MOVE and self.last_action.position: self.position = self.last_action.position # Pick next action from plan self._advance_to_next_plan_action(hhmm) # If this was the last action and there's a next action (from the plan), # it's not really the last — clear the flag if was_last and self.current_action is not None: self._entered_last_action = False # Detect first tick of the last action if self.current_action and self.is_last_action and not self._entered_last_action: self._entered_last_action = True # If still moving, advance along the path proportionally to elapsed # time, so the agent visibly walks the whole route (start -> destination) # over the move's duration instead of jumping. if ( self.current_action is not None and self.current_action.action_type == ActionType.MOVE and self.current_action.path ): path = self.current_action.path total = len(path) if total >= 2: start_min = self._hhmm_to_minutes(self.current_action.start_time) end_min = self._hhmm_to_minutes(self.current_action.end_time) dur = max(1, end_min - start_min) progress = (current_minute - start_min) / dur progress = min(1.0, max(0.0, progress)) # Fractional index so we move continuously *between* path points, # not just land on them — smooth even for short/sparse routes. fidx = progress * (total - 1) i = int(fidx) if i >= total - 1: nx, ny = path[-1] else: frac = fidx - i (ax, ay), (bx, by) = path[i], path[i + 1] nx = round(ax + (bx - ax) * frac) ny = round(ay + (by - ay) * frac) # A route has a destination, but the agent is not semantically # *at* that destination until arrival. Retaining the current # location prevents UI, replanning, and conversations from # treating an in-transit agent as already there. self.position = Position(x=nx, y=ny, location_id=self.position.location_id) self.current_action.path_index = i return self.current_action def get_state(self) -> Dict[str, Any]: """Return current state as a dict (for frontend/API).""" return { "agent_id": self.agent_id, "position": self.position.model_dump(), "last_action": self.last_action.model_dump() if self.last_action else None, "current_action": self.current_action.model_dump() if self.current_action else None, "next_action": self.next_action.model_dump() if self.next_action else None, } # --------------------------------------------------------------------------- # Self-test # --------------------------------------------------------------------------- if __name__ == "__main__": import argparse from src.core.log import setup_logging setup_logging(run_id="actions_test", console=True) from src.config import PERSONALITIES_DIR parser = argparse.ArgumentParser(description="Actions module self-test — simulate an agent's day") parser.add_argument( "persona", nargs="?", default="parv_singla", help="Persona name (e.g. parv_singla, tanishq, gurnoor_singh) or path to a persona JSON", ) parser.add_argument( "-t", "--ticks", type=int, default=600, help="Number of ticks to simulate (default: 600 = 10 hours)", ) parser.add_argument( "--step", type=int, default=5, help="Print status every N ticks (default: 5)", ) args = parser.parse_args() # Resolve persona path candidate = Path(args.persona) if candidate.exists(): persona_path = candidate elif candidate.suffix == ".json": matches = sorted(PERSONALITIES_DIR.glob(f"**/{candidate.name}")) persona_path = matches[0] if matches else candidate else: matches = sorted(PERSONALITIES_DIR.glob(f"**/{args.persona}/{args.persona}.json")) if not matches: matches = sorted(PERSONALITIES_DIR.glob(f"**/{args.persona}.json")) if not matches: available = sorted({p.parent.name for p in PERSONALITIES_DIR.glob("**/*.json")}) print(f"Persona '{args.persona}' not found. Available: {', '.join(available)}") sys.exit(1) persona_path = matches[0] persona = json.loads(persona_path.read_text()) persona_name = persona.get("Name", persona_path.stem) # Load day plan from Short_term from src.agents.Short_term import load_day_plan, date_from_simulation_time sim_date = date_from_simulation_time("2026-07-03 00:00") day_plan = load_day_plan(persona_name, sim_date) if not day_plan: print(f"No day plan found for {persona_name} on {sim_date}") print("Generate one first: python backend/src/core/Agent.py " + args.persona) sys.exit(1) print(f"Loaded day plan for {persona_name}: {len(day_plan)} actions") # Set up resolver and manager resolver = LocationResolver() initial_pos = resolver.resolve("Brahmaputra_Boys 1") or Position(x=185, y=547, location_id="Brahmaputra_Boys 1") manager = AgentActionManager( agent_id=persona_path.stem, day_plan=day_plan, initial_position=initial_pos, resolver=resolver, ) # Run simulation print(f"\nSimulating {args.ticks} ticks (step={args.step}) from 00:00...") print(f"{'Tick':>5} {'Time':>6} {'Type':<6} {'Description':<45} {'Location':<25} {'Position'}") print("-" * 110) for tick in range(0, args.ticks): hhmm = manager._minutes_to_hhmm(tick) action = manager.tick(tick) if tick % args.step == 0 and action: desc = action.description[:43] loc = action.location_id[:23] if action.location_id else "" pos = f"({manager.position.x},{manager.position.y})" print(f"{tick:>5} {hhmm:>6} {action.action_type.value:<6} {desc:<45} {loc:<25} {pos}") print(f"\nFinal state:") state = manager.get_state() print(f" Position: ({state['position']['x']}, {state['position']['y']}) at {state['position']['location_id']}") print(f" Current action: {state['current_action']['description'] if state['current_action'] else 'None'}") print("\nActions.py self-test passed.")