""" File I/O and data persistence for TV_APP Handles all JSON file operations and data archiving """ import json import os import glob from datetime import datetime # Base directory anchored to this file's location (project root), so paths work # regardless of the current working directory when the app is launched. _BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # File paths - organized in data directory SETTINGS_FILE = os.path.join(_BASE_DIR, 'data', 'camera_settings.json') PLAYERS_FILE = os.path.join(_BASE_DIR, 'data', 'players.json') TOURNAMENT_FILE = os.path.join(_BASE_DIR, 'data', 'tournament_state.json') RESULTS_FILE = os.path.join(_BASE_DIR, 'data', 'tournament_results.json') LEAGUE_FILE = os.path.join(_BASE_DIR, 'data', 'league_state.json') ARCHIVE_DIR = os.path.join(_BASE_DIR, 'data', 'tournament_archives') LEAGUE_ARCHIVE_DIR = os.path.join(_BASE_DIR, 'data', 'league_archives') # Default settings DEFAULT_SETTINGS = { 'camera_titles': { '1': 'Camera 1', '2': 'Camera 2', '3': 'Camera 3', '4': 'Camera 4', '5': 'Camera 5', '6': 'Camera 6' }, 'display_options': { 'show_titles': True, 'title_size': 1.1 } } class FileStorage: """Handle JSON file read/write operations""" @staticmethod def _ensure_directory(directory): """Ensure directory exists""" if directory and not os.path.exists(directory): os.makedirs(directory) @staticmethod def _read_json(filepath): """Read JSON file safely""" try: if os.path.exists(filepath): with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, IOError) as e: print(f"Error reading {filepath}: {e}") return None @staticmethod def _write_json(filepath, data): """Write JSON file safely""" try: directory = os.path.dirname(filepath) FileStorage._ensure_directory(directory) with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) return True except IOError as e: print(f"Error writing {filepath}: {e}") return False class SettingsStorage(FileStorage): """Manage camera and display settings""" @staticmethod def load_settings(): """Load settings from JSON file, create with defaults if not exists""" settings = FileStorage._read_json(SETTINGS_FILE) if settings is None: return DEFAULT_SETTINGS.copy() # Ensure all required keys exist (backwards compatibility) for key in DEFAULT_SETTINGS: if key not in settings: settings[key] = DEFAULT_SETTINGS[key] for camera_id in DEFAULT_SETTINGS['camera_titles']: if camera_id not in settings.get('camera_titles', {}): settings['camera_titles'][camera_id] = DEFAULT_SETTINGS['camera_titles'][camera_id] return settings @staticmethod def save_settings(settings): """Save settings to JSON file""" return FileStorage._write_json(SETTINGS_FILE, settings) class PlayerStorage(FileStorage): """Manage player data""" DEFAULT_PLAYERS = { 'players': [ {'id': i, 'name': f'Player {i}', 'enabled': True} for i in range(1, 7) # 6 cameras ] } @staticmethod def load_players(): """Load players from JSON file, create with defaults if not exists""" players = FileStorage._read_json(PLAYERS_FILE) if players is None: PlayerStorage.save_players(PlayerStorage.DEFAULT_PLAYERS) return PlayerStorage.DEFAULT_PLAYERS.copy() return players @staticmethod def save_players(players_data): """Save players to JSON file""" return FileStorage._write_json(PLAYERS_FILE, players_data) class TournamentStorage(FileStorage): """Manage tournament state""" @staticmethod def load_tournament_state(): """Load tournament state from JSON file""" return FileStorage._read_json(TOURNAMENT_FILE) @staticmethod def save_tournament_state(tournament_data): """Save tournament state to JSON file""" return FileStorage._write_json(TOURNAMENT_FILE, tournament_data) class ResultsStorage(FileStorage): """Manage tournament results""" @staticmethod def load_results(): """Load results from JSON file""" return FileStorage._read_json(RESULTS_FILE) @staticmethod def save_results(results_data): """Save results to JSON file""" return FileStorage._write_json(RESULTS_FILE, results_data) class LeagueStorage(FileStorage): """Manage league state""" @staticmethod def load_league_state(): """Load league state from JSON file""" return FileStorage._read_json(LEAGUE_FILE) @staticmethod def save_league_state(league_data): """Save league state to JSON file""" return FileStorage._write_json(LEAGUE_FILE, league_data) class ArchiveStorage(FileStorage): """Manage tournament and league archives""" @staticmethod def archive_tournament(tournament_data, results_data): """Archive completed tournament data. Returns (success, filename) tuple""" try: FileStorage._ensure_directory(ARCHIVE_DIR) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_filename = f"tournament_{timestamp}.json" archive_path = os.path.join(ARCHIVE_DIR, archive_filename) archive_data = { 'tournament': tournament_data, 'results': results_data, 'archived_at': datetime.now().isoformat() } success = FileStorage._write_json(archive_path, archive_data) if success: print(f"Tournament archived to: {archive_path}") return (True, archive_filename) return (False, None) except Exception as e: print(f"Error archiving tournament: {e}") return (False, None) @staticmethod def archive_league(league_data): """Archive completed league data. Returns (success, filename) tuple""" try: FileStorage._ensure_directory(LEAGUE_ARCHIVE_DIR) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_filename = f"league_{timestamp}.json" archive_path = os.path.join(LEAGUE_ARCHIVE_DIR, archive_filename) archive_data = { 'league': league_data, 'archived_at': datetime.now().isoformat() } success = FileStorage._write_json(archive_path, archive_data) if success: print(f"League archived to: {archive_path}") return (True, archive_filename) return (False, None) except Exception as e: print(f"Error archiving league: {e}") return (False, None) @staticmethod def get_archived_tournaments(): """Get list of standalone archived tournaments""" try: if not os.path.exists(ARCHIVE_DIR): return [] archives = [] for file_path in glob.glob(os.path.join(ARCHIVE_DIR, "tournament_*.json")): try: data = FileStorage._read_json(file_path) if not data: continue filename = os.path.basename(file_path) archived_at = data.get('archived_at', 'Unknown') tournament_data = data.get('tournament', {}) results_data = data.get('results', {}) # Skip tournaments that are part of a league if tournament_data.get('league_tournament_number') or results_data.get('league_tournament_number'): continue archive_info = { 'filename': filename, 'filepath': file_path, 'archived_at': archived_at, 'created_at': tournament_data.get('created_at', 'Unknown'), 'tournament_type': tournament_data.get('tournament_type', '20_targets'), 'total_players': tournament_data.get('total_players', 0), 'total_rounds': tournament_data.get('total_rounds', 0), 'tournament_finished': results_data.get('tournament_finished', False), 'participants_count': len(results_data.get('participants', {})) } archives.append(archive_info) except (json.JSONDecodeError, IOError) as e: print(f"Error reading archive {file_path}: {e}") continue # Sort by archived date (newest first) archives.sort(key=lambda x: x['archived_at'], reverse=True) return archives except Exception as e: print(f"Error getting archived tournaments: {e}") return [] @staticmethod def get_archived_leagues(): """Get list of all archived leagues""" try: if not os.path.exists(LEAGUE_ARCHIVE_DIR): return [] archives = [] for file_path in glob.glob(os.path.join(LEAGUE_ARCHIVE_DIR, "league_*.json")): try: data = FileStorage._read_json(file_path) if not data: continue filename = os.path.basename(file_path) archived_at = data.get('archived_at', 'Unknown') league_data = data.get('league', {}) archive_info = { 'filename': filename, 'filepath': file_path, 'archived_at': archived_at, 'created_at': league_data.get('created_at', 'Unknown'), 'league_id': league_data.get('league_id', 'Unknown'), 'tournament_type': league_data.get('tournament_type', '20_targets'), 'total_tournaments': league_data.get('total_tournaments', 5), 'participants_count': len(league_data.get('participants', {})), 'league_finished': league_data.get('league_finished', False), 'completed_tournaments': len(league_data.get('completed_tournaments', [])) } archives.append(archive_info) except (json.JSONDecodeError, IOError) as e: print(f"Error reading league archive {file_path}: {e}") continue # Sort by archived date (newest first) archives.sort(key=lambda x: x['archived_at'], reverse=True) return archives except Exception as e: print(f"Error getting archived leagues: {e}") return [] @staticmethod def load_archive_file(filepath): """Load a specific archive file""" return FileStorage._read_json(filepath)