From 68beb7b1c4961530fba9797cdcfbd5abd7a282a6 Mon Sep 17 00:00:00 2001 From: Jared Vititoe Date: Sat, 8 Feb 2025 00:11:28 -0500 Subject: [PATCH] more dynamic --- app.py | 113 ++++++++++------------------------------------------ config.json | 36 +---------------- 2 files changed, 23 insertions(+), 126 deletions(-) diff --git a/app.py b/app.py index 3bebd63..03ccff9 100644 --- a/app.py +++ b/app.py @@ -9,41 +9,16 @@ from flask import Flask, render_template, jsonify import requests from urllib3.exceptions import InsecureRequestWarning -# Configure logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) - -# Disable InsecureRequestWarning requests.packages.urllib3.disable_warnings(InsecureRequestWarning) - -# Initialize Flask app app = Flask(__name__) - -# Global state device_status = {} -# Configuration functions def load_config(): with open('config.json') as f: return json.load(f) -# Network utility functions -def ping(host): - param = '-n' if platform.system().lower() == 'windows' else '-c' - command = ['ping', param, '1', host] - return subprocess.call(command, stdout=subprocess.DEVNULL) == 0 - -def send_webhook(device, status, diagnostics): - config = load_config() - webhook_data = { - "device": device, - "status": status, - "timestamp": datetime.now().isoformat(), - "diagnostics": diagnostics - } - requests.post(config['webhook_url'], json=webhook_data) - -# UniFi API integration class UnifiAPI: def __init__(self, config): self.base_url = config['unifi']['controller'] @@ -74,10 +49,15 @@ class UnifiAPI: return [] def get_device_details(self, device_id): - url = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device/{device_id}" - response = self.session.get(url, headers=self.headers) - response.raise_for_status() - return response.json() + try: + url = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device/{device_id}" + response = self.session.get(url, headers=self.headers) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error(f"Failed to get device details: {e}") + return None + def get_device_diagnostics(self, device): details = self.get_device_details(device['device_id']) if not details: @@ -99,63 +79,22 @@ class UnifiAPI: 'ports': {}, 'radios': {} } - - for port in interfaces.get('ports', []): - result['ports'][f"port_{port['idx']}"] = { - 'state': port['state'], - 'type': port['connector'], + for port in interfaces: + result['ports'][f"port_{port['index']}"] = { + 'state': port['up'] and 'up' or 'down', 'speed': { - 'current': port['speedMbps'], - 'max': port['maxSpeedMbps'] + 'current': port.get('speed', 0), + 'max': port.get('max_speed', 0) } } - - for radio in interfaces.get('radios', []): - result['radios'][f"{radio['frequencyGHz']}GHz"] = { - 'standard': radio['wlanStandard'], - 'channel': radio['channel'], - 'width': f"{radio['channelWidthMHz']}MHz" - } - return result -def run_diagnostics(device): - try: - config = load_config() - unifi = UnifiAPI(config) - diagnostics = unifi.get_device_diagnostics(device) - @app.route('/') def home(): config = load_config() unifi = UnifiAPI(config) devices = unifi.get_devices() return render_template('index.html', devices=devices) - logger.debug(f"Got diagnostics for {device['name']}: {diagnostics}") - return diagnostics - except Exception as e: - return {"error": str(e)} - -def update_status(): - while True: - config = load_config() - for device in config['devices']: - current_status = ping(device['ip']) - previous_status = device_status.get(device['name'], True) - - if current_status != previous_status: - diagnostics = run_diagnostics(device) - send_webhook(device, current_status, diagnostics) - - device_status[device['name']] = current_status - time.sleep(config['check_interval']) - -# Flask routes -@app.route('/') -def home(): - config = load_config() - devices = config['devices'] - return render_template('index.html', devices=devices) @app.route('/api/status') def status(): @@ -163,24 +102,14 @@ def status(): @app.route('/api/diagnostics') def get_diagnostics(): - try: - config = load_config() - unifi = UnifiAPI(config) - devices = unifi.get_all_devices() - - diagnostics = {} - for device in config['devices']: - if device.get('device_id'): - device_details = unifi.get_device_details(device['device_id']) - diagnostics[device['name']] = device_details - - return jsonify(diagnostics) - except Exception as e: - logger.error(f"Error in diagnostics endpoint: {str(e)}") - logger.exception("Full traceback:") - return jsonify({"error": str(e)}), 500 + config = load_config() + unifi = UnifiAPI(config) + devices = unifi.get_devices() + diagnostics = {} + for device in devices: + diagnostics[device['name']] = unifi.get_device_diagnostics(device) + return jsonify(diagnostics) -# Application entry point if __name__ == '__main__': status_thread = threading.Thread(target=update_status, daemon=True) status_thread.start() diff --git a/config.json b/config.json index 1b06c3b..a2f1313 100644 --- a/config.json +++ b/config.json @@ -4,37 +4,5 @@ "api_key": "kyPfIsAVie3hwMD4Bc1MjAu8N7HVPIb8", "site_id": "default" }, - "devices": [ - { - "name": "UniFi Dream Machine Pro", - "ip": "10.10.10.1", - "type": "router", - "connection_type": "copper", - "critical": true - }, - { - "name": "USW Pro 24 PoE", - "ip": "10.10.10.139", - "type": "switch", - "connection_type": "fiber", - "critical": true - } - ], - "check_interval": 30, - "webhook_url": "https://your-webhook-url", - "alert_thresholds": { - "fiber": { - "optical_power_min": -10, - "optical_power_max": 0, - "error_rate_threshold": 0.001 - }, - "copper": { - "signal_quality_min": 70, - "max_cable_length": 100 - } - }, - "troubleshooting": { - "fiber_tests": ["optical_power", "light_levels", "error_rate", "sfp_diagnostics"], - "copper_tests": ["cable_length", "crosstalk", "signal_quality", "poe_status"] - } -} + "check_interval": 30 +} \ No newline at end of file