Spaces:

build-small-hackathon
/

mumbai-local

Paused

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

d4a1b20 verified 5 months ago

7.77 kB

"""
engine.py — Mumbai Local Panic (4-track edition): the canonical DATA MODEL + setup.
Defines the map, the four tracks, the train roster, the state dataclasses, the balance loader,
and new_game(). Pure data + construction only:
  - rules / legality / chaos effects live in rules.py
  - the per-turn simulation (movement, meters, win/loss) lives in simulation.py
  - the reference dispatcher lives in doctrine.py
Import-safe, headless, deterministic: same seed => same starting state.
THE LINE (8 stations, south -> north).  UP = toward Churchgate (direction -1).
DOWN = toward Virar (direction +1).  Majors are overflow-checked; interchanges are the only
places a train may cross between the slow and fast tracks (or turn around at a terminus).
"""
from __future__ import annotations
import json, os, random
from dataclasses import dataclass, field


# --------------------------------------------------------------- balance (tunable surface)
def load_balance(path=None, profile=None):
    path = path or os.path.join(os.path.dirname(__file__), "data", "balance.json")
    with open(path, encoding="utf-8") as f:
        cfg = json.load(f)
    return cfg[profile or cfg.get("_active", "canon")]


B = load_balance()

# --------------------------------------------------------------- the map
STATIONS = ["Churchgate", "Mumbai Central", "Dadar", "Bandra", "Andheri", "Borivali", "Vasai Road", "Virar"]
IDX = {s: i for i, s in enumerate(STATIONS)}
MAJORS = {"Dadar", "Andheri", "Borivali", "Churchgate"}        # overflow-checked
INTERCHANGES = set(B["interchanges"])                          # the only crossover points
BASE_INFLOW = {"Churchgate": 150, "Mumbai Central": 80, "Dadar": 180, "Bandra": 90,
               "Andheri": 160, "Borivali": 140, "Vasai Road": 70, "Virar": 50}

# --------------------------------------------------------------- the four tracks (lanes)
LANES = ("slow_up", "slow_down", "fast_up", "fast_down")
def lane_of(track, direction):      return f"{track}_{'up' if direction < 0 else 'down'}"
def opposing_lane(track, direction): return f"{track}_{'down' if direction < 0 else 'up'}"

# --------------------------------------------------------------- chaos cards + personnel
# traintrouble: a PEAK-only card (unlock R16, once) — injects the 2 reinforcement trains onto the
# line instead of them auto-spawning. Base 10, drops to 8 in peak (peak_card_cost) where it lives.
CARDS = {"monsoon_flood": 10, "signal_failure": 8, "vip_special": 12, "track_cow": 6,
         "festival_crowd": 9, "traintrouble": 10}


def card_cost(g, card):
    """A card's CURRENT chaos-energy cost. Base from CARDS, overridden by balance 'peak_card_cost'
    during PEAK (e.g. monsoon_flood drops to 8 in the rush). The single source of truth for cost —
    legal_cards / play_card / staged-apply all read this so display and charge never disagree."""
    base = CARDS[card]
    if g.phase == "peak":
        return B.get("peak_card_cost", {}).get(card, base)
    return base
INCIDENT_FIX = {"signal_failure": "signal_engineer", "monsoon_flood": "cleaner",
                "track_cow": "cow_team", "festival_crowd": "police"}   # vip_special: no unit clears it
SENSITIVE = {"ladies_special", "virar_express"}                       # holding these spikes anger


# --------------------------------------------------------------- state
@dataclass
class Station:
    name: str; cap: int; crowd: float = 0.0; anger: float = 0.0
    def pct(self): return 100.0 * self.crowd / self.cap


@dataclass
class Train:
    id: str; name: str; ttype: str; direction: int; pos: int; track: str = "slow"
    delay: int = 0; held: bool = False; stuck_turns: int = 0; just_switched: bool = False
    def stops_at(self, sname):                                   # fast trains stop at majors only
        return True if self.ttype in ("slow", "ladies", "cargo") else (sname in MAJORS)


