#!/usr/bin/env python3 """ Configuration Management System for Video Player Handles all configuration files, validation, and management """ import os import json import yaml import logging from pathlib import Path from typing import Dict, List, Optional, Any, Union from dataclasses import dataclass, asdict import shutil from dotenv import load_dotenv # Load environment variables load_dotenv() @dataclass class VideoPlayerConfig: """Video Player Configuration""" # Video settings video_folder: str = os.getenv("VIDEO_FOLDER", "/home/pi/Videos") supported_formats: List[str] = None default_channel: int = 1 auto_play: bool = True # Display settings fullscreen: bool = True window_width: int = 1920 window_height: int = 1080 hide_cursor: bool = True # Audio settings audio_device: str = "default" volume: int = 50 mute_on_start: bool = False # Channel settings channel_timeout: float = 3.0 multi_digit_timeout: float = 1.0 channel_display_timeout: float = 2.0 channel_refresh_interval: float = 30.0 # seconds channel_assignment_method: str = "alphabetical" # alphabetical, manual, custom # IR Remote settings ir_pin: int = 18 ir_protocols: List[str] = None ir_repeat_delay: float = 0.1 # VLC settings vlc_options: List[str] = None # Logging settings log_level: str = "INFO" log_file: str = "/var/log/video_player.log" log_max_size: int = 10485760 # 10MB log_backup_count: int = 5 # System settings check_interval: float = 1.0 restart_on_crash: bool = True max_restart_attempts: int = 3 def __post_init__(self): if self.supported_formats is None: self.supported_formats = ['.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.webm', '.m4v'] if self.ir_protocols is None: self.ir_protocols = ['NEC', 'RC5'] if self.vlc_options is None: self.vlc_options = [ "--fullscreen", "--no-video-title-show", "--no-audio-display", "--no-osd", "--quiet" ] @dataclass class ChannelConfig: """Channel Configuration""" number: int name: str path: str description: str = "" category: str = "general" enabled: bool = True priority: int = 0 @dataclass class IRMappingConfig: """IR Code Mapping Configuration""" ir_code: str command: str description: str = "" repeatable: bool = True class ConfigManager: """Configuration Manager for Video Player""" def __init__(self, config_dir: str = "/etc/video_player"): self.config_dir = Path(config_dir) self.config_dir.mkdir(parents=True, exist_ok=True) # Configuration file paths self.main_config_file = self.config_dir / "config.json" self.channels_file = self.config_dir / "channels.json" self.ir_mapping_file = self.config_dir / "ir_mapping.json" self.env_file = self.config_dir / ".env" self.logger = logging.getLogger(__name__) # Load configurations self.main_config = self.load_main_config() self.channels = self.load_channels() self.ir_mapping = self.load_ir_mapping() def load_main_config(self) -> VideoPlayerConfig: """Load main configuration""" try: if self.main_config_file.exists(): with open(self.main_config_file, 'r') as f: config_data = json.load(f) return VideoPlayerConfig(**config_data) else: config = VideoPlayerConfig() self.save_main_config(config) return config except Exception as e: self.logger.error(f"Error loading main config: {e}") return VideoPlayerConfig() def save_main_config(self, config: VideoPlayerConfig): """Save main configuration""" try: config_dict = asdict(config) with open(self.main_config_file, 'w') as f: json.dump(config_dict, f, indent=2) self.logger.info("Main configuration saved") except Exception as e: self.logger.error(f"Error saving main config: {e}") def load_channels(self) -> Dict[int, ChannelConfig]: """Load channel configuration""" try: if self.channels_file.exists(): with open(self.channels_file, 'r') as f: channels_data = json.load(f) channels = {} for channel_num, channel_info in channels_data.items(): channels[int(channel_num)] = ChannelConfig( number=int(channel_num), name=channel_info['name'], path=channel_info['path'], description=channel_info.get('description', ''), category=channel_info.get('category', 'general'), enabled=channel_info.get('enabled', True), priority=channel_info.get('priority', 0) ) return channels else: return {} except Exception as e: self.logger.error(f"Error loading channels: {e}") return {} def save_channels(self, channels: Dict[int, ChannelConfig]): """Save channel configuration""" try: channels_data = {} for channel_num, channel in channels.items(): channels_data[str(channel_num)] = asdict(channel) with open(self.channels_file, 'w') as f: json.dump(channels_data, f, indent=2) self.logger.info("Channel configuration saved") except Exception as e: self.logger.error(f"Error saving channels: {e}") def load_ir_mapping(self) -> Dict[str, IRMappingConfig]: """Load IR code mapping configuration""" try: if self.ir_mapping_file.exists(): with open(self.ir_mapping_file, 'r') as f: mapping_data = json.load(f) mapping = {} for ir_code, mapping_info in mapping_data.items(): if isinstance(mapping_info, str): # Simple string mapping mapping[ir_code] = IRMappingConfig( ir_code=ir_code, command=mapping_info ) else: # Detailed mapping mapping[ir_code] = IRMappingConfig( ir_code=ir_code, command=mapping_info['command'], description=mapping_info.get('description', ''), repeatable=mapping_info.get('repeatable', True) ) return mapping else: return self.create_default_ir_mapping() except Exception as e: self.logger.error(f"Error loading IR mapping: {e}") return {} def save_ir_mapping(self, mapping: Dict[str, IRMappingConfig]): """Save IR code mapping configuration""" try: mapping_data = {} for ir_code, mapping_config in mapping.items(): mapping_data[ir_code] = asdict(mapping_config) with open(self.ir_mapping_file, 'w') as f: json.dump(mapping_data, f, indent=2) self.logger.info("IR mapping configuration saved") except Exception as e: self.logger.error(f"Error saving IR mapping: {e}") def create_default_ir_mapping(self) -> Dict[str, IRMappingConfig]: """Create default IR code mapping""" default_mapping = { # NEC protocol examples "NEC_00FF_00FF": IRMappingConfig("NEC_00FF_00FF", "power_toggle", "Power button"), "NEC_00FF_807F": IRMappingConfig("NEC_00FF_807F", "channel_0", "Channel 0"), "NEC_00FF_40BF": IRMappingConfig("NEC_00FF_40BF", "channel_1", "Channel 1"), "NEC_00FF_C03F": IRMappingConfig("NEC_00FF_C03F", "channel_2", "Channel 2"), "NEC_00FF_20DF": IRMappingConfig("NEC_00FF_20DF", "channel_3", "Channel 3"), "NEC_00FF_A05F": IRMappingConfig("NEC_00FF_A05F", "channel_4", "Channel 4"), "NEC_00FF_609F": IRMappingConfig("NEC_00FF_609F", "channel_5", "Channel 5"), "NEC_00FF_E01F": IRMappingConfig("NEC_00FF_E01F", "channel_6", "Channel 6"), "NEC_00FF_10EF": IRMappingConfig("NEC_00FF_10EF", "channel_7", "Channel 7"), "NEC_00FF_906F": IRMappingConfig("NEC_00FF_906F", "channel_8", "Channel 8"), "NEC_00FF_50AF": IRMappingConfig("NEC_00FF_50AF", "channel_9", "Channel 9"), "NEC_00FF_00FF": IRMappingConfig("NEC_00FF_00FF", "play_pause", "Play/Pause"), "NEC_00FF_807F": IRMappingConfig("NEC_00FF_807F", "stop", "Stop"), "NEC_00FF_40BF": IRMappingConfig("NEC_00FF_40BF", "next_channel", "Next channel"), "NEC_00FF_C03F": IRMappingConfig("NEC_00FF_C03F", "prev_channel", "Previous channel"), "NEC_00FF_20DF": IRMappingConfig("NEC_00FF_20DF", "volume_up", "Volume up"), "NEC_00FF_A05F": IRMappingConfig("NEC_00FF_A05F", "volume_down", "Volume down"), # RC5 protocol examples "RC5_00_0C_0": IRMappingConfig("RC5_00_0C_0", "power_toggle", "Power button"), "RC5_00_00_0": IRMappingConfig("RC5_00_00_0", "channel_0", "Channel 0"), "RC5_00_01_0": IRMappingConfig("RC5_00_01_0", "channel_1", "Channel 1"), "RC5_00_02_0": IRMappingConfig("RC5_00_02_0", "channel_2", "Channel 2"), "RC5_00_03_0": IRMappingConfig("RC5_00_03_0", "channel_3", "Channel 3"), "RC5_00_04_0": IRMappingConfig("RC5_00_04_0", "channel_4", "Channel 4"), "RC5_00_05_0": IRMappingConfig("RC5_00_05_0", "channel_5", "Channel 5"), "RC5_00_06_0": IRMappingConfig("RC5_00_06_0", "channel_6", "Channel 6"), "RC5_00_07_0": IRMappingConfig("RC5_00_07_0", "channel_7", "Channel 7"), "RC5_00_08_0": IRMappingConfig("RC5_00_08_0", "channel_8", "Channel 8"), "RC5_00_09_0": IRMappingConfig("RC5_00_09_0", "channel_9", "Channel 9"), "RC5_00_35_0": IRMappingConfig("RC5_00_35_0", "play_pause", "Play/Pause"), "RC5_00_36_0": IRMappingConfig("RC5_00_36_0", "stop", "Stop"), "RC5_00_32_0": IRMappingConfig("RC5_00_32_0", "next_channel", "Next channel"), "RC5_00_33_0": IRMappingConfig("RC5_00_33_0", "prev_channel", "Previous channel"), "RC5_00_10_0": IRMappingConfig("RC5_00_10_0", "volume_up", "Volume up"), "RC5_00_11_0": IRMappingConfig("RC5_00_11_0", "volume_down", "Volume down"), # Repeat command "REPEAT": IRMappingConfig("REPEAT", "repeat_last", "Repeat last command") } self.save_ir_mapping(default_mapping) return default_mapping def validate_config(self, config: VideoPlayerConfig) -> List[str]: """Validate configuration and return list of errors""" errors = [] # Validate video folder video_folder = Path(config.video_folder) if not video_folder.exists(): errors.append(f"Video folder does not exist: {config.video_folder}") elif not video_folder.is_dir(): errors.append(f"Video folder is not a directory: {config.video_folder}") # Validate GPIO pin if not (1 <= config.ir_pin <= 40): errors.append(f"Invalid GPIO pin: {config.ir_pin}") # Validate channel number if config.default_channel < 1: errors.append(f"Invalid default channel: {config.default_channel}") # Validate timeouts if config.channel_timeout <= 0: errors.append(f"Invalid channel timeout: {config.channel_timeout}") if config.multi_digit_timeout <= 0: errors.append(f"Invalid multi-digit timeout: {config.multi_digit_timeout}") # Validate volume if not (0 <= config.volume <= 100): errors.append(f"Invalid volume: {config.volume}") # Validate log level valid_log_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] if config.log_level.upper() not in valid_log_levels: errors.append(f"Invalid log level: {config.log_level}") return errors def create_backup(self, backup_name: str = None): """Create backup of all configuration files""" if backup_name is None: backup_name = f"backup_{int(time.time())}" backup_dir = self.config_dir / "backups" / backup_name backup_dir.mkdir(parents=True, exist_ok=True) try: # Copy configuration files for config_file in [self.main_config_file, self.channels_file, self.ir_mapping_file]: if config_file.exists(): shutil.copy2(config_file, backup_dir / config_file.name) self.logger.info(f"Configuration backup created: {backup_dir}") return backup_dir except Exception as e: self.logger.error(f"Error creating backup: {e}") return None def restore_backup(self, backup_name: str): """Restore configuration from backup""" backup_dir = self.config_dir / "backups" / backup_name if not backup_dir.exists(): self.logger.error(f"Backup not found: {backup_name}") return False try: # Restore configuration files for config_file in [self.main_config_file, self.channels_file, self.ir_mapping_file]: backup_file = backup_dir / config_file.name if backup_file.exists(): shutil.copy2(backup_file, config_file) # Reload configurations self.main_config = self.load_main_config() self.channels = self.load_channels() self.ir_mapping = self.load_ir_mapping() self.logger.info(f"Configuration restored from backup: {backup_name}") return True except Exception as e: self.logger.error(f"Error restoring backup: {e}") return False def list_backups(self) -> List[str]: """List available configuration backups""" backups_dir = self.config_dir / "backups" if not backups_dir.exists(): return [] return [d.name for d in backups_dir.iterdir() if d.is_dir()] def export_config(self, export_file: str): """Export all configuration to a single file""" try: export_data = { 'main_config': asdict(self.main_config), 'channels': {str(k): asdict(v) for k, v in self.channels.items()}, 'ir_mapping': {k: asdict(v) for k, v in self.ir_mapping.items()} } with open(export_file, 'w') as f: json.dump(export_data, f, indent=2) self.logger.info(f"Configuration exported to: {export_file}") return True except Exception as e: self.logger.error(f"Error exporting configuration: {e}") return False def import_config(self, import_file: str): """Import configuration from a file""" try: with open(import_file, 'r') as f: import_data = json.load(f) # Import main config if 'main_config' in import_data: self.main_config = VideoPlayerConfig(**import_data['main_config']) self.save_main_config(self.main_config) # Import channels if 'channels' in import_data: channels = {} for channel_num, channel_data in import_data['channels'].items(): channels[int(channel_num)] = ChannelConfig(**channel_data) self.channels = channels self.save_channels(channels) # Import IR mapping if 'ir_mapping' in import_data: mapping = {} for ir_code, mapping_data in import_data['ir_mapping'].items(): mapping[ir_code] = IRMappingConfig(**mapping_data) self.ir_mapping = mapping self.save_ir_mapping(mapping) self.logger.info(f"Configuration imported from: {import_file}") return True except Exception as e: self.logger.error(f"Error importing configuration: {e}") return False # Configuration templates def create_config_templates(): """Create configuration file templates""" templates_dir = Path("templates") templates_dir.mkdir(exist_ok=True) # Main config template main_config_template = { "video_folder": "/home/pi/Videos", "supported_formats": [".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv", ".webm", ".m4v"], "default_channel": 1, "auto_play": True, "fullscreen": True, "window_width": 1920, "window_height": 1080, "hide_cursor": True, "audio_device": "default", "volume": 50, "mute_on_start": False, "channel_timeout": 3.0, "multi_digit_timeout": 1.0, "channel_display_timeout": 2.0, "channel_refresh_interval": 30.0, "channel_assignment_method": "alphabetical", "ir_pin": 18, "ir_protocols": ["NEC", "RC5"], "ir_repeat_delay": 0.1, "vlc_options": [ "--fullscreen", "--no-video-title-show", "--no-audio-display", "--no-osd", "--quiet" ], "log_level": "INFO", "log_file": "/var/log/video_player.log", "log_max_size": 10485760, "log_backup_count": 5, "check_interval": 1.0, "restart_on_crash": True, "max_restart_attempts": 3 } with open(templates_dir / "config.json.template", 'w') as f: json.dump(main_config_template, f, indent=2) # Channels template channels_template = { "1": { "number": 1, "name": "Sample Video 1", "path": "/home/pi/Videos/sample1.mp4", "description": "First sample video", "category": "general", "enabled": True, "priority": 0 }, "2": { "number": 2, "name": "Sample Video 2", "path": "/home/pi/Videos/sample2.mp4", "description": "Second sample video", "category": "general", "enabled": True, "priority": 0 } } with open(templates_dir / "channels.json.template", 'w') as f: json.dump(channels_template, f, indent=2) # IR mapping template ir_mapping_template = { "NEC_00FF_00FF": { "ir_code": "NEC_00FF_00FF", "command": "power_toggle", "description": "Power button", "repeatable": True }, "NEC_00FF_807F": { "ir_code": "NEC_00FF_807F", "command": "channel_0", "description": "Channel 0", "repeatable": True } } with open(templates_dir / "ir_mapping.json.template", 'w') as f: json.dump(ir_mapping_template, f, indent=2) print("Configuration templates created in templates/ directory") if __name__ == "__main__": # Create configuration templates create_config_templates() # Example usage config_manager = ConfigManager() # Print current configuration print("Current configuration:") print(f"Video folder: {config_manager.main_config.video_folder}") print(f"Default channel: {config_manager.main_config.default_channel}") print(f"IR pin: {config_manager.main_config.ir_pin}") print(f"Channels: {len(config_manager.channels)}") print(f"IR mappings: {len(config_manager.ir_mapping)}") # Validate configuration errors = config_manager.validate_config(config_manager.main_config) if errors: print("Configuration errors:") for error in errors: print(f" - {error}") else: print("Configuration is valid")