Spaces:

build-small-hackathon
/

mumbai-local

Paused

Deploy Mumbai Local Panic — Nemotron(ZeroGPU) dispatcher + VoxCPM2 voice

d4a1b20 verified 5 months ago

8.49 kB

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

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