Deephaven and asyncio for fantasy football

docker run --rm --name deephaven -p 10000:10000 --env START_OPTS=-DAuthHandlers=io.deephaven.auth.AnonymousAuthenticationHandler --env FPL_EMAIL="dev@mail.com" --env FPL_PASSWORD="Deephaven!" --env FPL_COOKIE="datadome=mduvowtanq4UMeaKA6aMj~0Uma~aLiBywaBSuvQLyTS6y8jIwvXUrZTb6HUqhZXclIxRUAOCjBQWbJ_ORlgZ16Z0tVAsHC1eK~uiVL76QfflE43G~oSCy6pCHAlzna;pl_profile=eyJzIjogIld6SXNPREUyTmpjMU5ERmQ6MXJmZjSyOkZIQ0FKREtkMFowYzJjY1pwckFSMUNJN2tkODdNc251dTFWWGNLaGstUkkiLCAidSI6IHsiaWQiOiA4MTY2NzU0MSwgImZuIjogIktlaXRoIiwgImxuIjogIk1jQ29uaWdseSIsICJmYyI6IDF9fQ==" ghcr.io/deephaven/server:0.34.2

import os

os.system("pip install aiohttp")
os.system("pip install fpl")
import aiohttp
from fpl import FPL
import asyncio
from threading import Thread
from dataclasses import dataclass
from typing import Callable
from concurrent.futures import CancelledError

from deephaven.table import Table
from deephaven.column import string_col, int_col
from deephaven import dtypes
from deephaven.table_factory import new_table
from deephaven.stream.table_publisher import table_publisher, TablePublisher


SLEEP_TIME = 1
SQUAD_SIZE = 15
GW_NUM = 0
BONUS_ADDED = False

@dataclass
class MiniLeagueEntry:
    id: int
    event_total: int
    player_name: str
    rank: int
    last_rank: int
    rank_sort: int
    total: int
    entry: int
    entry_name: str
    picks: dict
    transfers_cost: int
    live_game_week_points: int = 0
    live_total_points: int = 0


async def fetch(session, url, retries=10, cooldown=1):
    retries_count = 0
    while True:
        try:
            async with session.get(url) as response:
                result = await response.json()
                return result
        except aiohttp.client_exceptions.ContentTypeError:
            retries_count += 1

            if retries_count > retries:
                raise Exception(f"Could not fetch {url} after {retries} retries")

            if cooldown:
                await asyncio.sleep(cooldown)


async def get_event_status(session):
    stat = await fetch(
        session, "https://fantasy.premierleague.com/api/event-status/")
    global GW_NUM, BONUS_ADDED
    GW_NUM = stat["status"][0]["event"]
    BONUS_ADDED = stat["status"][0]["bonus_added"]


async def get_static_data(fpl_session, ml_id):
    # Get the current team standings from a mini-league at time of the last update
    teams = []
    classic_league = await fpl_session.get_classic_league(ml_id)
    for result in classic_league.standings["results"]:
        user = await fpl_session.get_user(result["entry"])
        picks = await user.get_picks(GW_NUM)
        result.update({"picks": picks})
        history = await user.get_user_history()
        result["transfers_cost"] = history[-1]["event_transfers_cost"]
        teams.append(MiniLeagueEntry(**result))
    return teams


async def calculate_live_score(ml_id, message_handler: Callable[[MiniLeagueEntry], None]):
    async with aiohttp.ClientSession() as session:
        await get_event_status(session)
        fpl_session = FPL(session)
        await fpl_session.login()
        list_of_teams = await get_static_data(fpl_session, ml_id)
        while True:
            game_week = await fpl_session.get_gameweek(GW_NUM, include_live=True, return_json=True)
            squad_range = range(SQUAD_SIZE)
            for team in list_of_teams:
                live_score = 0
                for position in squad_range:
                    player_stats = game_week["elements"][team.picks[GW_NUM][position]["element"]]["stats"]
                    live_score += player_stats["total_points"] * team.picks[GW_NUM][position]["multiplier"]
                team.live_game_week_points = live_score
                live_score -= (0 if BONUS_ADDED else team.transfers_cost)
                team.live_total_points = team.total + (live_score - team.event_total)
                message_handler(team)
            await asyncio.sleep(SLEEP_TIME)


def to_table(matches: list[MiniLeagueEntry]):
    return new_table(
        [
            int_col("Rank", [x.rank for x in matches]),
            int_col("LastRank", [x.last_rank for x in matches]),
            int_col("Entry", [x.entry for x in matches]),
            string_col("Team", [x.entry_name for x in matches]),
            string_col("Manager", [x.player_name.split()[0] for x in matches]),
            int_col("GW", [x.live_game_week_points for x in matches]),
            int_col("Hit", [x.transfers_cost for x in matches]),
            int_col("GWMinusHit", [(x.live_game_week_points - x.transfers_cost) for x in matches]),
            int_col("Total", [x.live_total_points for x in matches]),
        ]
    )


def create_matches(ml_id, event_loop) -> tuple[Table, Callable[[], None]]:
    on_shutdown_callbacks = []

    def on_shutdown():
        nonlocal on_shutdown_callbacks
        for c in on_shutdown_callbacks:
            c()

    my_matches: list[MiniLeagueEntry] = []

    def on_flush(tp: TablePublisher):
        nonlocal my_matches
        my_matches_copy = my_matches.copy()
        my_matches.clear()
        tp.add(to_table(my_matches_copy))

    table, publisher = table_publisher(
        f"Matches for {GW_NUM}",
        {
            "Rank": dtypes.int32,
            "LastRank": dtypes.int32,
            "Entry": dtypes.int32,
            "Team": dtypes.string,
            "Manager": dtypes.string,
            "GW": dtypes.int32,
            "Hit": dtypes.int32,
            "GWMinusHit": dtypes.int32,
            "Total": dtypes.int32,
        },
        on_flush_callback=on_flush,
        on_shutdown_callback=on_shutdown,
    )

    future = asyncio.run_coroutine_threadsafe(
        calculate_live_score(ml_id, my_matches.append), event_loop
    )

    def on_future_done(f):
        nonlocal publisher
        try:
            e = f.exception(timeout=0) or RuntimeError("completed")
        except CancelledError as c:
            e = RuntimeError("cancelled")
        publisher.publish_failure(e)

    future.add_done_callback(on_future_done)

    on_shutdown_callbacks.append(future.cancel)

    return table, future.cancel


my_event_loop = asyncio.new_event_loop()
Thread(target=my_event_loop.run_forever).start()


def subscribe_stats(ml_id):
    blink_table, on_done = create_matches(ml_id, my_event_loop)

    blink_table = (blink_table
                   .sort_descending(order_by="Total")
                   .update("Rank = i+1"))

    return blink_table, on_done


fpl_ml, fpl_cancel = subscribe_stats(291289)
fpl_ml = fpl_ml.format_columns(["Rank = Rank < LastRank ? DARK_BLUEGREEN : Rank > LastRank ? DARK_RED : NO_FORMATTING"]).layout_hints(hide=["LastRank","Entry"])

# call this to explicitly cancel
# fpl_cancel()