バックエンドAPI
This commit is contained in:
230
main.py
Normal file
230
main.py
Normal file
@@ -0,0 +1,230 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user