from __future__ import annotations from world_simulator.domain import CitizenGoal, CitizenMode, Npc, WorldState from world_simulator.simulation.mechanics import distance_between, has_hostile_intent, is_alive from world_simulator.simulation.perception import ( CitizenPerception, build_perception, nearby_threat_npc, recent_attacker_id, ) def refresh_citizen_motivation( world: WorldState, npc: Npc, ) -> tuple[CitizenPerception, CitizenGoal | None]: perception = build_perception(world, npc) _update_needs_and_emotions(npc, perception) goal = select_goal(world, npc, perception) npc.current_goal = goal npc.mode = _mode_for_goal(goal) return perception, goal def select_goal( world: WorldState, npc: Npc, perception: CitizenPerception, ) -> CitizenGoal | None: if not is_alive(npc): return None immediate_threat = next( (threat for threat in perception.nearby_threats if threat.get("can_attack_you")), None, ) if immediate_threat is not None: return CitizenGoal( goal_type="survive", target_npc_id=_str_or_none(immediate_threat.get("npc_id")), priority=100, reason=f"{immediate_threat['name']} is close enough to attack", ) hostile_target = resolve_hostile_directive_target(world, npc) if hostile_target is not None: return CitizenGoal( goal_type="pursue_and_attack_target", target_npc_id=hostile_target.id, priority=98, reason=f"Hostile directive targets {hostile_target.name}", ) attacker_id = recent_attacker_id(world, npc) if attacker_id is not None: attacker = _npc_by_id(world, attacker_id) if attacker is not None and npc.emotions.anger > npc.emotions.fear: return CitizenGoal( goal_type="retaliate", target_npc_id=attacker.id, priority=92, reason=f"{attacker.name} attacked recently", ) if attacker is not None and _wants_to_call_for_help(npc): return CitizenGoal( goal_type="call_for_help", target_npc_id=attacker.id, priority=90, reason=f"{attacker.name} attacked recently", ) return CitizenGoal( goal_type="survive", target_npc_id=attacker_id, priority=94, reason="This citizen was attacked recently", ) visible_threat = nearby_threat_npc(world, perception) if visible_threat is not None: return CitizenGoal( goal_type="survive", target_npc_id=visible_threat.id, priority=88, reason=f"{visible_threat.name} is remembered or acting as dangerous nearby", ) ally_threat = _attacker_from_observed_attack(world, npc, perception) if ally_threat is not None: return CitizenGoal( goal_type="help_or_defend", target_npc_id=ally_threat.id, priority=82, reason=f"{ally_threat.name} attacked someone nearby", ) if npc.needs.food >= 80: return CitizenGoal( goal_type="find_food", priority=60, reason="Food need is high", ) return CitizenGoal( goal_type="routine_life", priority=20, reason="No urgent threat or directive is present", ) def resolve_hostile_directive_target(world: WorldState, npc: Npc) -> Npc | None: if not has_hostile_intent(npc): return None directive = (npc.god_directive or "").lower() explicit_target = _explicit_directive_target(world, npc, directive) if explicit_target is not None: return explicit_target if _directive_allows_any_target(directive): return _nearest_living_target(world, npc) return _nearest_living_target(world, npc) def _update_needs_and_emotions(npc: Npc, perception: CitizenPerception) -> None: if not is_alive(npc): return if perception.under_threat: npc.needs.safety = max(npc.needs.safety, 95) npc.emotions.fear = max(npc.emotions.fear, 82) npc.emotions.stress = max(npc.emotions.stress, 72) return if has_hostile_intent(npc): npc.needs.safety = max(npc.needs.safety, 35) npc.emotions.anger = max(npc.emotions.anger, 70) npc.emotions.stress = max(npc.emotions.stress, 45) return npc.needs.safety = _move_toward(npc.needs.safety, 20, step=5) npc.emotions.fear = _move_toward(npc.emotions.fear, 0, step=5) npc.emotions.anger = _move_toward(npc.emotions.anger, 0, step=3) npc.emotions.stress = _move_toward(npc.emotions.stress, 0, step=5) def _mode_for_goal(goal: CitizenGoal | None) -> CitizenMode: if goal is None: return "routine" match goal.goal_type: case "survive" | "call_for_help": return "threatened" case "pursue_and_attack_target" | "retaliate": return "hostile_pursuit" case "help_or_defend": return "helping" case "respond_to_event" | "investigate": return "investigating" case _: return "routine" def _explicit_directive_target(world: WorldState, npc: Npc, directive: str) -> Npc | None: for candidate in world.living_npcs(): if candidate.id == npc.id or not is_alive(candidate): continue if candidate.id.lower() in directive or candidate.name.lower() in directive: return candidate return None def _directive_allows_any_target(directive: str) -> bool: return any( keyword in directive for keyword in ( "everyone", "everybody", "all", "anyone", "else", "visible", "citizen", "npc", ) ) def _nearest_living_target(world: WorldState, npc: Npc) -> Npc | None: candidates = [ (distance_between(npc, candidate), candidate.id, candidate) for candidate in world.living_npcs() if candidate.id != npc.id and is_alive(candidate) ] if not candidates: return None return min(candidates, key=lambda item: (item[0], item[1]))[2] def _attacker_from_observed_attack( world: WorldState, npc: Npc, perception: CitizenPerception, ) -> Npc | None: for episode in reversed(npc.structured_memory.episodes): if ( episode.kind == "attack" and episode.subject_kind == "npc" and episode.perspective == "witness" and episode.actor_id != npc.id ): attacker = _npc_by_id(world, episode.actor_id) if attacker is not None and is_alive(attacker): return attacker for text in reversed(perception.last_hostile_actions): lowered = text.lower() if "attacked you" in lowered: continue if "attacked" not in lowered: continue for candidate in world.living_npcs(): if candidate.id == npc.id: continue if candidate.name.lower() in lowered or candidate.id.lower() in lowered: return candidate return None def _npc_by_id(world: WorldState, npc_id: str) -> Npc | None: return next((npc for npc in world.living_npcs() if npc.id == npc_id), None) def _wants_to_call_for_help(npc: Npc) -> bool: personality = npc.personality.lower() return "social" in personality or "cautious" in personality or npc.role != "guard" def _str_or_none(raw: object) -> str | None: return raw if isinstance(raw, str) else None def _move_toward(value: int, target: int, *, step: int) -> int: if value < target: return min(value + step, target) if value > target: return max(value - step, target) return value