231 lines
8.1 KiB
Python
231 lines
8.1 KiB
Python
from fastapi import FastAPI, HTTPException, Response
|
|
from pydantic import BaseModel
|
|
from typing import Dict, List, Optional, Any
|
|
from diplomacy import Game, Map
|
|
import random
|
|
|
|
app = FastAPI(title="Diplomacy Calculation Engine API", description="Stateless Diplomacy engine API based on the 'diplomacy' library.")
|
|
|
|
# --- Data Models ---
|
|
|
|
class GameState(BaseModel):
|
|
"""Represents the complete state of a Diplomacy game."""
|
|
game_state: Dict[str, Any]
|
|
|
|
class OrderRequest(BaseModel):
|
|
"""Request for processing or validating orders."""
|
|
game_state: Dict[str, Any]
|
|
orders: Dict[str, List[str]] # Power Name -> List of Orders
|
|
|
|
class RenderRequest(BaseModel):
|
|
"""Request for rendering the game map."""
|
|
game_state: Dict[str, Any]
|
|
orders: Optional[Dict[str, List[str]]] = None
|
|
incl_orders: bool = True
|
|
incl_abbrev: bool = True
|
|
|
|
class DrawRequest(BaseModel):
|
|
"""Request to force a draw in the game."""
|
|
game_state: Dict[str, Any]
|
|
winners: Optional[List[str]] = None
|
|
|
|
# --- Helper Functions ---
|
|
|
|
def load_game(state_dict: Dict[str, Any]) -> Game:
|
|
"""Loads a Game object from a dictionary."""
|
|
try:
|
|
game = Game()
|
|
game.set_state(state_dict)
|
|
return game
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Failed to load game state: {str(e)}")
|
|
|
|
@app.get("/game/initial-state")
|
|
def get_initial_state(map_name: str = "standard"):
|
|
"""Returns the initial game state for a given map."""
|
|
try:
|
|
game = Game(map_name=map_name)
|
|
return {"game_state": game.get_state()}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Failed to create initial state for map '{map_name}': {str(e)}")
|
|
|
|
@app.get("/debug_heartbeat")
|
|
def debug_heartbeat():
|
|
"""Simple heartbeat to check if server code is updated."""
|
|
return {"status": "ok", "version": "debug_v1"}
|
|
|
|
# --- Endpoints ---
|
|
|
|
@app.get("/maps")
|
|
def list_maps():
|
|
"""List available maps (standard library maps)."""
|
|
# Simply return common ones for now, can be expanded
|
|
return {"maps": ["standard"]}
|
|
|
|
@app.get("/maps/{map_name}")
|
|
def get_map_details(map_name: str):
|
|
"""Get topology and details of a specific map."""
|
|
try:
|
|
d_map = Map(map_name)
|
|
return {
|
|
"name": map_name,
|
|
"provinces": d_map.locs,
|
|
"adjacencies": {loc: d_map.abut_list(loc) for loc in d_map.locs},
|
|
"powers": d_map.powers,
|
|
"centers": d_map.scs,
|
|
"homes": d_map.homes
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=404, detail=f"Map '{map_name}' not found: {str(e)}")
|
|
|
|
@app.post("/calculate/process")
|
|
def process_phase(request: OrderRequest):
|
|
"""Resolves orders and advances to the next phase."""
|
|
game = load_game(request.game_state)
|
|
|
|
# Set orders for each power
|
|
for power, orders in request.orders.items():
|
|
if power in game.powers:
|
|
game.set_orders(power, orders)
|
|
|
|
# Process the game
|
|
game.process()
|
|
|
|
return {"game_state": game.get_state()}
|
|
|
|
@app.post("/calculate/validate")
|
|
def validate_orders(request: OrderRequest):
|
|
"""Validates orders without advancing the phase."""
|
|
game = load_game(request.game_state)
|
|
|
|
results = {}
|
|
for power, orders in request.orders.items():
|
|
if power in game.powers:
|
|
game.set_orders(power, orders)
|
|
# diplomacy library validates during set_orders and process or via specific methods
|
|
# Here we can check if orders were actually accepted or get status
|
|
results[power] = game.get_orders(power)
|
|
|
|
return {"validated_orders": results, "errors": game.error}
|
|
|
|
@app.post("/game/draw")
|
|
def draw_game(request: DrawRequest):
|
|
"""Forces the game to a draw status."""
|
|
game = load_game(request.game_state)
|
|
|
|
# Process the draw
|
|
game.draw(winners=request.winners)
|
|
|
|
return {"game_state": game.get_state()}
|
|
|
|
@app.post("/calculate/possible-orders")
|
|
def get_possible_orders(request: GameState, power_name: Optional[str] = None, by_power: bool = False):
|
|
"""Get all possible valid orders for the current state."""
|
|
game = load_game(request.game_state)
|
|
|
|
all_possibilities = game.get_all_possible_orders()
|
|
|
|
if by_power:
|
|
results = {}
|
|
for power in game.powers:
|
|
# Get unit locations for this power
|
|
power_units = game.get_units(power)
|
|
power_locs = [u.split(' ')[1] for u in power_units]
|
|
|
|
# For adjustment phases, also include build locations (homes where units can be built)
|
|
if 'ADJUSTMENT' in game.phase:
|
|
# Get build information for this power
|
|
builds_info = game.get_state().get('builds', {}).get(power, {})
|
|
build_homes = builds_info.get('homes', [])
|
|
power_locs.extend(build_homes)
|
|
|
|
results[power] = {loc: orders for loc, orders in all_possibilities.items() if loc in power_locs}
|
|
|
|
if power_name:
|
|
power_name = power_name.upper()
|
|
if power_name not in results:
|
|
raise HTTPException(status_code=404, detail=f"Power '{power_name}' not found")
|
|
return {"power": power_name, "possible_orders": results[power_name]}
|
|
|
|
return {"possible_orders": results}
|
|
|
|
if power_name:
|
|
power_name = power_name.upper()
|
|
if power_name not in game.powers:
|
|
raise HTTPException(status_code=404, detail=f"Power '{power_name}' not found")
|
|
|
|
# Get unit locations for this power
|
|
power_units = game.get_units(power_name) # returns list like ['A PAR', 'F BRE']
|
|
power_locs = [u.split(' ')[1] for u in power_units]
|
|
|
|
# For adjustment phases, also include build locations
|
|
if 'ADJUSTMENT' in game.phase:
|
|
builds_info = game.get_state().get('builds', {}).get(power_name, {})
|
|
build_homes = builds_info.get('homes', [])
|
|
power_locs.extend(build_homes)
|
|
|
|
filtered = {loc: orders for loc, orders in all_possibilities.items() if loc in power_locs}
|
|
return {"power": power_name, "possible_orders": filtered}
|
|
|
|
return {"possible_orders": all_possibilities}
|
|
|
|
@app.post("/calculate/auto-orders")
|
|
def auto_orders(request: GameState, power_name: str):
|
|
"""Generates random valid orders for a specific power."""
|
|
game = load_game(request.game_state)
|
|
power_name = power_name.upper()
|
|
|
|
if power_name not in game.powers:
|
|
raise HTTPException(status_code=404, detail=f"Power '{power_name}' not found")
|
|
|
|
# Check for units
|
|
power_units = game.get_units(power_name)
|
|
if not power_units:
|
|
return {"orders": []}
|
|
|
|
possible_orders = game.get_all_possible_orders()
|
|
orders_to_submit = []
|
|
|
|
for unit in power_units:
|
|
loc = unit.split(' ')[1]
|
|
if loc in possible_orders:
|
|
unit_orders = possible_orders[loc]
|
|
if unit_orders:
|
|
orders_to_submit.append(random.choice(unit_orders))
|
|
|
|
return {"orders": orders_to_submit}
|
|
|
|
@app.post("/render")
|
|
def render_map(request: RenderRequest):
|
|
"""Renders the game map as an SVG."""
|
|
game = load_game(request.game_state)
|
|
|
|
# If orders are provided, set them temporarily for rendering
|
|
if request.orders:
|
|
for power, orders in request.orders.items():
|
|
if power in game.powers:
|
|
game.set_orders(power, orders)
|
|
|
|
svg_content = game.render(
|
|
incl_orders=request.incl_orders,
|
|
incl_abbrev=request.incl_abbrev,
|
|
output_format='svg'
|
|
)
|
|
|
|
response = Response(content=svg_content, media_type="image/svg+xml")
|
|
# Debug info
|
|
try:
|
|
all_units = game.get_units()
|
|
unit_count = sum(len(u) for u in all_units.values()) if all_units else 0
|
|
response.headers["X-Debug-Unit-Count"] = str(unit_count)
|
|
# Austria units as example
|
|
response.headers["X-Debug-Units-Austria"] = str(game.get_units("AUSTRIA"))
|
|
except Exception as e:
|
|
response.headers["X-Debug-Error"] = str(e)
|
|
|
|
return response
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|