gandalf/devices.py

"""Device inventory built from the monitor's cached state.

Pure functions with no I/O: app.py passes in the JSON blobs the monitor stored,
so search and the per-device pages always agree about what a device is.
"""
import re
from typing import Dict, List, Optional

# UniFi reports PDUs as switches or APs, so model codes take priority over type.
_PDU_MODELS = {'USPPDUP', 'UP6', 'USPDA2B'}
_KIND_BY_TYPE = {
    'usw': 'switch', 'udm': 'router', 'ugw': 'router', 'uxg': 'router',
    'uap': 'ap', 'usp': 'pdu',
}


def unifi_kind(unifi_type: str, model: str) -> str:
    if model in _PDU_MODELS:
        return 'pdu'
    return _KIND_BY_TYPE.get((unifi_type or '').lower(), 'device')


def _mac_forms(mac: str) -> List[str]:
    """Return a MAC with and without separators, so either spelling is searchable."""
    return [mac, re.sub(r'[^0-9a-fA-F]', '', mac)] if mac else []


def _search_text(*parts) -> str:
    flat = []
    for p in parts:
        if isinstance(p, (list, tuple, set)):
            flat.extend(str(x) for x in p if x)
        elif p:
            flat.append(str(p))
    return ' '.join(flat).lower()


def _port_neighbors(ports: dict) -> List[str]:
    return [p['lldp']['system_name'] for p in ports.values()
            if (p.get('lldp') or {}).get('system_name')]


def build_index(snapshot: dict, link_stats: dict, modem: Optional[dict]) -> List[dict]:
    """Return every known device as {name, kind, status, ip, model, ..., search}."""
    snapshot = snapshot or {}
    link_stats = link_stats or {}
    devices: Dict[str, dict] = {}

    for name, host in (snapshot.get('hosts') or {}).items():
        link_ifaces = (link_stats.get('hosts') or {}).get(name) or {}
        ifaces = sorted(set(host.get('interfaces') or {}) | set(link_ifaces))
        devices[name] = {
            'name': name, 'kind': 'host', 'status': host.get('status', 'unknown'),
            'ip': host.get('ip', ''), 'mac': '', 'model': '', 'source': host.get('source', ''),
            'interface_count': len(ifaces),
            'search': _search_text(name, host.get('ip'), 'host server', ifaces),
        }

    # Prometheus also sees hosts that aren't in config.json, e.g. LXC app servers.
    # They have link stats but no alerting, so their state is reported as unknown.
    for name, ifaces in (link_stats.get('hosts') or {}).items():
        if name not in devices:
            ip = next((i.get('host_ip') for i in ifaces.values() if i.get('host_ip')), '')
            devices[name] = {
                'name': name, 'kind': 'host', 'status': 'unknown', 'ip': ip, 'mac': '', 'model': '',
                'source': 'prometheus', 'interface_count': len(ifaces),
                'search': _search_text(name, ip, 'host server', list(ifaces)),
            }

    switches = link_stats.get('unifi_switches') or {}
    for d in snapshot.get('unifi') or []:
        name = d.get('name', '')
        ports = (switches.get(name) or {}).get('ports') or {}
        kind = unifi_kind(d.get('type', ''), d.get('model', ''))
        devices[name] = {
            'name': name, 'kind': kind, 'status': 'up' if d.get('connected') else 'down',
            'ip': d.get('ip', ''), 'mac': d.get('mac', ''), 'model': d.get('model', ''),
            'source': 'unifi', 'interface_count': len(ports),
            'search': _search_text(
                name, d.get('ip'), _mac_forms(d.get('mac', '')), d.get('model'),
                d.get('type'), kind, list(ports), _port_neighbors(ports),
            ),
        }

    # A switch can have link stats while the device list is unavailable this cycle.
    for name, sw in switches.items():
        if name not in devices:
            ports = sw.get('ports') or {}
            devices[name] = {
                'name': name, 'kind': unifi_kind('usw', sw.get('model', '')), 'status': 'unknown',
                'ip': sw.get('ip', ''), 'mac': '', 'model': sw.get('model', ''),
                'source': 'unifi', 'interface_count': len(ports),
                'search': _search_text(name, sw.get('ip'), sw.get('model'), list(ports), _port_neighbors(ports)),
            }

    if modem and modem.get('name'):
        name = modem['name']
        if not modem.get('reachable'):
            status = 'unknown'
        else:
            status = 'down' if modem.get('wan_problems') else 'up'
        devices[name] = {
            'name': name, 'kind': 'modem', 'status': status,
            'ip': modem.get('lan_ip', ''), 'mac': '', 'model': modem.get('model', ''),
            'source': 'modem', 'interface_count': 0,
            'search': _search_text(
                name, modem.get('lan_ip'), modem.get('wan_ip'), modem.get('model'),
                'modem gateway fiber wan at&t att',
            ),
        }

    return sorted(devices.values(), key=lambda d: d['name'].lower())


def uplinks_for_host(name: str, switches: dict) -> List[dict]:
    """Find the switch ports whose LLDP neighbor is this host."""
    found = []
    for sw_name, sw in (switches or {}).items():
        for port_name, port in (sw.get('ports') or {}).items():
            lldp = port.get('lldp') or {}
            if lldp.get('system_name') == name:
                found.append({
                    'switch': sw_name, 'port': port_name, 'port_idx': port.get('port_idx'),
                    'host_interface': lldp.get('port_id', ''), 'up': port.get('up', False),
                    'speed_mbps': port.get('speed_mbps', 0),
                })
    return found


def device_detail(name: str, snapshot: dict, link_stats: dict, modem: Optional[dict]) -> Optional[dict]:
    """Return everything known about one device, or None if it doesn't exist."""
    match = next((d for d in build_index(snapshot, link_stats, modem) if d['name'] == name), None)
    if match is None:
        return None
    link_stats = link_stats or {}
    detail = {k: v for k, v in match.items() if k != 'search'}
    switches = link_stats.get('unifi_switches') or {}

    if match['kind'] == 'host':
        states = ((snapshot or {}).get('hosts') or {}).get(name, {}).get('interfaces') or {}
        link_ifaces = (link_stats.get('hosts') or {}).get(name) or {}
        interfaces = {i: {'state': s} for i, s in states.items()}
        for iface, stats in link_ifaces.items():
            interfaces.setdefault(iface, {}).update(stats)
        detail['interfaces'] = interfaces
        detail['uplinks'] = uplinks_for_host(name, switches)
    elif match['kind'] == 'modem':
        pass  # the page renders the shared WAN / Fiber panel from /api/modem
    else:
        detail['ports'] = (switches.get(name) or {}).get('ports') or {}
        detail['info'] = (link_stats.get('unifi_devices') or {}).get(name) or {}
    detail['updated'] = link_stats.get('updated')
    return detail