""" DISASTER GRID — Autonomous AI Emergency Manager Streamlit dashboard — production-ready for Hugging Face Spaces """ import streamlit as st import random import time # ── Page config (must be the very first Streamlit call) ────────────────────── st.set_page_config( page_title="DISASTER GRID — Autonomous AI Emergency Manager", page_icon="⚡", layout="wide", initial_sidebar_state="collapsed", ) # ── Try real backend; graceful stub fallback ────────────────────────────────── try: from src.disaster_grid.environment import CityGrid from src.disaster_grid.models import ActionType REAL_ENV = True except ImportError: REAL_ENV = False if not REAL_ENV: class ActionType: MOVE_N = "MOVE_N" MOVE_S = "MOVE_S" MOVE_E = "MOVE_E" MOVE_W = "MOVE_W" REPAIR = "REPAIR" RECHARGE = "RECHARGE" WAIT = "WAIT" class CityGrid: def __init__(self): self.grid_health = [100] * 25 self.agent_pos = 0 self.agent_energy = 100 def reset(self, options=None): self.grid_health = [random.randint(60, 100) for _ in range(25)] self.grid_health[0] = 100 self.agent_pos = 0 self.agent_energy = 100 if options and "target_crises" in options: for idx in options["target_crises"]: if 0 <= idx < 25: self.grid_health[idx] = random.randint(5, 25) def step(self, action: dict): act = action.get("action", "WAIT") reward = -0.5 # entropy: one random sector degrades decay_idx = random.randint(1, 24) self.grid_health[decay_idx] = max( 0, self.grid_health[decay_idx] - random.randint(3, 8) ) self.agent_energy = max(0, self.agent_energy - 2) row, col = divmod(self.agent_pos, 5) if act == "MOVE_N" and row > 0: self.agent_pos -= 5 elif act == "MOVE_S" and row < 4: self.agent_pos += 5 elif act == "MOVE_W" and col > 0: self.agent_pos -= 1 elif act == "MOVE_E" and col < 4: self.agent_pos += 1 elif act == "REPAIR": if self.grid_health[self.agent_pos] < 30 and self.agent_energy >= 10: self.grid_health[self.agent_pos] = min( 100, self.grid_health[self.agent_pos] + 30 ) self.agent_energy -= 10 reward += 20 elif act == "RECHARGE": if self.agent_pos == 0: self.agent_energy = min(100, self.agent_energy + 40) reward += 5 avg_hp = sum(self.grid_health) / 25 done = avg_hp < 20 or self.agent_energy <= 0 info = {"avg_health": avg_hp, "reward": reward} return {}, reward, done, False, info # ── Heuristic agent policy ──────────────────────────────────────────────────── def pick_action(grid_health, agent_pos, agent_energy): """ Lightweight rule-based fallback policy. Returns (action_string, reasoning_string) — identical shape to the LLM agent. """ critical = [i for i, h in enumerate(grid_health) if h < 30] if agent_energy <= 20 and agent_pos == 0: return ActionType.RECHARGE, ( f"Energy critically low ({agent_energy}%) — recharging at base." ) if agent_energy <= 20: row, col = divmod(agent_pos, 5) act = ActionType.MOVE_N if row > 0 else ActionType.MOVE_W return act, f"Energy low ({agent_energy}%) — routing back to base for recharge." if agent_pos in critical: return ActionType.REPAIR, ( f"Sector {agent_pos} is critical (HP={grid_health[agent_pos]}) — " f"initiating emergency repair." ) if critical: target = critical[0] trow, tcol = divmod(target, 5) row, col = divmod(agent_pos, 5) if trow < row: act = ActionType.MOVE_N elif trow > row: act = ActionType.MOVE_S elif tcol < col: act = ActionType.MOVE_W else: act = ActionType.MOVE_E return act, ( f"Navigating toward critical sector {target} " f"(HP={grid_health[target]})." ) worst = min(range(25), key=lambda i: grid_health[i]) if worst == agent_pos: return ActionType.WAIT, "All sectors nominal — holding position, monitoring entropy." wrow, wcol = divmod(worst, 5) row, col = divmod(agent_pos, 5) if wrow < row: act = ActionType.MOVE_N elif wrow > row: act = ActionType.MOVE_S elif wcol < col: act = ActionType.MOVE_W else: act = ActionType.MOVE_E return act, ( f"Patrolling toward degraded sector {worst} " f"(HP={grid_health[worst]}) — entropy watch active." ) # ── Session-state initialisation ───────────────────────────────────────────── def _init_state(): if "env" not in st.session_state: env = CityGrid() env.reset() st.session_state.env = env e = st.session_state.env defaults = { "grid_health": list(e.grid_health), "agent_pos": e.agent_pos, "agent_energy": e.agent_energy, "step_count": 0, "total_reward": 0.0, "repair_count": 0, "recharge_count": 0, "auto_running": False, "game_over": False, "log_lines": [ "SYS › DISASTER GRID v1.0 — systems online.", "SYS › 25 sectors initialised — all readings nominal.", "SYS › Awaiting operator command...", ], } for k, v in defaults.items(): if k not in st.session_state: st.session_state[k] = v _init_state() # ── Helpers ─────────────────────────────────────────────────────────────────── def _sync_from_env(): e = st.session_state.env st.session_state.grid_health = list(e.grid_health) st.session_state.agent_pos = e.agent_pos st.session_state.agent_energy = e.agent_energy def _add_log(msg: str, kind: str = "sys"): prefix = { "sys": "SYS ›", "act": "ACT ›", "warn": "⚠ ›", "crit": "🔥 ›", "done": "✗ ›", } st.session_state.log_lines.append(f"{prefix.get(kind, ' ›')} {msg}") if len(st.session_state.log_lines) > 100: st.session_state.log_lines = st.session_state.log_lines[-100:] def _do_step(): if st.session_state.game_over: return action, reasoning = pick_action( st.session_state.grid_health, st.session_state.agent_pos, st.session_state.agent_energy, ) act_str = action.value if hasattr(action, "value") else str(action) _, reward, done, _, _ = st.session_state.env.step( {"action": act_str, "reasoning": reasoning} ) _sync_from_env() st.session_state.step_count += 1 st.session_state.total_reward += float(reward) if act_str == "REPAIR": st.session_state.repair_count += 1 if act_str == "RECHARGE": st.session_state.recharge_count += 1 _add_log(f'"{reasoning}"', "act") _add_log( f"Action: {act_str} | Reward: {reward:+.1f} " f"| Step #{st.session_state.step_count}", "sys", ) if done: st.session_state.game_over = True st.session_state.auto_running = False _add_log("SIMULATION ENDED — city collapsed or energy depleted.", "done") def _full_reset(): st.session_state.auto_running = False st.session_state.game_over = False st.session_state.step_count = 0 st.session_state.total_reward = 0.0 st.session_state.repair_count = 0 st.session_state.recharge_count = 0 st.session_state.log_lines = [ "SYS › City reset to baseline.", "SYS › All 25 sectors restored to nominal.", "SYS › Awaiting operator command...", ] st.session_state.env.reset() _sync_from_env() # ── CSS injection ───────────────────────────────────────────────────────────── THEME_CSS = """ """ st.markdown(THEME_CSS, unsafe_allow_html=True) # ── HTML builders ───────────────────────────────────────────────────────────── def _build_grid_html(grid_health: list, agent_pos: int) -> str: cells = [] for i in range(25): hp = grid_health[i] is_agent = i == agent_pos is_base = i == 0 is_crit = hp < 30 if is_agent: emoji = "🤖"; css_cls = "cell-agent" elif is_base: emoji = "🏢"; css_cls = "cell-base" elif is_crit: emoji = "🔥"; css_cls = "cell-crit" else: emoji = "🟩"; css_cls = "cell-ok" hp_color = ( "#ff4d4d" if hp < 30 else "#ffb800" if hp < 60 else "#00ff9d" ) cells.append(f"""
{i} {emoji}
""") return '
' + "".join(cells) + "
" def _build_log_html(log_lines: list) -> str: lines_html = [] for line in log_lines[-80:]: if line.startswith("ACT"): lines_html.append(f'{line}') elif line.startswith("⚠"): lines_html.append(f'{line}') elif line.startswith("🔥"): lines_html.append(f'{line}') elif line.startswith("✗"): lines_html.append(f'{line}') else: lines_html.append(f'{line}') inner = "
".join(lines_html) # tiny inline JS to keep the log scrolled to bottom scroll = ( "" ) return ( '
' f'
{inner}
' "
" + scroll ) # ── Derived values ──────────────────────────────────────────────────────────── gh = st.session_state.grid_health nrg = st.session_state.agent_energy avg = sum(gh) / 25 crit = sum(1 for h in gh if h < 30) steps = st.session_state.step_count rew = st.session_state.total_reward # ── Header ─────────────────────────────────────────────────────────────────── if st.session_state.game_over: badge_cls, badge_txt = "badge-gameover", "GAME OVER" elif st.session_state.auto_running: badge_cls, badge_txt = "badge-running", "● RUNNING" else: badge_cls, badge_txt = "badge-standby", "STANDBY" st.markdown(f"""
⚡ DISASTER GRID
Autonomous AI Emergency Manager — GRPO / Llama-3
{badge_txt}
STEP {steps:04d}
""", unsafe_allow_html=True) # ── Telemetry metrics row ───────────────────────────────────────────────────── mc = st.columns(6, gap="small") labels = ["Agent Energy", "City Avg HP", "Critical Sectors", "Total Reward", "Repairs", "Recharges"] values = [f"{nrg}", f"{avg:.1f}", str(crit), f"{rew:+.1f}", str(st.session_state.repair_count), str(st.session_state.recharge_count)] for col, lbl, val in zip(mc, labels, values): with col: st.metric(lbl, val) # ── Progress bars ───────────────────────────────────────────────────────────── pb1, pb2 = st.columns(2, gap="small") with pb1: st.markdown('
Agent Energy
', unsafe_allow_html=True) st.progress(max(0, min(100, nrg)) / 100) with pb2: st.markdown('
City Avg Health
', unsafe_allow_html=True) st.progress(max(0.0, min(100.0, avg)) / 100) st.markdown("
", unsafe_allow_html=True) # ── Main 3-column layout ────────────────────────────────────────────────────── left_col, center_col, right_col = st.columns([2, 5, 3], gap="medium") # ─────────────────────────────── LEFT PANEL ────────────────────────────────── with left_col: # ── Command Interface ── st.markdown('
◈ COMMAND INTERFACE
', unsafe_allow_html=True) st.markdown("") # breathing room if st.session_state.auto_running: st.markdown('
', unsafe_allow_html=True) if st.button("◼ Stop Agent", key="btn_stop"): st.session_state.auto_running = False _add_log("Agent halted by operator.", "sys") st.rerun() st.markdown("
", unsafe_allow_html=True) else: if not st.session_state.game_over: if st.button("▶ Run Autonomous Agent", key="btn_run"): st.session_state.auto_running = True _add_log("Autonomous agent activated — Llama-3 online.", "act") st.rerun() if not st.session_state.auto_running and not st.session_state.game_over: if st.button("⏭ Step Once", key="btn_step"): _do_step() st.rerun() st.markdown("
", unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) if st.button("🌋 Inject Random Disaster", key="btn_dis"): crises = random.sample(range(1, 25), 4) st.session_state.env.reset(options={"target_crises": crises}) # FIX: Tell Streamlit to sync with the newly updated physics engine! _sync_from_env() _add_log(f"RANDOM DISASTER: Sectors {crises} have caught fire!", "crit") if hasattr(st, "rerun"): st.rerun() else: st.experimental_rerun() st.markdown("
", unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) if st.button("🎯 Targeted Attack (Corners)", key="btn_corners"): corners = [4, 20, 24, 14] # FIX 2: Added 'options=' here so this button doesn't crash either! st.session_state.env.reset(options={"target_crises": corners}) _sync_from_env() _add_log("TARGETED STRIKE: Corners [4, 14, 20, 24] hit!", "crit") st.rerun() st.markdown("
", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) if st.button("↺ Reset City", key="btn_reset"): _full_reset() st.rerun() st.markdown("
", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # ── Legend ── st.markdown('
◈ LEGEND
', unsafe_allow_html=True) st.markdown("""
🤖  Agent position
🏢  Base (Sector 0)
🔥  Critical sector (<30 HP)
🟩  Healthy sector
▬  HP bar at cell base
""", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # ── Session stats ── st.markdown('
◈ SESSION STATS
', unsafe_allow_html=True) rew_cls = "s-danger" if rew < 0 else "s-amber" if rew == 0 else "" crit_cls = "s-danger" if crit > 3 else "s-amber" if crit > 0 else "" st.markdown(f"""
Critical {crit}
Reward {rew:+.1f}
Repairs {st.session_state.repair_count}
Recharges {st.session_state.recharge_count}
Steps {steps}
Avg HP {avg:.1f}
""", unsafe_allow_html=True) # ──────────────────────────── CENTER PANEL (Grid) ──────────────────────────── with center_col: st.markdown('
◈ CITY GRID — 5×5 OPERATIONAL THEATRE
', unsafe_allow_html=True) grid_html = _build_grid_html(st.session_state.grid_health, st.session_state.agent_pos) st.markdown(grid_html, unsafe_allow_html=True) # ──────────────────────────── RIGHT PANEL (Log) ────────────────────────────── with right_col: st.markdown('
◈ AGENT BRAIN LOG
', unsafe_allow_html=True) log_html = _build_log_html(st.session_state.log_lines) st.markdown(log_html, unsafe_allow_html=True) # ── Auto-run engine (at script bottom — runs ONE step then reruns) ──────────── if st.session_state.auto_running and not st.session_state.game_over: _do_step() time.sleep(0.38) # ~2.6 steps/sec — adjust to taste st.rerun()