@dataclass
class Incident:
    # track encodes the blocked lane(s): a specific lane ("slow_up") for cow/signal; a service
    # ("slow"/"fast") for flood/vip (both directions); "all" for a PEAK signal failure; "-" festival.
    id: str; itype: str; location: str; track: str; severity: int; age: int = 0
    assigned: str | None = None; arriving_in: int = 0; resolving_in: int = 0; duration: int = 3


@dataclass
class Unit:
    id: str; utype: str; loc: int; used: int = 0; busy: bool = False; assigned: str | None = None


@dataclass
class Game:
    seed: int
    turn: int = 0; phase: str = "normal"
    score: float = 0.0; delay: int = 0; safety: float = 100.0
    anger: float = 10.0; pressure: float = 0.0; patience: float = 100.0
    energy: int = 12
    stations: list = field(default_factory=list)
    trains: list = field(default_factory=list)
    incidents: list = field(default_factory=list)
    units: list = field(default_factory=list)
    inc_counter: int = 0; card_plays: dict = field(default_factory=dict); log: list = field(default_factory=list)
    over: bool = False; won: bool = False; reason: str = ""
    collisions_avoided: int = 0; crossover_blocks: int = 0          # 4-track telemetry
    consec: dict = field(default_factory=dict); played_this_turn: set = field(default_factory=set)  # chaos limiter


def lane(t):
    """The lane a train currently occupies, e.g. 'slow_up'."""
    return lane_of(t.track, t.direction)


# Trains shuttle Virar<->Churchgate. new_game spawns the first B['n_trains'] (6 default); PEAK
# adds the next B['peak_extra_trains'] (slots 7-8) as reinforcements (simulation._spawn_peak).
# ladies_special & virar_express are the anger-sensitive ones (engine.SENSITIVE). The two peak
# trains reuse the ladies/dabbawala SPRITES (via ttype) but are NOT in SENSITIVE (kept calm so
# the rush doesn't double the anger spike) — they're extra bodies for the dispatcher to juggle.
ROSTER = [
    ("slow_local",      "Slow Local",      "slow",   +1, "Churchgate",     "slow"),  # DOWN
    ("fast_842",        "8:42 Fast",       "fast",   -1, "Virar",          "fast"),  # UP
    ("ladies_special",  "Ladies Special",  "ladies", +1, "Dadar",          "slow"),  # DOWN  sensitive
    ("virar_express",   "Virar Express",   "fast",   +1, "Mumbai Central", "fast"),  # DOWN  sensitive
    ("cargo_dabbawala", "Dabbawala Cargo", "cargo",  -1, "Andheri",        "slow"),  # UP
    ("fast_borivali",   "Borivali Fast",   "fast",   -1, "Borivali",       "fast"),  # UP
    ("ladies_peak",     "Ladies Relief",   "ladies", -1, "Virar",          "slow"),  # UP   PEAK (7th)
    ("cargo_peak",      "Dabbawala Relief","cargo",  +1, "Churchgate",     "slow"),  # DOWN PEAK (8th)
]


