146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
import logging
|
|
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from itertools import product
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.sql.expression import func
|
|
from typing import Callable, List, Mapping, Optional
|
|
|
|
from app.routers import players
|
|
from app.sql import models
|
|
from ..sql.database import get_db
|
|
from ..templates import jinja_templates
|
|
|
|
api_router = APIRouter(prefix="/suggest", tags=["suggest"])
|
|
html_router = APIRouter(
|
|
prefix="/suggest", include_in_schema=False, default_response_class=HTMLResponse
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
@api_router.get("/matchup")
|
|
def suggest_game(
|
|
player_ids: Optional[str] = None,
|
|
excluded_decks: Optional[str] = None,
|
|
strategy_id: Optional[str] = None,
|
|
db=Depends(get_db),
|
|
):
|
|
if not strategy_id:
|
|
strategy_id = "smallest_score_spread"
|
|
if strategy_id not in STRATEGIES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Cannot generate a matchup recommendation with strategy {strategy_id}. Available options: {STRATEGIES.keys()}",
|
|
)
|
|
|
|
# TODO - this should be an error, surely!
|
|
player_ids_as_list = list(map(int, player_ids.split(","))) if player_ids else []
|
|
excluded_decks_as_list = (
|
|
list(map(int, excluded_decks.split(","))) if excluded_decks else []
|
|
)
|
|
LOGGER.info(f"{player_ids=}")
|
|
LOGGER.info(f"{excluded_decks=}")
|
|
return STRATEGIES[strategy_id].strategy(
|
|
player_ids_as_list, excluded_decks_as_list, db
|
|
)
|
|
|
|
|
|
def smallest_score_spread_strategy(
|
|
player_ids: List[int], excluded_decks: List[int], db: Session
|
|
):
|
|
row_number_column = (
|
|
func.row_number()
|
|
.over(
|
|
partition_by=[models.Deck.name],
|
|
order_by=models.EloScore.id.desc(),
|
|
)
|
|
.label("row_number")
|
|
)
|
|
|
|
sub_query = (
|
|
db.query(
|
|
models.Deck.id, models.Deck.name, models.Player.name, models.EloScore.score
|
|
)
|
|
.join(models.Player, models.Deck.owner_id == models.Player.id)
|
|
.join(models.EloScore, models.Deck.id == models.EloScore.deck_id)
|
|
.filter(models.Deck.owner_id.in_(player_ids))
|
|
.filter(models.Deck.id.not_in(excluded_decks))
|
|
.add_column(row_number_column)
|
|
)
|
|
|
|
sub_query = sub_query.subquery()
|
|
query = db.query(sub_query).filter(sub_query.c.row_number == 1)
|
|
results = query.all()
|
|
|
|
decks_by_player = defaultdict(list)
|
|
for row in results:
|
|
LOGGER.info(f"Processing {row}")
|
|
decks_by_player[row[2]].append(
|
|
{"id": row[0], "name": row[1], "latest_score": row[3]}
|
|
)
|
|
|
|
# Tried using `heapq` for this logic but it keps producing duplicates :shrug:
|
|
matchups = []
|
|
for info_tuple in product(*decks_by_player.values()):
|
|
scores = [info["latest_score"] for info in info_tuple]
|
|
score_spread = max([abs(x - y) for (x, y) in product(scores, repeat=2)])
|
|
matchup = Matchup(
|
|
# Oof, there's _definitely_ a better way of doing this...
|
|
participants=list(
|
|
zip(decks_by_player.keys(), [info["name"] for info in info_tuple])
|
|
),
|
|
score=100 - score_spread,
|
|
description=f"Score spread of {score_spread}, between {max(scores)} and {min(scores)}",
|
|
)
|
|
matchups.append((score_spread, matchup))
|
|
|
|
matchups.sort(key=lambda tup: tup[0])
|
|
return [item[1] for item in matchups[:5]]
|
|
|
|
|
|
# `order=True` so that tuples containing these can be put into a heap
|
|
@dataclass(order=True)
|
|
class Matchup:
|
|
# `participants` is a List of dictionaries containing `player_name` and `deck_name`
|
|
participants: List[Mapping]
|
|
score: int
|
|
description: str
|
|
|
|
|
|
@dataclass
|
|
class StrategyDefinition:
|
|
strategy_id: str
|
|
friendly_name: str
|
|
# Parameters are:
|
|
# `List[int]` of player_ids
|
|
# `List[int]` of excluded_decks
|
|
# `Session`, a db-connection
|
|
strategy: Callable[[List[int], List[int], Session], List[Matchup]]
|
|
description: Optional[str] = None
|
|
|
|
|
|
STRATEGIES: Mapping[str, StrategyDefinition] = {
|
|
strat_def.strategy_id: strat_def
|
|
for strat_def in [
|
|
StrategyDefinition(
|
|
strategy_id="smallest_score_spread",
|
|
friendly_name="Smallest Score Spread",
|
|
strategy=smallest_score_spread_strategy,
|
|
)
|
|
]
|
|
}
|
|
|
|
|
|
@html_router.get("/matchup")
|
|
def suggest_matchup_html(request: Request, db=Depends(get_db)):
|
|
return jinja_templates.TemplateResponse(
|
|
request,
|
|
"suggest/matchup.html",
|
|
{"players": players.list_players(db=db), "strategies": STRATEGIES},
|
|
)
|