Spaces:

MCP-1st-Birthday
/

Agentic-Codenames-Arena

Running

Agentic-Codenames-Arena / support /database.py

lucadipalma

adding files

aa2d45f 11 months ago

7.42 kB

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
from contextlib import contextmanager
import threading

Base = declarative_base()

# Thread-safe engine creation
engine = create_engine(
    'sqlite:///codenames_games.db',
    connect_args={'check_same_thread': False},
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=20
)

# Thread-local session
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
session_lock = threading.Lock()


@contextmanager
def get_db_session():
    """Thread-safe database session context manager"""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()


class Game(Base):
    __tablename__ = 'games'

    id = Column(Integer, primary_key=True, autoincrement=True)
    timestamp = Column(DateTime, default=datetime.utcnow, nullable=False)
    winner_team = Column(String(10), nullable=False)  # 'red' or 'blue'
    red_score = Column(Integer, nullable=False)  # remaining words
    blue_score = Column(Integer, nullable=False)  # remaining words
    reason = Column(String(50), nullable=True)  # None or 'killer'
    red_team_preset = Column(String(50), nullable=True)  # 'openai', 'google', etc.
    blue_team_preset = Column(String(50), nullable=True)

    # Relationships
    players = relationship("GamePlayer", back_populates="game", cascade="all, delete-orphan")


class GamePlayer(Base):
    __tablename__ = 'game_players'

    id = Column(Integer, primary_key=True, autoincrement=True)
    game_id = Column(Integer, ForeignKey('games.id'), nullable=False)
    player_name = Column(String(100), nullable=False)
    team = Column(String(10), nullable=False)  # 'red' or 'blue'
    role = Column(String(20), nullable=False)  # 'boss', 'captain', 'player'
    model_name = Column(String(100), nullable=False)
    is_human = Column(Boolean, default=False)

    # Relationship
    game = relationship("Game", back_populates="players")


def init_database():
    """Initialize the database tables"""
    Base.metadata.create_all(bind=engine)


def save_game_to_db(game, winner_and_score):
    """
    Save a completed game to the database
    Args:
        game: Game object with players and board info
        winner_and_score: tuple (winner_team, scores, reason)
    """
    winner_team = winner_and_score[0]
    scores = winner_and_score[1]
    reason = winner_and_score[2] if len(winner_and_score) > 2 else None

    with get_db_session() as session:
        # Create game record
        new_game = Game(
            winner_team=winner_team,
            red_score=scores[0],
            blue_score=scores[1],
            reason=reason,
            red_team_preset=game.red_team_choice,
            blue_team_preset=game.blue_team_choice
        )
        session.add(new_game)
        session.flush()  # Get the game ID

        # Create player records
        for player in game.players:
            is_human = player.model_name.lower() == "human brain"
            game_player = GamePlayer(
                game_id=new_game.id,
                player_name=player.name,
                team=player.team,
                role=player.role,
                model_name=player.model_name,
                is_human=is_human
            )
            session.add(game_player)

        session.commit()
        return new_game.id


