299 lines
11 KiB
Python
299 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Interactive Database Playground
|
|
|
|
Test and experiment with database models and operations.
|
|
Run this script to create test games and explore the database operations.
|
|
|
|
Usage:
|
|
python test_db_playground.py
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from uuid import uuid4, UUID
|
|
from typing import Optional
|
|
|
|
from app.database.operations import DatabaseOperations
|
|
from app.models.roster_models import PdRosterLinkData, SbaRosterLinkData
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DatabasePlayground:
|
|
"""Interactive playground for testing database operations."""
|
|
|
|
def __init__(self):
|
|
self.db_ops = DatabaseOperations()
|
|
self.test_game_id: Optional[UUID] = None
|
|
|
|
async def demo_pd_game(self):
|
|
"""Create and populate a PD league game with roster and lineup."""
|
|
logger.info("=" * 60)
|
|
logger.info("DEMO: Creating PD League Game")
|
|
logger.info("=" * 60)
|
|
|
|
# Create game
|
|
self.test_game_id = uuid4()
|
|
game = await self.db_ops.create_game(
|
|
game_id=self.test_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public",
|
|
home_team_is_ai=False,
|
|
away_team_is_ai=True,
|
|
ai_difficulty="balanced"
|
|
)
|
|
logger.info(f"✅ Created PD game: {game.id}")
|
|
logger.info(f" Home Team: {game.home_team_id} (Human)")
|
|
logger.info(f" Away Team: {game.away_team_id} (AI - {game.ai_difficulty})")
|
|
|
|
# Add cards to roster
|
|
logger.info("\n📋 Adding cards to roster...")
|
|
home_cards = [101, 102, 103, 104, 105, 106, 107, 108, 109]
|
|
away_cards = [201, 202, 203, 204, 205, 206, 207, 208, 209]
|
|
|
|
for card_id in home_cards:
|
|
roster_link = await self.db_ops.add_pd_roster_card(
|
|
game_id=self.test_game_id,
|
|
card_id=card_id,
|
|
team_id=1
|
|
)
|
|
logger.info(f" Added card {card_id} to home roster (link_id={roster_link.id})")
|
|
|
|
for card_id in away_cards:
|
|
roster_link = await self.db_ops.add_pd_roster_card(
|
|
game_id=self.test_game_id,
|
|
card_id=card_id,
|
|
team_id=2
|
|
)
|
|
logger.info(f" Added card {card_id} to away roster (link_id={roster_link.id})")
|
|
|
|
# Get roster
|
|
home_roster = await self.db_ops.get_pd_roster(self.test_game_id, team_id=1)
|
|
away_roster = await self.db_ops.get_pd_roster(self.test_game_id, team_id=2)
|
|
logger.info(f"\n✅ Home roster: {len(home_roster)} cards")
|
|
logger.info(f"✅ Away roster: {len(away_roster)} cards")
|
|
|
|
# Set lineup
|
|
logger.info("\n⚾ Setting starting lineup...")
|
|
positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF']
|
|
|
|
for i, (card_id, position) in enumerate(zip(home_cards, positions), start=1):
|
|
lineup = await self.db_ops.add_pd_lineup_card(
|
|
game_id=self.test_game_id,
|
|
team_id=1,
|
|
card_id=card_id,
|
|
position=position,
|
|
batting_order=i if position != 'P' else None,
|
|
is_starter=True
|
|
)
|
|
logger.info(f" {position}: Card {card_id} (batting {lineup.batting_order or 'N/A'})")
|
|
|
|
# Get active lineup
|
|
home_lineup = await self.db_ops.get_active_lineup(self.test_game_id, team_id=1)
|
|
logger.info(f"\n✅ Active lineup: {len(home_lineup)} players")
|
|
|
|
# Update game state
|
|
logger.info("\n🎮 Simulating game progress...")
|
|
await self.db_ops.update_game_state(
|
|
game_id=self.test_game_id,
|
|
inning=3,
|
|
half="bottom",
|
|
home_score=2,
|
|
away_score=1,
|
|
status="active"
|
|
)
|
|
logger.info(" Updated to: Bottom 3rd, Home 2 - Away 1")
|
|
|
|
# Save a play
|
|
play_data = {
|
|
'game_id': self.test_game_id,
|
|
'play_number': 1,
|
|
'inning': 3,
|
|
'half': 'bottom',
|
|
'outs_before': 1,
|
|
'batting_order': 3,
|
|
'away_score': 1,
|
|
'home_score': 2,
|
|
'batter_id': home_lineup[2].id,
|
|
'pitcher_id': home_lineup[0].id,
|
|
'catcher_id': home_lineup[1].id,
|
|
'dice_roll': '14+6',
|
|
'hit_type': 'FB',
|
|
'result_description': 'Deep fly ball to center field - Home Run!',
|
|
'outs_recorded': 0,
|
|
'runs_scored': 1,
|
|
'pa': 1,
|
|
'ab': 1,
|
|
'hit': 1,
|
|
'homerun': 1,
|
|
'rbi': 1,
|
|
'is_go_ahead': False,
|
|
'complete': True,
|
|
'locked': False
|
|
}
|
|
|
|
play = await self.db_ops.save_play(play_data)
|
|
logger.info(f"\n⚾ Saved play #{play.play_number}: {play.result_description}")
|
|
|
|
# Load game state
|
|
logger.info("\n🔄 Loading complete game state...")
|
|
game_state = await self.db_ops.load_game_state(self.test_game_id)
|
|
if game_state:
|
|
logger.info(f"✅ Loaded game state:")
|
|
logger.info(f" Status: {game_state['game']['status']}")
|
|
logger.info(f" Score: {game_state['game']['away_score']}-{game_state['game']['home_score']}")
|
|
logger.info(f" Lineups: {len(game_state['lineups'])} players")
|
|
logger.info(f" Plays: {len(game_state['plays'])} plays recorded")
|
|
|
|
return self.test_game_id
|
|
|
|
async def demo_sba_game(self):
|
|
"""Create and populate an SBA league game with roster and lineup."""
|
|
logger.info("\n" + "=" * 60)
|
|
logger.info("DEMO: Creating SBA League Game")
|
|
logger.info("=" * 60)
|
|
|
|
# Create game
|
|
game_id = uuid4()
|
|
game = await self.db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=11,
|
|
game_mode="ranked",
|
|
visibility="public",
|
|
home_team_is_ai=False,
|
|
away_team_is_ai=False
|
|
)
|
|
logger.info(f"✅ Created SBA game: {game.id}")
|
|
logger.info(f" Home Team: {game.home_team_id}")
|
|
logger.info(f" Away Team: {game.away_team_id}")
|
|
logger.info(f" Mode: {game.game_mode}")
|
|
|
|
# Add players to roster
|
|
logger.info("\n📋 Adding players to roster...")
|
|
home_players = [501, 502, 503, 504, 505, 506, 507, 508, 509]
|
|
away_players = [601, 602, 603, 604, 605, 606, 607, 608, 609]
|
|
|
|
for player_id in home_players:
|
|
roster_link = await self.db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
team_id=10
|
|
)
|
|
logger.info(f" Added player {player_id} to home roster (link_id={roster_link.id})")
|
|
|
|
for player_id in away_players:
|
|
roster_link = await self.db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
team_id=11
|
|
)
|
|
logger.info(f" Added player {player_id} to away roster (link_id={roster_link.id})")
|
|
|
|
# Get roster
|
|
home_roster = await self.db_ops.get_sba_roster(game_id, team_id=10)
|
|
away_roster = await self.db_ops.get_sba_roster(game_id, team_id=11)
|
|
logger.info(f"\n✅ Home roster: {len(home_roster)} players")
|
|
logger.info(f"✅ Away roster: {len(away_roster)} players")
|
|
|
|
# Set lineup
|
|
logger.info("\n⚾ Setting starting lineup...")
|
|
positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF']
|
|
|
|
for i, (player_id, position) in enumerate(zip(home_players, positions), start=1):
|
|
lineup = await self.db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=10,
|
|
player_id=player_id,
|
|
position=position,
|
|
batting_order=i if position != 'P' else None,
|
|
is_starter=True
|
|
)
|
|
logger.info(f" {position}: Player {player_id} (batting {lineup.batting_order or 'N/A'})")
|
|
|
|
# Create game session
|
|
session = await self.db_ops.create_game_session(game_id)
|
|
logger.info(f"\n✅ Created game session for game: {session.game_id}")
|
|
|
|
# Update session snapshot
|
|
await self.db_ops.update_session_snapshot(
|
|
game_id=game_id,
|
|
state_snapshot={'test': 'data', 'connected_users': ['user1', 'user2']}
|
|
)
|
|
logger.info("✅ Updated session snapshot")
|
|
|
|
return game_id
|
|
|
|
async def demo_cleanup(self):
|
|
"""Demonstrate querying capabilities."""
|
|
if not self.test_game_id:
|
|
logger.warning("No test game created yet")
|
|
return
|
|
|
|
logger.info("\n" + "=" * 60)
|
|
logger.info("DEMO: Query Capabilities")
|
|
logger.info("=" * 60)
|
|
|
|
# Get game
|
|
game = await self.db_ops.get_game(self.test_game_id)
|
|
if game:
|
|
logger.info(f"\n📊 Game Details:")
|
|
logger.info(f" ID: {game.id}")
|
|
logger.info(f" League: {game.league_id}")
|
|
logger.info(f" Status: {game.status}")
|
|
logger.info(f" Inning: {game.current_half} {game.current_inning}")
|
|
logger.info(f" Score: {game.away_score}-{game.home_score}")
|
|
|
|
# Get plays
|
|
plays = await self.db_ops.get_plays(self.test_game_id)
|
|
logger.info(f"\n⚾ Plays: {len(plays)} total")
|
|
for play in plays:
|
|
logger.info(f" Play #{play.play_number}: {play.result_description}")
|
|
|
|
# Get active lineup
|
|
home_lineup = await self.db_ops.get_active_lineup(self.test_game_id, team_id=1)
|
|
logger.info(f"\n👥 Active Lineup: {len(home_lineup)} players")
|
|
for player in sorted(home_lineup, key=lambda x: x.batting_order or 99):
|
|
logger.info(f" {player.position}: Card {player.card_id} (batting {player.batting_order or 'N/A'})")
|
|
|
|
|
|
async def main():
|
|
"""Run interactive playground demos."""
|
|
playground = DatabasePlayground()
|
|
|
|
try:
|
|
# Run PD game demo
|
|
pd_game_id = await playground.demo_pd_game()
|
|
|
|
# Run SBA game demo
|
|
sba_game_id = await playground.demo_sba_game()
|
|
|
|
# Show query capabilities
|
|
await playground.demo_cleanup()
|
|
|
|
logger.info("\n" + "=" * 60)
|
|
logger.info("✅ ALL DEMOS COMPLETED SUCCESSFULLY!")
|
|
logger.info("=" * 60)
|
|
logger.info(f"\nCreated games:")
|
|
logger.info(f" PD Game: {pd_game_id}")
|
|
logger.info(f" SBA Game: {sba_game_id}")
|
|
logger.info("\nYou can query these games in the database to explore the data!")
|
|
|
|
except Exception as e:
|
|
logger.error(f"\n❌ Error during demo: {e}", exc_info=True)
|
|
raise
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|