219 lines
8.0 KiB
Python
219 lines
8.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query, Response
|
|
from typing import Literal, Optional, List
|
|
import logging
|
|
import pydantic
|
|
from pydantic import validator, root_validator
|
|
|
|
from ..db_engine import db, PitchingCardRatings, model_to_dict, chunked, PitchingCard, Player, query_to_csv
|
|
from ..dependencies import oauth2_scheme, valid_token, LOG_DATA
|
|
|
|
logging.basicConfig(
|
|
filename=LOG_DATA['filename'],
|
|
format=LOG_DATA['format'],
|
|
level=LOG_DATA['log_level']
|
|
)
|
|
|
|
router = APIRouter(
|
|
prefix='/api/v2/pitchingcardratings',
|
|
tags=['pitchingcardratings']
|
|
)
|
|
|
|
|
|
class PitchingCardRatingsModel(pydantic.BaseModel):
|
|
pitchingcard_id: int
|
|
vs_hand: Literal['R', 'L', 'vR', 'vL']
|
|
homerun: float = 0.0
|
|
bp_homerun: float = 0.0
|
|
triple: float = 0.0
|
|
double_three: float = 0.0
|
|
double_two: float = 0.0
|
|
double_cf: float = 0.0
|
|
single_two: float = 0.0
|
|
single_one: float = 0.0
|
|
single_center: float = 0.0
|
|
bp_single: float = 0.0
|
|
hbp: float = 0.0
|
|
walk: float = 0.0
|
|
strikeout: float = 0.0
|
|
flyout_lf_b: float = 0.0
|
|
flyout_cf_b: float = 0.0
|
|
flyout_rf_b: float = 0.0
|
|
groundout_a: float = 0.0
|
|
groundout_b: float = 0.0
|
|
xcheck_p: float = 0.0
|
|
xcheck_c: float = 0.0
|
|
xcheck_1b: float = 0.0
|
|
xcheck_2b: float = 0.0
|
|
xcheck_3b: float = 0.0
|
|
xcheck_ss: float = 0.0
|
|
xcheck_lf: float = 0.0
|
|
xcheck_cf: float = 0.0
|
|
xcheck_rf: float = 0.0
|
|
avg: float = 0.0
|
|
obp: float = 0.0
|
|
slg: float = 0.0
|
|
|
|
@validator("avg", always=True)
|
|
def avg_validator(cls, v, values, **kwargs):
|
|
return (values['homerun'] + values['bp_homerun'] / 2 + values['triple'] + values['double_three'] +
|
|
values['double_two'] + values['double_cf'] + values['single_two'] + values['single_one'] +
|
|
values['single_center'] + values['bp_single'] / 2) / 108
|
|
|
|
@validator("obp", always=True)
|
|
def obp_validator(cls, v, values, **kwargs):
|
|
return ((values['hbp'] + values['walk']) / 108) + values['avg']
|
|
|
|
@validator("slg", always=True)
|
|
def slg_validator(cls, v, values, **kwargs):
|
|
return (values['homerun'] * 4 + values['bp_homerun'] * 2 + values['triple'] * 3 + values['double_three'] * 2 +
|
|
values['double_two'] * 2 + values['double_cf'] * 2 + values['single_two'] + values['single_one'] +
|
|
values['single_center'] + values['bp_single'] / 2) / 108
|
|
|
|
@root_validator
|
|
def validate_chance_total(cls, values):
|
|
total_chances = (
|
|
values['homerun'] + values['bp_homerun'] + values['triple'] + values['double_three'] +
|
|
values['double_two'] + values['double_cf'] + values['single_two'] + values['single_one'] +
|
|
values['single_center'] + values['bp_single'] + values['hbp'] + values['walk'] +
|
|
values['strikeout'] + values['flyout_lf_b'] + values['flyout_cf_b'] + values['flyout_rf_b'] +
|
|
values['groundout_a'] + values['groundout_b'] + values['xcheck_p'] + values['xcheck_c'] +
|
|
values['xcheck_1b'] + values['xcheck_2b'] + values['xcheck_3b'] + values['xcheck_ss'] +
|
|
values['xcheck_lf'] + values['xcheck_cf'] + values['xcheck_rf'])
|
|
|
|
if round(total_chances) != 108:
|
|
raise ValueError("Must have exactly 108 chances on the card")
|
|
return values
|
|
|
|
|
|
class RatingsList(pydantic.BaseModel):
|
|
ratings: List[PitchingCardRatingsModel]
|
|
|
|
|
|
@router.get('')
|
|
async def get_card_ratings(
|
|
pitchingcard_id: list = Query(default=None), vs_hand: Literal['R', 'L', 'vR', 'vL'] = None,
|
|
short_output: bool = False, csv: bool = False, cardset_id: list = Query(default=None),
|
|
token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to pull card ratings.'
|
|
)
|
|
|
|
all_ratings = PitchingCardRatings.select()
|
|
|
|
if pitchingcard_id is not None:
|
|
all_ratings = all_ratings.where(PitchingCardRatings.pitchingcard_id << pitchingcard_id)
|
|
if vs_hand is not None:
|
|
all_ratings = all_ratings.where(PitchingCardRatings.vs_hand == vs_hand[-1])
|
|
if cardset_id is not None:
|
|
set_players = Player.select(Player.player_id).where(Player.cardset_id << cardset_id)
|
|
set_cards = PitchingCard.select(PitchingCard.id).where(PitchingCard.player << set_players)
|
|
all_ratings = all_ratings.where(PitchingCardRatings.pitchingcard << set_cards)
|
|
|
|
if csv:
|
|
return_val = query_to_csv(all_ratings)
|
|
db.close()
|
|
return Response(content=return_val, media_type='text/csv')
|
|
|
|
else:
|
|
return_val = {'count': all_ratings.count(), 'ratings': [
|
|
model_to_dict(x, recurse=not short_output) for x in all_ratings
|
|
]}
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@router.get('/{ratings_id}')
|
|
async def get_one_rating(ratings_id: int, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to pull card ratings.'
|
|
)
|
|
|
|
this_rating = PitchingCardRatings.get_or_none(PitchingCardRatings.id == ratings_id)
|
|
if this_rating is None:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'PitchingCardRating id {ratings_id} not found')
|
|
|
|
r_data = model_to_dict(this_rating)
|
|
db.close()
|
|
return r_data
|
|
|
|
|
|
@router.get('/player/{player_id}')
|
|
async def get_player_ratings(player_id: int, variant: list = Query(default=None), short_output: bool = False):
|
|
all_cards = PitchingCard.select().where(PitchingCard.player_id == player_id).order_by(PitchingCard.variant)
|
|
if variant is not None:
|
|
all_cards = all_cards.where(PitchingCard.variant << variant)
|
|
|
|
all_ratings = PitchingCardRatings.select().where(PitchingCardRatings.pitchingcard << all_cards)
|
|
|
|
return_val = {'count': all_ratings.count(), 'ratings': [
|
|
model_to_dict(x, recurse=not short_output) for x in all_ratings
|
|
]}
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@router.put('')
|
|
async def put_ratings(ratings: RatingsList, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to post card ratings.'
|
|
)
|
|
|
|
new_ratings = []
|
|
updates = 0
|
|
for x in ratings.ratings:
|
|
try:
|
|
PitchingCardRatings.get(
|
|
(PitchingCardRatings.pitchingcard_id == x.pitchingcard_id) & (PitchingCardRatings.vs_hand == x.vs_hand)
|
|
)
|
|
updates += PitchingCardRatings.update(x.dict()).where(
|
|
(PitchingCardRatings.pitchingcard_id == x.pitchingcard_id) & (PitchingCardRatings.vs_hand == x.vs_hand)
|
|
).execute()
|
|
except PitchingCardRatings.DoesNotExist:
|
|
new_ratings.append(x.dict())
|
|
|
|
with db.atomic():
|
|
for batch in chunked(new_ratings, 30):
|
|
PitchingCardRatings.insert_many(batch).on_conflict_replace().execute()
|
|
|
|
db.close()
|
|
return f'Updated ratings: {updates}; new ratings: {len(new_ratings)}'
|
|
|
|
|
|
@router.delete('/{ratings_id}')
|
|
async def delete_rating(
|
|
ratings_id: int, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to post card ratings.'
|
|
)
|
|
|
|
this_rating = PitchingCardRatings.get_or_none(PitchingCardRatings.id == ratings_id)
|
|
if this_rating is None:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'PitchingCardRating id {ratings_id} not found')
|
|
|
|
count = this_rating.delete_instance()
|
|
db.close()
|
|
|
|
if count == 1:
|
|
return f'Rating {this_rating} has been deleted'
|
|
else:
|
|
raise HTTPException(status_code=500, detail=f'Rating {this_rating} could not be deleted')
|
|
|