Compare commits

...

1 Commits

Author SHA1 Message Date
8894c1be4b Feature: Add parallel device processing and improved error handling
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>
2025-12-30 11:35:25 -06:00
8 changed files with 380 additions and 119 deletions

View File

@ -4,6 +4,7 @@ import argparse
import logging
import sys
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
from utils import load_json_file, ensure_data_directory, get_data_file_path, is_valid_ip, match_pattern
from unifi_client import UnifiClient, AuthenticationError
@ -94,72 +95,154 @@ def setup_unifi_client(config: dict, logger: logging.Logger) -> Optional[UnifiCl
return None
def process_devices(devices: list, config_manager: ConfigurationManager,
console_manager: ConsoleSettingsManager, logger: logging.Logger):
def process_single_device(device: dict, config_manager: ConfigurationManager,
console_manager: ConsoleSettingsManager,
logger: logging.Logger,
report_gen: 'ReportGenerator') -> tuple:
"""
Process all devices for configuration.
Process a single device for configuration.
Args:
device: Device to process
config_manager: Configuration manager instance
console_manager: Console settings manager instance
logger: Logger instance
report_gen: Report generator for error collection
Returns:
Tuple of (device_info, success, messages)
"""
device_name = device.get('name', 'Unknown')
device_ip = device.get('ip', '')
messages = []
try:
logger.debug(f"Processing: {device_name} ({device_ip})")
# Get device details
device_details = config_manager.get_device_details(device_ip, device_name)
if not device_details:
messages.append(f" {device_name}: Could not get device details, skipping")
report_gen.add_error(device_name, f"Connection failed to {device_ip}")
return None, False, messages
# Check and update template
template_success = config_manager.check_and_update_template(device, device_details)
if template_success:
logger.debug(f" {device_name}: Template checked/updated")
# Refresh device details after template update
device_details = config_manager.get_device_details(device_ip, device_name)
# Configure MQTT
mqtt_success, mqtt_status = config_manager.configure_mqtt_settings(device, device_details)
if mqtt_success:
if mqtt_status == "Updated":
logger.debug(f" {device_name}: MQTT updated")
else:
logger.debug(f" {device_name}: MQTT already configured")
else:
messages.append(f" {device_name}: MQTT configuration failed - {mqtt_status}")
report_gen.add_error(device_name, f"MQTT configuration failed: {mqtt_status}")
# Apply console settings
console_success, console_status = console_manager.apply_console_settings(device, device_details)
if console_success:
if console_status == "Applied":
logger.debug(f" {device_name}: Console settings applied")
elif console_status != "No console settings" and console_status != "Empty console set":
logger.debug(f" {device_name}: Console settings - {console_status}")
else:
messages.append(f" {device_name}: Console settings failed - {console_status}")
report_gen.add_warning(device_name, f"Console settings failed: {console_status}")
# Save device details
device_info = {
**device,
'mqtt_status': mqtt_status,
'console_status': console_status,
'firmware': device_details.get('StatusFWR', {}).get('Version', 'Unknown')
}
logger.debug(f" {device_name}: ✓ Processing completed successfully")
return device_info, True, messages
except Exception as e:
messages.append(f" {device_name}: ✗ Error during processing: {e}")
report_gen.add_error(device_name, f"Unexpected error: {str(e)}")
return None, False, messages
def process_devices(devices: list, config_manager: ConfigurationManager,
console_manager: ConsoleSettingsManager, logger: logging.Logger,
report_gen: 'ReportGenerator', max_workers: int = 10):
"""
Process all devices for configuration in parallel.
Args:
devices: List of devices to process
config_manager: Configuration manager instance
console_manager: Console settings manager instance
logger: Logger instance
report_gen: Report generator for error collection
max_workers: Maximum number of parallel workers (default: 10)
"""
device_details_list = []
stats = {'processed': 0, 'mqtt_updated': 0, 'console_updated': 0, 'failed': 0}
all_messages = []
for device in devices:
device_name = device.get('name', 'Unknown')
device_ip = device.get('ip', '')
logger.info(f"\nProcessing {len(devices)} devices in parallel (max {max_workers} workers)...")
logger.info(f"\nProcessing: {device_name} ({device_ip})")
# Process devices in parallel using ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# Submit all device processing tasks
future_to_device = {
executor.submit(process_single_device, device, config_manager,
console_manager, logger, report_gen): device
for device in devices
}
try:
# Get device details
device_details = config_manager.get_device_details(device_ip, device_name)
# Collect results as they complete
for future in as_completed(future_to_device):
device = future_to_device[future]
device_name = device.get('name', 'Unknown')
if not device_details:
logger.warning(f"{device_name}: Could not get device details, skipping")
try:
device_info, success, messages = future.result()
# Store messages for reporting (only errors/warnings)
all_messages.extend(messages)
if success and device_info:
device_details_list.append(device_info)
stats['processed'] += 1
# Track MQTT and console updates
if device_info.get('mqtt_status') == 'Updated':
stats['mqtt_updated'] += 1
if device_info.get('console_status') == 'Applied':
stats['console_updated'] += 1
else:
stats['failed'] += 1
except Exception as e:
all_messages.append(f"{device_name}: Unexpected error: {e}")
report_gen.add_error(device_name, f"Processing exception: {str(e)}")
stats['failed'] += 1
continue
# Check and update template
template_success = config_manager.check_and_update_template(device, device_details)
# Refresh device details after template update
if template_success:
device_details = config_manager.get_device_details(device_ip, device_name)
# Configure MQTT
mqtt_success, mqtt_status = config_manager.configure_mqtt_settings(device, device_details)
if mqtt_success and mqtt_status == "Updated":
stats['mqtt_updated'] += 1
# Apply console settings
console_success, console_status = console_manager.apply_console_settings(device, device_details)
if console_success and console_status == "Applied":
stats['console_updated'] += 1
# Save device details
device_info = {
**device,
'mqtt_status': mqtt_status,
'console_status': console_status,
'firmware': device_details.get('StatusFWR', {}).get('Version', 'Unknown')
}
device_details_list.append(device_info)
stats['processed'] += 1
except Exception as e:
logger.error(f"{device_name}: Error during processing: {e}")
stats['failed'] += 1
# Print only error/warning messages (messages list now only contains issues)
if all_messages:
logger.info("\n" + "=" * 60)
logger.info("PROCESSING ISSUES")
logger.info("=" * 60)
for message in all_messages:
logger.info(message)
return device_details_list, stats
def find_device_by_identifier(devices: list, identifier: str, logger: logging.Logger) -> Optional[dict]:
"""
Find a device by IP address or hostname.
@ -286,7 +369,7 @@ def main():
# Process all devices
logger.info(f"\nProcessing {len(devices)} devices...")
device_details_list, stats = process_devices(devices, config_manager, console_manager, logger)
device_details_list, stats = process_devices(devices, config_manager, console_manager, logger, report_gen)
# Save device details
report_gen.save_device_details(device_details_list)
@ -301,6 +384,9 @@ def main():
console_manager.print_failure_summary()
# Print errors and warnings summary
report_gen.print_errors_and_warnings_summary()
logger.info("TasmotaManager completed")
return 0

View File

@ -128,13 +128,14 @@ class ConfigurationManager:
return False
def configure_mqtt_settings(self, device: dict, device_details: dict) -> Tuple[bool, str]:
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)
@ -154,6 +155,9 @@ class ConfigurationManager:
# 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', '') != ''
@ -168,39 +172,77 @@ class ConfigurationManager:
# Build list of settings to update
updates_needed = []
password_needs_update = False
# Check each MQTT setting
mqtt_host = mqtt_config.get('Host', '')
if mqtt_host and current_mqtt.get('MqttHost', '') != mqtt_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)
if current_mqtt.get('MqttPort', 0) != mqtt_port:
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', '')
if mqtt_user and current_mqtt.get('MqttUser', '') != mqtt_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', '')
# Note: Can't verify password from status, so always set it
if mqtt_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)
if current_mqtt.get('Topic', '') != mqtt_topic:
# 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 and current_mqtt.get('FullTopic', '') != mqtt_full_topic:
updates_needed.append(('FullTopic', mqtt_full_topic))
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'))
@ -208,8 +250,8 @@ class ConfigurationManager:
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")
# 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:

View File

@ -2,6 +2,7 @@
import logging
import time
import threading
from typing import Dict, List, Optional, Tuple
from utils import send_tasmota_command, retry_command, get_hostname_base
@ -21,6 +22,7 @@ class ConsoleSettingsManager:
self.config = config
self.logger = logger or logging.getLogger(__name__)
self.command_failures = {} # Track failed commands by device
self._lock = threading.Lock() # Thread-safe access to command_failures
def apply_console_settings(self, device: dict, device_details: dict) -> Tuple[bool, str]:
"""
@ -73,9 +75,10 @@ class ConsoleSettingsManager:
# 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)
with self._lock:
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"
@ -233,16 +236,19 @@ class ConsoleSettingsManager:
def print_failure_summary(self):
"""Print summary of all command failures."""
if not self.command_failures:
return
with self._lock:
if not self.command_failures:
return
self.logger.error("=" * 60)
self.logger.error("COMMAND FAILURE SUMMARY")
self.logger.error("=" * 60)
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}")
# Sort by device name for consistent output
for device_name in sorted(self.command_failures.keys()):
failed_commands = self.command_failures[device_name]
self.logger.error(f"\n{device_name}:")
for cmd in failed_commands:
self.logger.error(f" - {cmd}")
self.logger.error("=" * 60)
self.logger.error("=" * 60)

