import logging from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks from fastapi_camelcase import CamelModel from pydantic import conint, constr from sqlalchemy.orm import Session from .. import scoring from ..schemas import GameConfig, GameModeEnum, Guess, ScoreMethodEnum from ..db import get_db, queries, models from ..point_gen import points, ExhaustedSourceError, reverse_geocode logger = logging.getLogger(__name__) router = APIRouter() class LinkedGame(CamelModel): linked_game: str class JoinGame(CamelModel): player_name: constr(min_length=1) class GuessResult(CamelModel): distance: int = None score: int = 0 total_score: int @router.put("") async def create_game(config: GameConfig, bg: BackgroundTasks, db: Session = Depends(get_db)): try: coords = await points.get_points(config) except ExhaustedSourceError: logger.exception(f"Failed to generate enough points for {config}") raise HTTPException(status_code=501, detail="Sufficient points could not be generated quickly enough") game_id = queries.create_game(db, config, coords) bg.add_task(points.restock_source, config) return { "gameId": game_id } def get_game(game_id: str, db: Session = Depends(get_db)) -> models.Game: game = queries.get_game(db, game_id) if game is None: raise HTTPException(status_code=404, detail="Game not found") return game @router.get("/{game_id}/config", response_model=GameConfig) def get_game_config(game: models.Game = Depends(get_game)): return game @router.get("/{game_id}/coords") def get_game_coords(game: models.Game = Depends(get_game)): return { str(coord.round_number): { "lat": coord.latitude, "lng": coord.longitude, "country": coord.country_code, } for coord in game.coordinates } @router.post("/{game_id}/join") def join_game(game_id: str, join: JoinGame, db: Session = Depends(get_db)): get_game(game_id, db) # confirm this game exists player_id = queries.join_game(db, game_id, join.player_name) if player_id is None: raise HTTPException(status_code=409, detail="Player name in use") return { "playerId": player_id } @router.get("/{game_id}/players") def get_players(game: models.Game = Depends(get_game)): return { "players": [ { "name": p.player_name, "currentRound": queries.get_next_round_number(p), "totalScore": queries.get_total_score(p), "guesses": { str(g.round_number): { "lat": g.latitude, "lng": g.longitude, "country": g.country_code, "score": g.round_score, "timeRemaining": g.time_remaining, } for g in p.guesses } } for p in game.players ] } def get_player(game_id: str, player_id: str, db: Session = Depends(get_db)) -> models.Player: player = queries.get_player(db, player_id) if player is None: raise HTTPException(status_code=404, detail="Player not found") if player.game_id != game_id: raise HTTPException(status_code=404, detail="Player not in game") return player @router.get("/{game_id}/players/{player_id}/current") def get_current_round(db: Session = Depends(get_db), player: models.Player = Depends(get_player)): coord = queries.get_next_coordinate(db, player) if coord is None: return { "currentRound": None, "coord": None, "timer": None, } return { "currentRound": coord.round_number, "coord": { "lat": coord.latitude, "lng": coord.longitude, "country": coord.country_code, }, "timer": queries.get_next_round_time(player), } @router.put("/{game_id}/linked", status_code=204) def set_linked_game(game_id: str, linked_game: LinkedGame, db: Session = Depends(get_db)): queries.link_game(db, game_id, linked_game.linked_game) @router.get("/{game_id}/linked") def get_linked_game(game: models.Game = Depends(get_game)): return { "linkedGame": game.linked_game } @router.get("/{game_id}/round/{round_number}/first") def get_first_submitter(game_id: str, round_number: conint(gt=0), db: Session = Depends(get_db)): return { "first": queries.get_first_submitter(db, game_id, round_number) } @router.post("/{game_id}/round/{round_number}/guess/{player_id}", response_model=GuessResult) async def submit_guess(round_number: conint(gt=0), guess: Guess, db: Session = Depends(get_db), game: models.Game = Depends(get_game), player: models.Player = Depends(get_player)): target = queries.get_coordinate(db, player.game_id, round_number) country_code = await reverse_geocode(guess.lat, guess.lng) match game.score_method: case ScoreMethodEnum.country_distance: score_fn = scoring.score if country_code == target.country_code else scoring.score_pow score, distance = score_fn((target.latitude, target.longitude), (guess.lat, guess.lng)) case ScoreMethodEnum.country_race: score = scoring.score_country_race(target.country_code, country_code, guess.time_remaining, game.timer) distance = None case ScoreMethodEnum.hard: score, distance = scoring.score_hard((target.latitude, target.longitude), (guess.lat, guess.lng)) case ScoreMethodEnum.nightmare: score, distance = scoring.score_nightmare((target.latitude, target.longitude), (guess.lat, guess.lng)) case ScoreMethodEnum.ramp: score, distance = scoring.score((target.latitude, target.longitude), (guess.lat, guess.lng)) score *= 1 + ((round_number - 1) * 0.5) case ScoreMethodEnum.ramp_hard: score, distance = scoring.score_hard((target.latitude, target.longitude), (guess.lat, guess.lng)) score *= 1 + ((round_number - 1) * 0.5) case _: score, distance = scoring.score((target.latitude, target.longitude), (guess.lat, guess.lng)) if game.round_point_cap is not None: score = min(score, max(0, game.round_point_cap - sum(g.round_score for p in game.players for g in p.guesses if g.round_number == round_number))) if game.game_mode == GameModeEnum.gun_game and round_number == game.rounds and queries.get_first_submitter(db, game.game_id, round_number) is None: # first to submit in the last round of gun game gets 10k bonus points to ensure they are the winner of the whole game score += 10_000 added = queries.add_guess(db, guess, player, country_code, round_number, score) if not added: raise HTTPException(status_code=409, detail="Already submitted guess for this round") total_score = queries.get_total_score(player) return GuessResult(distance=distance, score=score, total_score=total_score) @router.post("/{game_id}/round/{round_number}/timeout/{player_id}", response_model=GuessResult, response_model_include={"total_score"}) def submit_timeout(round_number: conint(gt=0), db: Session = Depends(get_db), player: models.Player = Depends(get_player)): added = queries.add_timeout(db, player, round_number) if not added: raise HTTPException(status_code=409, detail="Already submitted guess for this round") total_score = queries.get_total_score(player) return GuessResult(total_score=total_score)