Coverage for netbox_data_import/trace_workbook.py: 96%
489 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +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, 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_CLASS_CLAIMED_KINDS,
22 PORT_CLASSES,
23 REAR_PORT_CLASSES,
24 same_device_and_cards,
25)
26from .trace_schema import TRACE_EXPORT_TIMESTAMP_MAX_LENGTH
27from .values import identity_text, source_text
29TRACE_PATH_SHEET = "Trace From To"
30TRACE_LIST_SHEET = "Trace List"
32IdentityKey = tuple[str, str, str, str]
34_PATH_HEADER = (
35 "Port",
36 "PortClass",
37 "Cards",
38 "Device",
39 "UPos",
40 "Rack",
41 "Location",
42 "CableClass",
43 "Port",
44 "PortClass",
45 "Cards",
46 "Device",
47 "UPos",
48 "Rack",
49 "Location",
50)
51_LIST_HEADER = ("Location", "Rack", "UPos", "Device", "Cards", "Port", "PortClass", "Cable")
52_BLOCK_LINE_MARKERS = ("From", "To")
55@dataclass(frozen=True)
56class TerminationReference:
57 """Name one source termination and retain its corroboration values."""
59 device: str
60 cards: str
61 port: str
62 port_class: str
63 u_position: str = ""
64 rack: str = ""
65 location: str = ""
67 @property
68 def identity_key(self) -> IdentityKey:
69 """Return the normalized device, cards, port, and claimed-kind identity."""
70 return (
71 identity_text(self.device),
72 identity_text(self.cards),
73 identity_text(self.port),
74 # An unrecognized PortClass keeps its own text, so two unknown values stay apart.
75 PORT_CLASS_CLAIMED_KINDS.get(self.port_class, identity_text(self.port_class)),
76 )
79@dataclass(frozen=True)
80class EndpointSummary:
81 """Retain the source From and To statements in their original direction."""
83 from_termination: TerminationReference
84 to_termination: TerminationReference
85 from_text: str
86 to_text: str
89@dataclass(frozen=True)
90class SegmentEvidence:
91 """Describe one source-claimed cable between two Termination References."""
93 left: TerminationReference
94 cable_class: str
95 right: TerminationReference
98@dataclass(frozen=True)
99class PassThroughClaim:
100 """Describe the source-claimed continuation through one device."""
102 device: str
103 cards: str
104 entry_port: str
105 exit_port: str
108@dataclass(frozen=True)
109class TraceProvenance:
110 """Locate one Source Trace occurrence in its workbook."""
112 workbook_fingerprint: str
113 sheet: str
114 block_ordinal: int
115 row_start: int
116 row_end: int
117 export_timestamp: str
118 from_text: str
119 to_text: str
120 direction: str
123@dataclass(frozen=True)
124class SourceTrace:
125 """Carry canonical Segment Evidence for one path or an Endpoint Summary fallback."""
127 endpoint_summary: EndpointSummary
128 segments: tuple[SegmentEvidence, ...]
129 pass_through_claims: tuple[PassThroughClaim, ...]
130 corroboration: tuple[TerminationReference, ...]
131 identity: str
132 content_fingerprint: str
133 provenance: tuple[TraceProvenance, ...]
134 errors: tuple[SourceDiagnostic, ...] = ()
136 @property
137 def valid(self) -> bool:
138 """Return whether source validation found no errors on this trace."""
139 return not self.errors
141 @property
142 def ends_at_rear_port(self) -> bool:
143 """Return whether the original To termination is a rear port."""
144 return self.endpoint_summary.to_termination.port_class in REAR_PORT_CLASSES
147@dataclass(frozen=True)
148class _RawRow:
149 """One spreadsheet row, holding the cells the sheet width defines."""
151 row_number: int
152 values: tuple[object, ...]
155@dataclass(frozen=True)
156class _Block:
157 """One From and To block with its header row, data rows, and workbook provenance."""
159 sheet: str
160 ordinal: int
161 row_start: int
162 row_end: int
163 from_text: str
164 to_text: str
165 has_to_line: bool
166 header: tuple[str, ...]
167 rows: tuple[_RawRow, ...]
168 export_timestamp: str
169 workbook_fingerprint: str
171 @property
172 def pair_key(self) -> tuple[str, str]:
173 """Return the From and To text that pairs this block with the other sheet."""
174 return source_text(self.from_text), source_text(self.to_text)
177@dataclass(frozen=True)
178class _ParsedSegment:
179 """One Segment Evidence entry with the row that stated it."""
181 evidence: SegmentEvidence
182 row_number: int
185@dataclass(frozen=True)
186class _ParsedVisit:
187 """One Trace List visit with the row that stated it."""
189 termination: TerminationReference
190 row_number: int
193_SegmentClaim = tuple[tuple[IdentityKey, IdentityKey], str]
196def parse_endpoint_line(line: str) -> TerminationReference:
197 """Parse ``Device > [Cards > ]Port (PortClass)`` into a Termination Reference."""
198 parts = source_text(line).split(" > ")
199 if len(parts) not in (2, 3):
200 raise ValueError("Endpoint line must contain a device, an optional cards label, and a port.")
201 device = source_text(parts[0])
202 cards = source_text(parts[1]) if len(parts) == 3 else ""
203 port_and_class = source_text(parts[-1])
204 marker = port_and_class.rfind(" (")
205 if marker <= 0 or not port_and_class.endswith(")"):
206 raise ValueError("Endpoint line must end with a PortClass in parentheses.")
207 port = source_text(port_and_class[:marker])
208 port_class = source_text(port_and_class[marker + 2 : -1])
209 if not device or not port or not port_class:
210 raise ValueError("Endpoint line contains an empty device, port, or PortClass.")
211 return TerminationReference(device=device, cards=cards, port=port, port_class=port_class)
214def canonical_trace_identity(
215 first: TerminationReference,
216 second: TerminationReference,
217) -> str:
218 """Return the sorted endpoint pair as compact canonical JSON."""
219 endpoints = sorted((first.identity_key, second.identity_key))
220 return json.dumps(endpoints, ensure_ascii=False, separators=(",", ":"))
223def canonical_orientation(
224 from_termination: TerminationReference,
225 to_termination: TerminationReference,
226 segments: Sequence[SegmentEvidence],
227) -> tuple[SegmentEvidence, ...]:
228 """Orient Segment Evidence from the endpoint whose identity key sorts first."""
229 if from_termination.identity_key <= to_termination.identity_key:
230 return tuple(segments)
231 return tuple(
232 SegmentEvidence(left=segment.right, cable_class=segment.cable_class, right=segment.left)
233 for segment in reversed(segments)
234 )
237def content_fingerprint(
238 from_termination: TerminationReference,
239 to_termination: TerminationReference,
240 segments: Sequence[SegmentEvidence],
241) -> str:
242 """Hash the canonical endpoints, Segment Evidence, and Pass-Through Claims."""
243 oriented = canonical_orientation(from_termination, to_termination, segments)
244 claims = _pass_through_claims(oriented)
245 endpoints = sorted((from_termination, to_termination), key=lambda termination: termination.identity_key)
246 payload = {
247 # An Endpoint Summary fallback states no segment, so the endpoints carry its whole content.
248 "endpoints": [[termination.identity_key, termination.port_class] for termination in endpoints],
249 "segments": [
250 [
251 segment.left.identity_key,
252 segment.left.port_class,
253 segment.cable_class,
254 segment.right.identity_key,
255 segment.right.port_class,
256 ]
257 for segment in oriented
258 ],
259 "pass_through_claims": [
260 [
261 identity_text(claim.device),
262 identity_text(claim.cards),
263 identity_text(claim.entry_port),
264 identity_text(claim.exit_port),
265 ]
266 for claim in claims
267 ],
268 }
269 serialized = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
270 return sha256(serialized.encode()).hexdigest()
273def _cell_text(value: object) -> str:
274 """Return the raw text of a cell value, empty for None."""
275 return "" if value is None else str(value)
278def _normalized_header(values: Iterable[object]) -> tuple[str, ...]:
279 """Return one header row as trimmed text."""
280 return tuple(source_text(value) for value in values)
283def _row_has_data(values: Iterable[object]) -> bool:
284 """Return whether any cell in the row carries text."""
285 return any(source_text(value) for value in values)
288def _row_values(sheet, row_number: int, width: int) -> tuple[object, ...]:
289 """Return the cells of one row across the sheet's fixed width."""
290 return tuple(sheet.cell(row=row_number, column=column).value for column in range(1, width + 1))
293def _last_data_row(values_by_row: Mapping[int, tuple[object, ...]], start: int, stop: int) -> int:
294 """Return the last row in the range that carries data."""
295 for row_number in range(stop, start - 1, -1):
296 if _row_has_data(values_by_row[row_number]):
297 return row_number
298 return start
301def _extract_blocks(sheet, workbook_fingerprint: str) -> tuple[_Block, ...]:
302 """Return every block on one sheet, including visits the next block's lines overwrite."""
303 width = len(_PATH_HEADER) if sheet.title == TRACE_PATH_SHEET else len(_LIST_HEADER)
304 # openpyxl builds a cell object on first access, so the sheet is read once and every step reuses it.
305 values_by_row = {row: _row_values(sheet, row, width) for row in range(1, sheet.max_row + 1)}
306 starts = [row for row, values in values_by_row.items() if source_text(values[0]) == "From"]
307 first_row = values_by_row.get(1, ())
308 export_timestamp = source_text(first_row[1]) if first_row and source_text(first_row[0]) == "Executed" else ""
309 blocks = []
310 for index, start in enumerate(starts):
311 boundary = starts[index + 1] if index + 1 < len(starts) else sheet.max_row + 1
312 to_row = start + 1
313 has_to_line = to_row < boundary and source_text(values_by_row[to_row][0]) == "To"
314 header_row = start + 2
315 header = _normalized_header(values_by_row[header_row]) if header_row < boundary else ()
316 rows = [
317 _RawRow(row_number, values_by_row[row_number])
318 for row_number in range(start + 3, boundary)
319 if _row_has_data(values_by_row[row_number])
320 ]
321 row_end = _last_data_row(values_by_row, start, boundary - 1)
322 # A block that fills its separator row keeps its last rows under the next block's lines.
323 for carry_row in (boundary, boundary + 1):
324 if carry_row > sheet.max_row:
325 continue
326 values = values_by_row[carry_row]
327 if source_text(values[0]) not in _BLOCK_LINE_MARKERS:
328 continue
329 if _row_has_data(values[2:]):
330 rows.append(_RawRow(carry_row, ("", "", *values[2:])))
331 row_end = carry_row
332 blocks.append(
333 _Block(
334 sheet=sheet.title,
335 ordinal=index + 1,
336 row_start=start,
337 row_end=row_end,
338 from_text=_cell_text(values_by_row[start][1]),
339 to_text=_cell_text(values_by_row[to_row][1]) if has_to_line else "",
340 has_to_line=has_to_line,
341 header=header,
342 rows=tuple(rows),
343 export_timestamp=export_timestamp,
344 workbook_fingerprint=workbook_fingerprint,
345 )
346 )
347 return tuple(blocks)
350def _location(block: _Block) -> str:
351 """Return the readable source position of one block."""
352 return f"{block.sheet} block {block.ordinal} (rows {block.row_start}-{block.row_end})"
355def _error(block: _Block, code: str, detail: str, row_number: int | None = None) -> SourceDiagnostic:
356 """Return one source diagnostic that names the block it came from."""
357 return SourceDiagnostic(
358 code=code,
359 message=f"{_location(block)}: {detail}",
360 row_number=block.row_start if row_number is None else row_number,
361 )
364def _metadata_errors(blocks: Iterable[_Block | None]) -> list[SourceDiagnostic]:
365 """Return diagnostics for raw workbook metadata that cannot fit its stored representation."""
366 errors = []
367 for block in blocks:
368 if block is not None and len(block.export_timestamp) > TRACE_EXPORT_TIMESTAMP_MAX_LENGTH:
369 errors.append(
370 _error(
371 block,
372 "trace.metadata_too_long",
373 f"The export timestamp exceeds {TRACE_EXPORT_TIMESTAMP_MAX_LENGTH} characters.",
374 row_number=1,
375 )
376 )
377 return errors
380def _endpoint_summary(block: _Block) -> tuple[EndpointSummary | None, list[SourceDiagnostic]]:
381 """Return the Endpoint Summary of one block, or the reason it has none."""
382 errors = []
383 if not block.has_to_line:
384 errors.append(_error(block, "trace.incomplete_block", "The block has no To line."))
385 return None, errors
386 try:
387 from_termination = parse_endpoint_line(block.from_text)
388 to_termination = parse_endpoint_line(block.to_text)
389 except ValueError as exc:
390 errors.append(_error(block, "trace.incomplete_block", str(exc)))
391 return None, errors
392 return (
393 EndpointSummary(
394 from_termination=from_termination,
395 to_termination=to_termination,
396 from_text=block.from_text,
397 to_text=block.to_text,
398 ),
399 errors,
400 )
403def _termination_from_path(values: tuple[object, ...], offset: int) -> TerminationReference:
404 """Return the Termination Reference one side of a Segment Evidence row states."""
405 termination = TerminationReference(
406 port=source_text(values[offset]),
407 port_class=source_text(values[offset + 1]),
408 cards=source_text(values[offset + 2]),
409 device=source_text(values[offset + 3]),
410 u_position=source_text(values[offset + 4]),
411 rack=source_text(values[offset + 5]),
412 location=source_text(values[offset + 6]),
413 )
414 if not termination.device or not termination.port or not termination.port_class:
415 raise ValueError("A Segment Evidence row has an empty device, port, or PortClass.")
416 return termination
419def _parse_segments(block: _Block) -> tuple[tuple[_ParsedSegment, ...], list[SourceDiagnostic]]:
420 """Return the Segment Evidence a path block states, and the rows it could not read."""
421 if block.header != _PATH_HEADER:
422 return (), [_error(block, "trace.incomplete_block", "The block has no recognized Segment Evidence header.")]
423 parsed = []
424 errors = []
425 for row in block.rows:
426 try:
427 left = _termination_from_path(row.values, 0)
428 cable_class = source_text(row.values[7])
429 right = _termination_from_path(row.values, 8)
430 except (IndexError, ValueError) as exc:
431 errors.append(_error(block, "trace.incomplete_block", str(exc), row.row_number))
432 continue
433 if not cable_class:
434 errors.append(
435 _error(
436 block, "trace.incomplete_block", "A Segment Evidence row has an empty CableClass.", row.row_number
437 )
438 )
439 continue
440 parsed.append(_ParsedSegment(SegmentEvidence(left, cable_class, right), row.row_number))
441 return tuple(parsed), errors
444def _parse_visits(block: _Block) -> tuple[tuple[_ParsedVisit, ...], list[SourceDiagnostic]]:
445 """Return the visits a Trace List block states, and the rows it could not read."""
446 if block.header != _LIST_HEADER:
447 return (), [_error(block, "trace.incomplete_block", "The block has no recognized Trace List header.")]
448 parsed = []
449 errors = []
450 for row in block.rows:
451 try:
452 termination = TerminationReference(
453 location=source_text(row.values[0]),
454 rack=source_text(row.values[1]),
455 u_position=source_text(row.values[2]),
456 device=source_text(row.values[3]),
457 cards=source_text(row.values[4]),
458 port=source_text(row.values[5]),
459 port_class=source_text(row.values[6]),
460 )
461 except IndexError:
462 termination = TerminationReference("", "", "", "")
463 if not termination.device or not termination.port or not termination.port_class:
464 errors.append(
465 _error(
466 block,
467 "trace.corroboration_mismatch",
468 "A Trace List visit has an empty device, port, or PortClass.",
469 row.row_number,
470 )
471 )
472 continue
473 parsed.append(_ParsedVisit(termination, row.row_number))
474 return tuple(parsed), errors
477def _unknown_port_class_error(
478 block: _Block,
479 terminations: Iterable[tuple[TerminationReference, int]],
480) -> SourceDiagnostic | None:
481 """Return the first PortClass outside the fixed vocabulary as a diagnostic."""
482 for termination, row_number in terminations:
483 if termination.port_class not in PORT_CLASSES:
484 return _error(
485 block,
486 "trace.unknown_port_class",
487 f"PortClass '{termination.port_class}' is outside the fixed vocabulary.",
488 row_number,
489 )
490 return None
493def _linearity_error(
494 block: _Block, summary: EndpointSummary, segments: Sequence[_ParsedSegment]
495) -> SourceDiagnostic | None:
496 """Return the first structural break in a path, or None when the path is linear."""
497 # A path states two terminations whether or not it states the segments between them.
498 if summary.from_termination.identity_key == summary.to_termination.identity_key:
499 return _error(
500 block,
501 "trace.non_linear_path",
502 "The From and To lines name one termination, so the path does not join two terminations.",
503 segments[-1].row_number if segments else block.row_start,
504 )
505 if not segments:
506 return None
507 if segments[0].evidence.left.identity_key != summary.from_termination.identity_key:
508 return _error(
509 block,
510 "trace.non_linear_path",
511 "The first Segment Evidence termination contradicts the From line.",
512 segments[0].row_number,
513 )
514 if segments[-1].evidence.right.identity_key != summary.to_termination.identity_key:
515 return _error(
516 block,
517 "trace.non_linear_path",
518 "The last Segment Evidence termination contradicts the To line.",
519 segments[-1].row_number,
520 )
521 seen_segments = set()
522 left_destinations: dict[IdentityKey, set[IdentityKey]] = defaultdict(set)
523 right_sources: dict[IdentityKey, set[IdentityKey]] = defaultdict(set)
524 for parsed in segments:
525 left_key = parsed.evidence.left.identity_key
526 right_key = parsed.evidence.right.identity_key
527 # One cable has no direction, so a reversed row restates the segment rather than adding one.
528 pair = tuple(sorted((left_key, right_key)))
529 if pair in seen_segments:
530 return _error(
531 block,
532 "trace.non_linear_path",
533 "The path repeats a Segment Evidence row.",
534 parsed.row_number,
535 )
536 seen_segments.add(pair)
537 left_destinations[left_key].add(right_key)
538 right_sources[right_key].add(left_key)
539 if len(left_destinations[left_key]) > 1 or len(right_sources[right_key]) > 1:
540 return _error(
541 block,
542 "trace.non_linear_path",
543 "The Segment Evidence rows branch.",
544 parsed.row_number,
545 )
546 for previous, following in itertools.pairwise(segments):
547 if not same_device_and_cards(previous.evidence.right, following.evidence.left):
548 return _error(
549 block,
550 "trace.non_linear_path",
551 "Consecutive Segment Evidence rows do not share a device and cards label.",
552 following.row_number,
553 )
554 return None
557def _pass_through_claims(segments: Sequence[SegmentEvidence]) -> tuple[PassThroughClaim, ...]:
558 """Return the continuation each pair of consecutive segments claims."""
559 claims = []
560 for previous, following in itertools.pairwise(segments):
561 if not same_device_and_cards(previous.right, following.left):
562 continue
563 claims.append(
564 PassThroughClaim(
565 device=previous.right.device,
566 cards=previous.right.cards,
567 entry_port=previous.right.port,
568 exit_port=following.left.port,
569 )
570 )
571 return tuple(claims)
574def _pass_through_error(block: _Block, segments: Sequence[_ParsedSegment]) -> SourceDiagnostic | None:
575 """Return a diagnostic for the first Pass-Through Claim at an interface PortClass."""
576 for previous, following in itertools.pairwise(segments):
577 if not same_device_and_cards(previous.evidence.right, following.evidence.left):
578 continue
579 if (
580 previous.evidence.right.port_class in INTERFACE_PORT_CLASSES
581 or following.evidence.left.port_class in INTERFACE_PORT_CLASSES
582 ):
583 return _error(
584 block,
585 "trace.pass_through_at_interface",
586 "A Pass-Through Claim enters or exits through an interface PortClass.",
587 following.row_number,
588 )
589 return None
592def _group_consecutive_visits(visits: Sequence[_ParsedVisit]) -> tuple[tuple[_ParsedVisit, ...], ...]:
593 """Group Trace List visits into one entry per visited device."""
594 groups: list[list[_ParsedVisit]] = []
595 for visit in visits:
596 device = identity_text(visit.termination.device)
597 if not groups or identity_text(groups[-1][0].termination.device) != device:
598 groups.append([])
599 groups[-1].append(visit)
600 return tuple(tuple(group) for group in groups)
603def _expected_visits(segments: Sequence[SegmentEvidence]) -> tuple[tuple[TerminationReference, ...], ...]:
604 """Return the terminations each device visit of the path presents."""
605 if not segments:
606 return ()
607 visits: list[tuple[TerminationReference, ...]] = [(segments[0].left,)]
608 visits.extend((previous.right, following.left) for previous, following in itertools.pairwise(segments))
609 visits.append((segments[-1].right,))
610 return tuple(visits)
613def _corroboration_error(
614 block: _Block,
615 segments: Sequence[SegmentEvidence],
616 visits: Sequence[_ParsedVisit],
617) -> SourceDiagnostic | None:
618 """Return the first contradiction between the Trace List and the path rows."""
619 if not visits or not segments:
620 return None
621 actual = _group_consecutive_visits(visits)
622 expected = _expected_visits(segments)
623 if segments[-1].right.port_class in REAR_PORT_CLASSES and len(actual) == len(expected) - 1:
624 expected = expected[:-1]
625 if len(actual) != len(expected):
626 return _error(
627 block,
628 "trace.corroboration_mismatch",
629 "The Trace List device sequence does not match the Segment Evidence visits.",
630 actual[0][0].row_number,
631 )
632 for actual_group, expected_group in zip(actual, expected, strict=True):
633 allowed = {termination.identity_key for termination in expected_group}
634 if any(visit.termination.identity_key not in allowed for visit in actual_group):
635 return _error(
636 block,
637 "trace.corroboration_mismatch",
638 "A Trace List termination contradicts its Segment Evidence visit.",
639 actual_group[0].row_number,
640 )
641 return None
644def _enrich_termination(
645 termination: TerminationReference,
646 corroboration: Iterable[TerminationReference],
647) -> TerminationReference:
648 """Return the termination with the corroboration values the source states for it."""
649 enriched = termination
650 # A path row is read before a Trace List visit, so an empty cell there must not hide a later one.
651 for candidate in corroboration:
652 if candidate.identity_key == termination.identity_key:
653 enriched = replace(
654 enriched,
655 u_position=enriched.u_position or candidate.u_position,
656 rack=enriched.rack or candidate.rack,
657 location=enriched.location or candidate.location,
658 )
659 return enriched
662def _enrich_summary(
663 summary: EndpointSummary,
664 corroboration: Iterable[TerminationReference],
665) -> EndpointSummary:
666 """Return the Endpoint Summary with corroboration on both endpoints."""
667 corroboration = tuple(corroboration)
668 return replace(
669 summary,
670 from_termination=_enrich_termination(summary.from_termination, corroboration),
671 to_termination=_enrich_termination(summary.to_termination, corroboration),
672 )
675def _provenance(block: _Block, summary: EndpointSummary) -> TraceProvenance:
676 """Return where one block stated the trace, and in which direction."""
677 direction = (
678 "canonical" if summary.from_termination.identity_key <= summary.to_termination.identity_key else "reversed"
679 )
680 return TraceProvenance(
681 workbook_fingerprint=block.workbook_fingerprint,
682 sheet=block.sheet,
683 block_ordinal=block.ordinal,
684 row_start=block.row_start,
685 row_end=block.row_end,
686 export_timestamp=block.export_timestamp,
687 from_text=block.from_text,
688 to_text=block.to_text,
689 direction=direction,
690 )
693def _deduplicate_errors(errors: Iterable[SourceDiagnostic]) -> tuple[SourceDiagnostic, ...]:
694 """Drop a repeated diagnostic, so every distinct finding keeps its own row."""
695 return tuple(dict.fromkeys(errors))
698def _path_trace(
699 path_block: _Block, list_block: _Block | None
700) -> tuple[SourceTrace | None, tuple[SourceDiagnostic, ...]]:
701 """Return the Source Trace one path block states, with any block-level diagnostic."""
702 summary, errors = _endpoint_summary(path_block)
703 errors.extend(_metadata_errors((path_block, list_block)))
704 if summary is None:
705 return None, tuple(errors)
706 parsed_segments, segment_errors = _parse_segments(path_block)
707 errors.extend(segment_errors)
708 parsed_visits: tuple[_ParsedVisit, ...] = ()
709 if list_block is not None:
710 parsed_visits, visit_errors = _parse_visits(list_block)
711 errors.extend(visit_errors)
712 endpoint_terms = (
713 (summary.from_termination, path_block.row_start),
714 (summary.to_termination, path_block.row_start + 1),
715 )
716 path_terms = tuple(
717 (termination, parsed.row_number)
718 for parsed in parsed_segments
719 for termination in (parsed.evidence.left, parsed.evidence.right)
720 )
721 visit_terms = tuple((visit.termination, visit.row_number) for visit in parsed_visits)
722 unknown = _unknown_port_class_error(path_block, (*endpoint_terms, *path_terms))
723 if unknown is None and list_block is not None:
724 unknown = _unknown_port_class_error(list_block, visit_terms)
725 if unknown is not None:
726 errors.append(unknown)
727 incomplete = any(error.code == "trace.incomplete_block" for error in errors)
728 linearity = None if incomplete else _linearity_error(path_block, summary, parsed_segments)
729 if linearity is not None:
730 errors.append(linearity)
731 if not incomplete and linearity is None:
732 pass_through = _pass_through_error(path_block, parsed_segments)
733 if pass_through is not None:
734 errors.append(pass_through)
735 if list_block is not None:
736 mismatch = _corroboration_error(
737 list_block,
738 tuple(parsed.evidence for parsed in parsed_segments),
739 parsed_visits,
740 )
741 if mismatch is not None:
742 errors.append(mismatch)
743 stated_segments = tuple(parsed.evidence for parsed in parsed_segments)
744 visits = tuple(visit.termination for visit in parsed_visits)
745 path_corroboration = tuple(
746 termination for segment in stated_segments for termination in (segment.left, segment.right)
747 )
748 summary = _enrich_summary(summary, (*path_corroboration, *visits))
749 segments = canonical_orientation(summary.from_termination, summary.to_termination, stated_segments)
750 provenance = [_provenance(path_block, summary)]
751 if list_block is not None:
752 provenance.append(_provenance(list_block, summary))
753 return (
754 SourceTrace(
755 endpoint_summary=summary,
756 segments=segments,
757 pass_through_claims=_pass_through_claims(segments),
758 corroboration=visits,
759 identity=canonical_trace_identity(summary.from_termination, summary.to_termination),
760 content_fingerprint=content_fingerprint(
761 summary.from_termination,
762 summary.to_termination,
763 stated_segments,
764 ),
765 provenance=tuple(provenance),
766 errors=_deduplicate_errors(errors),
767 ),
768 (),
769 )
772def _fallback_trace(block: _Block) -> tuple[SourceTrace | None, tuple[SourceDiagnostic, ...]]:
773 """Return the Endpoint Summary fallback an unpaired Trace List block states."""
774 summary, errors = _endpoint_summary(block)
775 errors.extend(_metadata_errors((block,)))
776 visits, visit_errors = _parse_visits(block)
777 errors.extend(visit_errors)
778 if summary is None or not visits:
779 return None, tuple(errors)
780 terms = (
781 (summary.from_termination, block.row_start),
782 (summary.to_termination, block.row_start + 1),
783 *((visit.termination, visit.row_number) for visit in visits),
784 )
785 unknown = _unknown_port_class_error(block, terms)
786 if unknown is not None:
787 errors.append(unknown)
788 loop = _linearity_error(block, summary, ())
789 if loop is not None:
790 errors.append(loop)
791 corroboration = tuple(visit.termination for visit in visits)
792 summary = _enrich_summary(summary, corroboration)
793 return (
794 SourceTrace(
795 endpoint_summary=summary,
796 segments=(),
797 pass_through_claims=(),
798 corroboration=corroboration,
799 identity=canonical_trace_identity(summary.from_termination, summary.to_termination),
800 content_fingerprint=content_fingerprint(summary.from_termination, summary.to_termination, ()),
801 provenance=(_provenance(block, summary),),
802 errors=_deduplicate_errors(errors),
803 ),
804 (),
805 )
808def _trace_selection_key(trace: SourceTrace) -> tuple[bool, bool, str]:
809 """Rank Segment Evidence first, then canonical direction and serialized occurrence content."""
810 summary = trace.endpoint_summary
811 # Provenance and errors carry the block position, which states where a trace was read, not what it says.
812 content = replace(trace, provenance=(), errors=())
813 return (
814 not bool(trace.segments),
815 summary.from_termination.identity_key > summary.to_termination.identity_key,
816 json.dumps(asdict(content), ensure_ascii=False, separators=(",", ":"), sort_keys=True),
817 )
820def _collapse_duplicates(traces: Sequence[SourceTrace]) -> tuple[SourceTrace, ...]:
821 """Collapse occurrences that share an identity, and flag differing evidence."""
822 by_identity: dict[str, list[SourceTrace]] = {}
823 for trace in traces:
824 by_identity.setdefault(trace.identity, []).append(trace)
825 collapsed = []
826 for occurrences in by_identity.values():
827 ordered = sorted(occurrences, key=_trace_selection_key)
828 selected = ordered[0]
829 provenance = tuple(dict.fromkeys(item for trace in ordered for item in trace.provenance))
830 # The fingerprint excludes Trace List data, so a later occurrence can state its own finding.
831 errors = _first_of_each_code(ordered)
832 # An endpoint-only fallback states no segments, so it contradicts no segment evidence.
833 compared = [trace for trace in ordered if trace.segments] or ordered
834 if len({trace.content_fingerprint for trace in compared}) > 1:
835 locations = "; ".join(_location_from_provenance(trace.provenance[0]) for trace in compared)
836 errors.append(
837 SourceDiagnostic(
838 code="trace.duplicate_conflict",
839 message=f"Source Trace occurrences have differing evidence: {locations}.",
840 row_number=min(item.row_start for item in provenance),
841 )
842 )
843 collapsed.append(replace(selected, provenance=provenance, errors=_deduplicate_errors(errors)))
844 return tuple(collapsed)
847def _first_of_each_code(occurrences: Sequence[SourceTrace]) -> list[SourceDiagnostic]:
848 """Return every occurrence's findings, reporting one repeated condition once."""
849 errors: list[SourceDiagnostic] = []
850 seen_codes: set[str] = set()
851 for occurrence in occurrences:
852 for error in occurrence.errors:
853 if error.code in seen_codes and occurrence is not occurrences[0]:
854 continue
855 seen_codes.add(error.code)
856 errors.append(error)
857 return errors
860def _location_from_provenance(provenance: TraceProvenance) -> str:
861 """Return the readable source position one provenance record holds."""
862 return f"{provenance.sheet} block {provenance.block_ordinal} (rows {provenance.row_start}-{provenance.row_end})"
865def _cross_trace_conflicts(traces: Sequence[SourceTrace]) -> tuple[SourceTrace, ...]:
866 """Flag every trace that shares a termination or disagrees about a CableClass."""
867 termination_claims: dict[IdentityKey, dict[int, set[_SegmentClaim]]] = defaultdict(lambda: defaultdict(set))
868 segment_classes: dict[tuple[IdentityKey, IdentityKey], dict[str, set[int]]] = defaultdict(lambda: defaultdict(set))
869 for index, trace in enumerate(traces):
870 claims: dict[IdentityKey, set[_SegmentClaim]] = defaultdict(set)
871 for segment in trace.segments:
872 ordered_pair = sorted((segment.left.identity_key, segment.right.identity_key))
873 segment_pair = ordered_pair[0], ordered_pair[1]
874 # The CableClass label keys its own mapping row, so it compares as the source states it.
875 cable_class = segment.cable_class
876 claim = segment_pair, cable_class
877 claims[segment.left.identity_key].add(claim)
878 claims[segment.right.identity_key].add(claim)
879 segment_classes[segment_pair][cable_class].add(index)
880 claims.setdefault(trace.endpoint_summary.from_termination.identity_key, set())
881 claims.setdefault(trace.endpoint_summary.to_termination.identity_key, set())
882 for termination, trace_claims in claims.items():
883 termination_claims[termination][index].update(trace_claims)
884 conflicts: dict[int, set[str]] = defaultdict(set)
885 for claims_by_owner in termination_claims.values():
886 owner_claims = list(claims_by_owner.values())
887 identical_shared_segment = bool(owner_claims[0]) and all(claims == owner_claims[0] for claims in owner_claims)
888 if len(claims_by_owner) > 1 and not identical_shared_segment:
889 for owner in claims_by_owner:
890 conflicts[owner].add("a termination is claimed by another Source Trace")
891 for classes in segment_classes.values():
892 owners = set().union(*classes.values())
893 if len(classes) > 1 and len(owners) > 1:
894 for owner in owners:
895 conflicts[owner].add("a shared segment has conflicting CableClass labels")
896 checked = []
897 for index, trace in enumerate(traces):
898 if index not in conflicts:
899 checked.append(trace)
900 continue
901 provenance = trace.provenance[0]
902 detail = "; ".join(sorted(conflicts[index]))
903 error = SourceDiagnostic(
904 code="trace.cross_trace_conflict",
905 message=f"{_location_from_provenance(provenance)}: {detail}.",
906 row_number=provenance.row_start,
907 )
908 checked.append(replace(trace, errors=_deduplicate_errors((*trace.errors, error))))
909 return tuple(checked)
912def interpret(content: bytes) -> tuple[tuple[SourceTrace, ...], tuple[SourceDiagnostic, ...]]:
913 """Return typed Source Traces and source diagnostics from workbook bytes."""
914 try:
915 book = openpyxl.load_workbook(BytesIO(content), data_only=True)
916 except Exception as exc:
917 raise SourceUnreadable(f"Cannot open Excel file: {exc}") from exc
918 try:
919 recognized = [name for name in (TRACE_PATH_SHEET, TRACE_LIST_SHEET) if name in book.sheetnames]
920 if not recognized:
921 diagnostic = SourceDiagnostic(
922 code="trace.no_recognized_sheet",
923 message=f"Workbook has neither '{TRACE_PATH_SHEET}' nor '{TRACE_LIST_SHEET}'.",
924 )
925 return (), (diagnostic,)
926 workbook_fingerprint = sha256(content).hexdigest()
927 path_blocks = (
928 _extract_blocks(book[TRACE_PATH_SHEET], workbook_fingerprint) if TRACE_PATH_SHEET in recognized else ()
929 )
930 list_blocks = (
931 _extract_blocks(book[TRACE_LIST_SHEET], workbook_fingerprint) if TRACE_LIST_SHEET in recognized else ()
932 )
933 finally:
934 # close() releases an archive only in read-only mode, but it stays paired with the load.
935 book.close()
936 lists_by_pair: dict[tuple[str, str], list[_Block]] = defaultdict(list)
937 for block in list_blocks:
938 lists_by_pair[block.pair_key].append(block)
939 traces: list[SourceTrace] = []
940 diagnostics: list[SourceDiagnostic] = []
941 paired_lists: set[tuple[str, int]] = set()
942 for path_block in path_blocks:
943 candidates = lists_by_pair[path_block.pair_key]
944 list_block = candidates.pop(0) if candidates else None
945 if list_block is not None:
946 paired_lists.add((list_block.sheet, list_block.ordinal))
947 trace, block_diagnostics = _path_trace(path_block, list_block)
948 diagnostics.extend(block_diagnostics)
949 if trace is not None:
950 traces.append(trace)
951 for block in list_blocks:
952 if (block.sheet, block.ordinal) in paired_lists:
953 continue
954 trace, block_diagnostics = _fallback_trace(block)
955 diagnostics.extend(block_diagnostics)
956 if trace is not None:
957 traces.append(trace)
958 checked = _cross_trace_conflicts(_collapse_duplicates(traces))
959 diagnostics.extend(error for trace in checked for error in trace.errors)
960 return checked, tuple(diagnostics)
963__all__ = (
964 "FRONT_PORT_CLASSES",
965 "INTERFACE_PORT_CLASSES",
966 "PORT_CLASSES",
967 "REAR_PORT_CLASSES",
968 "EndpointSummary",
969 "PassThroughClaim",
970 "SegmentEvidence",
971 "SourceTrace",
972 "TerminationReference",
973 "TraceProvenance",
974 "canonical_orientation",
975 "canonical_trace_identity",
976 "content_fingerprint",
977 "interpret",
978 "parse_endpoint_line",
979)