View File

@ -76,7 +76,9 @@ class TasmotaDiscovery:
exclude_patterns = network_config.get('exclude_patterns', [])
for pattern in exclude_patterns:
if match_pattern(device_name, pattern) or match_pattern(device_hostname, pattern):
# Use match_entire_string=False to allow partial matching with wildcards
if match_pattern(device_name, pattern, match_entire_string=False) or \
match_pattern(device_hostname, pattern, match_entire_string=False):
return True
return False
@ -228,8 +230,23 @@ class TasmotaDiscovery:
# Track deprecated devices
if previous_data:
# Handle case where previous_data might be a list directly or wrapped in a dict
if isinstance(previous_data, list):
previous_devices = previous_data
elif isinstance(previous_data, dict):
# Check if it has a 'devices' key (wrapped format)
if 'devices' in previous_data:
previous_devices = previous_data['devices']
else:
# Assume it's a single device dict, wrap it
previous_devices = [previous_data]
else:
# Invalid format, skip deprecated device tracking
self.logger.warning(f"Previous data has unexpected type: {type(previous_data)}")
return
current_ips = {d['ip'] for d in devices}
deprecated = [d for d in previous_data if d.get('ip') not in current_ips]
deprecated = [d for d in previous_devices if d.get('ip') not in current_ips]
if deprecated:
self.logger.info(f"Found {len(deprecated)} deprecated devices")