def get_stats():
    """
    Retrieve comprehensive statistics from the database
    Returns:
        dict: Contains total_games, game_history, model_stats, provider_stats
    """
    with get_db_session() as session:
        # Total games
        total_games = session.query(Game).count()

        # Game history
        games = session.query(Game).order_by(Game.timestamp.desc()).all()
        game_history = []

        for game in games:
            game_info = {
                'id': game.id,
                'timestamp': game.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
                'winner': game.winner_team,
                'red_score': game.red_score,
                'blue_score': game.blue_score,
                'reason': game.reason,
                'red_preset': game.red_team_preset,
                'blue_preset': game.blue_team_preset,
                'players': []
            }

            for player in game.players:
                game_info['players'].append({
                    'name': player.player_name,
                    'team': player.team,
                    'role': player.role,
                    'model': player.model_name,
                    'is_human': player.is_human
                })

            game_history.append(game_info)

        # Model stats (individual)
        model_stats = {}
        all_players = session.query(GamePlayer).all()

        for player in all_players:
            model = player.model_name
            if model not in model_stats:
                model_stats[model] = {'wins': 0, 'losses': 0, 'games': 0}

            model_stats[model]['games'] += 1
            if player.team == player.game.winner_team:
                model_stats[model]['wins'] += 1
            else:
                model_stats[model]['losses'] += 1

        # Provider stats (grouped)
        provider_stats = {
            'openai': {'wins': 0, 'losses': 0, 'games': 0},
            'google': {'wins': 0, 'losses': 0, 'games': 0},
            'anthropic': {'wins': 0, 'losses': 0, 'games': 0},
            'opensource': {'wins': 0, 'losses': 0, 'games': 0},
            'human': {'wins': 0, 'losses': 0, 'games': 0}
        }

        # Map models to providers
        model_to_provider = {
            'gpt-5': 'openai',
            'gpt-5-mini': 'openai',
            'gpt-5-nano': 'openai',
            'gpt-4.1-nano': 'openai',
            'gemini-2.5-pro': 'google',
            'gemini-2.5-flash': 'google',
            'gemini-2.0-flash-001': 'google',
            'gemini-2.0-flash-lite-001': 'google',
            'claude-sonnet-4-5-20250929':'anthropic',
            'claude-3-7-sonnet-20250219': 'anthropic',
            'claude-3-5-haiku-20241022': 'anthropic',
            'claude-3-haiku-20240307': 'anthropic',
            'deepseek-ai/DeepSeek-V3.1': 'opensource',
            'deepseek-ai/DeepSeek-R1': 'opensource',
            'moonshotai/Kimi-K2-Thinking': 'opensource',
            'Qwen/Qwen3-235B-A22B-Instruct-2507': 'opensource',
            'openai/gpt-oss-120b': 'opensource',
            'openai/gpt-oss-20b': 'opensource',
            'human brain': 'human'
        }

        for player in all_players:
            provider = model_to_provider.get(player.model_name.lower(), 'unknown')
            if provider in provider_stats:
                provider_stats[provider]['games'] += 1
                if player.team == player.game.winner_team:
                    provider_stats[provider]['wins'] += 1
                else:
                    provider_stats[provider]['losses'] += 1

        return {
            'total_games': total_games,
            'game_history': game_history,
            'model_stats': model_stats,
            'provider_stats': provider_stats
        }


