| """ |
| render.py — turn engine state into the terse prompt strings the templates expect. Pure view |
| layer: reads state + the turn's legal actions, returns strings. No model calls, no rules |
| decisions. Keeps the LLM contract in one place so prompt tweaks don't leak into the engine. |
| """ |
| from __future__ import annotations |
| import os |
| from engine import STATIONS, MAJORS, INTERCHANGES, CARDS, B |
| from rules import legal_cards, card_available |
|
|
| PROMPT_DIR = os.path.join(os.path.dirname(__file__), "prompts") |
|
|
|
|
| def load_prompts(prompt_dir=PROMPT_DIR): |
| |
| disp_sys = os.environ.get("MLP_DISPATCHER_SYSTEM", "dispatcher_system.txt") |
|
|
| def rd(name): |
| path = name if os.path.isabs(name) else os.path.join(prompt_dir, name) |
| with open(path, encoding="utf-8") as f: |
| return f.read() |
|
|
| return { |
| "dispatcher_system": rd(disp_sys), |
| "dispatcher_user": rd("dispatcher_user_template.txt"), |
| "chaos_system": rd("chaos_system.txt"), |
| "chaos_user": rd("chaos_user_template.txt"), |
| } |
|
|
|
|
| def _dir(d): |
| return "DOWN" if d == +1 else "UP" |
|
|
|
|
| def _fill(template, mapping): |
| out = template |
| for k, v in mapping.items(): |
| out = out.replace("{{" + k + "}}", str(v)) |
| return "\n".join(l for l in out.splitlines() if not l.lstrip().startswith("#")) |
|
|
|
|
| |
| def describe_action(a): |
| t = a["type"] |
| if t.startswith("deploy_"): |
| return f"{t} {a['unit']} -> {a['to']} ({a['target']}, resolves in {a['resolves_in']}t)" |
| if t == "move_train": |
| return f"move_train {a['train']} -> {a['to']}" |
| if t == "hold_train": |
| return f"hold_train {a['train']} @{a['at']}" |
| if t in ("switch_to_fast", "switch_to_slow"): |
| return f"{t} {a['train']} @{a['at']}" |
| if t == "prioritize_train": |
| return f"prioritize_train {a['train']}" |
| if t == "clear_platform_priority": |
| return f"clear_platform_priority @{a['at']}" |
| if t == "issue_announcement": |
| return f"issue_announcement {a.get('scope', 'line')}-wide" |
| extra = " ".join(f"{k}={v}" for k, v in a.items() if k not in ("action_id", "type")) |
| return f"{t} {extra}".rstrip() |
|
|
|
|
| |
| def _stations_block(g): |
| out = [] |
| for s in g.stations: |
| pct = s.pct() |
| flag = "!!" if pct > 100 else "!" if pct > 85 else "" |
| tag = " [MAJOR]" if s.name in MAJORS else "" |
| out.append(f"- {s.name}{tag} {int(s.crowd)}/{s.cap} ({pct:.0f}%{flag})") |
| return "\n".join(out) |
|
|
|
|
| def _incidents_block(g): |
| if not g.incidents: |
| return "- none" |
| return "\n".join(f"- [{i.itype}] {i.location} {i.track} sev{i.severity} age{i.age}t" |
| + (" (unit en route)" if i.assigned else "") for i in g.incidents) |
|
|
|
|
| def _trains_block(g): |
| return "\n".join(f'- {t.id} "{t.name}" [{t.ttype}] @{STATIONS[t.pos]} {_dir(t.direction)} {t.track} ' |
| f'delay+{t.delay} ' + ("held" if t.held else (f"stuck{t.stuck_turns}t" if t.stuck_turns else "running")) |
| for t in g.trains) |
|
|
|
|
| def _personnel_block(g): |
| return "\n".join(f"- {u.id} {u.utype} @{STATIONS[u.loc]} {'busy' if u.busy else 'available'} " |
| f"used {u.used}/{B['personnel_max']}" for u in g.units) |
|
|
|
|
| def _legal_actions_block(A): |
| return "\n".join(f"{a['action_id']} {describe_action(a)}" for a in A) |
|
|
|
|
| def _legal_cards_block(g): |
| cards = legal_cards(g) |
| if not cards: |
| return "- (no playable cards — hoard energy)" |
| return "\n".join(f"{c['card']} cost {c['cost']}" for c in cards) |
|
|
|
|
| def _peak_banner(g): |
| if g.phase != "peak": |
| return "" |
| return (">>> PEAK RUSH (the escalator): inflow +40%, anger growth +30%, a SIGNAL FAILURE now " |
| "blocks ALL tracks, TRAIN TROUBLE card unlocks (+2 trains), monsoon flood drops to 8, " |
| "rain won't quit <<<") |
|
|
|
|
| |
| def render_dispatcher_user(template, g, A): |
| return _fill(template, { |
| "turn": g.turn + 1, "total_turns": B["turns_to_win"], "phase": g.phase, |
| "peak_rush_banner": _peak_banner(g), |
| "score": f"{g.score:.0f}", "delay": f"{g.delay:.0f}", "safety": f"{g.safety:.0f}", |
| "anger": f"{g.anger:.0f}", "pressure": f"{g.pressure:.0f}", "patience": f"{g.patience:.0f}", |
| "incidents_block": _incidents_block(g), "stations_block": _stations_block(g), |
| "trains_block": _trains_block(g), "personnel_block": _personnel_block(g), |
| "legal_actions_block": _legal_actions_block(A), |
| }) |
|
|
|
|
| def render_chaos_user(template, g): |
| return _fill(template, { |
| "turn": g.turn + 1, "total_turns": B["turns_to_win"], "phase": g.phase, |
| "peak_rush_banner": _peak_banner(g), |
| "safety": f"{g.safety:.0f}", "anger": f"{g.anger:.0f}", "pressure": f"{g.pressure:.0f}", |
| "delay": f"{g.delay:.0f}", "stations_block": _stations_block(g), |
| "incidents_block": _incidents_block(g), "energy": g.energy, |
| "regen": B["chaos_energy_regen"], "legal_cards_block": _legal_cards_block(g), |
| }) |
|
|
