319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""
|
|
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)
|