""" simulation.py — the deterministic per-turn SIMULATION. apply(g, A, chosen) : apply the dispatcher's chosen action ids -> (announced, police_at) advance(g, ...) : advance the world one round — crowd inflow, train movement with 4-track collision avoidance, incident aging, meters, and win/loss. No model/IO calls (golden rule). Same seed + same inputs => same trace. COLLISION AVOIDANCE (resolved here, deterministically): * Headway — a train cannot enter a (station, lane) block held by another train on the same lane; it HOLDS instead of colliding (counted as collisions_avoided). * Crossover — a switch_to_* (set in apply) consumes the round: the switching train does NOT advance. It completes (stuck reset) unless the crossed/target lane is occupied at the interchange, in which case it's refused and HOLDS (counted as crossover_blocks). """ from __future__ import annotations import statistics from engine import (B, STATIONS, IDX, MAJORS, BASE_INFLOW, SENSITIVE, lane, opposing_lane) from rules import _blocked, dist def apply(g, A, chosen): """Apply chosen action ids to the state. Returns (announced, police_at).""" by = {a["action_id"]: a for a in A} announced = False; police_at = set() for cid in chosen: a = by[cid]; tp = a["type"] if tp.startswith("deploy_"): u = next(x for x in g.units if x.id == a["unit"]) inc = next(i for i in g.incidents if i.id == a["target"]) u.busy = True; u.used += 1; u.assigned = inc.id; inc.assigned = u.id inc.arriving_in = dist(u.loc, IDX[inc.location]); inc.resolving_in = B["personnel_resolve_turns"] if u.utype == "police": police_at.add(inc.location) elif tp == "move_train": next(x for x in g.trains if x.id == a["train"]).held = False elif tp == "hold_train": next(x for x in g.trains if x.id == a["train"]).held = True elif tp in ("switch_to_fast", "switch_to_slow"): t = next(x for x in g.trains if x.id == a["train"]) t.track = "fast" if tp.endswith("fast") else "slow" t.just_switched = True; t.held = False elif tp == "issue_announcement": announced = True elif tp == "clear_platform_priority": s = next(x for x in g.stations if x.name == a["at"]) s.crowd = max(0, s.crowd - B["clear_platform_amount"]); g.score += 2 return announced, police_at def _discharge(g, t): """A moving train discharges passengers at the stop it is leaving (halved if that station is flooded / festival-crowded).""" sname = STATIONS[t.pos] if not t.stops_at(sname): return thru = B["outflow_per_stop"] for inc in g.incidents: if inc.location == sname and inc.itype in ("monsoon_flood", "festival_crowd"): thru *= B["flood_throughput_mult"] st = g.stations[t.pos] disch = min(thru, st.crowd) st.crowd = max(0, st.crowd - thru) g.score += disch * 0.01 def advance(g, announced, police_at): g.turn += 1 # PEAK begins when the upcoming round (g.turn + 1, what the HUD shows) reaches peak_rush_turn, # so phase / banner / modifiers / persistent rain all switch on together at round 16. The 2 # reinforcement trains are no longer auto-spawned — they arrive only if the human plays the # peak-only TRAIN TROUBLE card (rules.apply_chaos). if g.phase != "peak" and g.turn + 1 >= B["peak_rush_turn"]: g.phase = "peak" im = B["inflow_peak_mult"] if g.phase == "peak" else 1.0 for s in g.stations: s.crowd += BASE_INFLOW[s.name] * im last = len(STATIONS) - 1 cd = B["collision_delay"] # 1) classify: crossovers (consume the round), incident/held holders, and movers holders, movers = [], [] for t in g.trains: if t.just_switched: here = {lane(o) for o in g.trains if o.pos == t.pos and o is not t} opp = opposing_lane(t.track, t.direction) unsafe = B["crossover_unsafe_holds"] and (opp in here or lane(t) in here) if unsafe: t.stuck_turns += 1; t.delay += cd; g.crossover_blocks += 1 else: t.stuck_turns = 0; t.delay += 1 holders.append(t) continue if t.held or _blocked(g, t): t.stuck_turns += 1; t.delay += cd holders.append(t) else: movers.append(t) # 2) resolve movers lead-first per lane (a lead train vacates its block for the follower) occ = {(t.pos, lane(t)) for t in holders} for t in sorted(movers, key=lambda t: (lane(t), -t.pos if t.direction > 0 else t.pos)): np_ = max(0, min(last, t.pos + t.direction)) if np_ != t.pos and (np_, lane(t)) in occ: # train ahead on same lane -> avoid collision t.stuck_turns += 1; t.delay += cd; g.collisions_avoided += 1 occ.add((t.pos, lane(t))) continue t.stuck_turns = 0 _discharge(g, t) t.pos = np_ if t.pos in (0, last): t.direction *= -1 # turnaround at the terminus occ.add((t.pos, lane(t))) for t in g.trains: t.just_switched = False # 3) incidents age / resolve (personnel resolves faster + safer; everything auto-expires) for inc in list(g.incidents): inc.age += 1 if inc.assigned: if inc.arriving_in > 0: inc.arriving_in -= 1 elif inc.resolving_in > 0: inc.resolving_in -= 1 if inc.resolving_in == 0: u = next(x for x in g.units if x.id == inc.assigned) u.busy = False; u.loc = IDX[inc.location]; u.assigned = None g.incidents.remove(inc); g.score += 8 g.safety = min(100, g.safety + B["safety_resolve_bonus"]); continue if inc.age >= inc.duration: if inc.assigned: u = next((x for x in g.units if x.id == inc.assigned), None) if u: u.busy = False; u.loc = IDX[inc.location]; u.assigned = None g.incidents.remove(inc); continue # 4) meters am = B["anger_peak_mult"] if g.phase == "peak" else 1.0 da = 0.0 for s in g.stations: if s.pct() > 85: da += (s.pct() - 85) * B["anger_over85_k"] * am elif s.pct() < 70: da -= B["anger_calm_below70"] da += sum(B["anger_held_sensitive_k"] for t in g.trains if t.id in SENSITIVE and (t.held or t.stuck_turns > 0)) * am da += sum(B["anger_festival_k"] for i in g.incidents if i.itype == "festival_crowd") * am if announced: da -= B["anger_announce_relief"] da -= B["anger_police_relief"] * len(police_at) da -= B["anger_passive_decay"] g.anger = max(0, min(100, g.anger + da)) g.safety = max(0, min(100, g.safety - sum(i.severity for i in g.incidents) * B["safety_incident_k"] + B["safety_recover"])) g.delay = sum(t.delay for t in g.trains) majpct = [s.pct() for s in g.stations if s.name in MAJORS] g.pressure = max(0, min(100, statistics.mean(majpct) * 0.5 + g.anger * 0.3 + len(g.incidents) * 5)) g.energy = min(40, g.energy + B["chaos_energy_regen"]) # 5) loss / win if g.anger >= 100: g.over, g.reason = True, "city anger collapse" elif g.safety < 20: g.over, g.reason = True, "safety collapse" elif sum(1 for t in g.trains if t.stuck_turns >= 3) >= 4: g.over, g.reason = True, "network lock" elif sum(1 for s in g.stations if s.name in MAJORS and s.pct() > 130) >= 2: g.over, g.reason = True, "double overflow" elif any(s.pct() > B.get("single_overflow_pct", 1e9) for s in g.stations): crushed = max(g.stations, key=lambda s: s.pct()) g.over, g.reason = True, f"platform crush @ {crushed.name} ({crushed.pct():.0f}%)" elif g.turn >= B["turns_to_win"]: g.over, g.won, g.reason = True, True, f"survived to turn {B['turns_to_win']}" # 6) chaos-limiter bookkeeping: a capped card NOT played this round resets its consecutive # counter (so cow/signal allow 2 in a row, then must rest 1 round before they're legal again). for c in B.get("consecutive_cap", {}): if c not in g.played_this_turn: g.consec[c] = 0 g.played_this_turn = set()