paper-dynasty-discord/in_game/data_cache.py

import datetime
from dataclasses import dataclass
import logging
from typing import Optional, Literal

from api_calls import db_get
from helpers import PD_SEASON

PLAYER_CACHE = {}
TEAM_CACHE = {}
BATTINGCARD_CACHE = {}  # { <player_id: int>: { <variant: int>: BattingWrapper } }
PITCHINGCARD_CACHE = {}  # { <player_id: int>: { <variant: int>: PitchingWrapper } }
logger = logging.getLogger("discord_app")


@dataclass
class Player:
    player_id: int
    p_name: str
    cost: int
    image: str
    mlbclub: str
    franchise: str
    cardset: dict
    set_num: int
    rarity: dict
    pos_1: str
    description: str
    quantity: Optional[int] = 999
    image2: Optional[str] = None
    pos_2: Optional[str] = None
    pos_3: Optional[str] = None
    pos_4: Optional[str] = None
    pos_5: Optional[str] = None
    pos_6: Optional[str] = None
    pos_7: Optional[str] = None
    pos_8: Optional[str] = None
    headshot: Optional[str] = None
    vanity_card: Optional[str] = None
    strat_code: Optional[str] = None
    bbref_id: Optional[str] = None
    fangr_id: Optional[str] = None
    mlbplayer: Optional[dict] = None
    created: datetime.datetime = datetime.datetime.now()

    def to_dict(self):
        return {
            "player_id": self.player_id,
            "p_name": self.p_name,
            "cost": self.cost,
            "image": self.image,
            "mlbclub": self.mlbclub,
            "franchise": self.franchise,
            "cardset": self.cardset,
            "set_num": self.set_num,
            "rarity": self.rarity,
            "pos_1": self.pos_1,
            "description": self.description,
            "quantity": self.quantity,
            "image2": self.image2,
            "pos_2": self.pos_2,
            "pos_3": self.pos_3,
            "pos_4": self.pos_4,
            "pos_5": self.pos_5,
            "pos_6": self.pos_6,
            "pos_7": self.pos_7,
            "pos_8": self.pos_8,
            "headshot": self.headshot,
            "vanity_card": self.vanity_card,
            "strat_code": self.strat_code,
            "bbref_id": self.bbref_id,
            "fangr_id": self.fangr_id,
            "mlbplayer": self.mlbplayer,
        }


@dataclass
class Team:
    id: int
    abbrev: str
    sname: str
    lname: str
    gmid: int
    gmname: str
    gsheet: str
    wallet: int
    team_value: int
    collection_value: int
    logo: str
    color: str
    season: int
    event: bool
    career: int
    ranking: int
    has_guide: bool
    is_ai: bool
    created: datetime.datetime = datetime.datetime.now()

    def to_dict(self):
        return {
            "id": self.id,
            "abbrev": self.abbrev,
            "sname": self.sname,
            "lname": self.lname,
            "gmid": self.gmid,
            "gmname": self.gmname,
            "gsheet": self.gsheet,
            "wallet": self.wallet,
            "team_value": self.team_value,
            "collection_value": self.collection_value,
            "logo": self.logo,
            "color": self.color,
            "season": self.season,
            "event": self.event,
            "career": self.career,
            "ranking": self.ranking,
            "has_guide": self.has_guide,
            "is_ai": self.is_ai,
        }


@dataclass
class BattingCard:
    player_id: int
    variant: int
    steal_low: int
    steal_high: int
    steal_auto: bool
    steal_jump: float
    bunting: str
    hit_and_run: str
    running: int
    offense_col: int
    hand: str


@dataclass
class CardPosition:
    player_id: int
    variant: int
    position: Literal["P", "C", "1B", "2B", "3B", "SS", "LF", "CF", "RF", "DH"]
    innings: int
    range: int
    error: int
    arm: Optional[int] = None
    pb: Optional[int] = None
    overthrow: Optional[int] = None

    def to_dict(self):
        return_val = {
            "player_id": self.player_id,
            "variant": self.variant,
            "position": self.position,
            "innings": self.innings,
            "range": self.range,
            "error": self.error,
        }
        if self.arm is not None:
            return_val["arm"] = self.arm
        if self.pb is not None:
            return_val["pb"] = self.pb
        if self.overthrow is not None:
            return_val["overthrow"] = self.overthrow

        return return_val


