232 lines
7.8 KiB
Python
232 lines
7.8 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from sqlmodel import SQLModel, select
|
|
|
|
from catan.game import GameConfig
|
|
from catan.sdk import Action, ActionType, CatanEnv
|
|
from services.common.db import engine, session_scope
|
|
from services.common.schemas import ReplayArchive, ReplayDetail, ReplayMeta
|
|
from services.game.models import Game, GameEvent
|
|
|
|
app = FastAPI(title="Catan Analytics")
|
|
|
|
|
|
@app.on_event("startup")
|
|
def _startup() -> None:
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
|
|
def _serialize_resources(resources: Dict[Any, int]) -> Dict[str, int]:
|
|
return {str(getattr(res, "value", res)): int(val) for res, val in resources.items()}
|
|
|
|
|
|
def _serialize_observation(observation: Dict[str, Any]) -> Dict[str, Any]:
|
|
game = dict(observation["game"])
|
|
players = {}
|
|
for name, info in game["players"].items():
|
|
pdata = dict(info)
|
|
if isinstance(pdata.get("resources"), dict):
|
|
pdata["resources"] = _serialize_resources(pdata["resources"])
|
|
players[name] = pdata
|
|
game["players"] = players
|
|
if "bank" in game:
|
|
game["bank"] = _serialize_resources(game["bank"])
|
|
return {"game": game, "board": observation["board"]}
|
|
|
|
|
|
def _load_game(game_id: str) -> Game:
|
|
with session_scope() as session:
|
|
game = session.get(Game, game_id)
|
|
if not game:
|
|
raise HTTPException(status_code=404, detail="Game not found")
|
|
return game
|
|
|
|
|
|
def _load_events(game_id: str) -> List[GameEvent]:
|
|
with session_scope() as session:
|
|
return session.exec(
|
|
select(GameEvent).where(GameEvent.game_id == game_id).order_by(GameEvent.idx)
|
|
).all()
|
|
|
|
|
|
def _replay_state(game: Game, events: List[GameEvent], step: int) -> Dict[str, Any]:
|
|
slots = game.slots.get("slots", [])
|
|
names = [slot.get("name") for slot in slots if slot.get("name")]
|
|
colors = [slot.get("color", "player") for slot in slots if slot.get("name")]
|
|
env = CatanEnv(GameConfig(player_names=names, colors=colors, seed=game.seed))
|
|
applied_events = [event for event in events if event.applied]
|
|
for event in applied_events[:step]:
|
|
action = Action(ActionType(event.action_type), event.payload)
|
|
env.step(action)
|
|
return _serialize_observation(env.observe())
|
|
|
|
|
|
def _build_archive(game: Game, events: List[GameEvent]) -> ReplayArchive:
|
|
slots = game.slots.get("slots", [])
|
|
players = [slot.get("name") for slot in slots if slot.get("name")]
|
|
actions = [
|
|
{
|
|
"idx": event.idx,
|
|
"ts": event.ts,
|
|
"actor": event.actor,
|
|
"action": {"type": event.action_type, "payload": event.payload},
|
|
"applied": event.applied,
|
|
"meta": event.debug_payload or {},
|
|
}
|
|
for event in events
|
|
]
|
|
return ReplayArchive(
|
|
id=game.id,
|
|
created_at=game.created_at,
|
|
seed=game.seed,
|
|
players=players,
|
|
winner=game.winner,
|
|
slots=slots,
|
|
actions=actions,
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> Dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/stats")
|
|
def stats(user: Optional[str] = None) -> Dict[str, Any]:
|
|
with session_scope() as session:
|
|
games = session.exec(select(Game)).all()
|
|
total_games = len(games)
|
|
finished_games = [g for g in games if g.status == "finished"]
|
|
avg_turns = 0.0
|
|
if finished_games:
|
|
with session_scope() as session:
|
|
counts = []
|
|
for game in finished_games:
|
|
count = session.exec(
|
|
select(GameEvent).where(GameEvent.game_id == game.id, GameEvent.applied == True)
|
|
).all()
|
|
counts.append(len(count))
|
|
avg_turns = sum(counts) / len(counts) if counts else 0.0
|
|
if not user:
|
|
return {
|
|
"total_games": total_games,
|
|
"finished_games": len(finished_games),
|
|
"avg_turns": avg_turns,
|
|
}
|
|
user_games = [g for g in games if user in [slot.get("name") for slot in g.slots.get("slots", [])]]
|
|
user_wins = sum(1 for g in games if g.winner == user)
|
|
return {
|
|
"total_games": total_games,
|
|
"finished_games": len(finished_games),
|
|
"avg_turns": avg_turns,
|
|
"user_games": len(user_games),
|
|
"user_wins": user_wins,
|
|
}
|
|
|
|
|
|
@app.get("/replays")
|
|
def list_replays() -> Dict[str, Any]:
|
|
with session_scope() as session:
|
|
games = session.exec(select(Game).where(Game.status == "finished").order_by(Game.created_at.desc())).all()
|
|
items: List[ReplayMeta] = []
|
|
with session_scope() as session:
|
|
for game in games:
|
|
actions = session.exec(
|
|
select(GameEvent).where(GameEvent.game_id == game.id, GameEvent.applied == True)
|
|
).all()
|
|
players = [slot.get("name") for slot in game.slots.get("slots", []) if slot.get("name")]
|
|
items.append(
|
|
ReplayMeta(
|
|
id=game.id,
|
|
created_at=game.created_at,
|
|
players=players,
|
|
winner=game.winner,
|
|
total_actions=len(actions),
|
|
)
|
|
)
|
|
return {"replays": [item.model_dump() for item in items]}
|
|
|
|
|
|
@app.get("/replays/{replay_id}")
|
|
def replay_detail(replay_id: str) -> Dict[str, Any]:
|
|
game = _load_game(replay_id)
|
|
events = _load_events(replay_id)
|
|
applied = [event for event in events if event.applied]
|
|
players = [slot.get("name") for slot in game.slots.get("slots", []) if slot.get("name")]
|
|
actions = [
|
|
{
|
|
"idx": event.idx,
|
|
"ts": event.ts,
|
|
"actor": event.actor,
|
|
"action": {"type": event.action_type, "payload": event.payload},
|
|
"applied": event.applied,
|
|
"meta": event.debug_payload or {},
|
|
}
|
|
for event in applied
|
|
]
|
|
detail = ReplayDetail(
|
|
id=game.id,
|
|
created_at=game.created_at,
|
|
seed=game.seed,
|
|
players=players,
|
|
winner=game.winner,
|
|
total_actions=len(applied),
|
|
actions=actions,
|
|
)
|
|
return detail.model_dump()
|
|
|
|
|
|
@app.get("/replays/{replay_id}/state")
|
|
def replay_state(replay_id: str, step: int = 0) -> Dict[str, Any]:
|
|
game = _load_game(replay_id)
|
|
events = _load_events(replay_id)
|
|
state = _replay_state(game, events, step)
|
|
return state
|
|
|
|
|
|
@app.get("/replays/{replay_id}/export")
|
|
def export_replay(replay_id: str) -> Dict[str, Any]:
|
|
game = _load_game(replay_id)
|
|
events = _load_events(replay_id)
|
|
archive = _build_archive(game, events)
|
|
return archive.model_dump()
|
|
|
|
|
|
@app.post("/replays/import")
|
|
def import_replay(payload: ReplayArchive) -> Dict[str, Any]:
|
|
with session_scope() as session:
|
|
existing = session.get(Game, payload.id)
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Replay already exists")
|
|
now = dt.datetime.now(dt.timezone.utc)
|
|
game = Game(
|
|
id=payload.id,
|
|
name=f"Replay {payload.id[:8]}",
|
|
status="finished",
|
|
max_players=len(payload.slots) or len(payload.players),
|
|
created_by="import",
|
|
created_at=payload.created_at,
|
|
updated_at=now,
|
|
seed=payload.seed,
|
|
slots={"slots": [slot.model_dump() for slot in payload.slots]},
|
|
winner=payload.winner,
|
|
)
|
|
session.add(game)
|
|
for action in payload.actions:
|
|
event = GameEvent(
|
|
game_id=payload.id,
|
|
idx=action.idx,
|
|
ts=action.ts,
|
|
actor=action.actor,
|
|
action_type=action.action.type,
|
|
payload=action.action.payload,
|
|
applied=action.applied,
|
|
debug_payload=action.meta or None,
|
|
)
|
|
session.add(event)
|
|
return {"status": "imported", "id": payload.id}
|