gandalf/app.py

import logging
import json
import platform
import subprocess
import threading
import time
from datetime import datetime
from flask import Flask, render_template, jsonify
import requests
from urllib3.exceptions import InsecureRequestWarning

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
app = Flask(__name__)
device_status = {}

def load_config():
    with open('config.json') as f:
        return json.load(f)

class UnifiAPI:
    def __init__(self, config):
        self.base_url = config['unifi']['controller']
        self.session = requests.Session()
        self.session.verify = False
        self.headers = {
            'X-API-KEY': config['unifi']['api_key'],
            'Accept': 'application/json'
        }
        self.site_id = "default"
        
    def get_devices(self):
        try:
            url = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device"
            response = self.session.get(url, headers=self.headers)
            response.raise_for_status()
            
            # Log raw response
            logger.debug(f"Response status: {response.status_code}")
            logger.debug(f"Response headers: {response.headers}")
            logger.debug(f"Raw response text: {response.text}")
            
            devices_data = response.json()
            logger.debug(f"Parsed JSON: {devices_data}")
            
            # Extract network_devices from the response
            network_devices = devices_data.get('network_devices', [])
            
            devices = []
            for device in network_devices:
                devices.append({
                    'name': device.get('name', 'Unknown'),
                    'ip': device.get('ip', '0.0.0.0'),
                    'type': device.get('type', 'unknown'),
                    'connection_type': 'fiber' if device.get('uplink', {}).get('media') == 'sfp' else 'copper',
                    'critical': True if device.get('type') in ['udm', 'usw'] else False,
                    'device_id': device.get('mac')
                })
                
            logger.debug(f"Processed devices: {devices}")
            return devices
            
        except Exception as e:
            logger.error(f"Error fetching devices: {e}")
            logger.exception("Full traceback:")
            return []    
            
    def get_device_details(self, device_id):
        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:
            return {'state': 'ERROR', 'error': 'Failed to fetch device details'}
            
        diagnostics = {
            'state': details.get('state', 'unknown'),
            'interfaces': {
                'ports': {}
            }
        }
        
        # Parse port information
        for port in details.get('port_table', []):
            diagnostics['interfaces']['ports'][f"Port {port.get('port_idx')}"] = {
                'state': 'up' if port.get('up') else 'down',
                'speed': {
                    'current': port.get('speed', 0),
                    'max': port.get('max_speed', 0)
                },
                'poe': port.get('poe_enable', False),
                'media': port.get('media', 'unknown')
            }
        
        return diagnostics

    def _parse_interfaces(self, interfaces):
        result = {
            'ports': {},
            'radios': {}
        }
        for port in interfaces:
            result['ports'][f"port_{port['index']}"] = {
                'state': port['up'] and 'up' or 'down',
                'speed': {
                    'current': port.get('speed', 0),
                    'max': port.get('max_speed', 0)
                }
            }
        return result

@app.route('/')
def home():
    config = load_config()
    unifi = UnifiAPI(config)
    devices = unifi.get_devices()
    return render_template('index.html', devices=devices)

@app.route('/api/status')
def status():
    return jsonify(device_status)

@app.route('/api/diagnostics')
def get_diagnostics():
    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)

if __name__ == '__main__':
    status_thread = threading.Thread(target=update_status, daemon=True)
    status_thread.start()
    app.run(debug=True)