"""CustomerDirector: spawn + in-shop AI.""" from __future__ import annotations import random from typing import Dict, List, Optional from game.sim.attract import ( compute_attract, displayed_tags_from_state, item_match, ) from game.sim.browse_ai import pick_display_target from game.sim.spawn_table import spawn_budget, p_regular_attempt from game.sim.accessibility import consider_duration as a11y_consider from game.sim.bark import BarkSelector from game.sim.customer import AgentState, CustomerAgent from game.sim.day_cycle import DayCycle, Phase class CustomerDirector: def __init__(self, db, economy, atmosphere, relationships, rng_seed: int = 0): self.db = db self.economy = economy self.atmosphere = atmosphere self.relationships = relationships self.rng = random.Random(rng_seed) self.bark = BarkSelector(self.rng) self.agents: List[CustomerAgent] = [] self._spawn_plan: List[float] = [] # elapsed times within phase self._phase_regulars_entered = set() self._spawned_this_phase = 0 self._agent_seq = 0 self._last_phase = None self.reduce_motion = False self.stats = { "enter_attempts": 0, "enters": 0, "regular_enters": {}, "by_phase": {}, "sales": 0, "barks": 0, "exits_patience": 0, } def on_phase(self, day: DayCycle) -> None: self._last_phase = day.phase self._phase_regulars_entered.clear() self._spawned_this_phase = 0 self._spawn_plan.clear() phase = day.phase.value self.bark.reset_phase(phase) budget = spawn_budget(phase, self.db.rules.get("spawn_budget")) limit = day.phase_limit() if not budget or not limit: return # poisson-ish: round budget with jitter count n = max(0, int(self.rng.gauss(budget, 0.4))) n = max(0, min(n, int(budget) + 2)) if budget > 0 and n == 0 and self.rng.random() < budget: n = 1 for i in range(n): # uniform-ish with ±20% on spacing t = (limit / max(1, n)) * (i + 0.5) t *= self.rng.uniform(0.8, 1.2) self._spawn_plan.append(min(limit * 0.95, max(0.1, t))) self._spawn_plan.sort() def tick(self, delta: float, day: DayCycle, state: dict) -> None: if self._last_phase != day.phase: self.on_phase(day) if day.paused: return phase = day.phase.value # spawn due due = [t for t in self._spawn_plan if t <= day.phase_elapsed] self._spawn_plan = [t for t in self._spawn_plan if t > day.phase_elapsed] for _ in due: self.try_spawn(day, state) # AI pace = day.pace_multiplier buy_th = float(self.db.rules.get("item_match_buy_threshold", 0.5)) consider_base = float(self.db.rules.get("consider_duration_base", 4.0)) consider_dur_default = a11y_consider(consider_base, pace, self.reduce_motion) for agent in list(self.agents): if agent.state == AgentState.EXIT: self.agents.remove(agent) continue if agent.state == AgentState.SPAWNING: agent.start(float(self.db.rules.get("patience_max_base", 60.0)) * pace) if agent.drain_patience(delta): self.stats["exits_patience"] += 1 # 30% polite leave bark if self.rng.random() < 0.3: self._maybe_bark(agent, phase) continue agent.tick_counter += 1 if agent.state == AgentState.BROWSE: agent.browse_ticks += 1 if agent.tick_counter % 8 == 0: agent.advance_path() if agent.browse_ticks >= 12: agent.state = AgentState.LINGER if agent.tick_counter % 4 == 0: self._try_consider(agent, buy_th, consider_dur_default) elif agent.state == AgentState.LINGER: agent.linger_ticks += 1 if agent.tick_counter % 4 == 0: self._try_consider(agent, buy_th, consider_dur_default) if agent.linger_ticks >= 40 and agent.state == AgentState.LINGER: # no match found — leave agent.state = AgentState.EXIT elif agent.state == AgentState.CONSIDER: agent.consider_left -= delta if agent.consider_left <= 0: agent.state = AgentState.QUEUE agent.path = ["counter"] agent.path_index = 0 elif agent.state == AgentState.QUEUE: # single counter: only first in QUEUE/PAY advances if self._is_front(agent): agent.state = AgentState.PAY elif agent.state == AgentState.PAY: self._pay(agent, day, state) agent.state = AgentState.EXIT elif agent.state == AgentState.TALK: pass # frozen until end_talk # refresh atmosphere crowding self.atmosphere.recompute( state, day.phase, n_customers=len(self.agents), max_customers=int(self.db.rules.get("max_simultaneous_customers", 5)), products=self.db.products, furniture=self.db.furniture, ) self.economy.sync_to_state(state) def interact(self, agent_id: str, phase: str) -> Optional[str]: """Player click → bark text (Deep Chat handled separately).""" for agent in self.agents: if agent.agent_id == agent_id: agent.begin_talk() text = self._maybe_bark(agent, phase) agent.end_talk() return text return None def _maybe_bark(self, agent: CustomerAgent, phase: str) -> Optional[str]: npc = self.db.npcs.get(agent.npc_id) or {} text = self.bark.pick(agent.npc_id, phase, list(npc.get("barks") or [])) if text: agent.last_bark = text self.stats["barks"] += 1 return text def _is_front(self, agent: CustomerAgent) -> bool: waiting = [a for a in self.agents if a.state in (AgentState.QUEUE, AgentState.PAY)] return waiting and waiting[0] is agent def _try_consider(self, agent: CustomerAgent, buy_th: float, consider_dur: float) -> None: best = pick_display_target(agent.desired_tags, self.economy.displays, self.db.products, threshold=buy_th) if best: agent.target_slot, agent.target_product, agent.consider_match = best agent.consider_left = consider_dur agent.state = AgentState.CONSIDER def _pay(self, agent: CustomerAgent, day: DayCycle, state: dict) -> None: if not agent.target_slot: return try: sale = self.economy.sell_from_display(agent.target_slot) except ValueError: return self.stats["sales"] = int(self.stats.get("sales", 0)) + 1 if agent.kind == "regular": self.relationships.apply_purchase(agent.npc_id, agent.consider_match, day.day_index) self.relationships.sync_to_state(state) def try_spawn(self, day: DayCycle, state: dict) -> Optional[CustomerAgent]: max_c = int(self.db.rules.get("max_simultaneous_customers", 5)) if len(self.agents) >= max_c: return None phase = day.phase.value self.stats["enter_attempts"] += 1 candidate = self._pick_candidate(phase) if not candidate: return None npc_id, kind, pref, tags, stage = candidate disp_tags = displayed_tags_from_state(state, self.db.products) attract = compute_attract( preferred_atmosphere=pref, shop_atmosphere={k: v for k, v in self.atmosphere.snapshot.items() if k != "crowding"}, desired_tags=tags, displayed_tags=disp_tags, has_free_browse_spot=True, current_customers=len(self.agents), stage=stage if kind == "regular" else "STRANGER", weights=self.db.rules.get("weights"), rel_table=self.db.rules.get("relationship_pull"), max_customers=max_c, ) thr = float(self.db.rules.get("enter_threshold", 0.45)) if attract < thr: return None self._agent_seq += 1 agent = CustomerAgent( agent_id=f"a{self._agent_seq}", npc_id=npc_id, kind=kind, preferred_atmosphere=pref, desired_tags=list(tags), stage=stage, entered_attract=attract, ) self.agents.append(agent) self.stats["enters"] += 1 self.stats["by_phase"].setdefault(phase, 0) self.stats["by_phase"][phase] += 1 if kind == "regular": self._phase_regulars_entered.add(npc_id) self.stats["regular_enters"].setdefault(phase, 0) self.stats["regular_enters"][phase] += 1 self.relationships.apply_visit(npc_id, day.day_index) self.relationships.sync_to_state(state) return agent def _pick_candidate(self, phase: str): p_reg = float(self.db.rules.get("p_regular_attempt", {}).get(phase, 0.0)) regulars = [] for nid, npc in self.db.npcs.items(): if npc.get("type") != "regular": continue if nid in self._phase_regulars_entered: continue w = float((npc.get("schedule") or {}).get(phase, 0) or 0) if w > 0: regulars.append((nid, w, npc)) use_regular = self.rng.random() < p_reg and regulars if use_regular: total = sum(w for _, w, _ in regulars) r = self.rng.random() * total acc = 0.0 chosen = regulars[-1] for item in regulars: acc += item[1] if r <= acc: chosen = item break nid, _, npc = chosen rel = self.relationships.ensure(nid) return nid, "regular", npc.get("preferred_atmosphere") or {}, npc.get("desired_tags") or [], rel["stage"] # passerby passers = [n for n in self.db.npcs.values() if n.get("type") == "passerby"] if not passers: return None npc = self.rng.choice(passers) return npc["id"], "passerby", npc.get("preferred_atmosphere") or {}, npc.get("desired_tags") or [], "STRANGER"