"""Template and MQTT configuration management.""" import logging import time import json from typing import Dict, Optional, Tuple, List from utils import send_tasmota_command, retry_command, get_hostname_base class ConfigurationManager: """Handles template and MQTT configuration for Tasmota devices.""" def __init__(self, config: dict, logger: Optional[logging.Logger] = None): """ Initialize configuration manager. Args: config: Configuration dictionary logger: Optional logger instance """ self.config = config self.logger = logger or logging.getLogger(__name__) def check_and_update_template(self, device: dict, device_details: dict) -> bool: """ Check and update device template if needed. Args: device: Device info dictionary device_details: Detailed device information Returns: bool: True if template was updated or already correct """ device_name = device.get('name', 'Unknown') device_ip = device.get('ip', '') if not device_ip: self.logger.warning(f"{device_name}: No IP address available") return False # Get hostname base for template matching hostname = device_details.get('StatusNET', {}).get('Hostname', device_name) hostname_base = get_hostname_base(hostname) # Check if we have a template for this device device_list = self.config.get('device_list', {}) template_config = None for template_name, template_data in device_list.items(): if hostname_base.lower() in template_name.lower(): template_config = template_data self.logger.debug(f"{device_name}: Matched template '{template_name}'") break if not template_config: self.logger.debug(f"{device_name}: No template match found for '{hostname_base}'") return True # No template to apply, consider it successful expected_template = template_config.get('template') if not expected_template: self.logger.debug(f"{device_name}: Template config has no template string") return True # Parse expected template try: expected_template_dict = json.loads(expected_template) except json.JSONDecodeError as e: self.logger.error(f"{device_name}: Invalid template JSON: {e}") return False # Get current template current_template = device_details.get('StatusSTS', {}).get('Template') if current_template == expected_template_dict: self.logger.debug(f"{device_name}: Template already correct") return True # Apply template self.logger.info(f"{device_name}: Applying template") # Send template command result, success = send_tasmota_command( device_ip, f"Template%20{expected_template}", timeout=10, logger=self.logger ) if not success: self.logger.error(f"{device_name}: Failed to set template") return False # Wait a moment for template to be applied time.sleep(2) # Send Module 0 to activate the template result, success = send_tasmota_command( device_ip, "Module%200", timeout=10, logger=self.logger ) if not success: self.logger.error(f"{device_name}: Failed to set Module 0") return False self.logger.info(f"{device_name}: Template applied, restarting device") # Restart device to apply changes send_tasmota_command(device_ip, "Restart%201", timeout=5, logger=self.logger) # Wait for device to restart time.sleep(10) # Verify template was applied result, success = send_tasmota_command( device_ip, "Status%200", timeout=10, logger=self.logger ) if success and result: new_template = result.get('StatusSTS', {}).get('Template') if new_template == expected_template_dict: self.logger.info(f"{device_name}: Template verified successfully") return True else: self.logger.warning(f"{device_name}: Template verification failed") return False return False def configure_mqtt_settings(self, device: dict, device_details: dict) -> Tuple[bool, str]: """ Configure MQTT settings on a device. Args: device: Device info dictionary device_details: Detailed device information Returns: Tuple of (success, status_message) """ device_name = device.get('name', 'Unknown') device_ip = device.get('ip', '') if not device_ip: return False, "No IP address" mqtt_config = self.config.get('mqtt', {}) # Get hostname base for Topic substitution hostname = device_details.get('StatusNET', {}).get('Hostname', device_name) hostname_base = get_hostname_base(hostname) # Get current MQTT settings current_mqtt = device_details.get('StatusMQT', {}) # Check if MQTT needs to be enabled mqtt_enabled = current_mqtt.get('MqttHost', '') != '' if not mqtt_enabled: self.logger.info(f"{device_name}: Enabling MQTT") result, success = send_tasmota_command( device_ip, "SetOption3%201", timeout=5, logger=self.logger ) if not success: return False, "Failed to enable MQTT" # Build list of settings to update updates_needed = [] # Check each MQTT setting mqtt_host = mqtt_config.get('Host', '') if mqtt_host and current_mqtt.get('MqttHost', '') != mqtt_host: updates_needed.append(('MqttHost', mqtt_host)) mqtt_port = mqtt_config.get('Port', 1883) if current_mqtt.get('MqttPort', 0) != mqtt_port: updates_needed.append(('MqttPort', mqtt_port)) mqtt_user = mqtt_config.get('User', '') if mqtt_user and current_mqtt.get('MqttUser', '') != mqtt_user: updates_needed.append(('MqttUser', mqtt_user)) mqtt_password = mqtt_config.get('Password', '') # Note: Can't verify password from status, so always set it if mqtt_password: updates_needed.append(('MqttPassword', mqtt_password)) # Handle Topic with %hostname_base% substitution mqtt_topic = mqtt_config.get('Topic', '') if mqtt_topic: mqtt_topic = mqtt_topic.replace('%hostname_base%', hostname_base) if current_mqtt.get('Topic', '') != mqtt_topic: updates_needed.append(('Topic', mqtt_topic)) mqtt_full_topic = mqtt_config.get('FullTopic', '') if mqtt_full_topic and current_mqtt.get('FullTopic', '') != mqtt_full_topic: updates_needed.append(('FullTopic', mqtt_full_topic)) # Handle NoRetain (SetOption62) no_retain = mqtt_config.get('NoRetain', False) current_no_retain = current_mqtt.get('NoRetain', False) if no_retain != current_no_retain: updates_needed.append(('SetOption62', '1' if no_retain else '0')) if not updates_needed: self.logger.debug(f"{device_name}: MQTT settings already correct") return True, "Already configured" # Apply updates self.logger.info(f"{device_name}: Updating {len(updates_needed)} MQTT settings") failed_updates = [] for setting_name, setting_value in updates_needed: command = f"{setting_name}%20{setting_value}" result, success = retry_command( lambda: send_tasmota_command(device_ip, command, timeout=5, logger=self.logger), max_attempts=3, delay=1.0, logger=self.logger, device_name=device_name ) if not success: failed_updates.append(setting_name) self.logger.warning(f"{device_name}: Failed to set {setting_name}") if failed_updates: return False, f"Failed to set: {', '.join(failed_updates)}" # Wait for settings to be applied time.sleep(2) self.logger.info(f"{device_name}: MQTT settings updated successfully") return True, "Updated" def get_device_details(self, device_ip: str, device_name: str = "Unknown") -> Optional[Dict]: """ Get detailed device information from Tasmota device. Args: device_ip: Device IP address device_name: Device name for logging Returns: dict: Device details or None if failed """ # Get Status 0 (all status info) result, success = send_tasmota_command( device_ip, "Status%200", timeout=10, logger=self.logger ) if not success or not result: self.logger.warning(f"{device_name}: Failed to get device details") return None return result