Coverage for netbox_data_import/trace_workbook.py: 98.40%
501 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Interpret trace workbooks without accessing the ORM."""
5from __future__ import annotations
7from collections import defaultdict
8from dataclasses import asdict, dataclass, field, replace
9from hashlib import sha256
10from io import BytesIO
11import itertools
12import json
13from typing import Iterable, Mapping, Sequence
15import openpyxl
17from .adapters import SourceDiagnostic, SourceUnreadable
18from .field_keys import (
19 FRONT_PORT_CLASSES,
20 INTERFACE_PORT_CLASSES,
21 PORT_CLASSES,
22 POWER_CLASSES,
23 REAR_PORT_CLASSES,
24 same_device_and_cards,
25)
26from .source_trace import (
27 EndpointSummary,
28 IdentityKey,
29 PassThroughClaim,
30 PortClassLabel,
31 SegmentEvidence,
32 SourceTrace,
33 TerminationReference,
34 TraceProvenance,
35)
36from .trace_schema import TRACE_EXPORT_TIMESTAMP_MAX_LENGTH
37from .identity import identity_text
38from .values import source_text
40TRACE_PATH_SHEET = "Trace From To"
41TRACE_LIST_SHEET = "Trace List"
43_PATH_HEADER = (
44 "Port",
45 "PortClass",
46 "Cards",
47 "Device",
48 "UPos",
49 "Rack",
50 "Location",
51 "CableClass",
52 "Port",
53 "PortClass",
54 "Cards",
55 "Device",
56 "UPos",
57 "Rack",
58 "Location",
59)
60_LIST_HEADER = ("Location", "Rack", "UPos", "Device", "Cards", "Port", "PortClass", "Cable")
61_BLOCK_LINE_MARKERS = ("From", "To")
64@dataclass(frozen=True)
65class TraceWorkbookConfig:
66 """The detached trace-workbook configuration: each mapped PortClass label and its fixed PortClass."""
68 port_class_map: Mapping[str, str] = field(default_factory=dict)
71@dataclass(frozen=True)
72class _RawRow:
73 """One spreadsheet row, holding the cells the sheet width defines."""
75 row_number: int
76 values: tuple[object, ...]
79@dataclass(frozen=True)
80class _Block:
81 """One From and To block with its header row, data rows, and workbook provenance."""
83 sheet: str
84 ordinal: int
85 row_start: int
86 row_end: int
87 from_text: str
88 to_text: str
89 has_to_line: bool
90 header: tuple[str, ...]
91 rows: tuple[_RawRow, ...]
92 export_timestamp: str
93 workbook_fingerprint: str
95 @property
96 def pair_key(self) -> tuple[str, str]:
97 """Return the From and To text that pairs this block with the other sheet."""
98 return source_text(self.from_text), source_text(self.to_text)
101@dataclass(frozen=True)
102class _ParsedSegment:
103 """One Segment Evidence entry with the row that stated it."""
105 evidence: SegmentEvidence
106 row_number: int
109@dataclass(frozen=True)
110class _ParsedVisit:
111 """One Trace List visit with the row that stated it."""
113 termination: TerminationReference
114 row_number: int
117_SegmentClaim = tuple[IdentityKey, IdentityKey]
120def parse_endpoint_line(line: str) -> TerminationReference:
121 """Parse ``Device > [Cards > ]Port (PortClass)`` into a Termination Reference."""
122 parts = source_text(line).split(" > ")
123 if len(parts) not in (2, 3):
124 raise ValueError("Endpoint line must contain a device, an optional cards label, and a port.")
125 device = source_text(parts[0])
126 cards = source_text(parts[1]) if len(parts) == 3 else ""
127 port_and_class = source_text(parts[-1])
128 marker = port_and_class.rfind(" (")
129 if marker <= 0 or not port_and_class.endswith(")"):
130 raise ValueError("Endpoint line must end with a PortClass in parentheses.")
131 port = source_text(port_and_class[:marker])
132 port_class = source_text(port_and_class[marker + 2 : -1])
133 if not device or not port or not port_class:
134 raise ValueError("Endpoint line contains an empty device, port, or PortClass.")
135 return TerminationReference(device=device, cards=cards, port=port, port_class=port_class)
138def canonical_trace_identity(
139 first: TerminationReference,
140 second: TerminationReference,
141) -> str:
142 """Return the sorted endpoint pair as compact canonical JSON."""
143 endpoints = sorted((first.identity_key, second.identity_key))
144 return json.dumps(endpoints, ensure_ascii=False, separators=(",", ":"))
147def canonical_orientation(
148 from_termination: TerminationReference,
149 to_termination: TerminationReference,
150 segments: Sequence[SegmentEvidence],
151) -> tuple[SegmentEvidence, ...]:
152 """Orient Segment Evidence from the endpoint whose identity key sorts first."""
153 if from_termination.identity_key <= to_termination.identity_key:
154 return tuple(segments)
155 return tuple(
156 SegmentEvidence(left=segment.right, cable_class=segment.cable_class, right=segment.left)
157 for segment in reversed(segments)
158 )
161def content_fingerprint(
162 from_termination: TerminationReference,
163 to_termination: TerminationReference,
164 segments: Sequence[SegmentEvidence],
165) -> str:
166 """Hash the canonical endpoints, Segment Evidence, and Pass-Through Claims."""
167 oriented = canonical_orientation(from_termination, to_termination, segments)
168 claims = _pass_through_claims(oriented)
169 endpoints = sorted((from_termination, to_termination), key=lambda termination: termination.identity_key)
170 payload = {
171 # An Endpoint Summary fallback states no segment, so the endpoints carry its whole content.
172 "endpoints": [[termination.identity_key, termination.port_class] for termination in endpoints],
173 "segments": [
174 [
175 segment.left.identity_key,
176 segment.left.port_class,
177 segment.cable_class,
178 segment.right.identity_key,
179 segment.right.port_class,
180 ]
181 for segment in oriented
182 ],
183 "pass_through_claims": [
184 [
185 identity_text(claim.device),
186 identity_text(claim.cards),
187 identity_text(claim.entry_port),
188 identity_text(claim.exit_port),
189 ]
190 for claim in claims
191 ],
192 }
193 serialized = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
194 return sha256(serialized.encode()).hexdigest()
197def _cell_text(value: object) -> str:
198 """Return the raw text of a cell value, empty for None."""
199 return "" if value is None else str(value)
202def _normalized_header(values: Iterable[object]) -> tuple[str, ...]:
203 """Return one header row as trimmed text."""
204 return tuple(source_text(value) for value in values)
207def _row_has_data(values: Iterable[object]) -> bool:
208 """Return whether any cell in the row carries text."""
209 return any(source_text(value) for value in values)
212def _row_values(sheet, row_number: int, width: int) -> tuple[object, ...]:
213 """Return the cells of one row across the sheet's fixed width."""
214 return tuple(sheet.cell(row=row_number, column=column).value for column in range(1, width + 1))
217def _last_data_row(values_by_row: Mapping[int, tuple[object, ...]], start: int, stop: int) -> int:
218 """Return the last row in the range that carries data."""
219 for row_number in range(stop, start - 1, -1):
220 if _row_has_data(values_by_row[row_number]):
221 return row_number
222 return start
225def _extract_blocks(sheet, workbook_fingerprint: str) -> tuple[_Block, ...]:
226 """Return every block on one sheet, including visits the next block's lines overwrite."""
227 width = len(_PATH_HEADER) if sheet.title == TRACE_PATH_SHEET else len(_LIST_HEADER)
228 # openpyxl builds a cell object on first access, so the sheet is read once and every step reuses it.
229 values_by_row = {row: _row_values(sheet, row, width) for row in range(1, sheet.max_row + 1)}
230 starts = [row for row, values in values_by_row.items() if source_text(values[0]) == "From"]
231 first_row = values_by_row.get(1, ())
232 export_timestamp = source_text(first_row[1]) if first_row and source_text(first_row[0]) == "Executed" else ""
233 blocks = []
234 for index, start in enumerate(starts):
235 boundary = starts[index + 1] if index + 1 < len(starts) else sheet.max_row + 1
236 to_row = start + 1
237 has_to_line = to_row < boundary and source_text(values_by_row[to_row][0]) == "To"
238 header_row = start + 2
239 header = _normalized_header(values_by_row[header_row]) if header_row < boundary else ()
240 rows = [
241 _RawRow(row_number, values_by_row[row_number])
242 for row_number in range(start + 3, boundary)
243 if _row_has_data(values_by_row[row_number])
244 ]
245 row_end = _last_data_row(values_by_row, start, boundary - 1)
246 # A block that fills its separator row keeps its last rows under the next block's lines.
247 for carry_row in (boundary, boundary + 1):
248 if carry_row > sheet.max_row:
249 continue
250 values = values_by_row[carry_row]
251 if source_text(values[0]) not in _BLOCK_LINE_MARKERS:
252 continue
253 if _row_has_data(values[2:]):
254 rows.append(_RawRow(carry_row, ("", "", *values[2:])))
255 row_end = carry_row
256 blocks.append(
257 _Block(
258 sheet=sheet.title,
259 ordinal=index + 1,
260 row_start=start,
261 row_end=row_end,
262 from_text=_cell_text(values_by_row[start][1]),
263 to_text=_cell_text(values_by_row[to_row][1]) if has_to_line else "",
264 has_to_line=has_to_line,
265 header=header,
266 rows=tuple(rows),
267 export_timestamp=export_timestamp,
268 workbook_fingerprint=workbook_fingerprint,
269 )
270 )
271 return tuple(blocks)
274def _location(block: _Block) -> str:
275 """Return the readable source position of one block."""
276 return f"{block.sheet} block {block.ordinal} (rows {block.row_start}-{block.row_end})"
279def _error(block: _Block, code: str, detail: str, row_number: int | None = None) -> SourceDiagnostic:
280 """Return one source diagnostic that names the block it came from."""
281 return SourceDiagnostic(
282 code=code,
283 message=f"{_location(block)}: {detail}",
284 row_number=block.row_start if row_number is None else row_number,
285 )
288def _metadata_errors(blocks: Iterable[_Block | None]) -> list[SourceDiagnostic]:
289 """Return diagnostics for raw workbook metadata that cannot fit its stored representation."""
290 errors = []
291 for block in blocks:
292 if block is not None and len(block.export_timestamp) > TRACE_EXPORT_TIMESTAMP_MAX_LENGTH:
293 errors.append(
294 _error(
295 block,
296 "trace.metadata_too_long",
297 f"The export timestamp exceeds {TRACE_EXPORT_TIMESTAMP_MAX_LENGTH} characters.",
298 row_number=1,
299 )
300 )
301 return errors
304def _endpoint_summary(block: _Block) -> tuple[EndpointSummary | None, list[SourceDiagnostic]]:
305 """Return the Endpoint Summary of one block, or the reason it has none."""
306 errors = []
307 if not block.has_to_line:
308 errors.append(_error(block, "trace.incomplete_block", "The block has no To line."))
309 return None, errors
310 try:
311 from_termination = parse_endpoint_line(block.from_text)
312 to_termination = parse_endpoint_line(block.to_text)
313 except ValueError as exc:
314 errors.append(_error(block, "trace.incomplete_block", str(exc)))
315 return None, errors
316 return (
317 EndpointSummary(
318 from_termination=from_termination,
319 to_termination=to_termination,
320 from_text=block.from_text,
321 to_text=block.to_text,
322 ),
323 errors,
324 )
327def _termination_from_path(values: tuple[object, ...], offset: int) -> TerminationReference:
328 """Return the Termination Reference one side of a Segment Evidence row states."""
329 termination = TerminationReference(
330 port=source_text(values[offset]),
331 port_class=source_text(values[offset + 1]),
332 cards=source_text(values[offset + 2]),
333 device=source_text(values[offset + 3]),
334 u_position=source_text(values[offset + 4]),
335 rack=source_text(values[offset + 5]),
336 location=source_text(values[offset + 6]),
337 )
338 if not termination.device or not termination.port or not termination.port_class:
339 raise ValueError("A Segment Evidence row has an empty device, port, or PortClass.")
340 return termination
343def _parse_segments(block: _Block) -> tuple[tuple[_ParsedSegment, ...], list[SourceDiagnostic]]:
344 """Return the Segment Evidence a path block states, and the rows it could not read."""
345 if block.header != _PATH_HEADER:
346 return (), [_error(block, "trace.incomplete_block", "The block has no recognized Segment Evidence header.")]
347 parsed = []
348 errors = []
349 for row in block.rows:
350 try:
351 left = _termination_from_path(row.values, 0)
352 cable_class = source_text(row.values[7])
353 right = _termination_from_path(row.values, 8)
354 except (IndexError, ValueError) as exc:
355 errors.append(_error(block, "trace.incomplete_block", str(exc), row.row_number))
356 continue
357 if not cable_class:
358 errors.append(
359 _error(
360 block, "trace.incomplete_block", "A Segment Evidence row has an empty CableClass.", row.row_number
361 )
362 )
363 continue
364 parsed.append(_ParsedSegment(SegmentEvidence(left, cable_class, right), row.row_number))
365 return tuple(parsed), errors
368def _parse_visits(block: _Block) -> tuple[tuple[_ParsedVisit, ...], list[SourceDiagnostic]]:
369 """Return the visits a Trace List block states, and the rows it could not read."""
370 if block.header != _LIST_HEADER:
371 return (), [_error(block, "trace.incomplete_block", "The block has no recognized Trace List header.")]
372 parsed = []
373 errors = []
374 for row in block.rows:
375 try:
376 termination = TerminationReference(
377 location=source_text(row.values[0]),
378 rack=source_text(row.values[1]),
379 u_position=source_text(row.values[2]),
380 device=source_text(row.values[3]),
381 cards=source_text(row.values[4]),
382 port=source_text(row.values[5]),
383 port_class=source_text(row.values[6]),
384 )
385 except IndexError:
386 termination = TerminationReference("", "", "", "")
387 if not termination.device or not termination.port or not termination.port_class:
388 errors.append(
389 _error(
390 block,
391 "trace.corroboration_mismatch",
392 "A Trace List visit has an empty device, port, or PortClass.",
393 row.row_number,
394 )
395 )
396 continue
397 parsed.append(_ParsedVisit(termination, row.row_number))
398 return tuple(parsed), errors
401def _unknown_port_class_error(
402 block: _Block,
403 terminations: Iterable[tuple[TerminationReference, int]],
404) -> SourceDiagnostic | None:
405 """Return the first PortClass outside the fixed vocabulary as a diagnostic."""
406 for termination, row_number in terminations:
407 if termination.port_class not in PORT_CLASSES:
408 return _error(
409 block,
410 "trace.unknown_port_class",
411 f"PortClass '{termination.port_class}' is outside the fixed vocabulary.",
412 row_number,
413 )
414 return None
417def _port_class_labels(
418 terminations: Iterable[TerminationReference], port_class_map: Mapping[str, str]
419) -> tuple[PortClassLabel, ...]:
420 """Return each stated label outside the fixed vocabulary, with the PortClass the map gives it."""
421 labels = {termination.port_class for termination in terminations if termination.port_class not in PORT_CLASSES}
422 return tuple(PortClassLabel(label, port_class_map.get(label, "")) for label in sorted(labels))
425def _mapped(termination: TerminationReference, port_class_map: Mapping[str, str]) -> TerminationReference:
426 """Return the termination with the fixed PortClass its source label maps to."""
427 port_class = port_class_map.get(termination.port_class)
428 return termination if port_class is None else replace(termination, port_class=port_class)
431def _mapped_segments(
432 segments: Iterable[_ParsedSegment], port_class_map: Mapping[str, str]
433) -> tuple[_ParsedSegment, ...]:
434 """Return Segment Evidence rows whose two ends carry their mapped PortClass."""
435 return tuple(
436 _ParsedSegment(
437 SegmentEvidence(
438 _mapped(parsed.evidence.left, port_class_map),
439 parsed.evidence.cable_class,
440 _mapped(parsed.evidence.right, port_class_map),
441 ),
442 parsed.row_number,
443 )
444 for parsed in segments
445 )
448def _mapped_visits(visits: Iterable[_ParsedVisit], port_class_map: Mapping[str, str]) -> tuple[_ParsedVisit, ...]:
449 """Return Trace List visits that carry their mapped PortClass."""
450 return tuple(_ParsedVisit(_mapped(visit.termination, port_class_map), visit.row_number) for visit in visits)
453def _mapped_summary(summary: EndpointSummary, port_class_map: Mapping[str, str]) -> EndpointSummary:
454 """Return the Endpoint Summary whose two terminations carry their mapped PortClass."""
455 return replace(
456 summary,
457 from_termination=_mapped(summary.from_termination, port_class_map),
458 to_termination=_mapped(summary.to_termination, port_class_map),
459 )
462def _termination_text(termination: TerminationReference) -> str:
463 """Return one termination in the endpoint-line form the source export writes."""
464 parts = [termination.device, termination.cards, f"{termination.port} ({termination.port_class})"]
465 return " > ".join(part for part in parts if part)
468def _linearity_error(
469 block: _Block, summary: EndpointSummary, segments: Sequence[_ParsedSegment]
470) -> SourceDiagnostic | None:
471 """Return the first structural break in a path, or None when the path is linear."""
472 # A path states two terminations whether or not it states the segments between them.
473 if summary.from_termination.identity_key == summary.to_termination.identity_key:
474 return _error(
475 block,
476 "trace.non_linear_path",
477 "The From and To lines name one termination, so the path does not join two terminations.",
478 segments[-1].row_number if segments else block.row_start,
479 )
480 if not segments:
481 return None
482 if segments[0].evidence.left.identity_key != summary.from_termination.identity_key:
483 return _error(
484 block,
485 "trace.non_linear_path",
486 "The first Segment Evidence termination contradicts the From line.",
487 segments[0].row_number,
488 )
489 if segments[-1].evidence.right.identity_key != summary.to_termination.identity_key:
490 return _error(
491 block,
492 "trace.non_linear_path",
493 "The last Segment Evidence termination contradicts the To line.",
494 segments[-1].row_number,
495 )
496 seen_segments = set()
497 left_destinations: dict[IdentityKey, set[IdentityKey]] = defaultdict(set)
498 right_sources: dict[IdentityKey, set[IdentityKey]] = defaultdict(set)
499 for parsed in segments:
500 left_key = parsed.evidence.left.identity_key
501 right_key = parsed.evidence.right.identity_key
502 # One cable has no direction, so a reversed row restates the segment rather than adding one.
503 pair = tuple(sorted((left_key, right_key)))
504 if pair in seen_segments:
505 return _error(
506 block,
507 "trace.non_linear_path",
508 "The path repeats a Segment Evidence row.",
509 parsed.row_number,
510 )
511 seen_segments.add(pair)
512 left_destinations[left_key].add(right_key)
513 right_sources[right_key].add(left_key)
514 if len(left_destinations[left_key]) > 1 or len(right_sources[right_key]) > 1:
515 return _error(
516 block,
517 "trace.non_linear_path",
518 "The Segment Evidence rows branch.",
519 parsed.row_number,
520 )
521 for previous, following in itertools.pairwise(segments):
522 if not same_device_and_cards(previous.evidence.right, following.evidence.left):
523 return _error(
524 block,
525 "trace.non_linear_path",
526 "Consecutive Segment Evidence rows do not share a device and cards label: "
527 f"row {previous.row_number} ends at {_termination_text(previous.evidence.right)}, "
528 f"but row {following.row_number} starts at {_termination_text(following.evidence.left)}.",
529 following.row_number,
530 )
531 return None
534def _power_evidence(
535 block: _Block, summary: EndpointSummary, segments: Sequence[_ParsedSegment]
536) -> tuple[EndpointSummary, tuple[_ParsedSegment, ...], SourceDiagnostic | None]:
537 """Return the one cable a power block names, from its From termination to that row's peer.
539 A power block states a PDU fan-out, not a path. Only the row that touches From names a cable of
540 this trace, and its peer replaces the To line, which names an arbitrary leaf of the fan-out.
541 """
542 from_key = summary.from_termination.identity_key
543 touching = []
544 for parsed in segments:
545 evidence = parsed.evidence
546 if evidence.left.identity_key != from_key and evidence.right.identity_key == from_key:
547 evidence = SegmentEvidence(left=evidence.right, cable_class=evidence.cable_class, right=evidence.left)
548 if evidence.left.identity_key == from_key:
549 touching.append(_ParsedSegment(evidence, parsed.row_number))
550 if len(touching) == 1:
551 (selected,) = touching
552 return replace(summary, to_termination=selected.evidence.right), (selected,), None
553 if touching:
554 detail = (
555 f"{len(touching)} Segment Evidence rows touch the From termination, "
556 "so the power block names no single cable."
557 )
558 row_number = touching[1].row_number
559 else:
560 detail = "No Segment Evidence row touches the From termination, so the power block names no cable."
561 row_number = block.row_start
562 return summary, (), _error(block, "trace.non_linear_path", detail, row_number)
565def _power_corroboration_error(
566 block: _Block, segment: SegmentEvidence, visits: Sequence[_ParsedVisit]
567) -> SourceDiagnostic | None:
568 """Return a contradiction when a non-empty Trace List does not visit both ends of the power cable."""
569 visited = {visit.termination.identity_key for visit in visits}
570 if not visits or {segment.left.identity_key, segment.right.identity_key} <= visited:
571 return None
572 return _error(
573 block,
574 "trace.corroboration_mismatch",
575 "The Trace List does not visit both terminations of the cable at the From termination.",
576 visits[0].row_number,
577 )
580def _endpoint_visits(
581 summary: EndpointSummary, visits: Iterable[TerminationReference]
582) -> tuple[TerminationReference, ...]:
583 """Return the visits of the two endpoints, which drops the rest of a PDU fan-out."""
584 endpoints = {summary.from_termination.identity_key, summary.to_termination.identity_key}
585 return tuple(visit for visit in visits if visit.identity_key in endpoints)
588def _pass_through_claims(segments: Sequence[SegmentEvidence]) -> tuple[PassThroughClaim, ...]:
589 """Return the continuation each pair of consecutive segments claims."""
590 claims = []
591 for previous, following in itertools.pairwise(segments):
592 if not same_device_and_cards(previous.right, following.left):
593 continue
594 claims.append(
595 PassThroughClaim(
596 device=previous.right.device,
597 cards=previous.right.cards,
598 entry_port=previous.right.port,
599 exit_port=following.left.port,
600 )
601 )
602 return tuple(claims)
605def _pass_through_error(block: _Block, segments: Sequence[_ParsedSegment]) -> SourceDiagnostic | None:
606 """Return a diagnostic for the first Pass-Through Claim at an interface PortClass."""
607 for previous, following in itertools.pairwise(segments):
608 if not same_device_and_cards(previous.evidence.right, following.evidence.left):
609 continue
610 if (
611 previous.evidence.right.port_class in INTERFACE_PORT_CLASSES
612 or following.evidence.left.port_class in INTERFACE_PORT_CLASSES
613 ):
614 entry = previous.evidence.right
615 exit_reference = following.evidence.left
616 return _error(
617 block,
618 "trace.pass_through_at_interface",
619 (
620 f"The path continues through {entry.device} from {entry.port} ({entry.port_class}) "
621 f"to {exit_reference.port} ({exit_reference.port_class}). "
622 "An interface PortClass can terminate a trace, "
623 "but it cannot join two cable segments. Correct the source path or PortClass."
624 ),
625 following.row_number,
626 )
627 return None
630def _group_consecutive_visits(visits: Sequence[_ParsedVisit]) -> tuple[tuple[_ParsedVisit, ...], ...]:
631 """Group Trace List visits into one entry per visited device."""
632 groups: list[list[_ParsedVisit]] = []
633 for visit in visits:
634 device = identity_text(visit.termination.device)
635 if not groups or identity_text(groups[-1][0].termination.device) != device:
636 groups.append([])
637 groups[-1].append(visit)
638 return tuple(tuple(group) for group in groups)
641def _expected_visits(segments: Sequence[SegmentEvidence]) -> tuple[tuple[TerminationReference, ...], ...]:
642 """Return the terminations each device visit of the path presents."""
643 if not segments:
644 return ()
645 visits: list[tuple[TerminationReference, ...]] = [(segments[0].left,)]
646 visits.extend((previous.right, following.left) for previous, following in itertools.pairwise(segments))
647 visits.append((segments[-1].right,))
648 return tuple(visits)
651def _corroboration_error(
652 block: _Block,
653 segments: Sequence[SegmentEvidence],
654 visits: Sequence[_ParsedVisit],
655) -> SourceDiagnostic | None:
656 """Return the first contradiction between the Trace List and the path rows."""
657 if not visits or not segments:
658 return None
659 actual = _group_consecutive_visits(visits)
660 expected = _expected_visits(segments)
661 if segments[-1].right.port_class in REAR_PORT_CLASSES and len(actual) == len(expected) - 1:
662 expected = expected[:-1]
663 if len(actual) != len(expected):
664 return _error(
665 block,
666 "trace.corroboration_mismatch",
667 "The Trace List device sequence does not match the Segment Evidence visits.",
668 actual[0][0].row_number,
669 )
670 for actual_group, expected_group in zip(actual, expected, strict=True):
671 allowed = {termination.identity_key for termination in expected_group}
672 if any(visit.termination.identity_key not in allowed for visit in actual_group):
673 return _error(
674 block,
675 "trace.corroboration_mismatch",
676 "A Trace List termination contradicts its Segment Evidence visit.",
677 actual_group[0].row_number,
678 )
679 return None
682def _enrich_termination(
683 termination: TerminationReference,
684 corroboration: Iterable[TerminationReference],
685) -> TerminationReference:
686 """Return the termination with the corroboration values the source states for it."""
687 enriched = termination
688 # A path row is read before a Trace List visit, so an empty cell there must not hide a later one.
689 for candidate in corroboration:
690 if candidate.identity_key == termination.identity_key:
691 enriched = replace(
692 enriched,
693 u_position=enriched.u_position or candidate.u_position,
694 rack=enriched.rack or candidate.rack,
695 location=enriched.location or candidate.location,
696 )
697 return enriched
700def _enrich_summary(
701 summary: EndpointSummary,
702 corroboration: Iterable[TerminationReference],
703) -> EndpointSummary:
704 """Return the Endpoint Summary with corroboration on both endpoints."""
705 corroboration = tuple(corroboration)
706 return replace(
707 summary,
708 from_termination=_enrich_termination(summary.from_termination, corroboration),
709 to_termination=_enrich_termination(summary.to_termination, corroboration),
710 )
713def _provenance(block: _Block, summary: EndpointSummary) -> TraceProvenance:
714 """Return where one block stated the trace, and in which direction."""
715 direction = (
716 "canonical" if summary.from_termination.identity_key <= summary.to_termination.identity_key else "reversed"
717 )
718 return TraceProvenance(
719 workbook_fingerprint=block.workbook_fingerprint,
720 sheet=block.sheet,
721 block_ordinal=block.ordinal,
722 row_start=block.row_start,
723 row_end=block.row_end,
724 export_timestamp=block.export_timestamp,
725 from_text=block.from_text,
726 to_text=block.to_text,
727 direction=direction,
728 )
731def _deduplicate_errors(errors: Iterable[SourceDiagnostic]) -> tuple[SourceDiagnostic, ...]:
732 """Drop a repeated diagnostic, so every distinct finding keeps its own row."""
733 return tuple(dict.fromkeys(errors))
736def _path_trace(
737 path_block: _Block, list_block: _Block | None, port_class_map: Mapping[str, str]
738) -> tuple[SourceTrace | None, tuple[SourceDiagnostic, ...]]:
739 """Return the Source Trace one path block states, with any block-level diagnostic."""
740 summary, errors = _endpoint_summary(path_block)
741 errors.extend(_metadata_errors((path_block, list_block)))
742 if summary is None:
743 return None, tuple(errors)
744 parsed_segments, segment_errors = _parse_segments(path_block)
745 errors.extend(segment_errors)
746 parsed_visits: tuple[_ParsedVisit, ...] = ()
747 if list_block is not None:
748 parsed_visits, visit_errors = _parse_visits(list_block)
749 errors.extend(visit_errors)
750 port_class_labels = _port_class_labels(
751 (
752 summary.from_termination,
753 summary.to_termination,
754 *(end for parsed in parsed_segments for end in (parsed.evidence.left, parsed.evidence.right)),
755 *(visit.termination for visit in parsed_visits),
756 ),
757 port_class_map,
758 )
759 # Every identity key below reads the mapped PortClass, so the map applies before any of them.
760 summary = _mapped_summary(summary, port_class_map)
761 parsed_segments = _mapped_segments(parsed_segments, port_class_map)
762 parsed_visits = _mapped_visits(parsed_visits, port_class_map)
763 endpoint_terms = (
764 (summary.from_termination, path_block.row_start),
765 (summary.to_termination, path_block.row_start + 1),
766 )
767 path_terms = tuple(
768 (termination, parsed.row_number)
769 for parsed in parsed_segments
770 for termination in (parsed.evidence.left, parsed.evidence.right)
771 )
772 visit_terms = tuple((visit.termination, visit.row_number) for visit in parsed_visits)
773 unknown = _unknown_port_class_error(path_block, (*endpoint_terms, *path_terms))
774 if unknown is None and list_block is not None:
775 unknown = _unknown_port_class_error(list_block, visit_terms)
776 if unknown is not None:
777 errors.append(unknown)
778 broken = any(error.code == "trace.incomplete_block" for error in errors)
779 power = summary.from_termination.port_class in POWER_CLASSES
780 if power:
781 summary, parsed_segments, power_error = (
782 (summary, (), None) if broken else _power_evidence(path_block, summary, parsed_segments)
783 )
784 if power_error is not None:
785 errors.append(power_error)
786 broken = broken or power_error is not None
787 linearity = None if broken else _linearity_error(path_block, summary, parsed_segments)
788 if linearity is not None:
789 errors.append(linearity)
790 if not broken and linearity is None:
791 pass_through = _pass_through_error(path_block, parsed_segments)
792 if pass_through is not None:
793 errors.append(pass_through)
794 if list_block is not None:
795 stated = tuple(parsed.evidence for parsed in parsed_segments)
796 mismatch = (
797 _power_corroboration_error(list_block, stated[0], parsed_visits)
798 if power
799 else _corroboration_error(list_block, stated, parsed_visits)
800 )
801 if mismatch is not None:
802 errors.append(mismatch)
803 stated_segments = tuple(parsed.evidence for parsed in parsed_segments)
804 visits = tuple(visit.termination for visit in parsed_visits)
805 if power:
806 visits = _endpoint_visits(summary, visits)
807 path_corroboration = tuple(
808 termination for segment in stated_segments for termination in (segment.left, segment.right)
809 )
810 summary = _enrich_summary(summary, (*path_corroboration, *visits))
811 segments = canonical_orientation(summary.from_termination, summary.to_termination, stated_segments)
812 provenance = [_provenance(path_block, summary)]
813 if list_block is not None:
814 provenance.append(_provenance(list_block, summary))
815 return (
816 SourceTrace(
817 endpoint_summary=summary,
818 segments=segments,
819 pass_through_claims=_pass_through_claims(segments),
820 corroboration=visits,
821 identity=canonical_trace_identity(summary.from_termination, summary.to_termination),
822 content_fingerprint=content_fingerprint(
823 summary.from_termination,
824 summary.to_termination,
825 stated_segments,
826 ),
827 provenance=tuple(provenance),
828 errors=_deduplicate_errors(errors),
829 port_class_labels=port_class_labels,
830 ),
831 (),
832 )
835def _fallback_trace(
836 block: _Block, port_class_map: Mapping[str, str]
837) -> tuple[SourceTrace | None, tuple[SourceDiagnostic, ...]]:
838 """Return the Endpoint Summary fallback an unpaired Trace List block states."""
839 summary, errors = _endpoint_summary(block)
840 errors.extend(_metadata_errors((block,)))
841 visits, visit_errors = _parse_visits(block)
842 errors.extend(visit_errors)
843 if summary is None or not visits:
844 return None, tuple(errors)
845 port_class_labels = _port_class_labels(
846 (summary.from_termination, summary.to_termination, *(visit.termination for visit in visits)),
847 port_class_map,
848 )
849 summary = _mapped_summary(summary, port_class_map)
850 visits = _mapped_visits(visits, port_class_map)
851 terms = (
852 (summary.from_termination, block.row_start),
853 (summary.to_termination, block.row_start + 1),
854 *((visit.termination, visit.row_number) for visit in visits),
855 )
856 unknown = _unknown_port_class_error(block, terms)
857 if unknown is not None:
858 errors.append(unknown)
859 loop = _linearity_error(block, summary, ())
860 if loop is not None:
861 errors.append(loop)
862 corroboration = tuple(visit.termination for visit in visits)
863 if summary.from_termination.port_class in POWER_CLASSES:
864 errors.append(
865 _error(
866 block,
867 "trace.non_linear_path",
868 "A Trace List block states no Segment Evidence, "
869 "so no row names the cable at the power From termination.",
870 )
871 )
872 corroboration = _endpoint_visits(summary, corroboration)
873 summary = _enrich_summary(summary, corroboration)
874 return (
875 SourceTrace(
876 endpoint_summary=summary,
877 segments=(),
878 pass_through_claims=(),
879 corroboration=corroboration,
880 identity=canonical_trace_identity(summary.from_termination, summary.to_termination),
881 content_fingerprint=content_fingerprint(summary.from_termination, summary.to_termination, ()),
882 provenance=(_provenance(block, summary),),
883 errors=_deduplicate_errors(errors),
884 port_class_labels=port_class_labels,
885 ),
886 (),
887 )
890def _trace_selection_key(trace: SourceTrace) -> tuple[bool, bool, str]:
891 """Rank Segment Evidence first, then canonical direction and serialized occurrence content."""
892 summary = trace.endpoint_summary
893 # Provenance and errors carry the block position, and collapse merges the labels of every occurrence.
894 content = {
895 key: value for key, value in asdict(trace).items() if key not in ("provenance", "errors", "port_class_labels")
896 }
897 return (
898 not bool(trace.segments),
899 summary.from_termination.identity_key > summary.to_termination.identity_key,
900 json.dumps(content, ensure_ascii=False, separators=(",", ":"), sort_keys=True),
901 )
904def _collapse_duplicates(traces: Sequence[SourceTrace]) -> tuple[SourceTrace, ...]:
905 """Collapse occurrences that share an identity, and flag differing evidence."""
906 by_identity: dict[str, list[SourceTrace]] = {}
907 for trace in traces:
908 by_identity.setdefault(trace.identity, []).append(trace)
909 collapsed = []
910 for occurrences in by_identity.values():
911 ordered = sorted(occurrences, key=_trace_selection_key)
912 selected = ordered[0]
913 provenance = tuple(dict.fromkeys(item for trace in ordered for item in trace.provenance))
914 # The fingerprint excludes Trace List data, so a later occurrence can state its own finding.
915 errors = _first_of_each_code(ordered)
916 # An endpoint-only fallback states no segments, so it contradicts no segment evidence.
917 compared = [trace for trace in ordered if trace.segments] or ordered
918 if len({trace.content_fingerprint for trace in compared}) > 1:
919 locations = "; ".join(_location_from_provenance(trace.provenance[0]) for trace in compared)
920 errors.append(
921 SourceDiagnostic(
922 code="trace.duplicate_conflict",
923 message=f"Source Trace occurrences have differing evidence: {locations}.",
924 row_number=min(item.row_start for item in provenance),
925 )
926 )
927 labels = tuple(sorted({label for trace in ordered for label in trace.port_class_labels}))
928 collapsed.append(
929 replace(selected, provenance=provenance, errors=_deduplicate_errors(errors), port_class_labels=labels)
930 )
931 return tuple(collapsed)
934def _first_of_each_code(occurrences: Sequence[SourceTrace]) -> list[SourceDiagnostic]:
935 """Return every occurrence's findings, reporting one repeated condition once."""
936 errors: list[SourceDiagnostic] = []
937 seen_codes: set[str] = set()
938 for occurrence in occurrences:
939 for error in occurrence.errors:
940 if error.code in seen_codes and occurrence is not occurrences[0]:
941 continue
942 seen_codes.add(error.code)
943 errors.append(error)
944 return errors
947def _location_from_provenance(provenance: TraceProvenance) -> str:
948 """Return the readable source position one provenance record holds."""
949 return f"{provenance.sheet} block {provenance.block_ordinal} (rows {provenance.row_start}-{provenance.row_end})"
952def _claimed_endpoints(trace: SourceTrace) -> tuple[TerminationReference, ...]:
953 """Return the endpoints a trace claims, which leaves out the To leaf of a power block with no cable."""
954 summary = trace.endpoint_summary
955 if summary.from_termination.port_class in POWER_CLASSES and not trace.segments:
956 return (summary.from_termination,)
957 return summary.from_termination, summary.to_termination
960def _endpoint_claims(occurrences: Iterable[SourceTrace]) -> dict[str, set[IdentityKey]]:
961 """Return the endpoints every occurrence of each identity claims, before collapse keeps one summary."""
962 claims: dict[str, set[IdentityKey]] = defaultdict(set)
963 for occurrence in occurrences:
964 claims[occurrence.identity].update(endpoint.identity_key for endpoint in _claimed_endpoints(occurrence))
965 return claims
968def _cross_trace_conflicts(
969 traces: Sequence[SourceTrace], endpoint_claims: Mapping[str, set[IdentityKey]]
970) -> tuple[SourceTrace, ...]:
971 """Flag every trace that claims a termination another Source Trace claims for another segment."""
972 termination_claims: dict[IdentityKey, dict[int, set[_SegmentClaim]]] = defaultdict(lambda: defaultdict(set))
973 for index, trace in enumerate(traces):
974 claims: dict[IdentityKey, set[_SegmentClaim]] = defaultdict(set)
975 for segment in trace.segments:
976 ordered_pair = sorted((segment.left.identity_key, segment.right.identity_key))
977 # Two labels for one pair can resolve to one policy, so the label decides nothing here.
978 segment_pair = ordered_pair[0], ordered_pair[1]
979 claims[segment.left.identity_key].add(segment_pair)
980 claims[segment.right.identity_key].add(segment_pair)
981 for endpoint in endpoint_claims[trace.identity]:
982 claims.setdefault(endpoint, set())
983 for termination, trace_claims in claims.items():
984 termination_claims[termination][index].update(trace_claims)
985 conflicts: dict[int, set[str]] = defaultdict(set)
986 for claims_by_owner in termination_claims.values():
987 owner_claims = list(claims_by_owner.values())
988 identical_shared_segment = bool(owner_claims[0]) and all(claims == owner_claims[0] for claims in owner_claims)
989 if len(claims_by_owner) > 1 and not identical_shared_segment:
990 for owner in claims_by_owner:
991 conflicts[owner].add("a termination is claimed by another Source Trace")
992 checked = []
993 for index, trace in enumerate(traces):
994 if index not in conflicts:
995 checked.append(trace)
996 continue
997 provenance = trace.provenance[0]
998 detail = "; ".join(sorted(conflicts[index]))
999 error = SourceDiagnostic(
1000 code="trace.cross_trace_conflict",
1001 message=f"{_location_from_provenance(provenance)}: {detail}.",
1002 row_number=provenance.row_start,
1003 )
1004 checked.append(replace(trace, errors=_deduplicate_errors((*trace.errors, error))))
1005 return tuple(checked)
1008def interpret(
1009 content: bytes, config: TraceWorkbookConfig
1010) -> tuple[tuple[SourceTrace, ...], tuple[SourceDiagnostic, ...]]:
1011 """Return typed Source Traces and source diagnostics from workbook bytes."""
1012 try:
1013 book = openpyxl.load_workbook(BytesIO(content), data_only=True)
1014 except Exception as exc:
1015 raise SourceUnreadable(f"Cannot open Excel file: {exc}") from exc
1016 try:
1017 recognized = [name for name in (TRACE_PATH_SHEET, TRACE_LIST_SHEET) if name in book.sheetnames]
1018 if not recognized:
1019 diagnostic = SourceDiagnostic(
1020 code="trace.no_recognized_sheet",
1021 message=f"Workbook has neither '{TRACE_PATH_SHEET}' nor '{TRACE_LIST_SHEET}'.",
1022 )
1023 return (), (diagnostic,)
1024 workbook_fingerprint = sha256(content).hexdigest()
1025 path_blocks = (
1026 _extract_blocks(book[TRACE_PATH_SHEET], workbook_fingerprint) if TRACE_PATH_SHEET in recognized else ()
1027 )
1028 list_blocks = (
1029 _extract_blocks(book[TRACE_LIST_SHEET], workbook_fingerprint) if TRACE_LIST_SHEET in recognized else ()
1030 )
1031 finally:
1032 # close() releases an archive only in read-only mode, but it stays paired with the load.
1033 book.close()
1034 lists_by_pair: dict[tuple[str, str], list[_Block]] = defaultdict(list)
1035 for block in list_blocks:
1036 lists_by_pair[block.pair_key].append(block)
1037 traces: list[SourceTrace] = []
1038 diagnostics: list[SourceDiagnostic] = []
1039 paired_lists: set[tuple[str, int]] = set()
1040 for path_block in path_blocks:
1041 candidates = lists_by_pair[path_block.pair_key]
1042 list_block = candidates.pop(0) if candidates else None
1043 if list_block is not None:
1044 paired_lists.add((list_block.sheet, list_block.ordinal))
1045 trace, block_diagnostics = _path_trace(path_block, list_block, config.port_class_map)
1046 diagnostics.extend(block_diagnostics)
1047 if trace is not None:
1048 traces.append(trace)
1049 for block in list_blocks:
1050 if (block.sheet, block.ordinal) in paired_lists:
1051 continue
1052 trace, block_diagnostics = _fallback_trace(block, config.port_class_map)
1053 diagnostics.extend(block_diagnostics)
1054 if trace is not None:
1055 traces.append(trace)
1056 checked = _cross_trace_conflicts(_collapse_duplicates(traces), _endpoint_claims(traces))
1057 diagnostics.extend(error for trace in checked for error in trace.errors)
1058 return checked, tuple(diagnostics)
1061__all__ = (
1062 "FRONT_PORT_CLASSES",
1063 "INTERFACE_PORT_CLASSES",
1064 "PORT_CLASSES",
1065 "REAR_PORT_CLASSES",
1066 "TraceWorkbookConfig",
1067 "canonical_orientation",
1068 "canonical_trace_identity",
1069 "content_fingerprint",
1070 "interpret",
1071 "parse_endpoint_line",
1072)