@dataclass
class BattingRatings:
    homerun: float
    bp_homerun: float
    triple: float
    double_three: float
    double_two: float
    double_pull: float
    single_two: float
    single_one: float
    single_center: float
    bp_single: float
    hbp: float
    walk: float
    strikeout: float
    lineout: float
    popout: float
    flyout_a: float
    flyout_bq: float
    flyout_lf_b: float
    flyout_rf_b: float
    groundout_a: float
    groundout_b: float
    groundout_c: float
    avg: float
    obp: float
    slg: float
    pull_rate: float
    center_rate: float
    slap_rate: float


@dataclass
class BattingWrapper:
    card: BattingCard
    ratings_vl: BattingRatings
    ratings_vr: BattingRatings
    created: datetime.datetime = datetime.datetime.now()


@dataclass
class PitchingCard:
    player_id: int
    variant: int
    balk: int
    wild_pitch: int
    hold: int
    starter_rating: int
    relief_rating: int
    batting: str
    offense_col: int
    hand: str
    closer_rating: Optional[int] = None


@dataclass
class PitchingRatings:
    homerun: float
    bp_homerun: float
    triple: float
    double_three: float
    double_two: float
    double_cf: float
    single_two: float
    single_one: float
    single_center: float
    bp_single: float
    hbp: float
    walk: float
    strikeout: float
    flyout_lf_b: float
    flyout_cf_b: float
    flyout_rf_b: float
    groundout_a: float
    groundout_b: float
    xcheck_p: float
    xcheck_c: float
    xcheck_1b: float
    xcheck_2b: float
    xcheck_3b: float
    xcheck_ss: float
    xcheck_lf: float
    xcheck_cf: float
    xcheck_rf: float
    avg: float
    obp: float
    slg: float


@dataclass
class PitchingWrapper:
    card: PitchingCard
    ratings_vl: PitchingRatings
    ratings_vr: PitchingRatings
    created: datetime.datetime = datetime.datetime.now()


async def get_pd_player(
    player_id, as_dict: Optional[bool] = True, skip_cache: bool = False
):
    if player_id in PLAYER_CACHE and not skip_cache:
        tdelta = datetime.datetime.now() - PLAYER_CACHE[player_id].created
        if tdelta.total_seconds() < 1209600:
            logger.debug(f"this_player: {PLAYER_CACHE[player_id]}")
            if as_dict:
                return PLAYER_CACHE[player_id].to_dict()
            else:
                return PLAYER_CACHE[player_id]
        else:
            logger.error(f"Refreshing player {player_id} in cache...")

    this_player = await db_get("players", object_id=player_id)
    for bad_key in ["mlbplayer", "paperdex"]:
        if bad_key in this_player:
            del this_player[bad_key]
    logger.debug(f"this_player: {this_player}")
    PLAYER_CACHE[player_id] = Player(**this_player)

    if as_dict:
        return this_player
    return PLAYER_CACHE[player_id]


async def get_pd_team(team_or_gm_id, as_dict: bool = True, skip_cache: bool = False):
    if team_or_gm_id in TEAM_CACHE and not skip_cache:
        tdelta = datetime.datetime.now() - TEAM_CACHE[team_or_gm_id].created
        if tdelta.total_seconds() < 1209600:
            logger.info(f"this_team: {TEAM_CACHE[team_or_gm_id]}")
            if as_dict:
                return TEAM_CACHE[team_or_gm_id].to_dict()
            else:
                return TEAM_CACHE[team_or_gm_id]
        else:
            logger.error(f"Refreshing team {team_or_gm_id} in cache...")

    this_team = await db_get(
        "teams", object_id=team_or_gm_id, params=[("inc_packs", False)]
    )
    if this_team is None:
        t_query = await db_get(
            "teams", params=[("season", PD_SEASON), ("gm_id", team_or_gm_id)]
        )
        if t_query is None:
            raise KeyError(f"No team found for ID {team_or_gm_id}")
        this_team = t_query["teams"][0]

    logger.info(f"this_team: {this_team}")
    TEAM_CACHE[team_or_gm_id] = Team(**this_team)

    if as_dict:
        return this_team
    return TEAM_CACHE[team_or_gm_id]


