#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ bcm_mibpair_report_V1.1.0.py Author : ETWen Date : 20260730 Purpose : Extract MIB_TPOK / MIB_RPOK counters from a Broadcom drivshell log and lay them out horizontally, one loopback pair per block: Port TX RX PASS/FAIL cd1 723629081133 581693309458 PASS cd33 581693309458 723629081133 PASS Pair membership and pair order come from the "vlan remove pbm=cdA,cdB" lines in the same log, so the report always follows the order the test actually ran in. Verdict per pair comes from the loopback cross-check: for a pair, cdA.TPOK must equal cdB.RPOK and cdA.RPOK must equal cdB.TPOK. PASS - both directions match (within --tolerance) FAIL - at least one direction is off; delta is listed in the trailing summary NA - the pair has no counters in the log Usage : ./bcm_mibpair_report_V1.1.0.py [options] -a, --all also emit pairs that have no counters in the log (values shown as "-") -b, --block repeat the header and blank-separate each pair -g, --grouped keep the SDK's comma grouping (1,234,567) -p, --with-port render "cd1(2)" instead of "cd1" -t, --tsv tab-separated output (for Excel / pandas) -s, --no-status drop the PASS/FAIL column -T, --tolerance N allowed |delta| in packets before a pair is FAIL (default 0). A +/-1 skew is normal: the SDK reads counters port-by-port while traffic still runs, so -T 1 is a reasonable production setting. -q, --quiet suppress the trailing summary comments -o, --output write to a file instead of stdout Notes : - The trailing "+delta" column printed by the SDK is stripped. - A cd port missing from the dump usually means the console scrolled or auto-logged out mid-"show c", NOT that the port was down. Cross-check against "ps" before drawing conclusions. - A +/-1 packet delta between a pair is normal snapshot skew: the SDK reads counters port-by-port while traffic is still running. Exit : 0 = every pair had counters and cross-checked clean 1 = usage / input error 2 = at least one pair is FAIL or NA (see summary) Version History V1.0.0 20260730 Initial Version (parity with bcm_mibpair_report_V1.2.0.sh) V1.0.1 20260730 Handle BrokenPipeError quietly when piped into head V1.1.0 20260730 Add PASS/FAIL/NA verdict column and --tolerance """ from __future__ import annotations import argparse import re import sys from dataclasses import dataclass, field from pathlib import Path from typing import Iterable, TextIO EXIT_OK = 0 EXIT_USAGE = 1 EXIT_WARN = 2 # "vlan remove 31 pbm=cd1,cd33" RE_PBM = re.compile(r"pbm=(cd\d+),(cd\d+)") # "MIB_TPOK.cd1(2) : 723,629,081,133 +723,629,081,133" RE_MIB = re.compile( r"^MIB_(?PTPOK|RPOK)\.(?Pcd\d+)\((?P\d+)\)\s*:\s*(?P[\d,]+)" ) MISSING = "-" @dataclass class PortCounters: """TPOK / RPOK for a single cd port.""" lport: int | None = None tx: int | None = None rx: int | None = None @property def complete(self) -> bool: return self.tx is not None and self.rx is not None @dataclass class LogData: """Everything the report needs, extracted from one drivshell log.""" pairs: list[tuple[str, str]] = field(default_factory=list) ports: dict[str, PortCounters] = field(default_factory=dict) def counters(self, cd: str) -> PortCounters: return self.ports.setdefault(cd, PortCounters()) def parse_log(lines: Iterable[str]) -> LogData: """Pull pbm pair definitions and TPOK/RPOK counters out of a drivshell log.""" data = LogData() seen_pairs: set[tuple[str, str]] = set() for raw in lines: line = raw.rstrip("\r\n") for a, b in RE_PBM.findall(line): if (a, b) not in seen_pairs: seen_pairs.add((a, b)) data.pairs.append((a, b)) # drivshell echoes commands, so a MIB line may be prefixed with # "drivshell>" noise; anchor on the MIB_ token instead of the line start. idx = line.find("MIB_") if idx < 0: continue m = RE_MIB.match(line[idx:]) if not m: continue pc = data.counters(m.group("cd")) pc.lport = int(m.group("lport")) value = int(m.group("val").replace(",", "")) if m.group("kind") == "TPOK": pc.tx = value else: pc.rx = value return data def fmt_value(value: int | None, grouped: bool) -> str: if value is None: return MISSING return f"{value:,}" if grouped else str(value) def fmt_port(cd: str, pc: PortCounters, with_port: bool) -> str: if with_port and pc.lport is not None: return f"{cd}({pc.lport})" return cd PASS, FAIL, NA = "PASS", "FAIL", "NA" def pair_deltas(data: LogData, a: str, b: str) -> tuple[int, int] | None: """(cdA.TX - cdB.RX, cdA.RX - cdB.TX), or None if counters are missing.""" pa, pb = data.counters(a), data.counters(b) if not (pa.complete and pb.complete): return None return pa.tx - pb.rx, pa.rx - pb.tx def verdict(data: LogData, a: str, b: str, tolerance: int) -> str: d = pair_deltas(data, a, b) if d is None: return NA return PASS if max(abs(d[0]), abs(d[1])) <= tolerance else FAIL def render(data: LogData, args: argparse.Namespace, out: TextIO) -> int: status = not args.no_status rows: list[tuple[str, ...]] = [] blocks: list[list[tuple[str, ...]]] = [] missing: list[tuple[str, str]] = [] failed: list[tuple[str, str, int, int]] = [] tally = {PASS: 0, FAIL: 0, NA: 0} for a, b in data.pairs: pa, pb = data.counters(a), data.counters(b) v = verdict(data, a, b, args.tolerance) tally[v] += 1 if v == NA: missing.append((a, b)) if not args.all: continue elif v == FAIL: d_tx, d_rx = pair_deltas(data, a, b) failed.append((a, b, d_tx, d_rx)) block = [] for cd, pc in ((a, pa), (b, pb)): row = ( fmt_port(cd, pc, args.with_port), fmt_value(pc.tx, args.grouped), fmt_value(pc.rx, args.grouped), ) block.append(row + (v,) if status else row) blocks.append(block) rows.extend(block) header = ("Port", "TX", "RX", "PASS/FAIL") if status else ("Port", "TX", "RX") if args.tsv: if not args.block: print("\t".join(header), file=out) for block in blocks: if args.block: print("\t".join(header), file=out) for row in block: print("\t".join(row), file=out) if args.block: print(file=out) else: # size columns to the widest cell actually emitted ncol = len(header) width = [ max([len(r[i]) for r in rows] + [len(header[i])]) + 2 for i in range(ncol) ] def line(row: tuple[str, ...]) -> str: cells = [f"{row[0]:<{width[0]}}"] cells += [f"{row[i]:>{width[i]}}" for i in range(1, ncol)] return "".join(cells).rstrip() if not args.block: print(line(header), file=out) for block in blocks: if args.block: print(line(header), file=out) for row in block: print(line(row), file=out) if args.block: print(file=out) if not args.quiet: if not args.block: print(file=out) print(f"# pairs in pbm list : {len(data.pairs)}", file=out) print(f"# pairs reported : {len(blocks)}", file=out) print( f"# PASS {tally[PASS]} FAIL {tally[FAIL]} NA {tally[NA]}" f" (tolerance {args.tolerance})", file=out, ) if missing: joined = " ".join(f"{a}/{b}" for a, b in missing) print(f"# NA (no counters): {joined}", file=out) for a, b, d_tx, d_rx in failed: print( f"# FAIL {a}/{b}: {a}.TX-{b}.RX={d_tx:+d} {a}.RX-{b}.TX={d_rx:+d}", file=out, ) return EXIT_WARN if (tally[FAIL] or tally[NA]) else EXIT_OK def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Render Broadcom drivshell MIB_TPOK/RPOK counters as a " "loopback-pair table.", epilog="Exit 2 means missing counters and/or a cross-check mismatch.", ) p.add_argument("log", type=Path, help="drivshell log file ('-' for stdin)") p.add_argument("-a", "--all", action="store_true", help="also emit pairs with no counters") p.add_argument("-b", "--block", action="store_true", help="repeat header and blank-separate each pair") p.add_argument("-g", "--grouped", action="store_true", help="keep comma grouping in values") p.add_argument("-p", "--with-port", action="store_true", help="render cd1(2) instead of cd1") p.add_argument("-t", "--tsv", action="store_true", help="tab-separated output") p.add_argument("-s", "--no-status", action="store_true", help="drop the PASS/FAIL column") p.add_argument("-T", "--tolerance", type=int, default=0, metavar="N", help="allowed |delta| in packets before a pair is FAIL " "(default 0; -T 1 absorbs normal snapshot skew)") p.add_argument("-q", "--quiet", action="store_true", help="suppress trailing summary comments") p.add_argument("-o", "--output", type=Path, default=None, help="write to a file instead of stdout") return p def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if str(args.log) == "-": data = parse_log(sys.stdin) else: # console captures are frequently not clean UTF-8 with args.log.open("r", encoding="utf-8", errors="replace") as fh: data = parse_log(fh) except OSError as exc: print(f"error: cannot read log: {exc}", file=sys.stderr) return EXIT_USAGE if not data.pairs: print("error: no 'pbm=cdA,cdB' pair definition found in log", file=sys.stderr) return EXIT_USAGE try: if args.output: with args.output.open("w", encoding="utf-8", newline="\n") as fh: return render(data, args, fh) return render(data, args, sys.stdout) except BrokenPipeError: # downstream closed the pipe (e.g. "| head") - not an error try: sys.stdout.close() except BrokenPipeError: pass return EXIT_OK except OSError as exc: print(f"error: cannot write output: {exc}", file=sys.stderr) return EXIT_USAGE if __name__ == "__main__": sys.exit(main())