def new_game(seed):
    """Construct a fresh, seeded game. Deterministic: same seed => identical start."""
    g = Game(seed=seed, energy=B["chaos_energy_start"], safety=B["safety_start"])
    for s in STATIONS:
        cap = B["cap_major"] if s in MAJORS else B["cap_minor"]
        g.stations.append(Station(s, cap, crowd=cap * random.Random(seed + IDX[s]).uniform(.25, .5)))
    for tid, name, ttype, d, start, track in ROSTER[:int(B.get("n_trains", 6))]:
        g.trains.append(Train(tid, name, ttype, d, IDX[start], track))
    # 8 units: two of each type (deployments are capped PER TYPE at B['personnel_max'], shared
    # across the pair — see rules.legal_actions). Second set seeded north for line coverage.
    g.units = [Unit("police_1", "police", IDX["Dadar"]),     Unit("police_2", "police", IDX["Andheri"]),
               Unit("cleaner_1", "cleaner", IDX["Andheri"]), Unit("cleaner_2", "cleaner", IDX["Borivali"]),
               Unit("cow_1", "cow_team", IDX["Borivali"]),   Unit("cow_2", "cow_team", IDX["Vasai Road"]),
               Unit("signal_1", "signal_engineer", IDX["Dadar"]), Unit("signal_2", "signal_engineer", IDX["Borivali"])]
    return g

	"""
	engine.py — Mumbai Local Panic (4-track edition): the canonical DATA MODEL + setup.

	Defines the map, the four tracks, the train roster, the state dataclasses, the balance loader,
	and new_game(). Pure data + construction only:
	- rules / legality / chaos effects live in rules.py
	- the per-turn simulation (movement, meters, win/loss) lives in simulation.py
	- the reference dispatcher lives in doctrine.py

	Import-safe, headless, deterministic: same seed => same starting state.

	THE LINE (8 stations, south -> north). UP = toward Churchgate (direction -1).
	DOWN = toward Virar (direction +1). Majors are overflow-checked; interchanges are the only
	places a train may cross between the slow and fast tracks (or turn around at a terminus).
	"""
	from __future__ import annotations
	import json, os, random
	from dataclasses import dataclass, field


	# --------------------------------------------------------------- balance (tunable surface)
	def load_balance(path=None, profile=None):
	path = path or os.path.join(os.path.dirname(__file__), "data", "balance.json")
	with open(path, encoding="utf-8") as f:
	cfg = json.load(f)
	return cfg[profile or cfg.get("_active", "canon")]


	B = load_balance()

	# --------------------------------------------------------------- the map
	STATIONS = ["Churchgate", "Mumbai Central", "Dadar", "Bandra", "Andheri", "Borivali", "Vasai Road", "Virar"]
	IDX = {s: i for i, s in enumerate(STATIONS)}
	MAJORS = {"Dadar", "Andheri", "Borivali", "Churchgate"} # overflow-checked
	INTERCHANGES = set(B["interchanges"]) # the only crossover points
	BASE_INFLOW = {"Churchgate": 150, "Mumbai Central": 80, "Dadar": 180, "Bandra": 90,
	"Andheri": 160, "Borivali": 140, "Vasai Road": 70, "Virar": 50}

	# --------------------------------------------------------------- the four tracks (lanes)
	LANES = ("slow_up", "slow_down", "fast_up", "fast_down")
	def lane_of(track, direction): return f"{track}_{'up' if direction < 0 else 'down'}"
	def opposing_lane(track, direction): return f"{track}_{'down' if direction < 0 else 'up'}"

	# --------------------------------------------------------------- chaos cards + personnel
	# traintrouble: a PEAK-only card (unlock R16, once) — injects the 2 reinforcement trains onto the
	# line instead of them auto-spawning. Base 10, drops to 8 in peak (peak_card_cost) where it lives.
	CARDS = {"monsoon_flood": 10, "signal_failure": 8, "vip_special": 12, "track_cow": 6,
	"festival_crowd": 9, "traintrouble": 10}


	def card_cost(g, card):
	"""A card's CURRENT chaos-energy cost. Base from CARDS, overridden by balance 'peak_card_cost'
	during PEAK (e.g. monsoon_flood drops to 8 in the rush). The single source of truth for cost —
	legal_cards / play_card / staged-apply all read this so display and charge never disagree."""
	base = CARDS[card]
	if g.phase == "peak":
	return B.get("peak_card_cost", {}).get(card, base)
	return base
	INCIDENT_FIX = {"signal_failure": "signal_engineer", "monsoon_flood": "cleaner",
	"track_cow": "cow_team", "festival_crowd": "police"} # vip_special: no unit clears it
	SENSITIVE = {"ladies_special", "virar_express"} # holding these spikes anger


	# --------------------------------------------------------------- state
	@dataclass
	class Station:
	name: str; cap: int; crowd: float = 0.0; anger: float = 0.0
	def pct(self): return 100.0 * self.crowd / self.cap


	@dataclass
	class Train:
	id: str; name: str; ttype: str; direction: int; pos: int; track: str = "slow"
	delay: int = 0; held: bool = False; stuck_turns: int = 0; just_switched: bool = False
	def stops_at(self, sname): # fast trains stop at majors only
	return True if self.ttype in ("slow", "ladies", "cargo") else (sname in MAJORS)


	@dataclass
	class Incident:
	# track encodes the blocked lane(s): a specific lane ("slow_up") for cow/signal; a service
	# ("slow"/"fast") for flood/vip (both directions); "all" for a PEAK signal failure; "-" festival.
	id: str; itype: str; location: str; track: str; severity: int; age: int = 0
	assigned: str \| None = None; arriving_in: int = 0; resolving_in: int = 0; duration: int = 3


	@dataclass
	class Unit:
	id: str; utype: str; loc: int; used: int = 0; busy: bool = False; assigned: str \| None = None


	@dataclass
	class Game:
	seed: int
	turn: int = 0; phase: str = "normal"
	score: float = 0.0; delay: int = 0; safety: float = 100.0
	anger: float = 10.0; pressure: float = 0.0; patience: float = 100.0
	energy: int = 12
	stations: list = field(default_factory=list)
	trains: list = field(default_factory=list)
	incidents: list = field(default_factory=list)
	units: list = field(default_factory=list)
	inc_counter: int = 0; card_plays: dict = field(default_factory=dict); log: list = field(default_factory=list)
	over: bool = False; won: bool = False; reason: str = ""
	collisions_avoided: int = 0; crossover_blocks: int = 0 # 4-track telemetry
	consec: dict = field(default_factory=dict); played_this_turn: set = field(default_factory=set) # chaos limiter


	def lane(t):
	"""The lane a train currently occupies, e.g. 'slow_up'."""
	return lane_of(t.track, t.direction)


	# Trains shuttle Virar<->Churchgate. new_game spawns the first B['n_trains'] (6 default); PEAK
	# adds the next B['peak_extra_trains'] (slots 7-8) as reinforcements (simulation._spawn_peak).
	# ladies_special & virar_express are the anger-sensitive ones (engine.SENSITIVE). The two peak
	# trains reuse the ladies/dabbawala SPRITES (via ttype) but are NOT in SENSITIVE (kept calm so
	# the rush doesn't double the anger spike) — they're extra bodies for the dispatcher to juggle.
	ROSTER = [
	("slow_local", "Slow Local", "slow", +1, "Churchgate", "slow"), # DOWN
	("fast_842", "8:42 Fast", "fast", -1, "Virar", "fast"), # UP
	("ladies_special", "Ladies Special", "ladies", +1, "Dadar", "slow"), # DOWN sensitive
	("virar_express", "Virar Express", "fast", +1, "Mumbai Central", "fast"), # DOWN sensitive
	("cargo_dabbawala", "Dabbawala Cargo", "cargo", -1, "Andheri", "slow"), # UP
	("fast_borivali", "Borivali Fast", "fast", -1, "Borivali", "fast"), # UP
	("ladies_peak", "Ladies Relief", "ladies", -1, "Virar", "slow"), # UP PEAK (7th)
	("cargo_peak", "Dabbawala Relief","cargo", +1, "Churchgate", "slow"), # DOWN PEAK (8th)
	]


	def new_game(seed):
	"""Construct a fresh, seeded game. Deterministic: same seed => identical start."""
	g = Game(seed=seed, energy=B["chaos_energy_start"], safety=B["safety_start"])
	for s in STATIONS:
	cap = B["cap_major"] if s in MAJORS else B["cap_minor"]
	g.stations.append(Station(s, cap, crowd=cap * random.Random(seed + IDX[s]).uniform(.25, .5)))
	for tid, name, ttype, d, start, track in ROSTER[:int(B.get("n_trains", 6))]:
	g.trains.append(Train(tid, name, ttype, d, IDX[start], track))
	# 8 units: two of each type (deployments are capped PER TYPE at B['personnel_max'], shared
	# across the pair — see rules.legal_actions). Second set seeded north for line coverage.
	g.units = [Unit("police_1", "police", IDX["Dadar"]), Unit("police_2", "police", IDX["Andheri"]),
	Unit("cleaner_1", "cleaner", IDX["Andheri"]), Unit("cleaner_2", "cleaner", IDX["Borivali"]),
	Unit("cow_1", "cow_team", IDX["Borivali"]), Unit("cow_2", "cow_team", IDX["Vasai Road"]),
	Unit("signal_1", "signal_engineer", IDX["Dadar"]), Unit("signal_2", "signal_engineer", IDX["Borivali"])]
	return g