Mike Geppert
9c22168f79
- Created modular Python files (main, utils, discovery, configuration, console_settings, unknown_devices, reporting, unifi_client) - Moved documentation files to docs/ - Moved data files to data/ - Removed old monolithic TasmotaManager.py and TasmotaManager_fixed.py - Updated .gitignore and pyproject.toml - All functionality preserved, command-line interface unchanged Version: 2.0.0
249 lines
8.8 KiB
Python
249 lines
8.8 KiB
Python
"""Console settings and parameter management."""
|
|
|
|
import logging
|
|
import time
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
from utils import send_tasmota_command, retry_command, get_hostname_base
|
|
|
|
|
|
class ConsoleSettingsManager:
|
|
"""Handles console parameter configuration for Tasmota devices."""
|
|
|
|
def __init__(self, config: dict, logger: Optional[logging.Logger] = None):
|
|
"""
|
|
Initialize console settings manager.
|
|
|
|
Args:
|
|
config: Configuration dictionary
|
|
logger: Optional logger instance
|
|
"""
|
|
self.config = config
|
|
self.logger = logger or logging.getLogger(__name__)
|
|
self.command_failures = {} # Track failed commands by device
|
|
|
|
def apply_console_settings(self, device: dict, device_details: dict) -> Tuple[bool, str]:
|
|
"""
|
|
Apply console settings to 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"
|
|
|
|
# Get hostname base for template matching
|
|
hostname = device_details.get('StatusNET', {}).get('Hostname', device_name)
|
|
hostname_base = get_hostname_base(hostname)
|
|
|
|
# Find which console_set to use for this device
|
|
console_set_name = self._get_console_set_name(hostname_base)
|
|
|
|
if not console_set_name:
|
|
self.logger.debug(f"{device_name}: No console settings configured")
|
|
return True, "No console settings"
|
|
|
|
# Get the console command list
|
|
console_commands = self._get_console_commands(console_set_name)
|
|
|
|
if not console_commands:
|
|
self.logger.debug(f"{device_name}: Console set '{console_set_name}' is empty")
|
|
return True, "Empty console set"
|
|
|
|
self.logger.info(f"{device_name}: Applying {len(console_commands)} console settings from '{console_set_name}'")
|
|
|
|
# Apply each console command
|
|
failed_commands = []
|
|
|
|
for command in console_commands:
|
|
if not command or not command.strip():
|
|
continue # Skip empty commands
|
|
|
|
success = self._apply_single_command(device_ip, device_name, command)
|
|
|
|
if not success:
|
|
failed_commands.append(command)
|
|
|
|
# Track failures for summary
|
|
if failed_commands:
|
|
if device_name not in self.command_failures:
|
|
self.command_failures[device_name] = []
|
|
self.command_failures[device_name].extend(failed_commands)
|
|
|
|
if failed_commands:
|
|
return False, f"Failed: {len(failed_commands)} commands"
|
|
|
|
self.logger.info(f"{device_name}: All console settings applied successfully")
|
|
return True, "Applied"
|
|
|
|
def _get_console_set_name(self, hostname_base: str) -> Optional[str]:
|
|
"""
|
|
Get the console_set name for a device based on hostname.
|
|
|
|
Args:
|
|
hostname_base: Base hostname of device
|
|
|
|
Returns:
|
|
str: Console set name or None
|
|
"""
|
|
device_list = self.config.get('device_list', {})
|
|
|
|
for template_name, template_data in device_list.items():
|
|
if hostname_base.lower() in template_name.lower():
|
|
return template_data.get('console_set')
|
|
|
|
return None
|
|
|
|
def _get_console_commands(self, console_set_name: str) -> List[str]:
|
|
"""
|
|
Get console commands from a named console set.
|
|
|
|
Args:
|
|
console_set_name: Name of the console set
|
|
|
|
Returns:
|
|
list: List of console commands
|
|
"""
|
|
console_set = self.config.get('console_set', {})
|
|
|
|
if isinstance(console_set, dict):
|
|
commands = console_set.get(console_set_name, [])
|
|
if isinstance(commands, list):
|
|
return commands
|
|
|
|
return []
|
|
|
|
def _apply_single_command(self, device_ip: str, device_name: str, command: str) -> bool:
|
|
"""
|
|
Apply a single console command to a device.
|
|
|
|
Args:
|
|
device_ip: Device IP address
|
|
device_name: Device name for logging
|
|
command: Console command to apply
|
|
|
|
Returns:
|
|
bool: True if successful
|
|
"""
|
|
# Parse command into parameter and value
|
|
parts = command.split(None, 1)
|
|
if not parts:
|
|
return True # Empty command, skip
|
|
|
|
param_name = parts[0]
|
|
param_value = parts[1] if len(parts) > 1 else ""
|
|
|
|
self.logger.debug(f"{device_name}: Setting {param_name} = {param_value}")
|
|
|
|
# Handle Retain parameters - set opposite first, then desired state
|
|
if param_name.endswith('Retain'):
|
|
opposite_value = 'Off' if param_value.lower() in ['on', '1', 'true'] else 'On'
|
|
|
|
# Set opposite first
|
|
opposite_command = f"{param_name}%20{opposite_value}"
|
|
result, success = send_tasmota_command(
|
|
device_ip, opposite_command, timeout=5, logger=self.logger
|
|
)
|
|
|
|
if not success:
|
|
self.logger.warning(f"{device_name}: Failed to set {param_name} to opposite state")
|
|
|
|
time.sleep(0.5) # Brief delay between commands
|
|
|
|
# Send the actual command
|
|
escaped_command = command.replace(' ', '%20')
|
|
|
|
result, success = retry_command(
|
|
lambda: send_tasmota_command(device_ip, escaped_command, timeout=5, logger=self.logger),
|
|
max_attempts=3,
|
|
delay=1.0,
|
|
logger=self.logger,
|
|
device_name=device_name
|
|
)
|
|
|
|
if not success:
|
|
self.logger.error(f"{device_name}: Failed to set {param_name} after 3 attempts")
|
|
return False
|
|
|
|
# Verify the command was applied (if possible)
|
|
if not self._verify_command(device_ip, device_name, param_name, param_value):
|
|
self.logger.warning(f"{device_name}: Verification failed for {param_name}")
|
|
# Don't return False here - some commands can't be verified
|
|
|
|
# Check if this is a rule definition - if so, enable it
|
|
if param_name.lower().startswith('rule'):
|
|
rule_number = param_name.lower().replace('rule', '')
|
|
if rule_number.isdigit():
|
|
enable_command = f"Rule{rule_number}%201"
|
|
|
|
self.logger.debug(f"{device_name}: Enabling rule{rule_number}")
|
|
|
|
result, success = send_tasmota_command(
|
|
device_ip, enable_command, timeout=5, logger=self.logger
|
|
)
|
|
|
|
if not success:
|
|
self.logger.warning(f"{device_name}: Failed to enable rule{rule_number}")
|
|
|
|
time.sleep(0.3) # Brief delay between commands
|
|
return True
|
|
|
|
def _verify_command(self, device_ip: str, device_name: str,
|
|
param_name: str, expected_value: str) -> bool:
|
|
"""
|
|
Verify a command was applied (where possible).
|
|
|
|
Args:
|
|
device_ip: Device IP address
|
|
device_name: Device name for logging
|
|
param_name: Parameter name
|
|
expected_value: Expected value
|
|
|
|
Returns:
|
|
bool: True if verified or verification not possible
|
|
"""
|
|
# Only verify certain parameters
|
|
verifiable = ['PowerOnState', 'SetOption']
|
|
|
|
if not any(param_name.startswith(v) for v in verifiable):
|
|
return True # Can't verify, assume success
|
|
|
|
# Get current value
|
|
result, success = send_tasmota_command(
|
|
device_ip, param_name, timeout=5, logger=self.logger
|
|
)
|
|
|
|
if not success or not result:
|
|
return True # Can't verify, assume success
|
|
|
|
# Check if value matches
|
|
current_value = result.get(param_name, '')
|
|
|
|
if str(current_value) == str(expected_value):
|
|
return True
|
|
|
|
return False
|
|
|
|
def print_failure_summary(self):
|
|
"""Print summary of all command failures."""
|
|
if not self.command_failures:
|
|
return
|
|
|
|
self.logger.error("=" * 60)
|
|
self.logger.error("COMMAND FAILURE SUMMARY")
|
|
self.logger.error("=" * 60)
|
|
|
|
for device_name, failed_commands in self.command_failures.items():
|
|
self.logger.error(f"\n{device_name}:")
|
|
for cmd in failed_commands:
|
|
self.logger.error(f" - {cmd}")
|
|
|
|
self.logger.error("=" * 60)
|