Lint / JS (eslint) (push) Successful in 8s
Lint / Python (flake8) (push) Successful in 1m0s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m9s
Lint / Python (flake8) (pull_request) Successful in 57s
Test / Python Tests (pytest) (push) Successful in 1m14s
Lint / JS (eslint) (pull_request) Successful in 8s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 56s
Test / Python Tests (pytest) (pull_request) Successful in 1m5s
Gear on the rear posts usually still faces the front (the StarTech's RPDU and ZimaBoard shelf), so "which posts" and "which way the ports face" are now separate fields (mount, facing). The page has three views: Front (front posts), Rear posts (rear-post gear seen from the front, i.e. what is behind the front gear) and Rear (from behind). An item seen from the side it doesn't face shows its back. Facing defaults to front, so existing items keep their meaning. Shelves are drawn at true scale with their contents ordered by a new "Position on shelf". The IceWhale ZimaBoard 832 is traced from IceWhale's labelled photo of its port end (ZimaDocs): Mini-DP, two RJ45-over-USB 3.0 stacks with link/activity LEDs, DC jack, gunmetal case, gold PCB edge and black base. Its RJ45 link lights follow the host's physical NICs and the status glow follows the host. New cutsheet_items columns facing and slot, added on first use. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
698 lines
31 KiB
Python
698 lines
31 KiB
Python
"""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')
|
|
# Which side of a 4-post rack an item is mounted from. A front and a rear item can share
|
|
# a unit; a full-depth one (a server) takes the unit on both sides.
|
|
MOUNTS = ('front', 'rear', 'full')
|
|
# Which way an item's ports face. Gear on the rear posts usually still faces the front.
|
|
FACINGS = ('front', 'rear')
|
|
# Rack views: the front posts seen from the front, the rear posts seen from the front
|
|
# (what is behind the front gear), and the rack seen from behind
|
|
RACK_VIEWS = ('front', 'behind', 'rear')
|
|
RACK_UNITS_MAX = 60
|
|
# Drawn fronts for gear without a live faceplate (keys of RACK_FRONTS in static/fronts.js)
|
|
CHASSIS = {
|
|
'unvr-pro': 'UniFi NVR Pro (UNVR Pro)',
|
|
'uacc-patch-24': 'UniFi Patch Panel Blank 24 (UACC-Rack-Panel-Patch-Blank-24)',
|
|
'sliger-cx4712': 'Sliger CX4712 4U (10-bay)',
|
|
'rosewill-rsv-l4500u': 'Rosewill RSV-L4500U 4U (15-bay)',
|
|
'zimaboard-832': 'IceWhale ZimaBoard 832 (shelf gear)',
|
|
'server-1u': 'Generic 1U server',
|
|
'server-2u': 'Generic 2U server',
|
|
'server-4u': 'Generic 4U server',
|
|
}
|
|
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'),
|
|
'mount': _text(data, 'mount', 10) or 'front',
|
|
'chassis': _text(data, 'chassis', 40),
|
|
'facing': _text(data, 'facing', 10) or 'front',
|
|
'slot': _int(data, 'slot', 1, 99, 'Position on the shelf'),
|
|
'shelf': _text(data, 'shelf', NAME_MAX),
|
|
'notes': _text(data, 'notes', NOTES_MAX),
|
|
}
|
|
if item['chassis'] and item['chassis'] not in CHASSIS:
|
|
raise ValidationError(f'Unknown front drawing {item["chassis"]}')
|
|
if item['facing'] not in FACINGS:
|
|
raise ValidationError(f'facing must be one of {", ".join(FACINGS)}')
|
|
if item['slot'] and not item['shelf']:
|
|
raise ValidationError('A position applies only to an item on a shelf')
|
|
if item['mount'] not in MOUNTS:
|
|
raise ValidationError(f'mount must be one of {", ".join(MOUNTS)}')
|
|
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 and side
|
|
item['mount'] = 'front'
|
|
item['facing'] = 'front'
|
|
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 _sides(item: dict) -> Tuple[str, ...]:
|
|
mount = item.get('mount') or 'front'
|
|
return ('front', 'rear') if mount == 'full' else (mount,)
|
|
|
|
|
|
def rack_layout(items: List[dict], units: int, face: str = 'front') -> dict:
|
|
"""Place items that have a rack unit, top of the rack first, for one view.
|
|
|
|
Front-post and rear-post items may share a unit; full-depth items take it on both.
|
|
Views (RACK_VIEWS): 'front' shows the front posts from the front; 'behind' shows the
|
|
rear posts from the front (what sits behind the front gear); 'rear' shows the rear posts
|
|
from behind. Full-depth items appear in all three.
|
|
An item that would share a unit and post with one already placed, or runs past the
|
|
top of the rack, is reported (whatever the view) rather than drawn over another.
|
|
Returns {'placed': [item + 'rack_top' + 'through'], 'overlaps': [{'name', 'units', 'with'}],
|
|
'outside': [name]}, where 'through' marks an item seen from the side it doesn't face
|
|
(its back, rather than its ports).
|
|
"""
|
|
side = 'front' if face == 'front' else 'rear'
|
|
looking = 'rear' if face == 'rear' else 'front'
|
|
taken: Dict[tuple, 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
|
|
sides = _sides(item)
|
|
cells = [(u, s) for u in range(low, high + 1) for s in sides]
|
|
clash = sorted({taken[c] for c in cells if c in taken})
|
|
if clash:
|
|
overlaps.append({'name': item['name'], 'units': span, 'with': clash})
|
|
continue
|
|
for c in cells:
|
|
taken[c] = item['name']
|
|
if side in sides:
|
|
placed.append(dict(item, rack_top=high, through=(item.get('facing') or 'front') != looking))
|
|
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
|
|
|
|
|
|
def validate_rack(data: dict) -> dict:
|
|
"""A rack from the rack settings form."""
|
|
rack = {
|
|
'name': _text(data, 'name', RACK_NAME_MAX, required=True, label='Rack name'),
|
|
'units': _int(data, 'units', 1, RACK_UNITS_MAX, 'Height'),
|
|
'notes': _text(data, 'notes', NOTES_MAX),
|
|
}
|
|
if not rack['units']:
|
|
raise ValidationError('Set the rack height in units')
|
|
return rack
|