mgeppert/TasmotaManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: extract console settings into apply_console_settings(); add apply_config_other wrapper; update get_device_details to use wrapper. Docs: fix Configuration example to separate console and config_other; clarify console section usage. Add dead_functions_summary.md audit confirming no dead functions.

Browse Source
This commit is contained in:
Mike Geppert 2025-08-08 22:32:09 -05:00
parent 126cd39555
commit 5784a0b414
3 changed files with 394 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
README.md
View File

@ -63,26 +63,34 @@ Create a `network_configuration.json` file with the following structure:
"Password": "mqtt-password", "Password": "mqtt-password",
"Topic": "%hostname_base%", "Topic": "%hostname_base%",
"FullTopic": "%prefix%/%topic%/", "FullTopic": "%prefix%/%topic%/",
"NoRetain": false, "NoRetain": false
"console": { },
"SwitchRetain": "Off", "config_other": {
"ButtonRetain": "Off", "Example_Device_Template": "{\"NAME\":\"Example\",\"GPIO\":[0],\"FLAG\":0,\"BASE\":18}"
"PowerOnState": "3", },
"PowerRetain": "On", "console": {
"SetOption1": "0", "SwitchRetain": "Off",
"SetOption3": "1", "ButtonRetain": "Off",
"SetOption4": "1", "PowerOnState": "3",
"SetOption13": "0", "PowerRetain": "On",
"SetOption19": "0", "SetOption1": "0",
"SetOption32": "8", "SetOption3": "1",
"SetOption53": "1", "SetOption4": "1",
"SetOption73": "1", "SetOption13": "0",
"rule1": "on button1#state=10 do power0 toggle endon" "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 ## Usage
Basic usage: Basic usage:
@ -180,7 +188,7 @@ This feature helps automate the setup of new Tasmota devices that haven't been p
## Console Parameters ## 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.) - Configure device behavior (PowerOnState, SetOptions, etc.)
- Set up rules for button actions - Set up rules for button actions

625
TasmotaManager.py Normal file → Executable file
View File

@ -1325,309 +1325,7 @@ class TasmotaDiscovery:
self.logger.error(f"{name}: Error updating {setting}: {str(e)}") self.logger.error(f"{name}: Error updating {setting}: {str(e)}")
# Apply console settings # Apply console settings
console_updated = False console_updated = self.apply_console_settings(ip, name, with_retry)
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}")
# Reboot the device if requested # Reboot the device if requested
if reboot: if reboot:
@ -1644,6 +1342,323 @@ class TasmotaDiscovery:
self.logger.error(f"Error configuring device at {ip}: {str(e)}") self.logger.error(f"Error configuring device at {ip}: {str(e)}")
return False 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): def configure_unknown_device(self, ip, hostname):
"""Configure an unknown device with the given hostname and MQTT settings.""" """Configure an unknown device with the given hostname and MQTT settings."""
return self.configure_mqtt_settings( return self.configure_mqtt_settings(
@ -2085,8 +2100,8 @@ class TasmotaDiscovery:
# Check and update MQTT settings if needed # Check and update MQTT settings if needed
mqtt_updated = check_mqtt_settings(ip, name, mqtt_data) mqtt_updated = check_mqtt_settings(ip, name, mqtt_data)
# Check and update template if needed # Check and update template (config_other) if needed
template_updated = self.check_and_update_template(ip, name) template_updated = self.apply_config_other(ip, name)
# Console settings are now applied in configure_mqtt_settings # Console settings are now applied in configure_mqtt_settings
console_updated = mqtt_updated console_updated = mqtt_updated

49
dead_functions_summary.md Normal file
View File

@ -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.