src/dbt_data_client.py

import duckdb
import os
import pandas as pd
import subprocess
import streamlit as st


DBT_DIR = os.path.join(os.path.dirname(__file__), "dbt_yfdash")
DUCKDB_PATH = os.path.join(DBT_DIR, "dev.duckdb")


FANTASY_POSITIONS = [
    "QB",
    "RB",
    "WR",
    "TE",
    "FB",
    "K",
]


def get_db_conn():
    return duckdb.connect(DUCKDB_PATH)


def run_dbt_build():
    process = subprocess.Popen(
        args=["dbt", "build"],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=DBT_DIR,
    )
    stdout, _ = process.communicate()
    st.text("\n".join(stdout.decode().split("\n")[1:][:-1]))


def execute_db_command_df(db_command: str):
    with get_db_conn() as db_conn:
        return db_conn.sql(db_command).df()


def get_current_tables() -> list[str]:
    current_tables_df = execute_db_command_df("SHOW TABLES")
    return current_tables_df["name"].tolist()


def get_snap_counts() -> pd.DataFrame:
    df = execute_db_command_df("SELECT * from snap_counts")
    df["fantasy_position"] = df["position"].isin(FANTASY_POSITIONS)
    return df


def get_play_by_play() -> pd.DataFrame:
    df = execute_db_command_df("SELECT * from play_by_play")
    return df


def get_player_stats() -> pd.DataFrame:
    df = execute_db_command_df("SELECT * from player_stats")
    return df


def get_ftn_charting() -> pd.DataFrame:
    df = execute_db_command_df("SELECT * from ftn_charting")
    return df


def get_depth_charts() -> pd.DataFrame:
    df = execute_db_command_df("SELECT * from depth_charts")
    return df


def get_nextgen_stats(stat_category: str) -> pd.DataFrame:
    df = execute_db_command_df(f"SELECT * from ngs_{stat_category}")
    return df


def get_pbp_participation() -> pd.DataFrame:
    # TODO - wire this up to actual play by play participation
    return pd.DataFrame()