View File

@ -10,14 +10,14 @@
"name": "NoT",
"subnet": "192.168.8",
"exclude_patterns": [
"^homeassistant*",
"^.*sonos.*"
"homeassistant*",
"*sonos*"
],
"unknown_device_patterns": [
"^tasmota_*",
"^tasmota-*",
"^esp-*",
"^ESP-*"
"tasmota_*",
"tasmota-*",
"esp-*",
"ESP-*"
]
}
}

View File

@ -24,6 +24,53 @@ class ReportGenerator:
self.config = config
self.discovery = discovery
self.logger = logger or logging.getLogger(__name__)
self.errors_and_warnings = [] # Collect errors and warnings
def add_error(self, device_name: str, message: str):
"""Add an error message to the collection."""
self.errors_and_warnings.append(('ERROR', device_name, message))
def add_warning(self, device_name: str, message: str):
"""Add a warning message to the collection."""
self.errors_and_warnings.append(('WARNING', device_name, message))
def print_errors_and_warnings_summary(self):
"""Print summary of all errors and warnings that require user attention."""
if not self.errors_and_warnings:
return
self.logger.info("")
self.logger.error("=" * 60)
self.logger.error("ERRORS AND WARNINGS REQUIRING ATTENTION")
self.logger.error("=" * 60)
# Sort by severity (ERROR first, then WARNING) and then by device name
sorted_issues = sorted(self.errors_and_warnings,
key=lambda x: (0 if x[0] == 'ERROR' else 1, x[1]))
for severity, device_name, message in sorted_issues:
if severity == 'ERROR':
self.logger.error(f"{device_name}: {message}")
else:
self.logger.warning(f"{device_name}: {message}")
# Print action items
self.logger.error("")
self.logger.error("ACTION REQUIRED:")
# Group by issue type
connection_errors = [x for x in sorted_issues if 'connection' in x[2].lower() or 'refused' in x[2].lower()]
mqtt_errors = [x for x in sorted_issues if 'mqtt' in x[2].lower()]
other_errors = [x for x in sorted_issues if x not in connection_errors and x not in mqtt_errors]
if connection_errors:
self.logger.error(f"{len(connection_errors)} device(s) unreachable - check if devices are online")
if mqtt_errors:
self.logger.error(f"{len(mqtt_errors)} device(s) with MQTT issues - review configuration")
if other_errors:
self.logger.error(f"{len(other_errors)} device(s) with other issues - review above details")
self.logger.error("=" * 60)
def generate_unifi_hostname_report(self) -> Dict:
"""

58
test_unifi_connection.py Normal file
View File

@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""Test UniFi connection and authentication."""
import requests
import urllib3
import json
urllib3.disable_warnings()
# Load your actual configuration
with open('network_configuration.json', 'r') as f:
config = json.load(f)
host = config['unifi']['host']
username = config['unifi']['username']
password = config['unifi']['password']
site = config['unifi'].get('site', 'default')
print(f'Testing connection to: {host}')
print(f'Username: {username}')
print(f'Site: {site}')
print('=' * 60)
# Test UniFi OS login (modern)
print('\n1. Attempting UniFi OS login (/api/auth/login)...')
try:
session = requests.Session()
response = session.post(
f'{host}/api/auth/login',
json={'username': username, 'password': password},
verify=False,
timeout=10
)
print(f' Status code: {response.status_code}')
print(f' Response: {response.text[:200]}')
if response.status_code == 200:
print(' ✓ UniFi OS authentication successful!')
except Exception as e:
print(f' ✗ Error: {e}')
# Test legacy UniFi Controller login (older controllers)
print('\n2. Attempting legacy UniFi Controller login (/api/login)...')
try:
session2 = requests.Session()
response2 = session2.post(
f'{host}/api/login',
json={'username': username, 'password': password},
verify=False,
timeout=10
)
print(f' Status code: {response2.status_code}')
print(f' Response: {response2.text[:200]}')
if response2.status_code == 200:
print(' ✓ Legacy UniFi authentication successful!')
except Exception as e:
print(f' ✗ Error: {e}')
print('\n' + '=' * 60)

