Spaces:
Running
Running
| from fastapi import FastAPI, HTTPException # type: ignore | |
| from fastapi.middleware.cors import CORSMiddleware # type: ignore | |
| from fastapi.staticfiles import StaticFiles # type: ignore | |
| from fastapi.responses import FileResponse # type: ignore | |
| from pydantic import BaseModel # type: ignore | |
| import random | |
| import uuid | |
| import uvicorn # type: ignore | |
| import os | |
| from typing import Any, List, Dict, Optional, cast | |
| from engine import ( | |
| GameState, INITIAL_FUNDS, SALE_PRICE, STORAGE_COST, | |
| CONTAINER_SIZE, SHIPPING_COST, LATE_PENALTY_RATE, | |
| INITIAL_FACTORIES, INITIAL_PORTS, WAREHOUSES, INITIAL_DEMAND | |
| ) | |
| app = FastAPI() | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| state = GameState() | |
| def generate_smart_alerts(): | |
| alerts = [] | |
| delayed_ports = [p for p in state.portDelays if p["currentDelay"] > p["baseDelay"] or p["isClosed"]] | |
| if delayed_ports: | |
| best_port = min((p for p in state.portDelays if not p["isClosed"]), key=lambda x: x["currentDelay"], default=None) | |
| if best_port: | |
| alerts.append({ | |
| "type": "warning", | |
| "msg": f"Risk Flag: {len(delayed_ports)} ports are experiencing delays. Recommendation: Route via {best_port['name']} (Wait: {best_port['currentDelay']} mo)." | |
| }) | |
| for region, demand in state.demands.items(): | |
| base_demand = INITIAL_DEMAND[region] | |
| if demand < base_demand * 0.8: | |
| alerts.append({"type": "warning", "msg": f"Anomaly: {region.upper()} demand dropped severely to {demand}. Reduce future order volume!"}) | |
| elif demand > base_demand * 1.2: | |
| alerts.append({"type": "success", "msg": f"Opportunity: {region.upper()} demand spiked to {demand}. Increase inventory to maximize revenue."}) | |
| total_inv = sum(state.inventory.values()) | |
| if total_inv * STORAGE_COST > getattr(state, 'last_revenue', 1000) * 0.3: | |
| alerts.append({"type": "danger", "msg": f"High storage cost warning! Holding {total_inv} units costs ${total_inv * STORAGE_COST}/mo. Avoid over-ordering."}) | |
| return alerts | |
| # --- ROUTES --- | |
| def get_state(): | |
| state.alerts = generate_smart_alerts() | |
| leaderboard = [] | |
| for f in state.factoryCosts: | |
| for p in state.portDelays: | |
| if not p["isClosed"]: | |
| cost_score = f["currentCost"] * 100 + SHIPPING_COST | |
| reliability = 100 - (p["currentDelay"] * 5) | |
| if state.scenario == "crisis": reliability -= 10 | |
| leaderboard.append({ | |
| "route": f"{f['name'].split(' ')[0]} → {p['name'].replace('Port of ','')}", | |
| "cost": cost_score, | |
| "time": p["currentDelay"], | |
| "reliability": reliability | |
| }) | |
| leaderboard = sorted(leaderboard, key=lambda x: (x["cost"], x["time"])) | |
| leaderboard = leaderboard[:5] # type: ignore | |
| return { | |
| "turn": state.turn, | |
| "scenario": state.scenario, | |
| "balance": state.balance, | |
| "benchmark_balance": state.benchmark_balance, | |
| "inventory": state.inventory, | |
| "demands": state.demands, | |
| "factoryCosts": state.factoryCosts, | |
| "portDelays": state.portDelays, | |
| "inTransit": state.inTransit, | |
| "history": state.history, | |
| "events": state.events[-50:] if len(state.events) > 50 else state.events, # type: ignore | |
| "alerts": state.alerts, | |
| "leaderboard": leaderboard | |
| } | |
| class OrderRequest(BaseModel): | |
| factory: str | |
| quantity: int | |
| port: str | |
| dest: str | |
| def place_order(order: OrderRequest): | |
| factory = next((f for f in state.factoryCosts if f["id"] == order.factory), None) | |
| port = next((p for p in state.portDelays if p["id"] == order.port), None) | |
| if factory is None or port is None: | |
| raise HTTPException(status_code=400, detail="Invalid factory or port") | |
| if cast(Dict[str, Any], port).get("isClosed"): # type: ignore | |
| raise HTTPException(status_code=400, detail=f"Cannot route through {port['name']}, port is closed!") | |
| current_cost = cast(float, factory["currentCost"]) | |
| cost_of_goods = current_cost * order.quantity | |
| containers = (order.quantity + CONTAINER_SIZE - 1) // CONTAINER_SIZE | |
| shipping = containers * SHIPPING_COST | |
| total_cost = cost_of_goods + shipping | |
| if state.balance < total_cost: | |
| raise HTTPException(status_code=400, detail=f"Insufficient funds. Need {total_cost}, have {state.balance}") | |
| state.balance -= total_cost | |
| state.month_shipping_costs += shipping | |
| state.month_goods_costs += cost_of_goods | |
| uid_str = str(uuid.uuid4()) | |
| state.inTransit.append({ | |
| "id": uid_str[:8], # type: ignore | |
| "factory": factory["id"], # type: ignore | |
| "port": port["id"], # type: ignore | |
| "dest": order.dest, | |
| "quantity": order.quantity, | |
| "delayLeft": port["currentDelay"], # type: ignore | |
| "totalDelay": port["currentDelay"], # type: ignore | |
| "costBasis": total_cost | |
| }) | |
| state.events.append({ | |
| "turn": state.turn, | |
| "type": "info", | |
| "msg": f"Ordered {order.quantity} units from {factory['name']} via {port['name']}. Cost: ${total_cost:,.0f}" # type: ignore | |
| }) | |
| return {"status": "success", "balance": state.balance} | |
| class ScenarioRequest(BaseModel): | |
| scenario: str | |
| def set_scenario(req: ScenarioRequest): | |
| state.scenario = req.scenario | |
| state.events.append({ | |
| "turn": state.turn, | |
| "type": "warning" if req.scenario == "crisis" else "success", | |
| "msg": f"SCENARIO CHANGED: Rules updated to {req.scenario.upper()} mode!" | |
| }) | |
| return get_state() | |
| def advance_turn(): | |
| turn_revenue = 0 | |
| turn_penalty = 0 | |
| turn_events = [] | |
| # 1. Process shipments | |
| new_in_transit = [] | |
| for shipment in state.inTransit: | |
| if shipment["delayLeft"] <= 1: | |
| state.inventory[shipment["dest"]] += shipment["quantity"] | |
| warehouse = next((w for w in WAREHOUSES if w["id"] == shipment["dest"]), None) | |
| warehouse_name = warehouse["name"] if warehouse else "Unknown" | |
| turn_events.append({ | |
| "turn": state.turn, | |
| "type": "success", | |
| "msg": f"{shipment['quantity']} units arrived at {warehouse_name}" | |
| }) | |
| else: | |
| shipment["delayLeft"] -= 1 | |
| new_in_transit.append(shipment) | |
| state.inTransit = new_in_transit | |
| # 2. Fulfill demand | |
| new_demands = dict(state.demands) | |
| for region in state.demands: | |
| demand = new_demands[region] | |
| avail = state.inventory[region] | |
| sold = min(demand, avail) | |
| missed = demand - sold | |
| state.inventory[region] -= sold | |
| turn_revenue += sold * SALE_PRICE | |
| if missed > 0: | |
| penalty = missed * (SALE_PRICE * LATE_PENALTY_RATE) | |
| turn_penalty += penalty # type: ignore | |
| if missed > demand * 0.5: | |
| turn_events.append({ | |
| "turn": state.turn, | |
| "type": "danger", | |
| "msg": f"{region.upper()} CRITICAL STOCKOUT! Missed {missed} units. Penalty: ${penalty:,.0f}" | |
| }) | |
| # 3. Holding costs | |
| holding_cost = sum(amt * STORAGE_COST for amt in state.inventory.values()) | |
| state.last_revenue = turn_revenue | |
| # 4. Update balance | |
| period_profit = turn_revenue - holding_cost - turn_penalty - state.month_goods_costs - state.month_shipping_costs # type: ignore | |
| state.balance += turn_revenue - holding_cost - turn_penalty # type: ignore | |
| # Compute Benchmark AI Balance | |
| # Benchmark buys exactly the demand each month at lowest fixed rate. | |
| benchmark_revenue = sum(state.demands.values()) * SALE_PRICE | |
| benchmark_cost = sum(state.demands.values()) * (50 + 30) # 50 = china max cheap, 30 = generic average shipping | |
| benchmark_profit = benchmark_revenue - benchmark_cost | |
| if state.scenario == "crisis": | |
| benchmark_profit *= 0.6 # AI suffers in crisis too | |
| state.benchmark_balance += benchmark_profit | |
| # 5. Shocks & Fluctuation | |
| demand_shock_chance = 0.9 if state.scenario == "normal" else 0.8 | |
| is_competitor_shock = random.random() > demand_shock_chance | |
| demand_multiplier = random.uniform(0.8, 1.2) | |
| if state.scenario == "growth": demand_multiplier += 0.2 | |
| if is_competitor_shock: | |
| demand_multiplier = 0.5 | |
| turn_events.append({"turn": state.turn + 1, "type": "warning", "msg": "Competitors undercut prices! Demand collapsed."}) | |
| for region in INITIAL_DEMAND: | |
| state.demands[region] = int(INITIAL_DEMAND[region] * demand_multiplier) | |
| price_spike_chance = 0.9 if state.scenario == "normal" else 0.75 # more spikes in crisis | |
| is_price_spike = random.random() > price_spike_chance | |
| if is_price_spike: | |
| turn_events.append({"turn": state.turn + 1, "type": "danger", "msg": "Global component shortage! Factory costs spiked +40%."}) | |
| for f in state.factoryCosts: | |
| base = f["baseCost"] | |
| if is_price_spike: | |
| f["currentCost"] = int(base * 1.4) | |
| else: | |
| f["currentCost"] = int(base * random.uniform(0.85, 1.15)) | |
| closure_chance = 0.95 if state.scenario == "normal" else 0.85 | |
| for p in state.portDelays: | |
| is_closed = random.random() > closure_chance | |
| if is_closed: | |
| turn_events.append({"turn": state.turn + 1, "type": "danger", "msg": f"GEOPOLITICAL SHOCK: {p['name']} closed for 2 months!"}) | |
| p["isClosed"] = is_closed | |
| p["currentDelay"] = max(2, p["baseDelay"] + 2) if is_closed else p["baseDelay"] + (1 if random.random() > 0.5 else 0) | |
| # SET STATE | |
| state.turn += 1 | |
| state.history.append({ | |
| "month": state.turn - 1, | |
| "profit": period_profit, | |
| "revenue": turn_revenue, | |
| "storage_costs": holding_cost, | |
| "shipping_costs": state.month_shipping_costs, | |
| "penalties": turn_penalty, | |
| "balance": state.balance, | |
| "benchmark": state.benchmark_balance | |
| }) | |
| # Reset monthly trackers | |
| state.month_shipping_costs = 0 | |
| state.month_goods_costs = 0 | |
| state.events.extend(turn_events) | |
| return get_state() | |
| def get_oracle_advice(): | |
| # Generate some strategic advice based on current state | |
| closed_ports = [p["name"] for p in state.portDelays if p["isClosed"]] | |
| delayed_ports = [p["name"] for p in state.portDelays if p["currentDelay"] > p["baseDelay"] and not p["isClosed"]] | |
| advice = "Market conditions are stable. Maintain standard logistics routing." | |
| action = "Continue optimal ordering." | |
| if closed_ports: | |
| advice = f"CRITICAL: {', '.join(closed_ports)} is closed! This will severely disrupt supply chains." | |
| action = "Reroute all shipments immediately to alternative ports to avoid stockouts." | |
| elif delayed_ports: | |
| advice = f"WARNING: Delays detected at {', '.join(delayed_ports)}." | |
| action = "Consider holding more inventory to buffer against transit delays." | |
| elif state.scenario == "crisis": | |
| advice = "Global geopolitical crisis is increasing closure probabilities." | |
| action = "Pre-stock warehouses to maximum capacity now before ports close." | |
| elif state.scenario == "growth": | |
| advice = "Aggressive growth scenario detected. Demand is inflating rapidly." | |
| action = "Scale up order volumes by 20% to meet surging demand and capture market share." | |
| return {"advice": advice, "action": action} | |
| def auto_play_turn(): | |
| # Simple AI agent: for each region, if inventory < demand, order the difference | |
| # Find cheapest available factory and port | |
| open_ports = [p for p in state.portDelays if not p["isClosed"]] | |
| if not open_ports: | |
| # Cannot order anything, just advance turn | |
| return advance_turn() | |
| best_port = min(open_ports, key=lambda x: x["currentDelay"]) | |
| best_factory = min(state.factoryCosts, key=lambda x: x["currentCost"]) | |
| for region, demand in state.demands.items(): | |
| if state.inventory[region] < demand: | |
| needed = demand - state.inventory[region] | |
| # Order 'needed' amount | |
| cost_of_goods = best_factory["currentCost"] * needed | |
| containers = (needed + CONTAINER_SIZE - 1) // CONTAINER_SIZE | |
| shipping = containers * SHIPPING_COST | |
| total_cost = cost_of_goods + shipping | |
| if state.balance >= total_cost: | |
| state.balance -= total_cost | |
| state.month_shipping_costs += shipping | |
| state.month_goods_costs += cost_of_goods | |
| uid_str = str(uuid.uuid4()) | |
| state.inTransit.append({ | |
| "id": uid_str[:8], | |
| "factory": best_factory["id"], | |
| "port": best_port["id"], | |
| "dest": region, | |
| "quantity": needed, | |
| "delayLeft": best_port["currentDelay"], | |
| "totalDelay": best_port["currentDelay"], | |
| "costBasis": total_cost | |
| }) | |
| state.events.append({ | |
| "turn": state.turn, | |
| "type": "info", | |
| "msg": f"AI ORACLE ordered {needed} units from {best_factory['name']} via {best_port['name']} for {region.upper()}." | |
| }) | |
| return advance_turn() | |
| def reset_game(): | |
| state.reset() | |
| return get_state() | |
| # --- SERVE FRONTEND --- | |
| async def serve_index(): | |
| return FileResponse("dist/index.html") | |
| async def serve_static(full_path: str): | |
| path = os.path.join("dist", full_path) | |
| if os.path.isfile(path): | |
| return FileResponse(path) | |
| return FileResponse("dist/index.html") | |
| if os.path.isdir("dist"): | |
| app.mount("/assets", StaticFiles(directory="dist/assets"), name="assets") | |
| # Serve static files if 'dist' folder exists (for Docker/production) | |
| if os.path.isdir("dist"): | |
| app.mount("/assets", StaticFiles(directory="dist/assets"), name="assets") | |
| async def serve_frontend(full_path: str): | |
| path = os.path.join("dist", full_path) | |
| if os.path.isfile(path): | |
| return FileResponse(path) | |
| return FileResponse("dist/index.html") | |
| if __name__ == "__main__": | |
| port = int(os.environ.get("PORT", 8000)) | |
| uvicorn.run("main:app", host="0.0.0.0", port=port, reload=True) | |