jaredandClaude Opus 5.5
415d2222e7
Lint / JS (eslint) (push) Successful in 21s
Lint / Python (flake8) (push) Successful in 2m36s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Test / Python Tests (pytest) (push) Successful in 2m46s
Lint / JS (eslint) (pull_request) Successful in 15s
Lint / Python (flake8) (pull_request) Successful in 2m14s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (push) Successful in 1m33s
Security / Python Security (bandit) (pull_request) Successful in 54s
Test / Python Tests (pytest) (pull_request) Successful in 1m3s
- Each PDU, UPS and power strip card lists its outlets that feed something, draw power or are off, with live watts. Each device fed gets a power badge with what it draws. Hovering a power card or a device draws its feeds as dotted lines and keeps both ends lit. - What an outlet feeds: a saved assignment, else the outlet's UniFi name when it names a device (or the modem), else a MAC in its name that belongs to a host NIC (the monitor's host_macs) or UniFi device. - Admins click an outlet to set what it feeds, nothing, or automatic. Saved in a new power_feeds table (created with the cutsheet tables); PUT /api/topology/power/<device>/<outlet>, admin and JSON only, with the device, outlet and target checked. UniFi is never changed. - An outlet that is off but feeds something is listed under "to look at". - Power gear is shown by default again, now that it has something to say. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
482 lines
23 KiB
Python
482 lines
23 KiB
Python
"""The live network topology: every device and every physical cable between them.
|
|
|
|
Pure functions with no I/O. app.py passes in the monitor's cached state and the
|
|
cutsheet, and the /topology page draws what build_graph returns.
|
|
|
|
The graph comes from the switches' point of view: each UniFi port with a neighbour
|
|
is one end of a cable. The same switch-to-switch cable is seen from both ends, so
|
|
those are merged into one link.
|
|
"""
|
|
import re
|
|
from collections import deque
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
import devices
|
|
|
|
INTERNET = 'internet'
|
|
|
|
# How UniFi devices are drawn and grouped. Power gear goes in its own rail.
|
|
_POWER_MODELS = {'USPPDUP': 'pdu', 'USPDA2B': 'ups', 'UP6': 'strip'}
|
|
POWER_KINDS = ('pdu', 'ups', 'strip')
|
|
|
|
_SEVERITY_RANK = {'critical': 3, 'warning': 2, 'info': 1}
|
|
_PORT_REF_RE = re.compile(r'(?:^|\s)port\s*(\d+)$', re.IGNORECASE)
|
|
_TOKEN_RE = re.compile(r'[a-z0-9]+')
|
|
|
|
|
|
def _norm(name: str) -> str:
|
|
"""A name with case, spaces, dashes and underscores ignored: "Large1" == "large-1"."""
|
|
return ''.join(_TOKEN_RE.findall((name or '').lower()))
|
|
|
|
|
|
def _node_kind(entry: dict) -> str:
|
|
if entry['kind'] == 'pdu':
|
|
return _POWER_MODELS.get(entry.get('model', ''), 'pdu')
|
|
return entry['kind']
|
|
|
|
|
|
def _medium(port: dict) -> str:
|
|
"""dac, fibre, sfp (a module that doesn't say which) or copper."""
|
|
sfp = port.get('sfp') or {}
|
|
if sfp.get('found') or sfp.get('part'):
|
|
kind = sfp.get('kind') or ''
|
|
if kind == 'dac':
|
|
return 'dac'
|
|
if kind == 'optic':
|
|
return 'fibre'
|
|
return 'sfp'
|
|
if str(port.get('media') or '').upper().startswith('SFP'):
|
|
return 'sfp' if port.get('up') else 'copper'
|
|
return 'copper'
|
|
|
|
|
|
def _rate(value) -> float:
|
|
try:
|
|
return max(float(value or 0), 0.0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _utilisation(a_to_b: float, b_to_a: float, speed_mbps) -> Optional[float]:
|
|
try:
|
|
speed = float(speed_mbps or 0)
|
|
except (TypeError, ValueError):
|
|
speed = 0.0
|
|
if speed <= 0:
|
|
return None
|
|
return round(min(max(a_to_b, b_to_a) * 8 / (speed * 1e6), 1.0), 4)
|
|
|
|
|
|
def _port_stats(port: dict) -> dict:
|
|
sfp = port.get('sfp') or {}
|
|
return {
|
|
'speed_mbps': port.get('speed_mbps') or 0,
|
|
'up': bool(port.get('up')),
|
|
'medium': _medium(port),
|
|
'sfp_part': sfp.get('part') or '',
|
|
'sfp_vendor': sfp.get('vendor') or '',
|
|
'sfp_serial': sfp.get('serial') or '',
|
|
'errors': round(_rate(port.get('rx_errs_rate')) + _rate(port.get('tx_errs_rate')), 3),
|
|
'drops': round(_rate(port.get('rx_drops_rate')) + _rate(port.get('tx_drops_rate')), 3),
|
|
}
|
|
|
|
|
|
def _end(node: str, port_name: str = '', port_idx=None, iface: str = '') -> dict:
|
|
return {'node': node, 'port_idx': port_idx, 'port_name': port_name, 'iface': iface}
|
|
|
|
|
|
def _far_port(switches: dict, here: str, port: dict, neighbour: str) -> Optional[int]:
|
|
"""Which port on a neighbouring switch the cable from `port` lands on, if it can be told."""
|
|
peer = (port.get('sfp') or {}).get('cable_peer') or {}
|
|
if peer.get('switch') == neighbour and peer.get('port_idx') is not None:
|
|
return peer['port_idx']
|
|
far = switches.get(neighbour)
|
|
if not far:
|
|
return None
|
|
ports = [p for p in (far.get('ports') or {}).values() if p.get('port_idx') is not None]
|
|
# LLDP's port ID names the neighbour's own port ("Port 3", "local Port 7")
|
|
m = _PORT_REF_RE.search(str((port.get('lldp') or {}).get('port_id') or '').strip())
|
|
if m and any(p['port_idx'] == int(m.group(1)) for p in ports):
|
|
return int(m.group(1))
|
|
back = [p['port_idx'] for p in ports if (p.get('lldp') or {}).get('system_name') == here]
|
|
if len(back) == 1:
|
|
return back[0]
|
|
# Gear that sends no LLDP of its own (a PDU): its single uplink
|
|
uplinks = [p['port_idx'] for p in ports if p.get('is_uplink')]
|
|
return uplinks[0] if len(uplinks) == 1 else None
|
|
|
|
|
|
def _is_generic_port_name(name: str) -> bool:
|
|
return bool(re.fullmatch(r'(port|sfp\+?|uplink|downlink)\s*\d*', (name or '').strip(), re.IGNORECASE))
|
|
|
|
|
|
def name_mismatch(port_name: str, neighbour: str, own: str, known: Dict[str, str]) -> Optional[dict]:
|
|
"""A port named after a different device than the one plugged into it.
|
|
|
|
known: {normalised name: display name} of every device. Runs of up to four words of
|
|
the port name are compared with them, so "Large1 Management" names large1 and
|
|
"Downlink to 24RJ45-POE" names 24RJ45-POE. Ports named only "Port 4" never match.
|
|
"""
|
|
if not port_name or _is_generic_port_name(port_name):
|
|
return None
|
|
named = _named_device(port_name, known, {_norm(neighbour), _norm(own)})
|
|
return {'named': named, 'actual': neighbour} if named else None
|
|
|
|
|
|
def _named_device(text: str, known: Dict[str, str], skip=()) -> Optional[str]:
|
|
"""The device a free-text name refers to: the longest run of up to four of its words that
|
|
is a device name (ignoring case, spaces, dashes and underscores), or None."""
|
|
words = _TOKEN_RE.findall((text or '').lower())
|
|
for size in range(min(4, len(words)), 0, -1):
|
|
for i in range(len(words) - size + 1):
|
|
run = ''.join(words[i:i + size])
|
|
if run in known and run not in skip:
|
|
return known[run]
|
|
return None
|
|
|
|
|
|
_MAC_IN_TEXT = re.compile(r'\b([0-9a-f]{2}(?:[:-][0-9a-f]{2}){5})\b', re.IGNORECASE)
|
|
|
|
|
|
def build_power(nodes: Dict[str, dict], link_stats: dict, feeds: List[dict]) -> List[dict]:
|
|
"""What every PDU, UPS and power strip outlet feeds.
|
|
|
|
In order: an assignment saved on the topology page (target '' means nothing), then the
|
|
outlet's name if it names a device ("storage-01", "Dream Machine Pro", or "AT&T Modem" for
|
|
the modem), then a MAC in its name that belongs to a host NIC or a UniFi device.
|
|
Returns [{node, power_w, outlets: [{index, name, on, power_w, auto_power_cycle, target,
|
|
match: saved|cleared|name|mac|none}]}].
|
|
"""
|
|
known = {_norm(n['name']): n['name'] for n in nodes.values() if n['kind'] not in ('stub', 'internet')}
|
|
modem = next((n['name'] for n in nodes.values() if n['kind'] == 'modem'), None)
|
|
macs = {str(mac).lower(): owner[0] for mac, owner in (link_stats.get('host_macs') or {}).items() if owner}
|
|
for name, info in (link_stats.get('unifi_devices') or {}).items():
|
|
if info.get('mac'):
|
|
macs.setdefault(info['mac'].lower(), name)
|
|
saved = {}
|
|
for f in feeds or []:
|
|
try:
|
|
saved[(f['source'], int(f['outlet']))] = f.get('target') or ''
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
|
|
sources = []
|
|
for node in sorted((n for n in nodes.values() if n.get('power')), key=lambda n: n['name'].lower()):
|
|
rows = []
|
|
for o in node.get('outlets') or []:
|
|
idx = o.get('index')
|
|
text = str(o.get('name') or '')
|
|
target, how = None, 'none'
|
|
if (node['name'], idx) in saved:
|
|
target = saved[(node['name'], idx)] or None
|
|
how = 'saved' if target else 'cleared'
|
|
else:
|
|
target = _named_device(text, known, {_norm(node['name'])})
|
|
if not target and modem and 'modem' in _TOKEN_RE.findall(text.lower()):
|
|
target = modem
|
|
if target:
|
|
how = 'name'
|
|
else:
|
|
m = _MAC_IN_TEXT.search(text)
|
|
target = macs.get(m.group(1).lower().replace('-', ':')) if m else None
|
|
how = 'mac' if target else 'none'
|
|
rows.append({'index': idx, 'name': text, 'on': o.get('on') is not False,
|
|
'power_w': o.get('power_w'), 'auto_power_cycle': bool(o.get('auto_power_cycle')),
|
|
'target': target, 'match': how})
|
|
sources.append({'node': node['name'], 'power_w': node.get('power_w'), 'outlets': rows})
|
|
return sources
|
|
|
|
|
|
def _cable_labels(items: List[dict], cables: List[dict]) -> Dict[Tuple[str, str], dict]:
|
|
"""{(device, port): cable} for cutsheet cables still in place. A cable can name the
|
|
device itself or a cutsheet item linked to it."""
|
|
device_of = {i['name']: (i.get('device') or i['name']) for i in items or []}
|
|
out: Dict[Tuple[str, str], dict] = {}
|
|
for c in cables or []:
|
|
if c.get('status') == 'retired':
|
|
continue
|
|
for end in ('a', 'b'):
|
|
dev = c.get(f'{end}_device') or ''
|
|
if not dev:
|
|
continue
|
|
ref = {'id': c.get('id'), 'label': c.get('label') or '', 'type': c.get('cable_type') or '',
|
|
'color': c.get('color') or ''}
|
|
for name in {dev, device_of.get(dev, dev)}:
|
|
out[(name, str(c.get(f'{end}_port') or ''))] = ref
|
|
return out
|
|
|
|
|
|
def _alerts_by_target(events: List[dict]) -> Dict[str, List[dict]]:
|
|
out: Dict[str, List[dict]] = {}
|
|
for ev in events or []:
|
|
if ev.get('is_suppressed'):
|
|
continue
|
|
out.setdefault(ev.get('target_name') or '', []).append({
|
|
'id': ev.get('id'), 'severity': ev.get('severity') or 'info',
|
|
'type': ev.get('event_type') or '', 'detail': ev.get('target_detail') or '',
|
|
'description': ev.get('description') or '',
|
|
})
|
|
return out
|
|
|
|
|
|
def _worst(alerts: List[dict]) -> str:
|
|
return max((a['severity'] for a in alerts), key=lambda s: _SEVERITY_RANK.get(s, 0), default='')
|
|
|
|
|
|
def build_graph(snapshot: dict, link_stats: dict, modem: Optional[dict], events: List[dict],
|
|
cutsheet_items: List[dict], cutsheet_cables: List[dict],
|
|
known_uplinks: Optional[dict] = None, power_feeds: Optional[List[dict]] = None) -> dict:
|
|
"""Every device and cable, as {nodes, links, power, warnings, updated}.
|
|
|
|
snapshot, link_stats, modem: the monitor's saved state.
|
|
events: active events, with is_suppressed set (suppressed ones are left out).
|
|
known_uplinks: the monitor's record of where each device was last plugged in,
|
|
drawn as "last seen" links for a device with no live link.
|
|
power_feeds: outlet assignments saved on the topology page (see build_power).
|
|
"""
|
|
snapshot = snapshot or {}
|
|
link_stats = link_stats or {}
|
|
switches = link_stats.get('unifi_switches') or {}
|
|
unifi_info = link_stats.get('unifi_devices') or {}
|
|
alerts = _alerts_by_target(events)
|
|
labels = _cable_labels(cutsheet_items, cutsheet_cables)
|
|
chassis = {}
|
|
for item in cutsheet_items or []:
|
|
chassis.setdefault(item.get('device') or item['name'], item.get('chassis') or '')
|
|
warnings: List[str] = []
|
|
|
|
# ── Nodes ────────────────────────────────────────────────────────────
|
|
nodes: Dict[str, dict] = {}
|
|
configured = set(snapshot.get('hosts') or {})
|
|
for entry in devices.build_index(snapshot, link_stats, modem):
|
|
name = entry['name']
|
|
if entry['kind'] == 'host' and name not in configured:
|
|
continue # IP-only Prometheus targets: shown only as a stub where a port sees them
|
|
kind = _node_kind(entry)
|
|
node = {
|
|
'id': name, 'name': name, 'kind': kind, 'status': entry['status'], 'model': entry.get('model', ''),
|
|
'ip': entry.get('ip', ''), 'url': f'/device/{name}', 'power': kind in POWER_KINDS,
|
|
}
|
|
if kind == 'host':
|
|
node['chassis'] = chassis.get(name, '')
|
|
node['interfaces'] = devices.physical_interfaces(name, snapshot)
|
|
for nic in node['interfaces']:
|
|
nic_alerts = [a for a in alerts.get(name, []) if a['detail'] == nic['name']]
|
|
nic['alert'] = _worst(nic_alerts)
|
|
elif name in switches or kind in ('router', 'switch'):
|
|
node['ports'] = {
|
|
str(p['port_idx']): {'name': pname, 'up': bool(p.get('up')), 'speed_mbps': p.get('speed_mbps') or 0,
|
|
'is_uplink': bool(p.get('is_uplink')), 'media': p.get('media') or '',
|
|
'poe_power': p.get('poe_power'), 'sfp': p.get('sfp') or None,
|
|
'tx_bytes_rate': _rate(p.get('tx_bytes_rate')),
|
|
'rx_bytes_rate': _rate(p.get('rx_bytes_rate')),
|
|
'neighbour': (p.get('lldp') or {}).get('system_name') or ''}
|
|
for pname, p in ((switches.get(name) or {}).get('ports') or {}).items()
|
|
if p.get('port_idx') is not None
|
|
}
|
|
if kind in POWER_KINDS:
|
|
info = unifi_info.get(name) or {}
|
|
node['power_w'] = info.get('power_w')
|
|
node['outlets'] = info.get('outlets') or []
|
|
node_alerts = alerts.get(name, [])
|
|
node['alerts'] = node_alerts[:10]
|
|
node['alert_count'] = len(node_alerts)
|
|
node['severity'] = _worst(node_alerts)
|
|
nodes[name] = node
|
|
|
|
router = next((n for n in nodes.values() if n['kind'] == 'router'), None)
|
|
modem_name = (modem or {}).get('name') if (modem or {}).get('name') in nodes else None
|
|
wan = ((unifi_info.get(router['name']) or {}).get('wan') or {}) if router else {}
|
|
wan_up = any(w.get('up') for w in wan.values()) if wan else None
|
|
if modem_name:
|
|
wan_up = (modem or {}).get('broadband_state', '').lower() == 'up' if wan_up is None else wan_up
|
|
nodes[INTERNET] = {
|
|
'id': INTERNET, 'name': 'Internet', 'kind': INTERNET, 'model': '', 'ip': '', 'url': '',
|
|
'status': 'unknown' if wan_up is None else ('up' if wan_up else 'down'),
|
|
'alerts': alerts.get('WAN', [])[:10], 'alert_count': len(alerts.get('WAN', [])),
|
|
'severity': _worst(alerts.get('WAN', [])), 'power': False,
|
|
}
|
|
known_names = {_norm(n['name']): n['name'] for n in nodes.values() if n['id'] != INTERNET}
|
|
|
|
# ── Links ────────────────────────────────────────────────────────────
|
|
links: List[dict] = []
|
|
claimed: Dict[Tuple[str, int], dict] = {}
|
|
|
|
def add_link(a: dict, b: dict, port: dict, **extra) -> dict:
|
|
link = {'id': '', 'a': a, 'b': b, 'ghost': False, 'name_mismatch': None,
|
|
'cable': labels.get((a['node'], str(a['port_idx'] if a['port_idx'] is not None else a['iface'])))
|
|
or labels.get((b['node'], str(b['port_idx'] if b['port_idx'] is not None else b['iface']))),
|
|
**_port_stats(port),
|
|
'_tx': _rate(port.get('tx_bytes_rate')), '_rx': _rate(port.get('rx_bytes_rate'))}
|
|
link.update(extra)
|
|
links.append(link)
|
|
for end in (a, b):
|
|
if end['port_idx'] is not None and end['node'] in switches:
|
|
claimed[(end['node'], end['port_idx'])] = link
|
|
return link
|
|
|
|
def ordered_ports(sw: dict):
|
|
return sorted(((pn, p) for pn, p in (sw.get('ports') or {}).items() if p.get('port_idx') is not None),
|
|
key=lambda kv: kv[1]['port_idx'])
|
|
|
|
# The router's internet port goes to the modem, then the modem to the internet
|
|
wan_port = None
|
|
if router and router['name'] in switches:
|
|
wan_port = next(((pn, p) for pn, p in ordered_ports(switches[router['name']]) if p.get('is_uplink')), None)
|
|
upstream = modem_name or INTERNET
|
|
if wan_port:
|
|
pn, p = wan_port
|
|
add_link(_end(router['name'], pn, p['port_idx']), _end(upstream, 'WAN' if modem_name else ''), p, wan=True)
|
|
elif router:
|
|
warnings.append(f'No internet port found on {router["name"]}')
|
|
if modem_name:
|
|
m = modem or {}
|
|
rates = m.get('rates') or {}
|
|
up = str(m.get('broadband_state') or '').lower() == 'up'
|
|
links.append({
|
|
'id': '', 'a': _end(INTERNET), 'b': _end(modem_name, str(m.get('source') or 'WAN').title()),
|
|
'ghost': False, 'name_mismatch': None, 'cable': None, 'wan': True,
|
|
'speed_mbps': m.get('line_speed_mbps') or 0, 'up': up,
|
|
'medium': 'fibre' if str(m.get('source') or '').upper() == 'FIBER' else 'copper',
|
|
'sfp_part': m.get('optics_part_no') or '', 'sfp_vendor': '', 'sfp_serial': '',
|
|
'errors': round(_rate(rates.get('rx_errors')) + _rate(rates.get('tx_errors')), 3),
|
|
'drops': round(_rate(rates.get('rx_drops')) + _rate(rates.get('tx_drops')), 3),
|
|
# Seen from the modem's broadband side: rx comes down from the internet
|
|
'_tx': _rate(rates.get('rx_bytes')), '_rx': _rate(rates.get('tx_bytes')),
|
|
})
|
|
elif not router:
|
|
warnings.append('No UniFi gateway reported yet, so the internet edge is not drawn')
|
|
|
|
# Device-to-device and device-to-host cables, from every switch port with a neighbour
|
|
for sw_name in sorted(switches):
|
|
for pname, port in ordered_ports(switches[sw_name]):
|
|
idx = port['port_idx']
|
|
if (sw_name, idx) in claimed:
|
|
continue
|
|
lldp = port.get('lldp') or {}
|
|
neighbour = lldp.get('system_name') or ''
|
|
here = _end(sw_name, pname, idx)
|
|
if neighbour in nodes and neighbour != sw_name:
|
|
nb = nodes[neighbour]
|
|
if nb['kind'] == 'host':
|
|
far = _end(neighbour, iface=lldp.get('port_id') or '')
|
|
else:
|
|
far_idx = _far_port(switches, sw_name, port, neighbour)
|
|
if far_idx is not None and (neighbour, far_idx) in claimed:
|
|
continue # the other end already drew this cable
|
|
far_port = next((pn for pn, p in ((switches.get(neighbour) or {}).get('ports') or {}).items()
|
|
if p.get('port_idx') == far_idx), '') if far_idx is not None else ''
|
|
far = _end(neighbour, far_port, far_idx)
|
|
link = add_link(here, far, port)
|
|
link['name_mismatch'] = name_mismatch(pname, neighbour, sw_name, known_names)
|
|
if link['name_mismatch'] is None and far['port_name']:
|
|
link['name_mismatch'] = name_mismatch(far['port_name'], sw_name, neighbour, known_names)
|
|
if link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'b'
|
|
elif link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'a'
|
|
continue
|
|
if port.get('is_uplink') and not neighbour:
|
|
continue # leads upstream to gear already drawn (a PDU's management port)
|
|
if not port.get('up') and not lldp:
|
|
continue
|
|
# Something is plugged in that Gandalf can't name: a grey stub
|
|
stub_id = f'stub:{sw_name}:{idx}'
|
|
mac = lldp.get('chassis_id') or ''
|
|
label = neighbour or ('' if _is_generic_port_name(pname) else pname) or mac or 'unknown device'
|
|
nodes[stub_id] = {
|
|
'id': stub_id, 'name': label, 'kind': 'stub', 'status': 'up' if port.get('up') else 'down',
|
|
'model': '', 'ip': neighbour if re.fullmatch(r'[\d.]+', neighbour) else '', 'mac': mac,
|
|
'url': f'/device/{sw_name}#port-{idx}', 'switch': sw_name, 'port_idx': idx, 'port_name': pname,
|
|
'iface': (lldp.get('port_id') or '') if neighbour else '', 'anonymous': not lldp,
|
|
'alerts': [], 'alert_count': 0, 'severity': '', 'power': False,
|
|
}
|
|
link = add_link(here, _end(stub_id, iface=nodes[stub_id]['iface']), port)
|
|
# A stub labelled by its port name would always "match" it, so compare with what was seen
|
|
link['name_mismatch'] = name_mismatch(pname, neighbour or mac or 'an unnamed device', sw_name, known_names)
|
|
if link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'a'
|
|
|
|
# Gear with no live cable: draw where it was last plugged in
|
|
linked = {e['node'] for link in links for e in (link['a'], link['b'])}
|
|
for name, entry in sorted((known_uplinks or {}).items()):
|
|
node = nodes.get(name)
|
|
if not node or name in linked or node['kind'] == 'stub':
|
|
continue
|
|
for up in entry.get('uplinks') or []:
|
|
sw, idx = up.get('switch'), up.get('port_idx')
|
|
if sw not in nodes or idx is None or (sw, idx) in claimed:
|
|
continue
|
|
port = next((p for p in ((switches.get(sw) or {}).get('ports') or {}).values()
|
|
if p.get('port_idx') == idx), {})
|
|
far = (_end(name, iface=up.get('host_interface') or '') if node['kind'] == 'host'
|
|
else _end(name, '', None))
|
|
link = add_link(_end(sw, up.get('port') or '', idx), far, port,
|
|
ghost=True, seen=entry.get('seen') or '')
|
|
link['up'] = False
|
|
|
|
# ── Depth: hops from the internet over device-to-device cables ───────
|
|
adjacency: Dict[str, set] = {}
|
|
for link in links:
|
|
a, b = link['a']['node'], link['b']['node']
|
|
adjacency.setdefault(a, set()).add(b)
|
|
adjacency.setdefault(b, set()).add(a)
|
|
depth: Dict[str, int] = {INTERNET: 0}
|
|
queue = deque([INTERNET])
|
|
while queue:
|
|
cur = queue.popleft()
|
|
for nxt in sorted(adjacency.get(cur, ())):
|
|
if nxt in depth:
|
|
continue
|
|
depth[nxt] = depth[cur] + 1
|
|
# Hosts and stubs are leaves: they never carry the path on to another device
|
|
if nodes[nxt]['kind'] not in ('host', 'stub'):
|
|
queue.append(nxt)
|
|
for node in nodes.values():
|
|
node['depth'] = depth.get(node['id'])
|
|
|
|
# Orient each cable from the internet side (a) to the far side (b), and count traffic that way
|
|
for i, link in enumerate(links):
|
|
da = depth.get(link['a']['node'], 1 << 20)
|
|
db_ = depth.get(link['b']['node'], 1 << 20)
|
|
# Rates were measured at end a as added: the reporting switch port, or the internet side
|
|
tx, rx = link.pop('_tx'), link.pop('_rx')
|
|
if db_ < da:
|
|
link['a'], link['b'] = link['b'], link['a']
|
|
if link.get('name_mismatch') and link['name_mismatch'].get('end'):
|
|
link['name_mismatch']['end'] = 'a' if link['name_mismatch']['end'] == 'b' else 'b'
|
|
tx, rx = rx, tx
|
|
link['a_to_b_bps'] = round(tx, 1)
|
|
link['b_to_a_bps'] = round(rx, 1)
|
|
link['utilisation'] = _utilisation(tx, rx, link['speed_mbps'])
|
|
link['id'] = f'l{i}'
|
|
|
|
# Which host NIC each cable lands on, so the drawing can light the right socket
|
|
for link in links:
|
|
for end in (link['a'], link['b']):
|
|
node = nodes[end['node']]
|
|
if node['kind'] == 'host' and end['iface']:
|
|
for nic in node.get('interfaces', []):
|
|
if nic['name'] == end['iface']:
|
|
nic.setdefault('links', []).append(link['id'])
|
|
|
|
# Power: what each outlet feeds, and on each device, the outlets feeding it
|
|
power = build_power({k: v for k, v in nodes.items() if v['kind'] != 'stub'}, link_stats, power_feeds or [])
|
|
for node in nodes.values():
|
|
node['power_in'] = []
|
|
for src in power:
|
|
for o in src['outlets']:
|
|
if o['target'] in nodes:
|
|
nodes[o['target']]['power_in'].append(
|
|
{'source': src['node'], 'outlet': o['index'], 'on': o['on'], 'power_w': o['power_w']})
|
|
|
|
if not switches:
|
|
warnings.append('No UniFi switch data yet: the monitor has not reported any ports')
|
|
return {
|
|
'nodes': sorted(nodes.values(), key=lambda n: (99 if n['depth'] is None else n['depth'], n['name'].lower())),
|
|
'links': links,
|
|
'power': power,
|
|
'warnings': warnings,
|
|
'updated': link_stats.get('updated'),
|
|
}
|