Files
gandalf/cutsheet.py
jaredandClaude Opus 5.5 bdd68a397d
Lint / JS (eslint) (push) Successful in 7s
Lint / Python (flake8) (push) Successful in 48s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 49s
Lint / Python (flake8) (pull_request) Successful in 51s
Test / Python Tests (pytest) (push) Successful in 1m2s
Lint / JS (eslint) (pull_request) Successful in 7s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 50s
Test / Python Tests (pytest) (pull_request) Successful in 53s
Rack: link items to devices, shelves, several racks, export; fix layout and light mode
Cutsheet items gain a "Gandalf device" link, so an item named after the
product (UDM-Pro) shows the live device's faceplate, ports and status.
Cable checks, suggestions, tickets and the device page treat an item and
its linked device as one. Without this, product-named items drew as bare
name plates.

Items can sit on a shelf instead of a rack unit; the shelf draws what is
on it with live status. Items have a rack, the page shows a tab per rack
(config "racks"), and a rack can be downloaded as SVG or PNG.

The rack is centred, long runs of empty units fold into one band (with a
toggle), every device fills the 19" width, the UPS shows its front only,
and the rack keeps its own palette so it reads in light mode.

New cutsheet_items columns rack, shelf and device are added on first use.
Deleting a shelf, or making it something else, is refused while it holds
items; renaming it carries over.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 21:44:59 -05:00