View File

@ -48,7 +48,7 @@ class UnifiClient:
self._login()
def _request_json(self, endpoint: str, method: str = 'GET',
data: Optional[dict] = None) -> dict:
data: Optional[dict] = None, check_meta: bool = True) -> dict:
"""
Make a request to the UniFi API and return JSON response.
@ -56,6 +56,7 @@ class UnifiClient:
endpoint: API endpoint path
method: HTTP method (GET, POST, etc.)
data: Optional data for POST requests
check_meta: Whether to check for meta.rc in response
Returns:
dict: JSON response
@ -80,11 +81,13 @@ class UnifiClient:
except ValueError:
raise UniFiDataError(f"Invalid JSON response from {endpoint}")
# Check for UniFi API error response
if isinstance(json_response, dict):
if json_response.get('meta', {}).get('rc') != 'ok':
error_msg = json_response.get('meta', {}).get('msg', 'Unknown error')
raise UniFiDataError(f"UniFi API error: {error_msg}")
# Check for UniFi API error response (only if check_meta is True)
if check_meta and isinstance(json_response, dict):
# Legacy UniFi controller format
if 'meta' in json_response:
if json_response.get('meta', {}).get('rc') != 'ok':
error_msg = json_response.get('meta', {}).get('msg', 'Unknown error')
raise UniFiDataError(f"UniFi API error: {error_msg}")
return json_response
@ -105,7 +108,9 @@ class UnifiClient:
}
try:
response = self._request_json('/api/auth/login', method='POST', data=login_data)
# UniFi OS doesn't return meta.rc for login, so don't check it
response = self._request_json('/api/auth/login', method='POST',
data=login_data, check_meta=False)
self.logger.debug("Successfully authenticated with UniFi controller")
except UniFiDataError as e:
@ -122,7 +127,7 @@ class UnifiClient:
Raises:
UniFiDataError: If request fails
"""
endpoint = f'/api/s/{self.site_id}/stat/sta'
endpoint = f'/proxy/network/api/s/{self.site_id}/stat/sta'
try:
response = self._request_json(endpoint)
@ -148,7 +153,7 @@ class UnifiClient:
Raises:
UniFiDataError: If request fails
"""
endpoint = f'/api/s/{self.site_id}/stat/device'
endpoint = f'/proxy/network/api/s/{self.site_id}/stat/device'
try:
response = self._request_json(endpoint)