#!/usr/bin/env python3 """ Live Supply Chain Map — visual dashboard for judges. Shows a world map with: - Ports (green=open, red=closed, yellow=reduced) - Shipping routes (blue=open, red=blocked) - Shipments moving along routes - Disruptions highlighted - Real-time stats panel Generates an HTML file that opens in browser. Run: python3 visualize.py """ import json import webbrowser import os from world import SupplyChainWorld from real_data import REAL_PORTS, REAL_ROUTES def generate_map_html(world: SupplyChainWorld, filename: str = "supply_chain_map.html") -> str: """Generate an interactive HTML map of the supply chain.""" # Port data with status colors ports_js = [] for p in REAL_PORTS: status = world.ports.get(p["id"], {}).get("status", "open") color = {"open": "#22c55e", "closed": "#ef4444", "reduced": "#eab308"}.get(status, "#666") throughput = p.get("throughput_teu", 0) ports_js.append({ "id": p["id"], "name": p["name"], "lat": p["lat"], "lon": p["lon"], "status": status, "color": color, "throughput": throughput, "country": p.get("country", ""), }) # Route data with status routes_js = [] for r in REAL_ROUTES: src_port = next((p for p in REAL_PORTS if p["id"] == r["from"]), None) dst_port = next((p for p in REAL_PORTS if p["id"] == r["to"]), None) if not src_port or not dst_port: continue route_info = world.routes.get(r["from"], {}).get(r["to"], {}) status = route_info.get("status", "open") color = "#3b82f6" if status == "open" else "#ef4444" routes_js.append({ "from_lat": src_port["lat"], "from_lon": src_port["lon"], "to_lat": dst_port["lat"], "to_lon": dst_port["lon"], "cost": r["cost_per_teu"], "days": r["transit_days"], "mode": r.get("mode", "ocean"), "status": status, "color": color, "carrier": r.get("carrier", ""), }) # Shipment positions shipments_js = [] for s in world.shipments.values(): port = world.ports.get(s.current_location, {}) p_data = next((p for p in REAL_PORTS if p["id"] == s.current_location), None) if p_data: color = {"pending": "#fbbf24", "in_transit": "#3b82f6", "delivered": "#22c55e", "lost": "#ef4444"}.get(s.status, "#666") shipments_js.append({ "id": s.id, "product": s.product, "value": s.value_usd, "lat": p_data["lat"], "lon": p_data["lon"], "status": s.status, "color": color, }) # Disruptions disruptions_js = [ {"type": d.type, "severity": d.severity, "desc": d.description, "active": d.active, "start": d.start_day, "end": d.end_day} for d in world.disruptions ] # Stats total_value = sum(s.value_usd for s in world.shipments.values()) stats = { "day": world.day, "total_days": world.total_days, "pending": sum(1 for s in world.shipments.values() if s.status == "pending"), "in_transit": sum(1 for s in world.shipments.values() if s.status == "in_transit"), "delivered": sum(1 for s in world.shipments.values() if s.status == "delivered"), "lost": sum(1 for s in world.shipments.values() if s.status == "lost"), "delivered_value": world.delivered_value, "lost_value": world.lost_value, "shipping_cost": world.total_shipping_cost, "total_value": total_value, } html = f""" SupplyChainEnv — Live Map

SupplyChainEnv — Global Trade Network

Day {stats['day']}/{stats['total_days']} | {world.difficulty.upper()}
{"".join(f'' for r in routes_js)} {"".join(f'{p["name"]}' for p in ports_js)} {"".join(f'' for s in shipments_js)}
Shipments Pending
{stats['pending']}
In Transit
{stats['in_transit']}
Delivered
{stats['delivered']} (${stats['delivered_value']:,.0f})
Lost
{stats['lost']} (${stats['lost_value']:,.0f})
Shipping Cost
${stats['shipping_cost']:,.0f}
Total Cargo Value
${stats['total_value']:,.0f}
Active Disruptions
{"".join(f'
{d["type"].upper()} [{d["severity"]}]
{d["desc"][:80]}
Day {d["start"]}-{d["end"]}
' for d in disruptions_js)}
Legend
● Port Open   ● Port Closed   ● Reduced
▬ Route Open   ▬ Route Blocked
● Pending   ● In Transit   ● Delivered
Data Sources
Port throughput: UNCTAD 2023
Shipping rates: Freightos Baltic Index Q1 2024
Disruptions: Lloyd's List, WHO, USGS 2017-2024
LPI scores: World Bank 2023
""" with open(filename, "w") as f: f.write(html) return os.path.abspath(filename) def main(): """Run the demo agent and generate map at each stage.""" from server.supply_chain_environment import SupplyChainEnvironment from models import SupplyChainAction def tool(name, args=None): return SupplyChainAction(action_type="ToolCallAction", tool_name=name, arguments=args or {}) env = SupplyChainEnvironment() env.reset(seed=42, difficulty="hard") # Route shipments obs = env.step(tool("view_shipments")) for s in obs.tool_result["shipments"]: if s["status"] == "pending": path_obs = env.step(tool("find_path", {"from_port": s["current_location"], "to_warehouse": s["destination"]})) path = path_obs.tool_result.get("path") if path: env.step(tool("route_shipment", {"shipment_id": s["id"], "route": path})) # Advance 15 days for interesting state for _ in range(15): obs = env.step(tool("advance_day")) if obs.done: break # Generate map path = generate_map_html(env._world) print(f"Map generated: {path}") webbrowser.open(f"file://{path}") if __name__ == "__main__": main()