636 lines
28 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Cutsheets: where every cable is meant to go, compared with what Gandalf sees.
Pure functions with no I/O, like devices.py. app.py loads rows from the
database and the monitor's cached link stats and passes them in.
Model
-----
* An **item** is something Gandalf can't discover: a patch panel, a wall jack,
a shelf, or a rack position for a device. Items of a passive kind (panels and
jacks) pass a signal straight through, from a port's front to the same port's
rear.
* A **cable** joins two endpoints. An endpoint is a device or item name, a port
(a switch's port number, a host's interface name, or a panel port), and for
passive items a side. The far end may be left blank for a cable that ends
nowhere yet, such as an unterminated fibre.
* A **path** is the chain of cables joined through passive items, e.g. switch
port 13 → panel port 5 (front) … panel port 5 (rear) → bedroom jack.
"""
import csv
import io
import re
from typing import Dict, List, Optional, Tuple
ITEM_KINDS = ('patch_panel', 'wall_jack', 'device', 'shelf', 'other')
PASSIVE_KINDS = ('patch_panel', 'wall_jack')
SIDES = ('front', 'rear')
STATUSES = ('planned', 'installed', 'dark', 'retired')
CABLE_TYPES = ('Cat5e', 'Cat6', 'Cat6a', 'DAC', 'AOC', 'OM3', 'OM4', 'OS2', 'Power', 'Other')
LABEL_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,31}$')
NAME_MAX = 100
PORT_MAX = 64
TEXT_MAX = 255
NOTES_MAX = 2000
RACK_NAME_MAX = 100
# Comparison outcomes, and whether each counts as a problem
VERDICTS = {
'match': 'Matches what Gandalf sees',
'wrong_neighbor': 'A different device is connected',
'no_link': 'Planned device is not linked',
'missing_port': 'That port is not reported by the device',
'unverified': "Gandalf can't confirm the far end",
'idle': 'Ends at a jack or panel with nothing linked',
'not_checked': 'No switch port to check against',
'planned': 'Planned, not installed yet',
'dark': 'Unused (dark)',
'retired': 'Retired',
'conflict': 'Another cable uses the same port',
}
PROBLEM_VERDICTS = ('wrong_neighbor', 'no_link', 'missing_port', 'conflict')
class ValidationError(ValueError):
"""A request the cutsheet can't accept, with a message for the user."""
# ── Validation ──────────────────────────────────────────────────────────────
def _text(data: dict, key: str, limit: int = TEXT_MAX, required: bool = False, label: str = '') -> str:
value = data.get(key)
value = '' if value is None else str(value).strip()
if required and not value:
raise ValidationError(f'{label or key} is required')
if len(value) > limit:
raise ValidationError(f'{label or key} must be {limit} characters or fewer')
return value
def _int(data: dict, key: str, low: int, high: int, label: str) -> Optional[int]:
value = data.get(key)
if value in (None, ''):
return None
try:
number = int(value)
except (TypeError, ValueError):
raise ValidationError(f'{label} must be a whole number')
if not low <= number <= high:
raise ValidationError(f'{label} must be between {low} and {high}')
return number
def validate_item(data: dict, items: Optional[Dict[str, dict]] = None) -> dict:
"""An item from a form. `items` (by name) is needed to check a shelf placement."""
kind = _text(data, 'kind', 20) or 'other'
if kind not in ITEM_KINDS:
raise ValidationError(f'kind must be one of {", ".join(ITEM_KINDS)}')
item = {
'name': _text(data, 'name', NAME_MAX, required=True, label='Name'),
'kind': kind,
'port_count': _int(data, 'port_count', 1, 512, 'Port count'),
'location': _text(data, 'location'),
'rack_u': _int(data, 'rack_u', 1, 60, 'Rack unit'),
'rack_height': _int(data, 'rack_height', 1, 20, 'Height'),
'rack': _text(data, 'rack', RACK_NAME_MAX),
'device': _text(data, 'device'),
'shelf': _text(data, 'shelf', NAME_MAX),
'notes': _text(data, 'notes', NOTES_MAX),
}
if item['shelf']:
shelf = (items or {}).get(item['shelf'])
if not shelf or shelf['kind'] != 'shelf':
raise ValidationError(f'{item["shelf"]} is not a shelf on the cutsheet')
if kind == 'shelf' or shelf['name'] == item['name']:
raise ValidationError('A shelf cannot sit on a shelf')
if item['rack_u'] or item['rack_height']:
raise ValidationError('An item on a shelf takes its rack position from the shelf; clear the rack unit')
item['rack'] = '' # the shelf's rack
if item['rack_height'] and not item['rack_u']:
raise ValidationError('Set a rack unit to go with the height')
if item['rack_u'] and not item['rack_height']:
item['rack_height'] = 1
if kind == 'patch_panel' and not item['port_count']:
raise ValidationError('A patch panel needs a port count')
return item
def _endpoint(data: dict, prefix: str, items: Dict[str, dict], required: bool) -> Tuple[str, str, str]:
device = _text(data, f'{prefix}_device', NAME_MAX, required=required,
label='From device' if prefix == 'a' else 'To device')
port = _text(data, f'{prefix}_port', PORT_MAX)
side = _text(data, f'{prefix}_side', 10)
if not device:
if port or side:
raise ValidationError('Choose a device before a port')
return '', '', ''
item = items.get(device)
passive = bool(item and item['kind'] in PASSIVE_KINDS)
if passive:
if side not in SIDES:
raise ValidationError(f'Choose the front or rear of {device}')
if not port:
raise ValidationError(f'Choose a port on {device}')
count = item.get('port_count')
if count and (not port.isdigit() or not 1 <= int(port) <= count):
raise ValidationError(f'{device} has ports 1 to {count}')
else:
side = ''
return device, port, side
def validate_cable(data: dict, items: Dict[str, dict]) -> dict:
label = _text(data, 'label', 32)
if label and not LABEL_RE.match(label):
raise ValidationError('Label may use letters, digits, dot, dash and underscore (up to 32)')
status = _text(data, 'status', 20) or 'installed'
if status not in STATUSES:
raise ValidationError(f'status must be one of {", ".join(STATUSES)}')
cable_type = _text(data, 'cable_type', 32)
if cable_type and cable_type not in CABLE_TYPES:
raise ValidationError(f'Type must be one of {", ".join(CABLE_TYPES)}')
length = data.get('length_m')
if length in (None, ''):
length = None
else:
try:
length = round(float(length), 2)
except (TypeError, ValueError):
raise ValidationError('Length must be a number of metres')
if not 0 < length <= 999:
raise ValidationError('Length must be between 0 and 999 metres')
a = _endpoint(data, 'a', items, required=True)
b = _endpoint(data, 'b', items, required=False)
if a == b:
raise ValidationError('Both ends are the same port')
return {
'label': label or None,
'a_device': a[0], 'a_port': a[1], 'a_side': a[2],
'b_device': b[0], 'b_port': b[1], 'b_side': b[2],
'cable_type': cable_type, 'length_m': length,
'color': _text(data, 'color', 32), 'status': status,
'notes': _text(data, 'notes', NOTES_MAX),
'label_applied': bool(data.get('label_applied')),
}
def next_label(cables: List[dict], prefix: str = 'C') -> str:
"""The next free cable ID: C01, C02, ... (wider once past 99)."""
used = set()
for c in cables:
m = re.match(rf'^{re.escape(prefix)}(\d+)$', c.get('label') or '', re.IGNORECASE)
if m:
used.add(int(m.group(1)))
n = 1
while n in used:
n += 1
return f'{prefix}{n:02d}'
# ── Paths ───────────────────────────────────────────────────────────────────
def _node(cable: dict, end: str) -> Optional[tuple]:
device = cable.get(f'{end}_device') or ''
if not device:
return None
return (device, cable.get(f'{end}_port') or '', cable.get(f'{end}_side') or '')
def _other_side(node: tuple, items: Dict[str, dict]) -> Optional[tuple]:
"""Through a passive item, the same port's other side; otherwise the path ends."""
item = items.get(node[0])
if not item or item['kind'] not in PASSIVE_KINDS or node[2] not in SIDES:
return None
return (node[0], node[1], 'rear' if node[2] == 'front' else 'front')
def _live(cable: dict) -> bool:
return cable.get('status') != 'retired'
def node_index(cables: List[dict]) -> Dict[tuple, List[dict]]:
"""Map each endpoint to the live cables plugged into it."""
index: Dict[tuple, List[dict]] = {}
for c in cables:
if not _live(c):
continue
for end in ('a', 'b'):
node = _node(c, end)
if node:
index.setdefault(node, []).append(c)
return index
def trace(cable: dict, cables: List[dict], items: Dict[str, dict],
index: Optional[Dict[tuple, List[dict]]] = None) -> dict:
"""Follow a cable through passive items in both directions.
Returns {'hops': [node, cable_id, node, ...], 'cable_ids': [...]} with the
hops ordered from one end of the run to the other. A node is
(device, port, side), or None for an unterminated end.
"""
index = index if index is not None else node_index(cables)
def walk(node: Optional[tuple], came_from: dict, seen: set) -> list:
hops = []
while node is not None:
through = _other_side(node, items)
if through is None:
break
nxt = next((c for c in index.get(through, []) if c['id'] != came_from['id'] and c['id'] not in seen),
None)
hops.append(through)
if nxt is None:
break
seen.add(nxt['id'])
hops.append(nxt['id'])
far = 'b' if _node(nxt, 'a') == through else 'a'
node = _node(nxt, far)
hops.append(node)
came_from = nxt
return hops
seen = {cable['id']}
toward_a = walk(_node(cable, 'a'), cable, seen)
toward_b = walk(_node(cable, 'b'), cable, seen)
hops = list(reversed(toward_a)) + [_node(cable, 'a'), cable['id'], _node(cable, 'b')] + toward_b
return {'hops': hops, 'cable_ids': [h for h in hops if isinstance(h, int)]}
# ── Plan vs. actual ─────────────────────────────────────────────────────────
def _switch_port(node: Optional[tuple], switches: dict) -> Optional[dict]:
"""The discovered port behind an endpoint, if it is a switch or router port."""
if not node or node[0] not in switches or not node[1].isdigit():
return None
for port in (switches[node[0]].get('ports') or {}).values():
if port.get('port_idx') == int(node[1]):
return port
return {} # the device is known but doesn't report this port
def _same(a: str, b: str) -> bool:
return bool(a) and bool(b) and a.strip().lower() == b.strip().lower()
def device_of(name: str, items: Dict[str, dict]) -> str:
"""The live device an item name stands for: its linked device, or the name itself."""
return (items.get(name) or {}).get('device') or name
def with_aliases(switches: dict, items: Dict[str, dict]) -> dict:
"""Switches also reachable by the name of an item linked to them (item "UDM-Pro" -> "Dream Machine Pro")."""
out = dict(switches or {})
for name, item in (items or {}).items():
device = item.get('device')
if device and device in out and name not in out:
out[name] = out[device]
return out
def _check_end(node: tuple, other: Optional[tuple], switches: dict, items: Dict[str, dict]) -> Optional[dict]:
port = _switch_port(node, switches)
if port is None:
return None
if port == {}:
return {'verdict': 'missing_port', 'detail': f'{node[0]} reports no port {node[1]}'}
other_passive = other is None or (items.get(other[0], {}).get('kind') in PASSIVE_KINDS)
expected = None if other_passive else device_of(other[0], items)
lldp = port.get('lldp') or {}
seen = lldp.get('system_name') or ''
sfp_peer = (port.get('sfp') or {}).get('cable_peer')
if sfp_peer and other is not None and not other_passive:
if _same(sfp_peer.get('switch', ''), device_of(other[0], items)) and str(sfp_peer.get('port_idx')) == other[1]:
return {'verdict': 'match', 'detail': 'Both ends report the same DAC serial'}
if not port.get('up'):
if expected:
return {'verdict': 'no_link', 'detail': f'{node[0]} port {node[1]} is down'}
return {'verdict': 'idle', 'detail': f'{node[0]} port {node[1]} is down'}
if expected:
if not seen:
return {'verdict': 'unverified', 'detail': f'Link up on port {node[1]}, but no neighbor is identified'}
if _same(seen, expected):
return {'verdict': 'match', 'detail': f'{node[0]} port {node[1]} sees {seen}'}
return {'verdict': 'wrong_neighbor', 'detail': f'{node[0]} port {node[1]} sees {seen}, not {expected}'}
detail = f'Link up on port {node[1]}' + (f', sees {seen}' if seen else '')
return {'verdict': 'unverified', 'detail': detail}
_VERDICT_RANK = ['wrong_neighbor', 'no_link', 'missing_port', 'unverified', 'idle', 'match']
def compare_path(path: dict, cables_by_id: Dict[int, dict], switches: dict, items: Dict[str, dict]) -> dict:
"""Check a traced run against discovery. Returns {'verdict', 'detail'}."""
run = [cables_by_id[i] for i in path['cable_ids']]
statuses = {c['status'] for c in run}
for status in ('retired', 'dark', 'planned'):
if status in statuses:
return {'verdict': status, 'detail': VERDICTS[status]}
ends = (path['hops'][0], path['hops'][-1])
checks = [c for c in (_check_end(ends[0], ends[1], switches, items),
_check_end(ends[1], ends[0], switches, items)) if c]
if not checks:
return {'verdict': 'not_checked', 'detail': VERDICTS['not_checked']}
# A switch-to-switch run is checked from both ends: report the worst
return min(checks, key=lambda c: _VERDICT_RANK.index(c['verdict']))
def evaluate(cables: List[dict], items: Dict[str, dict], switches: dict) -> Dict[int, dict]:
"""Trace and compare every cable. Returns {cable_id: {'path', 'verdict', 'detail'}}."""
switches = with_aliases(switches, items)
index = node_index(cables)
by_id = {c['id']: c for c in cables}
result: Dict[int, dict] = {}
for c in cables:
if not _live(c):
result[c['id']] = {'path': {'hops': [_node(c, 'a'), c['id'], _node(c, 'b')], 'cable_ids': [c['id']]},
'verdict': 'retired', 'detail': VERDICTS['retired']}
continue
conflict = next((n for n in (_node(c, 'a'), _node(c, 'b'))
if n and len([x for x in index.get(n, []) if x['status'] in ('installed', 'dark')]) > 1
and c['status'] in ('installed', 'dark')), None)
path = trace(c, cables, items, index)
if conflict:
others = [x.get('label') or f'#{x["id"]}' for x in index[conflict] if x['id'] != c['id']]
result[c['id']] = {'path': path, 'verdict': 'conflict',
'detail': f'{describe_node(conflict)} is also used by {", ".join(others)}'}
continue
result[c['id']] = dict(compare_path(path, by_id, switches, items), path=path)
return result
def describe_node(node: Optional[tuple]) -> str:
if node is None:
return 'unterminated'
device, port, side = node
text = device
if port:
text += f' port {port}' if port.isdigit() else f' {port}'
if side:
text += f' ({side})'
return text
# ── Undocumented links ──────────────────────────────────────────────────────
def undocumented(cables: List[dict], switches: dict, items: Optional[Dict[str, dict]] = None) -> List[dict]:
"""Linked switch ports that no live cable mentions, as suggestions to add."""
documented = set()
for c in cables:
if not _live(c):
continue
for end in ('a', 'b'):
node = _node(c, end)
if node:
documented.add((device_of(node[0], items or {}).lower(), node[1]))
suggestions = []
for sw_name, sw in sorted((switches or {}).items()):
for port_name, port in (sw.get('ports') or {}).items():
idx = port.get('port_idx')
if not port.get('up') or idx is None or (sw_name.lower(), str(idx)) in documented:
continue
lldp = port.get('lldp') or {}
sfp = port.get('sfp') or {}
peer = sfp.get('cable_peer')
suggestion = {
'switch': sw_name, 'port_idx': idx, 'port_name': port_name,
'neighbor': lldp.get('system_name') or '', 'neighbor_port': lldp.get('port_id') or '',
'speed_mbps': port.get('speed_mbps') or 0,
'cable_type': 'DAC' if sfp.get('kind') == 'dac' else '',
'via': 'lldp' if lldp.get('system_name') else '',
}
if lldp.get('source') == 'unifi-clients':
suggestion['via'] = 'client table'
if peer:
suggestion.update(neighbor=peer['switch'], neighbor_port=str(peer['port_idx']), via='DAC serial')
# Offer each DAC once, from the end that sorts first
if (peer['switch'].lower(), str(peer['port_idx'])) in documented:
continue
if (peer['switch'], peer['port_idx']) < (sw_name, idx):
continue
suggestions.append(suggestion)
return suggestions
# ── Labels to make ──────────────────────────────────────────────────────────
def labels_to_make(cables: List[dict], items: Dict[str, dict], port_labels: List[dict]) -> dict:
"""Cable IDs and panel-port labels that haven't been stuck on yet."""
cords = [
{'id': c['id'], 'text': c['label'], 'copies': 2 if c.get('b_device') else 1,
'where': f'{describe_node(_node(c, "a"))} ↔ {describe_node(_node(c, "b"))}'}
for c in sorted(cables, key=lambda c: _natural(c.get('label') or ''))
if c.get('label') and not c.get('label_applied') and c['status'] in ('installed', 'dark')
]
unlabeled = [c['id'] for c in cables if not c.get('label') and c['status'] in ('installed', 'dark')]
ports = [
{'item': p['item_name'], 'port': p['port'], 'text': p['label']}
for p in sorted(port_labels, key=lambda p: (p['item_name'].lower(), _natural(p['port'])))
if p.get('label') and not p.get('applied')
]
return {'cables': cords, 'ports': ports, 'unlabeled_cable_ids': unlabeled}
def _natural(text: str) -> list:
return [int(t) if t.isdigit() else t.lower() for t in re.split(r'(\d+)', text or '')]
# ── CSV ─────────────────────────────────────────────────────────────────────
CSV_FIELDS = ['label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side',
'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes']
def cables_to_csv(cables: List[dict]) -> str:
out = io.StringIO()
writer = csv.DictWriter(out, fieldnames=CSV_FIELDS, extrasaction='ignore', lineterminator='\n')
writer.writeheader()
for c in sorted(cables, key=lambda c: _natural(c.get('label') or '')):
row = {k: c.get(k) for k in CSV_FIELDS}
row['label_applied'] = 'yes' if c.get('label_applied') else ''
row['length_m'] = '' if c.get('length_m') is None else f'{float(c["length_m"]):g}'
# Keep spreadsheet apps from running a cell as a formula
writer.writerow({k: _csv_safe(v) for k, v in row.items()})
return out.getvalue()
def _csv_safe(value) -> str:
text = '' if value is None else str(value)
return "'" + text if text[:1] in ('=', '+', '-', '@', '\t', '\r') and not _is_number(text) else text
def _is_number(text: str) -> bool:
try:
float(text)
return True
except ValueError:
return False
def parse_csv(text: str, items: Dict[str, dict], max_rows: int = 2000) -> Tuple[List[dict], List[str]]:
"""Validate a cable CSV. Returns (cables, errors); nothing is applied on any error."""
reader = csv.DictReader(io.StringIO(text))
missing = [f for f in ('a_device',) if f not in (reader.fieldnames or [])]
if missing:
return [], [f'Missing column: {", ".join(missing)}']
rows, errors, labels = [], [], set()
for line, raw in enumerate(reader, start=2):
if line - 1 > max_rows:
errors.append(f'More than {max_rows} rows')
break
data = {k: (v or '').strip() for k, v in raw.items() if k}
for k in list(data):
if data[k].startswith("'") and data[k][1:2] in ('=', '+', '-', '@'):
data[k] = data[k][1:]
if not any(data.values()):
continue
data['label_applied'] = data.get('label_applied', '').lower() in ('yes', 'true', '1', 'y')
try:
cable = validate_cable(data, items)
except ValidationError as e:
errors.append(f'Line {line}: {e}')
continue
if cable['label']:
if cable['label'].lower() in labels:
errors.append(f'Line {line}: label {cable["label"]} appears twice')
continue
labels.add(cable['label'].lower())
rows.append(cable)
return rows, errors
# ── Problems, for the monitor ───────────────────────────────────────────────
def path_text(hops: list) -> str:
return ' → '.join(describe_node(h) for h in hops if not isinstance(h, int))
def runs_for(device: str, cables: List[dict], items: Dict[str, dict], switches: dict,
port: Optional[str] = None) -> List[dict]:
"""The live cables ending on a device (or one of its ports), each with its full run.
Returns [{'cable_id', 'label', 'port', 'status', 'verdict', 'detail', 'path'}].
"""
results = None
out = []
for c in sorted(cables, key=lambda c: _natural(c.get('label') or '')):
if not _live(c):
continue
for end in ('a', 'b'):
if not _same(device_of(c.get(f'{end}_device') or '', items), device):
continue
if port is not None and str(c.get(f'{end}_port') or '') != str(port):
continue
if results is None:
results = evaluate(cables, items, switches or {})
r = results[c['id']]
out.append({
'cable_id': c['id'], 'label': c.get('label') or f'#{c["id"]}',
'port': c.get(f'{end}_port') or '', 'status': c['status'],
'verdict': r['verdict'], 'detail': r['detail'], 'path': path_text(r['path']['hops']),
})
return out
def problems(cables: List[dict], items: Dict[str, dict], switches: dict) -> List[dict]:
"""One entry per installed run whose check found a problem.
A run through a panel is several cables with the same verdict; it is reported
once, under the cable on the switch port (the one people look for).
Returns [{'key', 'cable_id', 'label', 'switch', 'port', 'verdict', 'detail', 'path'}].
"""
switches = switches or {}
results = evaluate(cables, items, switches)
by_id = {c['id']: c for c in cables}
seen_runs = set()
found = []
for cid, r in sorted(results.items()):
if r['verdict'] not in PROBLEM_VERDICTS or by_id[cid]['status'] != 'installed':
continue
run = tuple(sorted(r['path']['cable_ids']))
if run in seen_runs and r['verdict'] != 'conflict':
continue
seen_runs.add(run)
hops = r['path']['hops']
ends = [n for n in (hops[0], hops[-1]) if n and n[0] in switches]
switch_end = ends[0] if ends else _node(by_id[cid], 'a')
# Name the run after the cable plugged into the switch, when there is one
lead = next((by_id[i] for i in r['path']['cable_ids']
if switch_end in (_node(by_id[i], 'a'), _node(by_id[i], 'b'))), by_id[cid])
if r['verdict'] == 'conflict':
lead = by_id[cid]
found.append({
'key': f'cable {lead["id"]}',
'cable_id': lead['id'],
'label': lead.get('label') or f'#{lead["id"]}',
'switch': switch_end[0] if switch_end else lead['a_device'],
'port': switch_end[1] if switch_end else lead['a_port'],
'verdict': r['verdict'],
'detail': r['detail'],
'path': path_text(hops),
})
return found
# ── Rack elevation ──────────────────────────────────────────────────────────
def rack_layout(items: List[dict], units: int) -> dict:
"""Place items that have a rack unit, top of the rack first.
An item that would share a unit with one already placed, or runs past the
top of the rack, is reported rather than drawn over another.
Returns {'placed': [item + 'rack_top'], 'overlaps': [{'name', 'units', 'with'}], 'outside': [name]}.
"""
taken: Dict[int, str] = {}
placed, overlaps, outside = [], [], []
racked = [i for i in items if i.get('rack_u')]
# Top first; where two claim the same top unit, the one recorded first is drawn
for item in sorted(racked, key=lambda i: (-(i['rack_u'] + (i.get('rack_height') or 1)), i.get('id') or 0)):
low = item['rack_u']
high = low + (item.get('rack_height') or 1) - 1
span = f'U{low}' if high == low else f'U{low}–{high}'
if high > units:
outside.append({'name': item['name'], 'units': span})
continue
clash = sorted({taken[u] for u in range(low, high + 1) if u in taken})
if clash:
overlaps.append({'name': item['name'], 'units': span, 'with': clash})
continue
for u in range(low, high + 1):
taken[u] = item['name']
placed.append(dict(item, rack_top=high))
return {'placed': placed, 'overlaps': overlaps, 'outside': outside}
_MARK_RANK = {'bad': 3, 'doc': 2, 'ok': 1}
def port_marks(cables: List[dict]) -> Dict[tuple, dict]:
"""Per (device, port): the worst check of the live cables plugged in there.
`cables` are evaluated cables (with 'verdict', 'detail' and 'run', the IDs of
every cable in its run). Returns {(device, port): {'mark', 'text', 'cable_id', 'run'}},
where mark is ok (matches), bad (a problem) or doc (documented, not confirmed).
"""
out: Dict[tuple, dict] = {}
for c in cables:
if not _live(c):
continue
mark = 'bad' if c['verdict'] in PROBLEM_VERDICTS else ('ok' if c['verdict'] == 'match' else 'doc')
label = c.get('label') or f'#{c["id"]}'
for end in ('a', 'b'):
device, port = c.get(f'{end}_device'), c.get(f'{end}_port')
if not device or not port:
continue
entry = out.setdefault((device, str(port)), {'mark': mark, 'cable_id': c['id'], 'run': [], 'by_detail': {}})
if _MARK_RANK[mark] > _MARK_RANK[entry['mark']]:
entry.update(mark=mark, cable_id=c['id'])
entry['by_detail'].setdefault(c['detail'], []).append(label)
entry['run'] = sorted(set(entry['run']) | set(c.get('run') or [c['id']]))
for entry in out.values():
# A panel port's front and rear cables share one check: say it once
entry['text'] = '; '.join(f'cable{"s" if len(labels) > 1 else ""} {", ".join(labels)}: {detail}'
for detail, labels in entry.pop('by_detail').items())
return out