# Initialize database on import
init_database()

	from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, Boolean
	from sqlalchemy.ext.declarative import declarative_base
	from sqlalchemy.orm import sessionmaker, relationship
	from datetime import datetime
	from contextlib import contextmanager
	import threading

	Base = declarative_base()

	# Thread-safe engine creation
	engine = create_engine(
	'sqlite:///codenames_games.db',
	connect_args={'check_same_thread': False},
	pool_pre_ping=True,
	pool_size=10,
	max_overflow=20
	)

	# Thread-local session
	SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
	session_lock = threading.Lock()


	@contextmanager
	def get_db_session():
	"""Thread-safe database session context manager"""
	session = SessionLocal()
	try:
	yield session
	session.commit()
	except Exception as e:
	session.rollback()
	raise e
	finally:
	session.close()


	class Game(Base):
	__tablename__ = 'games'

	id = Column(Integer, primary_key=True, autoincrement=True)
	timestamp = Column(DateTime, default=datetime.utcnow, nullable=False)
	winner_team = Column(String(10), nullable=False) # 'red' or 'blue'
	red_score = Column(Integer, nullable=False) # remaining words
	blue_score = Column(Integer, nullable=False) # remaining words
	reason = Column(String(50), nullable=True) # None or 'killer'
	red_team_preset = Column(String(50), nullable=True) # 'openai', 'google', etc.
	blue_team_preset = Column(String(50), nullable=True)

	# Relationships
	players = relationship("GamePlayer", back_populates="game", cascade="all, delete-orphan")


	class GamePlayer(Base):
	__tablename__ = 'game_players'

	id = Column(Integer, primary_key=True, autoincrement=True)
	game_id = Column(Integer, ForeignKey('games.id'), nullable=False)
	player_name = Column(String(100), nullable=False)
	team = Column(String(10), nullable=False) # 'red' or 'blue'
	role = Column(String(20), nullable=False) # 'boss', 'captain', 'player'
	model_name = Column(String(100), nullable=False)
	is_human = Column(Boolean, default=False)

	# Relationship
	game = relationship("Game", back_populates="players")


	def init_database():
	"""Initialize the database tables"""
	Base.metadata.create_all(bind=engine)


	def save_game_to_db(game, winner_and_score):
	"""
	Save a completed game to the database

	Args:
	game: Game object with players and board info
	winner_and_score: tuple (winner_team, scores, reason)
	"""
	winner_team = winner_and_score[0]
	scores = winner_and_score[1]
	reason = winner_and_score[2] if len(winner_and_score) > 2 else None

	with get_db_session() as session:
	# Create game record
	new_game = Game(
	winner_team=winner_team,
	red_score=scores[0],
	blue_score=scores[1],
	reason=reason,
	red_team_preset=game.red_team_choice,
	blue_team_preset=game.blue_team_choice
	)
	session.add(new_game)
	session.flush() # Get the game ID

	# Create player records
	for player in game.players:
	is_human = player.model_name.lower() == "human brain"
	game_player = GamePlayer(
	game_id=new_game.id,
	player_name=player.name,
	team=player.team,
	role=player.role,
	model_name=player.model_name,
	is_human=is_human
	)
	session.add(game_player)

	session.commit()
	return new_game.id


	def get_stats():
	"""
	Retrieve comprehensive statistics from the database

	Returns:
	dict: Contains total_games, game_history, model_stats, provider_stats
	"""
	with get_db_session() as session:
	# Total games
	total_games = session.query(Game).count()

	# Game history
	games = session.query(Game).order_by(Game.timestamp.desc()).all()
	game_history = []

	for game in games:
	game_info = {
	'id': game.id,
	'timestamp': game.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
	'winner': game.winner_team,
	'red_score': game.red_score,
	'blue_score': game.blue_score,
	'reason': game.reason,
	'red_preset': game.red_team_preset,
	'blue_preset': game.blue_team_preset,
	'players': []
	}

	for player in game.players:
	game_info['players'].append({
	'name': player.player_name,
	'team': player.team,
	'role': player.role,
	'model': player.model_name,
	'is_human': player.is_human
	})

	game_history.append(game_info)

	# Model stats (individual)
	model_stats = {}
	all_players = session.query(GamePlayer).all()

	for player in all_players:
	model = player.model_name
	if model not in model_stats:
	model_stats[model] = {'wins': 0, 'losses': 0, 'games': 0}

	model_stats[model]['games'] += 1
	if player.team == player.game.winner_team:
	model_stats[model]['wins'] += 1
	else:
	model_stats[model]['losses'] += 1

	# Provider stats (grouped)
	provider_stats = {
	'openai': {'wins': 0, 'losses': 0, 'games': 0},
	'google': {'wins': 0, 'losses': 0, 'games': 0},
	'anthropic': {'wins': 0, 'losses': 0, 'games': 0},
	'opensource': {'wins': 0, 'losses': 0, 'games': 0},
	'human': {'wins': 0, 'losses': 0, 'games': 0}
	}

	# Map models to providers
	model_to_provider = {
	'gpt-5': 'openai',
	'gpt-5-mini': 'openai',
	'gpt-5-nano': 'openai',
	'gpt-4.1-nano': 'openai',
	'gemini-2.5-pro': 'google',
	'gemini-2.5-flash': 'google',
	'gemini-2.0-flash-001': 'google',
	'gemini-2.0-flash-lite-001': 'google',
	'claude-sonnet-4-5-20250929':'anthropic',
	'claude-3-7-sonnet-20250219': 'anthropic',
	'claude-3-5-haiku-20241022': 'anthropic',
	'claude-3-haiku-20240307': 'anthropic',
	'deepseek-ai/DeepSeek-V3.1': 'opensource',
	'deepseek-ai/DeepSeek-R1': 'opensource',
	'moonshotai/Kimi-K2-Thinking': 'opensource',
	'Qwen/Qwen3-235B-A22B-Instruct-2507': 'opensource',
	'openai/gpt-oss-120b': 'opensource',
	'openai/gpt-oss-20b': 'opensource',
	'human brain': 'human'
	}

	for player in all_players:
	provider = model_to_provider.get(player.model_name.lower(), 'unknown')
	if provider in provider_stats:
	provider_stats[provider]['games'] += 1
	if player.team == player.game.winner_team:
	provider_stats[provider]['wins'] += 1
	else:
	provider_stats[provider]['losses'] += 1

	return {
	'total_games': total_games,
	'game_history': game_history,
	'model_stats': model_stats,
	'provider_stats': provider_stats
	}


	# Initialize database on import
	init_database()