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

Added ignore rules

Browse Source
This commit is contained in:
Mike Geppert 2025-10-26 12:45:08 -05:00
parent 3e66920c2a
commit 8603cd3e7e
3 changed files with 182 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -23,7 +23,7 @@ ENV/
*.log
# Local configuration that might contain sensitive information
network_configuration.json
#network_configuration.json
# Backup files
*.backup
@ -32,4 +32,4 @@ network_configuration.json
current.json
deprecated.json
TasmotaDevices.json
*.json.backup
*.json.backup

221
TasmotaManager.py
View File

@ -13,17 +13,78 @@ import argparse
# Disable SSL warnings
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
class AuthenticationError(Exception):
"""Raised when UniFi returns an authentication/authorization error (401/403)."""
pass
class UniFiDataError(Exception):
"""Raised when UniFi returns non-JSON or unexpected data across all fallbacks."""
pass
class UnifiClient:
def __init__(self, base_url, username, password, site_id, verify_ssl=True):
def __init__(self, base_url, username=None, password=None, site_id='default', verify_ssl=True, token=None):
self.base_url = base_url.rstrip('/')
self.username = username
self.password = password
self.site_id = site_id
self.token = token
self.session = requests.Session()
self.session.verify = verify_ssl
# Initialize cookie jar
self.session.cookies.clear()
# Set common headers for UniFi requests
self.session.headers.update({
'Accept': 'application/json',
'User-Agent': 'TasmotaManager/1.0',
'Referer': f'{self.base_url}/',
'Origin': self.base_url
})
# If using a Personal Access Token, set Authorization header
if self.token:
self.session.headers.update({
'Authorization': f'Bearer {self.token}'
})
def _request_json(self, url: str, method: str = 'GET', timeout: tuple = (5, 10)):
"""Perform an HTTP request and return parsed JSON or None.
Raises AuthenticationError on 401/403. Never logs sensitive tokens.
"""
logger = logging.getLogger(__name__)
try:
resp = self.session.request(method, url, timeout=timeout)
except requests.exceptions.RequestException as e:
# Network-level error
return None
status = getattr(resp, 'status_code', None)
# Auth errors
if status in (401, 403):
raise AuthenticationError(f"Auth error {status} for {url}")
# Other HTTP errors
if status and status >= 400:
return None
ctype = (resp.headers.get('Content-Type') or '').lower()
text = None
# If not JSON content-type, try to detect empty body, otherwise return None
if 'json' not in ctype:
try:
text = resp.text
except Exception:
text = ''
if not text:
return None
# Non-JSON body present (likely HTML); return None
logger.debug(f"Non-JSON response from {url} (status {status}, content-type {ctype}).")
return None
# Parse JSON safely
try:
return resp.json()
except ValueError:
# Empty or invalid JSON
logger.debug(f"Failed to parse JSON from {url} (status {status}, content-type {ctype}).")
return None
def _login(self) -> requests.Response: # Changed return type annotation
"""Authenticate with the UniFi Controller."""
@ -48,49 +109,61 @@ class UnifiClient:
self.session.headers['X-CSRF-Token'] = response.headers['X-CSRF-Token']
return response # Return the response object
except requests.exceptions.RequestException as e:
if hasattr(e, 'response') and e.response.status_code == 401:
if hasattr(e, 'response') and e.response is not None and e.response.status_code == 401:
raise Exception("Authentication failed. Please verify your username and password.") from e
raise
def get_clients(self) -> list:
"""Get all clients from the UniFi Controller."""
# Try the newer API endpoint first
url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/sta"
logger = logging.getLogger(__name__)
# 1) Preferred classic endpoint
url1 = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/sta"
data = self._request_json(url1)
if isinstance(data, dict):
return data.get('data', [])
if isinstance(data, list):
return data
# 2) v2 endpoint (often returns a list directly)
url2 = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/clients"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json().get('data', [])
except requests.exceptions.RequestException as e:
# If the newer endpoint fails, try the legacy endpoint
url = f"{self.base_url}/api/s/{self.site_id}/stat/sta"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json().get('data', [])
except requests.exceptions.RequestException as e:
# If both fail, try the v2 API endpoint
url = f"{self.base_url}/v2/api/site/{self.site_id}/clients"
response = self.session.get(url)
response.raise_for_status()
return response.json().get('data', [])
data = self._request_json(url2)
except AuthenticationError:
# Bubble up auth errors
raise
if isinstance(data, list):
return data
if isinstance(data, dict):
return data.get('data', [])
# 3) Legacy classic endpoint (no /proxy)
url3 = f"{self.base_url}/api/s/{self.site_id}/stat/sta"
data = self._request_json(url3)
if isinstance(data, dict):
return data.get('data', [])
if isinstance(data, list):
return data
raise UniFiDataError("UniFi returned no parsable JSON for clients across all endpoints")
def get_devices(self) -> list:
"""Get UniFi network devices (e.g., Access Points) from the Controller."""
# Try the newer API endpoint first
url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device"
logger = logging.getLogger(__name__)
# 1) Preferred classic endpoint
url1 = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json().get('data', [])
except requests.exceptions.RequestException:
# Try legacy endpoint
url = f"{self.base_url}/api/s/{self.site_id}/stat/device"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json().get('data', [])
except requests.exceptions.RequestException:
return []
data = self._request_json(url1)
except AuthenticationError:
raise
if isinstance(data, dict):
return data.get('data', [])
if isinstance(data, list):
return data
# 2) Legacy classic endpoint
url2 = f"{self.base_url}/api/s/{self.site_id}/stat/device"
data = self._request_json(url2)
if isinstance(data, dict):
return data.get('data', [])
if isinstance(data, list):
return data
raise UniFiDataError("UniFi returned no parsable JSON for devices across all endpoints")
class TasmotaDiscovery:
def __init__(self, debug: bool = False):
@ -130,33 +203,58 @@ class TasmotaDiscovery:
sys.exit(1)
def setup_unifi_client(self):
"""Set up the UniFi client with better error handling"""
"""Set up the UniFi client with better error handling and token support"""
self.logger.debug("Setting up UniFi client")
if not self.config or 'unifi' not in self.config:
raise ValueError("Missing UniFi configuration")
unifi_config = self.config['unifi']
required_fields = ['host', 'username', 'password', 'site']
missing_fields = [field for field in required_fields if field not in unifi_config]
host = unifi_config.get('host')
site = unifi_config.get('site', 'default')
token = unifi_config.get('token')
username = unifi_config.get('username')
password = unifi_config.get('password')
if missing_fields:
raise ValueError(f"Missing required UniFi configuration fields: {', '.join(missing_fields)}")
missing_core = []
if not host:
missing_core.append('host')
if not site:
missing_core.append('site')
if missing_core:
raise ValueError(f"Missing required UniFi configuration fields: {', '.join(missing_core)}")
if not token and (not username or not password):
raise ValueError("Provide either 'unifi.token' OR both 'unifi.username' and 'unifi.password'")
try:
self.logger.debug(f"Connecting to UniFi Controller at {unifi_config['host']}")
self.unifi_client = UnifiClient(
base_url=unifi_config['host'],
username=unifi_config['username'],
password=unifi_config['password'],
site_id=unifi_config['site'],
verify_ssl=False # Add this if using self-signed certificates
)
# Test the connection by making a simple request
response = self.unifi_client._login()
if not response:
raise ConnectionError(f"Failed to connect to UniFi controller: No response")
self.logger.debug(f"Connecting to UniFi Controller at {host} using {'token' if token else 'username/password'}")
# Instantiate client
if token:
self.unifi_client = UnifiClient(
base_url=host,
site_id=site,
verify_ssl=False,
token=token
)
# Test by calling a lightweight endpoint
try:
# Attempt to fetch devices (requires auth). Empty list is acceptable.
_ = self.unifi_client.get_devices()
except Exception as test_err:
raise ConnectionError(f"Token authentication failed: {test_err}")
else:
self.unifi_client = UnifiClient(
base_url=host,
username=username,
password=password,
site_id=site,
verify_ssl=False
)
# Test the connection by logging in
response = self.unifi_client._login()
if not response:
raise ConnectionError("Failed to connect to UniFi controller: No response")
self.logger.debug("UniFi client setup successful")
@ -623,7 +721,7 @@ class TasmotaDiscovery:
return devices
except Exception as e:
self.logger.error(f"Error getting devices from UniFi controller: {e}")
return []
raise
def save_tasmota_config(self, devices: list) -> None:
"""Save Tasmota device information to a JSON file with device tracking."""
@ -2830,21 +2928,18 @@ def main():
print("- Device list saved to: current.json")
print("- Detailed information saved to: TasmotaDevices.json")
except ConnectionError as e:
print(f"Connection Error: {str(e)}")
print("\nTrying to proceed with existing current.json...")
try:
discovery.get_device_details(use_current_json=True)
print("\nSuccessfully retrieved device details from existing current.json")
except Exception as inner_e:
print(f"Error processing existing devices: {str(inner_e)}")
return 1
except Exception as e:
except (AuthenticationError, UniFiDataError, ConnectionError) as e:
print(f"Error: {str(e)}")
if args.debug:
import traceback
traceback.print_exc()
return 1
except Exception as e:
print(f"Unexpected error: {str(e)}")
if args.debug:
import traceback
traceback.print_exc()
return 1
return 0

