"""Common utility functions used across the TasmotaManager modules.""" import re import logging import time import os import json from typing import Tuple, Optional, Any, Dict import requests def match_pattern(text: str, pattern: str, match_entire_string: bool = True) -> bool: """ Match a text string against a pattern that may contain wildcards. Args: text: The text to match against pattern: The pattern which may contain * wildcards match_entire_string: If True, pattern must match the entire string Returns: bool: True if the pattern matches, False otherwise """ if not text: return False # Convert glob pattern to regex escaped = re.escape(pattern) regex_pattern = escaped.replace(r'\*', '.*') if match_entire_string: regex_pattern = f'^{regex_pattern}$' return bool(re.match(regex_pattern, text, re.IGNORECASE)) def get_hostname_base(hostname: str) -> str: """ Extract the base hostname (everything before the first dash). Args: hostname: Full hostname (e.g., "KitchenLamp-1234") Returns: str: Base hostname (e.g., "KitchenLamp") """ if '-' in hostname: return hostname.split('-')[0] return hostname def send_tasmota_command(ip: str, command: str, timeout: int = 5, logger: Optional[logging.Logger] = None) -> Tuple[Optional[dict], bool]: """ Send a command to a Tasmota device via HTTP API. Args: ip: Device IP address command: Tasmota command to send timeout: Request timeout in seconds logger: Optional logger for debug output Returns: Tuple of (response_dict, success_bool) """ url = f"http://{ip}/cm?cmnd={command}" try: if logger: logger.debug(f"Sending command to {ip}: {command}") response = requests.get(url, timeout=timeout) response.raise_for_status() result = response.json() if logger: logger.debug(f"Response from {ip}: {result}") return result, True except requests.exceptions.Timeout: if logger: logger.warning(f"Timeout sending command to {ip}: {command}") return None, False except requests.exceptions.RequestException as e: if logger: logger.warning(f"Error sending command to {ip}: {e}") return None, False except Exception as e: if logger: logger.error(f"Unexpected error sending command to {ip}: {e}") return None, False def retry_command(func, max_attempts: int = 3, delay: float = 1.0, logger: Optional[logging.Logger] = None, device_name: str = "") -> Tuple[Any, bool]: """ Retry a command function multiple times with delay between attempts. Args: func: Function to call (should return tuple of (result, success)) max_attempts: Maximum number of attempts delay: Delay in seconds between attempts logger: Optional logger for output device_name: Device name for logging Returns: Tuple of (result, success) """ for attempt in range(1, max_attempts + 1): result, success = func() if success: return result, True if attempt < max_attempts: if logger: logger.debug(f"{device_name}: Retry attempt {attempt}/{max_attempts}") time.sleep(delay) return None, False def format_device_info(device: dict) -> str: """ Format device information for display. Args: device: Device dictionary Returns: str: Formatted device info string """ name = device.get('name', 'Unknown') ip = device.get('ip', 'Unknown') mac = device.get('mac', 'Unknown') connection = device.get('connection', 'Unknown') return f"{name} ({ip}) - MAC: {mac}, Connection: {connection}" def load_json_file(filepath: str, logger: Optional[logging.Logger] = None) -> Optional[dict]: """ Load and parse a JSON file. Args: filepath: Path to JSON file logger: Optional logger for error output Returns: dict or None if file doesn't exist or can't be parsed """ if not os.path.exists(filepath): if logger: logger.debug(f"File not found: {filepath}") return None try: with open(filepath, 'r') as f: return json.load(f) except json.JSONDecodeError as e: if logger: logger.error(f"Error parsing JSON file {filepath}: {e}") return None except Exception as e: if logger: logger.error(f"Error reading file {filepath}: {e}") return None def save_json_file(filepath: str, data: dict, logger: Optional[logging.Logger] = None) -> bool: """ Save data to a JSON file. Args: filepath: Path to save JSON file data: Data to save logger: Optional logger for error output Returns: bool: True if successful, False otherwise """ try: # Ensure directory exists os.makedirs(os.path.dirname(filepath) if os.path.dirname(filepath) else '.', exist_ok=True) with open(filepath, 'w') as f: json.dump(data, f, indent=4) return True except Exception as e: if logger: logger.error(f"Error saving JSON file {filepath}: {e}") return False def is_valid_ip(ip_string: str) -> bool: """ Validate if a string is a valid IP address. Args: ip_string: String to validate Returns: bool: True if valid IP address """ import socket try: socket.inet_aton(ip_string) return True except socket.error: return False def ensure_data_directory(): """Ensure the data directory exists.""" os.makedirs('data', exist_ok=True) os.makedirs('data/temp', exist_ok=True) def get_data_file_path(filename: str) -> str: """ Get the full path for a data file. Args: filename: Name of the file Returns: str: Full path in the data directory """ return os.path.join('data', filename)