Mike Geppert
2f3fd5d711
Major changes: - Implement parallel device processing using ThreadPoolExecutor (10 workers) - Add comprehensive error and warning tracking in ReportGenerator - Fix MQTT configuration verification (query Topic/FullTopic directly) - Improve console settings thread safety with locks - Fix UniFi client for UniFi OS API endpoints - Normalize FullTopic handling (strip URL-encoded spaces) - Update network exclude patterns to support wildcards - Add test_unifi_connection.py for debugging UniFi connectivity Performance improvements: - Process devices concurrently for faster execution - Reduced verbose logging during parallel processing Bug fixes: - Handle deprecated.json format correctly (list vs dict) - Fix exclude_patterns matching with partial string support - Fix UniFi API authentication and endpoint paths for UniFi OS 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""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, force_password_update: bool = False) -> Tuple[bool, str]:
|
|
"""
|
|
Configure MQTT settings on a device.
|
|
|
|
Args:
|
|
device: Device info dictionary
|
|
device_details: Detailed device information
|
|
force_password_update: Force update of password even if user hasn't changed
|
|
|
|
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', {})
|
|
|
|
# Log current MQTT state for debugging
|
|
self.logger.debug(f"{device_name}: Current MQTT settings from device: {current_mqtt}")
|
|
|
|
# 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 = []
|
|
password_needs_update = False
|
|
|
|
# Check each MQTT setting
|
|
mqtt_host = mqtt_config.get('Host', '')
|
|
current_host = current_mqtt.get('MqttHost', '')
|
|
self.logger.debug(f"{device_name}: Comparing MqttHost: current='{current_host}' vs expected='{mqtt_host}'")
|
|
if mqtt_host and current_host != mqtt_host:
|
|
updates_needed.append(('MqttHost', mqtt_host))
|
|
|
|
mqtt_port = mqtt_config.get('Port', 1883)
|
|
current_port = current_mqtt.get('MqttPort', 0)
|
|
self.logger.debug(f"{device_name}: Comparing MqttPort: current={current_port} vs expected={mqtt_port}")
|
|
if current_port != mqtt_port:
|
|
updates_needed.append(('MqttPort', mqtt_port))
|
|
|
|
mqtt_user = mqtt_config.get('User', '')
|
|
current_user = current_mqtt.get('MqttUser', '')
|
|
self.logger.debug(f"{device_name}: Comparing MqttUser: current='{current_user}' vs expected='{mqtt_user}'")
|
|
if mqtt_user and current_user != mqtt_user:
|
|
updates_needed.append(('MqttUser', mqtt_user))
|
|
password_needs_update = True # If user changed, update password too
|
|
|
|
# Only update password if:
|
|
# 1. force_password_update is True, OR
|
|
# 2. The username is being updated (password likely needs to match)
|
|
mqtt_password = mqtt_config.get('Password', '')
|
|
if mqtt_password and (force_password_update or password_needs_update):
|
|
updates_needed.append(('MqttPassword', mqtt_password))
|
|
self.logger.debug(f"{device_name}: Password will be updated (force={force_password_update}, user_changed={password_needs_update})")
|
|
|
|
# Handle Topic with %hostname_base% substitution
|
|
# Note: Topic is not always in StatusMQT, so query it directly
|
|
mqtt_topic = mqtt_config.get('Topic', '')
|
|
if mqtt_topic:
|
|
mqtt_topic = mqtt_topic.replace('%hostname_base%', hostname_base)
|
|
|
|
# Query current Topic value directly
|
|
result, success = send_tasmota_command(device_ip, "Topic", timeout=5, logger=self.logger)
|
|
current_topic = result.get('Topic', '') if success and result else ''
|
|
|
|
self.logger.debug(f"{device_name}: Comparing Topic: current='{current_topic}' vs expected='{mqtt_topic}'")
|
|
if current_topic != mqtt_topic:
|
|
updates_needed.append(('Topic', mqtt_topic))
|
|
|
|
# Handle FullTopic
|
|
# Note: FullTopic is not always in StatusMQT, so query it directly
|
|
mqtt_full_topic = mqtt_config.get('FullTopic', '')
|
|
if mqtt_full_topic:
|
|
# Query current FullTopic value directly
|
|
result, success = send_tasmota_command(device_ip, "FullTopic", timeout=5, logger=self.logger)
|
|
current_full_topic = result.get('FullTopic', '') if success and result else ''
|
|
|
|
self.logger.debug(f"{device_name}: Raw FullTopic from device: '{current_full_topic}'")
|
|
|
|
# Normalize: remove any URL-encoded spaces from the beginning of current value
|
|
# This handles the case where the device returns '%20%prefix%' instead of '%prefix%'
|
|
while current_full_topic.startswith('%20'):
|
|
current_full_topic = current_full_topic[3:]
|
|
|
|
# Also normalize expected value in case config has leading spaces
|
|
mqtt_full_topic_normalized = mqtt_full_topic.lstrip()
|
|
|
|
self.logger.debug(f"{device_name}: Comparing FullTopic: current='{current_full_topic}' vs expected='{mqtt_full_topic_normalized}'")
|
|
|
|
if current_full_topic != mqtt_full_topic_normalized:
|
|
updates_needed.append(('FullTopic', mqtt_full_topic_normalized))
|
|
|
|
# Handle NoRetain (SetOption62)
|
|
no_retain = mqtt_config.get('NoRetain', False)
|
|
current_no_retain = current_mqtt.get('NoRetain', False)
|
|
self.logger.debug(f"{device_name}: Comparing NoRetain: current={current_no_retain} vs expected={no_retain}")
|
|
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"
|
|
|
|
# Log what will be updated
|
|
self.logger.info(f"{device_name}: Updating {len(updates_needed)} MQTT settings: {[name for name, _ in updates_needed]}")
|
|
|
|
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
|