diff --git a/README.md b/README.md index 33dd984..a132fb4 100644 --- a/README.md +++ b/README.md @@ -63,26 +63,34 @@ Create a `network_configuration.json` file with the following structure: "Password": "mqtt-password", "Topic": "%hostname_base%", "FullTopic": "%prefix%/%topic%/", - "NoRetain": false, - "console": { - "SwitchRetain": "Off", - "ButtonRetain": "Off", - "PowerOnState": "3", - "PowerRetain": "On", - "SetOption1": "0", - "SetOption3": "1", - "SetOption4": "1", - "SetOption13": "0", - "SetOption19": "0", - "SetOption32": "8", - "SetOption53": "1", - "SetOption73": "1", - "rule1": "on button1#state=10 do power0 toggle endon" - } + "NoRetain": false + }, + "config_other": { + "Example_Device_Template": "{\"NAME\":\"Example\",\"GPIO\":[0],\"FLAG\":0,\"BASE\":18}" + }, + "console": { + "SwitchRetain": "Off", + "ButtonRetain": "Off", + "PowerOnState": "3", + "PowerRetain": "On", + "SetOption1": "0", + "SetOption3": "1", + "SetOption4": "1", + "SetOption13": "0", + "SetOption19": "0", + "SetOption32": "8", + "SetOption53": "1", + "SetOption73": "1", + "rule1": "on button1#state=10 do power0 toggle endon" } } ``` +Note: +- In the mqtt section, the Topic supports the placeholder "%hostname_base%". The script will replace this with the base of the device's hostname (everything before the first dash). For example, for a device named "KitchenLamp-1234", the Topic will be set to "KitchenLamp". +- NoRetain controls Tasmota's SetOption62 (true = No Retain, false = Use Retain). +- FullTopic typically remains "%prefix%/%topic%/" and is applied according to Tasmota's command format. + ## Usage Basic usage: @@ -180,7 +188,7 @@ This feature helps automate the setup of new Tasmota devices that haven't been p ## Console Parameters -The script supports setting Tasmota console parameters via the `console` section in the MQTT configuration. After verifying and updating MQTT settings, the script will apply all console parameters to each device. This allows you to: +The script supports setting Tasmota console parameters via the `console` section in the configuration. After verifying and updating MQTT settings, the script will apply all console parameters to each device. This allows you to: - Configure device behavior (PowerOnState, SetOptions, etc.) - Set up rules for button actions diff --git a/TasmotaManager.py b/TasmotaManager.py old mode 100644 new mode 100755 index c86280b..9411d54 --- a/TasmotaManager.py +++ b/TasmotaManager.py @@ -1325,309 +1325,7 @@ class TasmotaDiscovery: self.logger.error(f"{name}: Error updating {setting}: {str(e)}") # Apply console settings - console_updated = False - console_params = self.config.get('console', {}) - if console_params: - self.logger.info(f"{name}: Setting console parameters from configuration") - - # Special handling for Retain parameters - need to send opposite state first, then final state - # This is necessary because the changes are what create the update of the Retain state at the MQTT server - retain_params = ["ButtonRetain", "SwitchRetain", "PowerRetain"] - - # Process Retain parameters first - for param in retain_params: - if param in console_params: - try: - final_value = console_params[param] - # Set opposite state first - opposite_value = "On" if final_value.lower() == "off" else "Off" - - if with_retry: - # First command (opposite state) - with retry logic - url = f"http://{ip}/cm?cmnd={param}%20{opposite_value}" - success = False - attempts = 0 - max_attempts = 3 - last_error = None - - while not success and attempts < max_attempts: - attempts += 1 - try: - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.debug(f"{name}: Set {param} to {opposite_value} (step 1 of 2 to update MQTT broker retain settings)") - console_updated = True - success = True - else: - self.logger.warning(f"{name}: Failed to set {param} to {opposite_value} (attempt {attempts}/{max_attempts})") - last_error = f"HTTP {response.status_code}" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.Timeout as e: - self.logger.warning(f"{name}: Timeout setting {param} to {opposite_value} (attempt {attempts}/{max_attempts})") - last_error = "Timeout" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.RequestException as e: - self.logger.warning(f"{name}: Error setting {param} to {opposite_value}: {str(e)} (attempt {attempts}/{max_attempts})") - last_error = str(e) - if attempts < max_attempts: - time.sleep(1) # Wait before retry - - if not success: - self.logger.error(f"{name}: Failed to set {param} to {opposite_value} after {max_attempts} attempts. Last error: {last_error}") - # Track the failure for later reporting - if not hasattr(self, 'command_failures'): - self.command_failures = [] - self.command_failures.append({ - "device": name, - "ip": ip, - "command": f"{param} {opposite_value}", - "error": last_error - }) - else: - # First command (opposite state) - without retry logic - url = f"http://{ip}/cm?cmnd={param}%20{opposite_value}" - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.debug(f"{name}: Set {param} to {opposite_value} (step 1 of 2 to update MQTT broker retain settings)") - else: - self.logger.error(f"{name}: Failed to set {param} to {opposite_value}") - - # Small delay to ensure commands are processed in order - time.sleep(0.5) - - if with_retry: - # Second command (final state) - with retry logic - url = f"http://{ip}/cm?cmnd={param}%20{final_value}" - success = False - attempts = 0 - last_error = None - - while not success and attempts < max_attempts: - attempts += 1 - try: - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.debug(f"{name}: Set {param} to {final_value} (step 2 of 2 to update MQTT broker retain settings)") - console_updated = True - success = True - else: - self.logger.warning(f"{name}: Failed to set {param} to {final_value} (attempt {attempts}/{max_attempts})") - last_error = f"HTTP {response.status_code}" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.Timeout as e: - self.logger.warning(f"{name}: Timeout setting {param} to {final_value} (attempt {attempts}/{max_attempts})") - last_error = "Timeout" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.RequestException as e: - self.logger.warning(f"{name}: Error setting {param} to {final_value}: {str(e)} (attempt {attempts}/{max_attempts})") - last_error = str(e) - if attempts < max_attempts: - time.sleep(1) # Wait before retry - - if not success: - self.logger.error(f"{name}: Failed to set {param} to {final_value} after {max_attempts} attempts. Last error: {last_error}") - # Track the failure for later reporting - if not hasattr(self, 'command_failures'): - self.command_failures = [] - self.command_failures.append({ - "device": name, - "ip": ip, - "command": f"{param} {final_value}", - "error": last_error - }) - else: - # Second command (final state) - without retry logic - url = f"http://{ip}/cm?cmnd={param}%20{final_value}" - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.debug(f"{name}: Set {param} to {final_value} (step 2 of 2 to update MQTT broker retain settings)") - else: - self.logger.error(f"{name}: Failed to set {param} to {final_value}") - except Exception as e: - self.logger.error(f"{name}: Unexpected error setting {param} commands: {str(e)}") - # Track the failure for later reporting if using retry logic - if with_retry: - if not hasattr(self, 'command_failures'): - self.command_failures = [] - self.command_failures.append({ - "device": name, - "ip": ip, - "command": f"{param} (both steps)", - "error": str(e) - }) - - # Process all other console parameters - # Track rules that need to be enabled - rules_to_enable = {} - - for param, value in console_params.items(): - # Skip Retain parameters as they're handled specially above - if param in retain_params: - continue - - # Check if this is a rule definition (lowercase rule1, rule2, etc.) - if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): - # Store the rule number for later enabling - rule_num = param[-1] - rules_to_enable[rule_num] = True - if with_retry: - self.logger.info(f"{name}: Detected rule definition {param}='{value}', will auto-enable") - else: - self.logger.debug(f"{name}: Detected rule definition {param}, will auto-enable") - - # Skip Rule1, Rule2, etc. if we're auto-enabling rules and using retry logic - if with_retry and param.lower().startswith('rule') and param.lower() != param and param[-1].isdigit(): - # If this is in the config, we'll respect it, but log that it's not needed - self.logger.debug(f"{name}: Note: {param} is not needed with auto-enable feature") - - # Regular console parameter - # Special handling for rule parameters to properly encode the URL - if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): - # For rule commands, we need to URL encode the entire value to preserve special characters - import urllib.parse - encoded_value = urllib.parse.quote(value) - url = f"http://{ip}/cm?cmnd={param}%20{encoded_value}" - self.logger.info(f"{name}: Sending rule command: {url}") - else: - url = f"http://{ip}/cm?cmnd={param}%20{value}" - - if with_retry: - # With retry logic - success = False - attempts = 0 - max_attempts = 3 - last_error = None - - while not success and attempts < max_attempts: - attempts += 1 - try: - response = requests.get(url, timeout=5) - if response.status_code == 200: - # Special logging for rule parameters - if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): - self.logger.info(f"{name}: Rule command response: {response.text}") - self.logger.info(f"{name}: Set rule {param} to '{value}'") - else: - self.logger.debug(f"{name}: Set console parameter {param} to {value}") - console_updated = True - success = True - else: - self.logger.warning(f"{name}: Failed to set console parameter {param} (attempt {attempts}/{max_attempts})") - last_error = f"HTTP {response.status_code}" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.Timeout as e: - self.logger.warning(f"{name}: Timeout setting console parameter {param} (attempt {attempts}/{max_attempts})") - last_error = "Timeout" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.RequestException as e: - self.logger.warning(f"{name}: Error setting console parameter {param}: {str(e)} (attempt {attempts}/{max_attempts})") - last_error = str(e) - if attempts < max_attempts: - time.sleep(1) # Wait before retry - - if not success: - self.logger.error(f"{name}: Failed to set console parameter {param} after {max_attempts} attempts. Last error: {last_error}") - # Track the failure for later reporting - if not hasattr(self, 'command_failures'): - self.command_failures = [] - self.command_failures.append({ - "device": name, - "ip": ip, - "command": f"{param} {value}", - "error": last_error - }) - else: - # Without retry logic - response = requests.get(url, timeout=5) - if response.status_code == 200: - if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): - self.logger.info(f"{name}: Rule command response: {response.text}") - self.logger.info(f"{name}: Set rule {param} to '{value}'") - else: - self.logger.debug(f"{name}: Set console parameter {param} to {value}") - else: - self.logger.error(f"{name}: Failed to set console parameter {param}") - - # Auto-enable any rules that were defined - if with_retry: - self.logger.info(f"{name}: Rules to enable: {rules_to_enable}") - - for rule_num in rules_to_enable: - rule_enable_param = f"Rule{rule_num}" - - # Skip if the rule enable command was already in the config - if with_retry: - # Check if the uppercase version (Rule1) is in the config - if rule_enable_param in console_params: - self.logger.info(f"{name}: Skipping {rule_enable_param} as it's already in config (uppercase version)") - continue - - # If we're here, it means we found a rule definition earlier and added it to rules_to_enable - # No need to check again if it's in console_params - self.logger.info(f"{name}: Will enable {rule_enable_param} for rule definition found in config") - else: - # Simple check for any version of the rule enable command - if any(p.lower() == rule_enable_param.lower() for p in console_params): - continue - - # Rule auto-enabling - url = f"http://{ip}/cm?cmnd={rule_enable_param}%201" - - if with_retry: - # With retry logic - success = False - attempts = 0 - max_attempts = 3 - last_error = None - - while not success and attempts < max_attempts: - attempts += 1 - try: - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.info(f"{name}: Auto-enabled {rule_enable_param}") - console_updated = True - success = True - else: - self.logger.warning(f"{name}: Failed to auto-enable {rule_enable_param} (attempt {attempts}/{max_attempts})") - last_error = f"HTTP {response.status_code}" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.Timeout as e: - self.logger.warning(f"{name}: Timeout auto-enabling {rule_enable_param} (attempt {attempts}/{max_attempts})") - last_error = "Timeout" - if attempts < max_attempts: - time.sleep(1) # Wait before retry - except requests.exceptions.RequestException as e: - self.logger.warning(f"{name}: Error auto-enabling {rule_enable_param}: {str(e)} (attempt {attempts}/{max_attempts})") - last_error = str(e) - if attempts < max_attempts: - time.sleep(1) # Wait before retry - - if not success: - self.logger.error(f"{name}: Failed to auto-enable {rule_enable_param} after {max_attempts} attempts. Last error: {last_error}") - # Track the failure for later reporting - if not hasattr(self, 'command_failures'): - self.command_failures = [] - self.command_failures.append({ - "device": name, - "ip": ip, - "command": f"{rule_enable_param} 1", - "error": last_error - }) - else: - # Without retry logic - response = requests.get(url, timeout=5) - if response.status_code == 200: - self.logger.info(f"{name}: Auto-enabled {rule_enable_param}") - else: - self.logger.error(f"{name}: Failed to auto-enable {rule_enable_param}") + console_updated = self.apply_console_settings(ip, name, with_retry) # Reboot the device if requested if reboot: @@ -1644,6 +1342,323 @@ class TasmotaDiscovery: self.logger.error(f"Error configuring device at {ip}: {str(e)}") return False + def apply_console_settings(self, ip, name, with_retry=False): + """Apply console parameters from configuration to the device. + Returns True if any setting was updated, False otherwise. + """ + console_updated = False + console_params = self.config.get('console', {}) + if not console_params: + return False + + self.logger.info(f"{name}: Setting console parameters from configuration") + + # Special handling for Retain parameters - need to send opposite state first, then final state + # This is necessary because the changes are what create the update of the Retain state at the MQTT server + retain_params = ["ButtonRetain", "SwitchRetain", "PowerRetain"] + + # Process Retain parameters first + for param in retain_params: + if param in console_params: + try: + final_value = console_params[param] + # Set opposite state first + opposite_value = "On" if final_value.lower() == "off" else "Off" + + if with_retry: + # First command (opposite state) - with retry logic + url = f"http://{ip}/cm?cmnd={param}%20{opposite_value}" + success = False + attempts = 0 + max_attempts = 3 + last_error = None + + while not success and attempts < max_attempts: + attempts += 1 + try: + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.debug(f"{name}: Set {param} to {opposite_value} (step 1 of 2 to update MQTT broker retain settings)") + console_updated = True + success = True + else: + self.logger.warning(f"{name}: Failed to set {param} to {opposite_value} (attempt {attempts}/{max_attempts})") + last_error = f"HTTP {response.status_code}" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.Timeout as e: + self.logger.warning(f"{name}: Timeout setting {param} to {opposite_value} (attempt {attempts}/{max_attempts})") + last_error = "Timeout" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.RequestException as e: + self.logger.warning(f"{name}: Error setting {param} to {opposite_value}: {str(e)} (attempt {attempts}/{max_attempts})") + last_error = str(e) + if attempts < max_attempts: + time.sleep(1) # Wait before retry + + if not success: + self.logger.error(f"{name}: Failed to set {param} to {opposite_value} after {max_attempts} attempts. Last error: {last_error}") + # Track the failure for later reporting + if not hasattr(self, 'command_failures'): + self.command_failures = [] + self.command_failures.append({ + "device": name, + "ip": ip, + "command": f"{param} {opposite_value}", + "error": last_error + }) + else: + # First command (opposite state) - without retry logic + url = f"http://{ip}/cm?cmnd={param}%20{opposite_value}" + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.debug(f"{name}: Set {param} to {opposite_value} (step 1 of 2 to update MQTT broker retain settings)") + else: + self.logger.error(f"{name}: Failed to set {param} to {opposite_value}") + + # Small delay to ensure commands are processed in order + time.sleep(0.5) + + if with_retry: + # Second command (final state) - with retry logic + url = f"http://{ip}/cm?cmnd={param}%20{final_value}" + success = False + attempts = 0 + max_attempts = 3 + last_error = None + + while not success and attempts < max_attempts: + attempts += 1 + try: + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.debug(f"{name}: Set {param} to {final_value} (step 2 of 2 to update MQTT broker retain settings)") + console_updated = True + success = True + else: + self.logger.warning(f"{name}: Failed to set {param} to {final_value} (attempt {attempts}/{max_attempts})") + last_error = f"HTTP {response.status_code}" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.Timeout as e: + self.logger.warning(f"{name}: Timeout setting {param} to {final_value} (attempt {attempts}/{max_attempts})") + last_error = "Timeout" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.RequestException as e: + self.logger.warning(f"{name}: Error setting {param} to {final_value}: {str(e)} (attempt {attempts}/{max_attempts})") + last_error = str(e) + if attempts < max_attempts: + time.sleep(1) # Wait before retry + + if not success: + self.logger.error(f"{name}: Failed to set {param} to {final_value} after {max_attempts} attempts. Last error: {last_error}") + # Track the failure for later reporting + if not hasattr(self, 'command_failures'): + self.command_failures = [] + self.command_failures.append({ + "device": name, + "ip": ip, + "command": f"{param} {final_value}", + "error": last_error + }) + else: + # Second command (final state) - without retry logic + url = f"http://{ip}/cm?cmnd={param}%20{final_value}" + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.debug(f"{name}: Set {param} to {final_value} (step 2 of 2 to update MQTT broker retain settings)") + else: + self.logger.error(f"{name}: Failed to set {param} to {final_value}") + except Exception as e: + self.logger.error(f"{name}: Unexpected error setting {param} commands: {str(e)}") + # Track the failure for later reporting if using retry logic + if with_retry: + if not hasattr(self, 'command_failures'): + self.command_failures = [] + self.command_failures.append({ + "device": name, + "ip": ip, + "command": f"{param} (both steps)", + "error": str(e) + }) + + # Process all other console parameters + # Track rules that need to be enabled + rules_to_enable = {} + + for param, value in console_params.items(): + # Skip Retain parameters as they're handled specially above + if param in retain_params: + continue + + # Check if this is a rule definition (lowercase rule1, rule2, etc.) + if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): + # Store the rule number for later enabling + rule_num = param[-1] + rules_to_enable[rule_num] = True + if with_retry: + self.logger.info(f"{name}: Detected rule definition {param}='{value}', will auto-enable") + else: + self.logger.debug(f"{name}: Detected rule definition {param}, will auto-enable") + + # Skip Rule1, Rule2, etc. if we're auto-enabling rules and using retry logic + if with_retry and param.lower().startswith('rule') and param.lower() != param and param[-1].isdigit(): + # If this is in the config, we'll respect it, but log that it's not needed + self.logger.debug(f"{name}: Note: {param} is not needed with auto-enable feature") + + # Regular console parameter + # Special handling for rule parameters to properly encode the URL + if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): + # For rule commands, we need to URL encode the entire value to preserve special characters + import urllib.parse + encoded_value = urllib.parse.quote(value) + url = f"http://{ip}/cm?cmnd={param}%20{encoded_value}" + self.logger.info(f"{name}: Sending rule command: {url}") + else: + url = f"http://{ip}/cm?cmnd={param}%20{value}" + + if with_retry: + # With retry logic + success = False + attempts = 0 + max_attempts = 3 + last_error = None + + while not success and attempts < max_attempts: + attempts += 1 + try: + response = requests.get(url, timeout=5) + if response.status_code == 200: + # Special logging for rule parameters + if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): + self.logger.info(f"{name}: Rule command response: {response.text}") + self.logger.info(f"{name}: Set rule {param} to '{value}'") + else: + self.logger.debug(f"{name}: Set console parameter {param} to {value}") + console_updated = True + success = True + else: + self.logger.warning(f"{name}: Failed to set console parameter {param} (attempt {attempts}/{max_attempts})") + last_error = f"HTTP {response.status_code}" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.Timeout as e: + self.logger.warning(f"{name}: Timeout setting console parameter {param} (attempt {attempts}/{max_attempts})") + last_error = "Timeout" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.RequestException as e: + self.logger.warning(f"{name}: Error setting console parameter {param}: {str(e)} (attempt {attempts}/{max_attempts})") + last_error = str(e) + if attempts < max_attempts: + time.sleep(1) # Wait before retry + + if not success: + self.logger.error(f"{name}: Failed to set console parameter {param} after {max_attempts} attempts. Last error: {last_error}") + # Track the failure for later reporting + if not hasattr(self, 'command_failures'): + self.command_failures = [] + self.command_failures.append({ + "device": name, + "ip": ip, + "command": f"{param} {value}", + "error": last_error + }) + else: + # Without retry logic + response = requests.get(url, timeout=5) + if response.status_code == 200: + if param.lower().startswith('rule') and param.lower() == param and param[-1].isdigit(): + self.logger.info(f"{name}: Rule command response: {response.text}") + self.logger.info(f"{name}: Set rule {param} to '{value}'") + else: + self.logger.debug(f"{name}: Set console parameter {param} to {value}") + else: + self.logger.error(f"{name}: Failed to set console parameter {param}") + + # Auto-enable any rules that were defined + if with_retry: + self.logger.info(f"{name}: Rules to enable: {rules_to_enable}") + + for rule_num in rules_to_enable: + rule_enable_param = f"Rule{rule_num}" + + # Skip if the rule enable command was already in the config + if with_retry: + # Check if the uppercase version (Rule1) is in the config + if rule_enable_param in console_params: + self.logger.info(f"{name}: Skipping {rule_enable_param} as it's already in config (uppercase version)") + continue + + # If we're here, it means we found a rule definition earlier and added it to rules_to_enable + # No need to check again if it's in console_params + self.logger.info(f"{name}: Will enable {rule_enable_param} for rule definition found in config") + else: + # Simple check for any version of the rule enable command + if any(p.lower() == rule_enable_param.lower() for p in console_params): + continue + + # Rule auto-enabling + url = f"http://{ip}/cm?cmnd={rule_enable_param}%201" + + if with_retry: + # With retry logic + success = False + attempts = 0 + max_attempts = 3 + last_error = None + + while not success and attempts < max_attempts: + attempts += 1 + try: + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.info(f"{name}: Auto-enabled {rule_enable_param}") + console_updated = True + success = True + else: + self.logger.warning(f"{name}: Failed to auto-enable {rule_enable_param} (attempt {attempts}/{max_attempts})") + last_error = f"HTTP {response.status_code}" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.Timeout as e: + self.logger.warning(f"{name}: Timeout auto-enabling {rule_enable_param} (attempt {attempts}/{max_attempts})") + last_error = "Timeout" + if attempts < max_attempts: + time.sleep(1) # Wait before retry + except requests.exceptions.RequestException as e: + self.logger.warning(f"{name}: Error auto-enabling {rule_enable_param}: {str(e)} (attempt {attempts}/{max_attempts})") + last_error = str(e) + if attempts < max_attempts: + time.sleep(1) # Wait before retry + + if not success: + self.logger.error(f"{name}: Failed to auto-enable {rule_enable_param} after {max_attempts} attempts. Last error: {last_error}") + # Track the failure for later reporting + if not hasattr(self, 'command_failures'): + self.command_failures = [] + self.command_failures.append({ + "device": name, + "ip": ip, + "command": f"{rule_enable_param} 1", + "error": last_error + }) + else: + # Without retry logic + response = requests.get(url, timeout=5) + if response.status_code == 200: + self.logger.info(f"{name}: Auto-enabled {rule_enable_param}") + else: + self.logger.error(f"{name}: Failed to auto-enable {rule_enable_param}") + + return console_updated + + def apply_config_other(self, ip, name): + """Wrapper for applying config_other (template) settings.""" + return self.check_and_update_template(ip, name) + def configure_unknown_device(self, ip, hostname): """Configure an unknown device with the given hostname and MQTT settings.""" return self.configure_mqtt_settings( @@ -2085,8 +2100,8 @@ class TasmotaDiscovery: # Check and update MQTT settings if needed mqtt_updated = check_mqtt_settings(ip, name, mqtt_data) - # Check and update template if needed - template_updated = self.check_and_update_template(ip, name) + # Check and update template (config_other) if needed + template_updated = self.apply_config_other(ip, name) # Console settings are now applied in configure_mqtt_settings console_updated = mqtt_updated diff --git a/dead_functions_summary.md b/dead_functions_summary.md new file mode 100644 index 0000000..341e0c7 --- /dev/null +++ b/dead_functions_summary.md @@ -0,0 +1,49 @@ +# Dead Functions Audit for TasmotaManager.py + +Date: 2025-08-08 22:07 +Scope: /home/mgeppert/git_work/scripts/TasmotaManager/TasmotaManager.py + +Summary: No dead (unused) functions were found in TasmotaManager.py. All class methods and the top-level main() function are referenced either by other methods, the CLI entry flow, or the test suite. + +Method usage highlights (non-exhaustive references): + +- UnifiClient + - __init__: Instantiated in main() via TasmotaDiscovery.setup_unifi_client() + - _login: Called by TasmotaDiscovery.setup_unifi_client() (line ~134) + - get_clients: Used in TasmotaDiscovery.get_tasmota_devices() and process_single_device() + +- TasmotaDiscovery + - __init__: Instantiated in main() + - load_config: Used in tests and main() + - setup_unifi_client: Used in main() and process_single_device() + - is_tasmota_device: Used in get_tasmota_devices() + - _match_pattern: Used by is_hostname_unknown, is_device_excluded, and hostname bug handling logic + - get_device_hostname: Used in get_device_details() and unknown-device logic; exercised by tests + - is_hostname_unknown: Used in multiple flows; exercised by tests + - is_device_excluded: Used in get_tasmota_devices(), get_device_details(), process_single_device(); exercised by tests + - get_tasmota_devices: Used in main(); exercised by tests + - save_tasmota_config: Used in main() + - get_unknown_devices: Used by process_unknown_devices() + - process_unknown_devices: Invoked when --process-unknown is provided; referenced in main() and docs + - check_and_update_template: Called via apply_config_other() and directly by tests + - configure_mqtt_settings: Called in get_device_details() (via check_mqtt_settings) and configure_unknown_device() + - apply_console_settings: Called from configure_mqtt_settings() + - apply_config_other: Called from get_device_details() + - configure_unknown_device: Called from unknown device flows and process_single_device() + - is_ip_in_network_filter: Used by process_single_device() + - process_single_device: Used by main() and tests (unknown device flows) + - get_device_details: Used by main() and process_single_device() + +- Module-level + - main(): Called by the if __name__ == '__main__' guard + +Notes: +- Project-wide search across source and tests confirmed usage for each method. Example search hits include: + - is_hostname_unknown: test_pattern_matching.py, test_is_hostname_unknown.py, unifi_hostname_bug_* docs + - get_tasmota_devices: test_get_tasmota_devices.py and main() + - process_unknown_devices: main() and summary docs + - check_and_update_template: multiple tests including test_template_matching.py and test_blank_template_value.py + - get_device_hostname: test_get_device_hostname.py and internal flows + - is_device_excluded: test_is_device_excluded.py and internal flows + +Conclusion: No dead functions identified; no removals performed.