39
network_configuration.json
View File

@ -3,6 +3,7 @@
"host": "https://192.168.6.1",
"username": "Tasmota",
"password": "TasmotaManager12!@",
"site": "default",
"network_filter": {
"NoT_network": {
@ -33,48 +34,48 @@
"device_list": {
"TreatLife_SW_SS01S": {
"template": "{\"NAME\":\"TL SS01S Swtch\",\"GPIO\":[0,0,0,0,52,158,0,0,21,17,0,0,0],\"FLAG\":0,\"BASE\":18}",
"console_set": "Default"
"console_set": "Traditional"
},
"TreatLife_SW_SS02S": {
"template": "{\"NAME\":\"Treatlife SS02\",\"GPIO\":[0,0,0,0,288,576,0,0,224,32,0,0,0,0],\"FLAG\":0,\"BASE\":18}",
"console_set": "Default"
"console_set": "Traditional"
},
"TreatLife_SW_SS02S_Orig": {
"template": "{\"NAME\":\"Treatlife SS02\",\"GPIO\":[0,0,0,0,289,0,0,0,224,32,0,0,0,0],\"FLAG\":0,\"BASE\":18}",
"console_set": "Default"
"console_set": "Traditional"
},
"TreatLife_DIM_DS02S": {
"template": "{\"NAME\":\"DS02S Dimmer\",\"GPIO\":[0,107,0,108,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":54}",
"console_set": "Default"
"console_set": "Traditional"
},
"CloudFree_SW1": {
"template": "{\"NAME\":\"CloudFree SW1\",\"GPIO\":[0,224,0,32,320,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":1}",
"console_set": "Default"
"console_set": "Traditional"
},
"Gosund_WP5_Plug": {
"template": "{\"NAME\":\"Gosund-WP5\",\"GPIO\":[0,0,0,0,17,0,0,0,56,57,21,0,0],\"FLAG\":0,\"BASE\":18}",
"console_set": "Default"
"console_set": "Traditional"
},
"Gosund_Plug": {
"template": "{\"NAME\":\"Gosund-WP5\",\"GPIO\":[0,0,0,0,32,0,0,0,320,321,224,0,0,0],\"FLAG\":0,\"BASE\":18}",
"console_set": "Default"
"console_set": "Traditional"
},
"CloudFree_X10S_Plug": {
"template": "{\"NAME\":\"Aoycocr X10S\",\"GPIO\":[56,0,57,0,21,134,0,0,131,17,132,0,0],\"FLAG\":0,\"BASE\":45}",
"console_set": "alt"
"console_set": "Traditional"
},
"Sonoff_S31_PM_Plug": {
"template": "{\"NAME\":\"Sonoff S31\",\"GPIO\":[17,145,0,146,0,0,0,0,21,56,0,0,0],\"FLAG\":0,\"BASE\":41}",
"console_set": "Default"
"console_set": "Traditional"
},
"Sonoff S31": {
"template": "",
"console_set": "Default"
"Sonoff TX Ultimate 1": {
"template": "{\"NAME\":\"Sonoff T5-1C-120\",\"GPIO\":[0,0,7808,0,7840,3872,0,0,0,1376,0,7776,0,0,224,3232,0,480,3200,0,0,0,3840,0,0,0,0,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":1}",
"console_set": "SONOFF_ULTIMATE"
}
},
"console_set": {
"Default": [
"Traditional": [
"",
"SwitchRetain Off",
"ButtonRetain Off",
"PowerRetain On",
@ -85,24 +86,28 @@
"SetOption13 0",
"SetOption19 0",
"SetOption32 8",
"SetOption40 40",
"SetOption53 1",
"SetOption73 1",
"rule1 on button1#state=10 do power0 toggle endon"
],
"alt": [
"SONOFF_ULTIMATE": [
"SwitchRetain Off",
"ButtonRetain Off",
"PowerRetain On",
"PowerOnState 3",
"Pixels 32",
"SetOption1 0",
"SetOption3 1",
"SetOption4 1",
"SetOption13 0",
"SetOption19 0",
"SetOption32 8",
"SetOption40 40",
"SetOption53 1",
"SetOption73 1",
"rule1 on button1#state=10 do power0 toggle endon"
"SetOption73 1"
]
}
}