async def get_pd_battingcard(player_id: int, variant: Optional[int] = 0):
    if player_id in BATTINGCARD_CACHE and variant in BATTINGCARD_CACHE[player_id]:
        tdelta = datetime.datetime.now() - BATTINGCARD_CACHE[player_id][variant].created
        if tdelta.total_seconds() < 609600:
            logger.info(f"this_battingcard: {BATTINGCARD_CACHE[player_id][variant]}")
            return BATTINGCARD_CACHE[player_id][variant]

    vl_data, vr_data = None, None
    r_query = await db_get(f"battingcardratings/player/{player_id}")
    if r_query["count"] > 0:
        for row in r_query["ratings"]:
            if row["battingcard"]["variant"] == variant:
                if row["vs_hand"].lower() == "r":
                    vr_data = row
                elif row["vs_hand"].lower() == "l":
                    vl_data = row
            if vl_data is not None and vr_data is not None:
                break

    if None in [vl_data, vr_data]:
        raise KeyError(
            f"Could not find batting card ratings for player {player_id} variant {variant}"
        )

    logger.info("prepping the batting card")
    bc_data = vl_data["battingcard"]
    bc_data["player_id"] = player_id
    del bc_data["id"], bc_data["player"]
    this_bc = BattingCard(**bc_data)

    logger.info("prepping the vl ratings")
    vl_ratings = vl_data
    del vl_ratings["battingcard"], vl_ratings["id"], vl_ratings["vs_hand"]
    this_vl = BattingRatings(**vl_ratings)

    logger.info("prepping the vl ratings")
    vr_ratings = vr_data
    del vr_ratings["battingcard"], vr_ratings["id"], vr_ratings["vs_hand"]
    this_vr = BattingRatings(**vr_ratings)

    logger.info("prepping the wrapper")
    this_wrapper = BattingWrapper(card=this_bc, ratings_vl=this_vl, ratings_vr=this_vr)

    if player_id in BATTINGCARD_CACHE:
        BATTINGCARD_CACHE[player_id][variant] = this_wrapper
    else:
        BATTINGCARD_CACHE[player_id] = {variant: this_wrapper}

    return this_wrapper


async def get_pd_pitchingcard(player_id: int, variant: Optional[int] = 0):
    if player_id in PITCHINGCARD_CACHE and variant in PITCHINGCARD_CACHE[player_id]:
        tdelta = (
            datetime.datetime.now() - PITCHINGCARD_CACHE[player_id][variant].created
        )
        if tdelta.total_seconds() < 609600:
            logger.debug(f"this_pitchingcard: {PITCHINGCARD_CACHE[player_id][variant]}")
            return PITCHINGCARD_CACHE[player_id][variant]

    vl_data, vr_data = None, None
    r_query = await db_get(f"pitchingcardratings/player/{player_id}")
    if r_query["count"] > 0:
        for row in r_query["ratings"]:
            if row["pitchingcard"]["variant"] == variant:
                if row["vs_hand"].lower() == "r":
                    vr_data = row
                elif row["vs_hand"].lower() == "l":
                    vl_data = row
            if vl_data is not None and vr_data is not None:
                break

    if None in [vl_data, vr_data]:
        raise KeyError(
            f"Could not find pitching card ratings for player {player_id} variant {variant}"
        )

    logger.debug("prepping the pitching card")
    pc_data = vl_data["pitchingcard"]
    pc_data["player_id"] = player_id
    del pc_data["id"], pc_data["player"]
    this_pc = PitchingCard(**pc_data)

    logger.debug("prepping the vl ratings")
    vl_ratings = vl_data
    del vl_ratings["pitchingcard"], vl_ratings["id"], vl_ratings["vs_hand"]
    this_vl = PitchingRatings(**vl_ratings)

    logger.debug("prepping the vr ratings")
    vr_ratings = vr_data
    del vr_ratings["pitchingcard"], vr_ratings["id"], vr_ratings["vs_hand"]
    this_vr = PitchingRatings(**vr_ratings)

    logger.debug("prepping the wrapper")
    this_wrapper = PitchingWrapper(card=this_pc, ratings_vl=this_vl, ratings_vr=this_vr)

    if player_id in PITCHINGCARD_CACHE:
        PITCHINGCARD_CACHE[player_id][variant] = this_wrapper
    else:
        PITCHINGCARD_CACHE[player_id] = {variant: this_wrapper}

    return this_wrapper