"""Unknown device processing and interactive setup.""" import logging import time from typing import Optional from utils import send_tasmota_command, get_hostname_base from configuration import ConfigurationManager class UnknownDeviceProcessor: """Handles processing of unknown/unconfigured Tasmota devices.""" def __init__(self, config: dict, config_manager: ConfigurationManager, logger: Optional[logging.Logger] = None): """ Initialize unknown device processor. Args: config: Configuration dictionary config_manager: Configuration manager instance logger: Optional logger instance """ self.config = config self.config_manager = config_manager self.logger = logger or logging.getLogger(__name__) def process_unknown_devices(self, devices: list): """ Interactively process unknown devices. Args: devices: List of unknown devices """ if not devices: self.logger.info("No unknown devices to process") return self.logger.info(f"Found {len(devices)} unknown devices to process") for device in devices: self._process_single_unknown_device(device) def _process_single_unknown_device(self, device: dict): """ Process a single unknown device interactively. Args: device: Device info dictionary """ device_name = device.get('name', 'Unknown') device_ip = device.get('ip', '') if not device_ip: self.logger.warning(f"{device_name}: No IP address, skipping") return self.logger.info(f"\n{'='*60}") self.logger.info(f"Processing unknown device: {device_name} ({device_ip})") self.logger.info(f"{'='*60}") # Check if device has a power control result, success = send_tasmota_command(device_ip, "Power", timeout=5, logger=self.logger) if not success: self.logger.warning(f"{device_name}: Cannot communicate with device, skipping") return # Check if device has power control capability has_power = 'POWER' in result or 'POWER1' in result if not has_power: self.logger.warning(f"{device_name}: Device has no power control, skipping toggle") new_hostname = self._prompt_for_hostname(device_name, device_ip, toggle=False) else: # Start toggling and prompt for hostname new_hostname = self._prompt_for_hostname_with_toggle(device_name, device_ip) if not new_hostname: self.logger.info(f"{device_name}: Skipped (no hostname entered)") return # Configure the device with new hostname self._configure_device(device_ip, device_name, new_hostname) def _prompt_for_hostname_with_toggle(self, device_name: str, device_ip: str) -> Optional[str]: """ Prompt for hostname while toggling device power. Args: device_name: Current device name device_ip: Device IP address Returns: str: New hostname or None if cancelled """ import threading self.logger.info(f"{device_name}: Toggling power to help identify device...") self.logger.info("The device will toggle on/off every 2 seconds") # Flag to control toggle thread stop_toggle = threading.Event() def toggle_power(): """Toggle power in background thread.""" while not stop_toggle.is_set(): send_tasmota_command(device_ip, "Power%20Toggle", timeout=3) time.sleep(2) # Start toggle thread toggle_thread = threading.Thread(target=toggle_power, daemon=True) toggle_thread.start() try: # Prompt for hostname new_hostname = self._prompt_for_hostname(device_name, device_ip, toggle=True) finally: # Stop toggling stop_toggle.set() toggle_thread.join(timeout=3) # Turn off the device send_tasmota_command(device_ip, "Power%20Off", timeout=3) return new_hostname def _prompt_for_hostname(self, device_name: str, device_ip: str, toggle: bool = False) -> Optional[str]: """ Prompt user for new hostname. Args: device_name: Current device name device_ip: Device IP address toggle: Whether device is currently toggling Returns: str: New hostname or None if cancelled """ print(f"\n{'='*60}") print(f"Unknown Device Found:") print(f" Current Name: {device_name}") print(f" IP Address: {device_ip}") if toggle: print(f" Status: Device is toggling to help identify it") print(f"{'='*60}") print(f"Enter new hostname for this device (or press Enter to skip):") try: new_hostname = input("> ").strip() if not new_hostname: return None return new_hostname except (KeyboardInterrupt, EOFError): print("\nCancelled") return None def _configure_device(self, device_ip: str, old_name: str, new_hostname: str): """ Configure device with new hostname and MQTT settings. Args: device_ip: Device IP address old_name: Old device name new_hostname: New hostname to set """ self.logger.info(f"{old_name}: Configuring device with hostname '{new_hostname}'") # Set Friendly Name 1 command = f"FriendlyName1%20{new_hostname}" result, success = send_tasmota_command(device_ip, command, timeout=5, logger=self.logger) if not success: self.logger.error(f"{old_name}: Failed to set hostname") return self.logger.info(f"{old_name}: Hostname set to '{new_hostname}'") # Set DeviceName (for MQTT) command = f"DeviceName%20{new_hostname}" send_tasmota_command(device_ip, command, timeout=5, logger=self.logger) # Get device details for MQTT configuration device_details = self.config_manager.get_device_details(device_ip, new_hostname) if not device_details: self.logger.warning(f"{new_hostname}: Could not get device details") else: # Configure MQTT settings device_info = {'name': new_hostname, 'ip': device_ip} success, status = self.config_manager.configure_mqtt_settings(device_info, device_details) if success: self.logger.info(f"{new_hostname}: MQTT settings configured") else: self.logger.warning(f"{new_hostname}: MQTT configuration incomplete: {status}") # Restart device to apply all changes self.logger.info(f"{new_hostname}: Restarting device to apply changes") send_tasmota_command(device_ip, "Restart%201", timeout=5, logger=self.logger) self.logger.info(f"{new_hostname}: Configuration complete")