532 lines
20 KiB
Python
532 lines
20 KiB
Python
#!/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")
|