From 24c4beecc96152ae4c3942b12a9e6460a1b6f2c1 Mon Sep 17 00:00:00 2001 From: Michele Bigi Date: Tue, 22 Sep 2026 12:37:31 +0200 Subject: [PATCH] Add protocol-certification M5 Protocolli report UI (2.69.0). MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instances, L0–L3 max level, measured/limit/margin/source/method, traceability chain. FAIL listed first, not under Warning. Visible skips. No pack numbers, no L3 solver. --- .../src/backend/periscopex/protocol_l0.py | 22 +- .../src/backend/periscopex/protocol_report.py | 198 ++++++++++++++++ periscope/src/frontend/content/changelog.md | 7 + periscope/src/frontend/package.json | 4 +- .../components/report/protocol-section.tsx | 193 +++++++++++---- .../frontend/src/lib/protocol-report.test.ts | 77 ++++++ .../src/frontend/src/lib/protocol-report.ts | 113 +++++++++ tests/pcb/test_protocol_l0_m2.py | 4 +- tests/pcb/test_protocol_l1_m3.py | 2 +- tests/pcb/test_protocol_l2_m4.py | 2 +- tests/pcb/test_protocol_report_m5.py | 222 ++++++++++++++++++ 11 files changed, 785 insertions(+), 59 deletions(-) create mode 100644 periscope/src/backend/periscopex/protocol_report.py create mode 100644 periscope/src/frontend/src/lib/protocol-report.test.ts create mode 100644 periscope/src/frontend/src/lib/protocol-report.ts create mode 100644 tests/pcb/test_protocol_report_m5.py diff --git a/periscope/src/backend/periscopex/protocol_l0.py b/periscope/src/backend/periscopex/protocol_l0.py index 8f9fd56..bbfb89b 100644 --- a/periscope/src/backend/periscopex/protocol_l0.py +++ b/periscope/src/backend/periscopex/protocol_l0.py @@ -300,10 +300,11 @@ def protocol_exam( impedance_nets: list[dict] | dict | None = None, ) -> tuple[ProtocolCertificationSection, list[Finding]]: from backend.periscopex.protocol_l1 import certify_l1 - from backend.periscopex.protocol_l2 import ( - PACK_MACROPHASE as L2_MACRO, - LEVEL as L2_LEVEL, - certify_l2, + from backend.periscopex.protocol_l2 import LEVEL as L2_LEVEL, certify_l2 + from backend.periscopex.protocol_report import ( + PACK_MACROPHASE as L5_MACRO, + attach_instance_report, + result_rank, ) if graph is None: @@ -312,7 +313,7 @@ def protocol_exam( instances = recognize_physical_buses(graph, cat) if not instances: sec = empty_protocol_section() - sec.macrophase = L2_MACRO + sec.macrophase = L5_MACRO return sec, [] l0, findings = certify_l0(graph, instances, cat) l1, l1_findings_out = certify_l1(graph, instances, layout, cat) @@ -324,16 +325,25 @@ def protocol_exam( impedance_nets=impedance_nets, ) findings.extend(l2_findings_out) + ifaces = {p.id: p for p in cat.physical_interfaces} dumps: list[dict[str, Any]] = [] for inst in instances: row = inst.model_dump() row["l0_checks"] = [c.model_dump() for c in l0.get(inst.instance_id, [])] row["l1_checks"] = [c.model_dump() for c in l1.get(inst.instance_id, [])] row["l2_checks"] = [c.model_dump() for c in l2.get(inst.instance_id, [])] + row.update(attach_instance_report( + inst, + ifaces.get(inst.physical_interface_id), + l0.get(inst.instance_id, []), + l1.get(inst.instance_id, []), + l2.get(inst.instance_id, []), + )) dumps.append(row) + dumps.sort(key=lambda r: result_rank(str(r.get("worst_result") or "UNKNOWN"))) sec = ProtocolCertificationSection( schema_version=SCHEMA_VERSION, - macrophase=L2_MACRO, + macrophase=L5_MACRO, recognized_instances=dumps, message="", max_level_reached=L2_LEVEL, diff --git a/periscope/src/backend/periscopex/protocol_report.py b/periscope/src/backend/periscopex/protocol_report.py new file mode 100644 index 0000000..cb2ac5f --- /dev/null +++ b/periscope/src/backend/periscopex/protocol_report.py @@ -0,0 +1,198 @@ +"""M5: protocol report view — checks, chain, FAIL before WARNING, visible skips. + +Does not invent Z. Does not run L3/OpenEMS. Does not add pack numbers. +L0/L1 are not electrical certification. +""" + +from __future__ import annotations + +from typing import Any + +from backend.periscopex.protocol_catalog import PhysicalInterface, ProtocolConstraint +from backend.periscopex.protocol_l0 import _constraint_for +from backend.periscopex.protocol_recognize import PhysicalBusInstance + +PACK_MACROPHASE = "M5" + +_SKIP = frozenset({ + "UNKNOWN", "MISSING_SOURCE", "VENDOR_DEPENDENT", + "CONTROLLER_DEPENDENT", "PHY_DEPENDENT", +}) +L3_SKIP_NOTE = ( + "L3 channel not run (no channel/S-param data; OpenEMS out of scope). Not PASS." +) + + +def result_rank(result: str) -> int: + """FAIL is always first. Never bury FAIL under WARNING (HubAudio).""" + if result == "FAIL": + return 0 + if result == "WARNING" or result in _SKIP: + return 1 + if result == "NOT_APPLICABLE": + return 2 + return 3 + + +def instance_worst_result(results: list[str]) -> str: + if not results: + return "UNKNOWN" + return min(results, key=result_rank) + + +def _unit(dump: dict[str, Any]) -> str: + if dump.get("measured_mm") is not None or dump.get("limit_mm") is not None: + return "mm" + if dump.get("measured_ohm") is not None or dump.get("limit_ohm") is not None: + return "ohm" + return "" + + +def _measured(dump: dict[str, Any]) -> float | None: + for k in ("measured_ohm", "measured_mm"): + v = dump.get(k) + if isinstance(v, (int, float)): + return float(v) + return None + + +def _limit(dump: dict[str, Any]) -> float | None: + for k in ("limit_ohm", "limit_mm", "limit_ohm_max"): + v = dump.get(k) + if isinstance(v, (int, float)): + return float(v) + return None + + +def _margin(dump: dict[str, Any]) -> float | None: + for k in ("margin_ohm", "margin_mm"): + v = dump.get(k) + if isinstance(v, (int, float)): + return float(v) + return None + + +def _group_for_net(inst: PhysicalBusInstance, net: str | None) -> str: + if not net: + if inst.groups: + return inst.groups[0].id or inst.groups[0].kind + return "" + for g in inst.groups: + if net in g.nets: + return g.id or g.kind + return "" + + +def _cite(cons: ProtocolConstraint | None) -> tuple[str, str, str]: + if cons is None: + return "", "", "" + src = cons.source + doc = "" + section = "" + if src is not None: + doc = src.document or src.organization or "" + section = src.section or src.table or (src.page or "") + method = cons.measurement_method or "" + return doc, section, method + + +def check_to_report_row( + dump: dict[str, Any], + inst: PhysicalBusInstance, + iface: PhysicalInterface | None, +) -> dict[str, Any]: + check = str(dump.get("check") or "") + result = str(dump.get("result") or "") + cons = _constraint_for(iface, check) if iface is not None else None + doc, section, method = _cite(cons) + nets = list(dump.get("nets") or []) + net = nets[0] if nets else (inst.nets[0] if inst.nets else "") + group = _group_for_net(inst, net or None) + skip = result in _SKIP or "non certificata" in str(dump.get("notes") or "") + row = { + "check": check, + "level": dump.get("level") or "", + "result": result, + "mandatory": dump.get("mandatory") or "", + "measured": _measured(dump), + "limit": _limit(dump), + "margin": _margin(dump), + "unit": _unit(dump), + "source": doc, + "method": method, + "notes": dump.get("notes") or "", + "skip_visible": skip, + "rank": result_rank(result), + "chain": { + "net": net, + "group": group, + "physical_interface_id": inst.physical_interface_id, + "logical_protocol_id": inst.logical_protocol_id, + "constraint_id": cons.id if cons else "", + "document": doc, + "section": section, + }, + } + return row + + +def l3_not_run_row(inst: PhysicalBusInstance) -> dict[str, Any]: + return { + "check": "channel", + "level": "L3", + "result": "UNKNOWN", + "mandatory": "INFORMATIONAL", + "measured": None, + "limit": None, + "margin": None, + "unit": "", + "source": "", + "method": "", + "notes": L3_SKIP_NOTE, + "skip_visible": True, + "rank": result_rank("UNKNOWN"), + "chain": { + "net": inst.nets[0] if inst.nets else "", + "group": inst.groups[0].id if inst.groups else "", + "physical_interface_id": inst.physical_interface_id, + "logical_protocol_id": inst.logical_protocol_id, + "constraint_id": "", + "document": "", + "section": "", + }, + } + + +def format_chain(chain: dict[str, Any]) -> str: + keys = ( + "net", "group", "physical_interface_id", "logical_protocol_id", + "constraint_id", "document", "section", + ) + parts = [str(chain.get(k) or "—") for k in keys] + return " → ".join(parts) + + +def attach_instance_report( + inst: PhysicalBusInstance, + iface: PhysicalInterface | None, + l0: list[Any], + l1: list[Any], + l2: list[Any], +) -> dict[str, Any]: + dumps = ( + [c.model_dump() if hasattr(c, "model_dump") else dict(c) for c in l0] + + [c.model_dump() if hasattr(c, "model_dump") else dict(c) for c in l1] + + [c.model_dump() if hasattr(c, "model_dump") else dict(c) for c in l2] + ) + rows = [check_to_report_row(d, inst, iface) for d in dumps] + rows.append(l3_not_run_row(inst)) + rows.sort(key=lambda r: (r["rank"], str(r["level"]), str(r["check"]))) + worst = instance_worst_result([str(r["result"]) for r in rows if r["level"] != "L3"]) + # L3 not-run must not upgrade instance to PASS or hide FAIL + return { + "report_checks": rows, + "worst_result": worst, + "fail_count": sum(1 for r in rows if r["result"] == "FAIL"), + "skip_count": sum(1 for r in rows if r.get("skip_visible")), + "chain_example": format_chain(rows[0]["chain"]) if rows else "", + } diff --git a/periscope/src/frontend/content/changelog.md b/periscope/src/frontend/content/changelog.md index fec72a1..078402e 100644 --- a/periscope/src/frontend/content/changelog.md +++ b/periscope/src/frontend/content/changelog.md @@ -2,6 +2,13 @@ What's new in Periscope. +## 2.69.0 — 2026-09-22 — Protocol certification M5 (Protocolli report) + +Report section **Protocolli**: recognised instances, max level L0–L3 (L3 is a visible not-run skip, no solver), each check with measured / limit / margin / source / method, traceability chain net → group → physical interface → protocol → constraint → document → section. FAIL is listed first and styled as Error — not hidden under Warning (HubAudio). Visible skips stay on the page. No pack numbers, no invented Z, no OpenEMS. + +- [New] `protocol_report.py`; Protocolli table UI. +- [New] pytest `tests/pcb/test_protocol_report_m5.py`; `protocol-report.test.ts`. + ## 2.68.0 — 2026-09-22 — Protocol certification M4 (L2 electrical) L2 ELECTRICAL certifier: impedance, termination, voltage/levels, rise/fall when sources exist. Z only from datasheet / stackup / fabricator / **cited pack** — never invented 90 Ω USB. USB-IF/IEEE only if the pack has a cite. Silicon cross is max PCB × PHY subset when datasheet windows exist. Length is not delay. No L3/OpenEMS. L0/L1 are **not** electrical. RECOMMENDED never FAIL. MISSING_SOURCE / UNKNOWN when cite is absent. Visible skip when Z cannot be obtained. diff --git a/periscope/src/frontend/package.json b/periscope/src/frontend/package.json index 3334837..52e3021 100644 --- a/periscope/src/frontend/package.json +++ b/periscope/src/frontend/package.json @@ -1,6 +1,6 @@ { "name": "periscope-web", - "version": "2.68.0", + "version": "2.69.0", "private": true, "scripts": { "sync-version": "node scripts/sync-version.mjs", @@ -10,7 +10,7 @@ "build": "next build", "start": "next start", "lint": "eslint", - "test": "node --experimental-strip-types --test src/lib/findings-forest.test.ts" + "test": "node --experimental-strip-types --test src/lib/findings-forest.test.ts src/lib/protocol-report.test.ts" }, "dependencies": { "@base-ui/react": "^1.3.0", diff --git a/periscope/src/frontend/src/components/report/protocol-section.tsx b/periscope/src/frontend/src/components/report/protocol-section.tsx index f5d6e38..1afffa0 100644 --- a/periscope/src/frontend/src/components/report/protocol-section.tsx +++ b/periscope/src/frontend/src/components/report/protocol-section.tsx @@ -1,6 +1,42 @@ "use client"; import type { ProtocolCertificationSection } from "@/lib/types"; +import { StatusBadge } from "@/components/report/status-badge"; +import { + formatLimit, + formatMargin, + formatMeasured, + formatProtocolChain, + protocolResultToFindingStatus, + sortProtocolChecks, + type ProtocolReportCheck, +} from "@/lib/protocol-report"; +import { cn } from "@/lib/utils"; + +function asChecks(row: Record): ProtocolReportCheck[] { + const raw = row.report_checks; + if (Array.isArray(raw) && raw.length > 0) { + return sortProtocolChecks(raw as ProtocolReportCheck[]); + } + const fallback: ProtocolReportCheck[] = []; + for (const key of ["l0_checks", "l1_checks", "l2_checks"] as const) { + const list = row[key]; + if (!Array.isArray(list)) continue; + for (const c of list) { + const rec = c as Record; + fallback.push({ + check: String(rec.check ?? ""), + level: String(rec.level ?? ""), + result: String(rec.result ?? ""), + notes: String(rec.notes ?? ""), + skip_visible: + String(rec.result ?? "") !== "PASS" && + String(rec.result ?? "") !== "FAIL", + }); + } + } + return sortProtocolChecks(fallback); +} export function ProtocolSection({ section, @@ -9,64 +45,127 @@ export function ProtocolSection({ }) { const instances = section?.recognized_instances ?? []; const message = section?.message || "nessun protocollo riconosciuto"; + const maxLevel = section?.max_level_reached || "—"; return ( -
-

Protocolli

+
+
+

Protocolli

+

+ Livello massimo: {maxLevel} + {" "} + (L0–L3; L3 not run unless channel data exists) +

+

Logical protocol vs physical interface. L0 structural, L1 geometric, - L2 electrical (Z from datasheet/stackup/fab or a cited pack — never - invented 90 Ω). L0/L1 are not electrical. Length is not delay. + L2 electrical (Z only from datasheet/stackup/fab or a cited pack). + L0/L1 are not electrical. Length is not delay. FAIL is listed first — + it is not filed under Warning.

{instances.length === 0 ? (

{message}

) : ( -
    +
      {instances.map((row, i) => { - const phys = String(row.physical_interface_id ?? ""); - const log = String(row.logical_protocol_id ?? ""); - const status = String(row.recognition_status ?? ""); - const conf = row.confidence; - const groups = Array.isArray(row.groups) ? row.groups : []; + const rec = row as Record; + const phys = String(rec.physical_interface_id ?? ""); + const log = String(rec.logical_protocol_id ?? ""); + const status = String(rec.recognition_status ?? ""); + const worst = String(rec.worst_result ?? ""); + const checks = asChecks(rec); + const fail = Number(rec.fail_count ?? 0) > 0 || worst === "FAIL"; return ( -
    • -

      - {log} → {phys} - {status ? ` · ${status}` : ""} - {typeof conf === "number" ? ` · ${conf.toFixed(2)}` : ""} -

      - {groups.length > 0 ? ( -

      - {groups - .map((g) => { - const rec = g as Record; - return String(rec.kind ?? rec.id ?? "group"); - }) - .join(", ")} +

    • +
      +

      + {log} → {phys} + {status ? ` · ${status}` : ""} +

      + {worst ? ( + + ) : null} + {fail ? ( + + FAIL visible — not under Warning + + ) : null} +
      +
      + + + + + + + + + + + + + + + {checks.map((c, j) => { + const isFail = c.result === "FAIL"; + const skip = Boolean(c.skip_visible); + return ( + + + + + + + + + + + ); + })} + +
      LvCheckResultMeasuredLimitMarginSourceMethod
      {c.level}{c.check} + {c.result} + {skip ? " · skip" : ""} + {formatMeasured(c)}{formatLimit(c)}{formatMargin(c)}{c.source || "—"}{c.method || "—"}
      +
      + {String(rec.chain_example || "") ? ( +

      + Catena: {String(rec.chain_example)}

      ) : null} - {Array.isArray(row.l0_checks) && row.l0_checks.length > 0 ? ( -

      - L0{" "} - {(row.l0_checks as Record[]) - .map((c) => `${String(c.check)}=${String(c.result)}`) - .join("; ")} -

      - ) : null} - {Array.isArray(row.l1_checks) && row.l1_checks.length > 0 ? ( -

      - L1{" "} - {(row.l1_checks as Record[]) - .map((c) => `${String(c.check)}=${String(c.result)}`) - .join("; ")} -

      - ) : null} - {Array.isArray(row.l2_checks) && row.l2_checks.length > 0 ? ( -

      - L2{" "} - {(row.l2_checks as Record[]) - .map((c) => `${String(c.check)}=${String(c.result)}`) - .join("; ")} -

      + {checks + .filter((c) => c.result === "FAIL" || c.skip_visible) + .map((c, j) => + c.chain ? ( +

      + {c.level}/{c.check}: {formatProtocolChain(c.chain)} +

      + ) : null, + )} + {checks.some((c) => c.skip_visible && c.notes) ? ( +
        + {checks + .filter((c) => c.skip_visible && c.notes) + .map((c, j) => ( +
      • {c.notes}
        • + ))} +
      ) : null}
      • ); diff --git a/periscope/src/frontend/src/lib/protocol-report.test.ts b/periscope/src/frontend/src/lib/protocol-report.test.ts new file mode 100644 index 0000000..28dd828 --- /dev/null +++ b/periscope/src/frontend/src/lib/protocol-report.test.ts @@ -0,0 +1,77 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { + formatProtocolChain, + instanceWorstResult, + protocolResultRank, + protocolResultToFindingStatus, + sortProtocolChecks, +} from "./protocol-report.ts"; +import { groupByWorstStatus } from "./findings-forest.ts"; +import type { Finding } from "./types.ts"; + +function f(partial: Partial & Pick): Finding { + return { + finding_id: null, + mpn: "", + aspect: null, + finding: partial.finding ?? "", + why: "", + source_page: null, + reference: "", + ...partial, + }; +} + +test("FAIL ranks before WARNING and is not buried", () => { + assert.ok(protocolResultRank("FAIL") < protocolResultRank("WARNING")); + assert.ok(protocolResultRank("FAIL") < protocolResultRank("UNKNOWN")); + const sorted = sortProtocolChecks([ + { check: "z", result: "UNKNOWN", level: "L2" }, + { check: "topology", result: "FAIL", level: "L0" }, + { check: "length", result: "PASS", level: "L1" }, + ]); + assert.equal(sorted[0].result, "FAIL"); + assert.equal(sorted[0].check, "topology"); + assert.equal(instanceWorstResult(["UNKNOWN", "FAIL", "PASS"]), "FAIL"); + assert.equal(protocolResultToFindingStatus("FAIL"), "ERROR"); +}); + +test("chain is net → group → physical → protocol → constraint → document → section", () => { + const text = formatProtocolChain({ + net: "USB_D+", + group: "DIFF_USB", + physical_interface_id: "usb2-hs-dpair", + logical_protocol_id: "usb2-hs", + constraint_id: "usb2-hs-dpair-differential_impedance", + document: "", + section: "", + }); + assert.equal( + text, + "USB_D+ → DIFF_USB → usb2-hs-dpair → usb2-hs → usb2-hs-dpair-differential_impedance → — → —", + ); + assert.ok(!text.includes("90")); +}); + +test("protocol FAIL finding stays in Error folder with WARNING skips on same ref", () => { + const groups = groupByWorstStatus([ + f({ + designator: "U1", + status: "ERROR", + rule_id: "PE-PRT-L0-001", + source: "protocol_l0", + finding: "L0 FAIL topology", + }), + f({ + designator: "U1", + status: "WARNING", + rule_id: "PE-PRT-L2-001", + source: "protocol_l2", + finding: "non certificata a L2 per mancanza di Z", + }), + ]); + assert.equal(groups[0].label, "Error"); + assert.equal(groups[0].status, "ERROR"); + assert.ok(groups[0].findings.some((x) => x.rule_id === "PE-PRT-L0-001")); +}); diff --git a/periscope/src/frontend/src/lib/protocol-report.ts b/periscope/src/frontend/src/lib/protocol-report.ts new file mode 100644 index 0000000..7bf5f68 --- /dev/null +++ b/periscope/src/frontend/src/lib/protocol-report.ts @@ -0,0 +1,113 @@ +/** Protocol report ranking and chain. FAIL never ranks under WARNING. */ + +export const PROTOCOL_LEVELS = ["L0", "L1", "L2", "L3"] as const; + +export type ProtocolCheckResult = + | "PASS" + | "FAIL" + | "WARNING" + | "UNKNOWN" + | "NOT_APPLICABLE" + | "VENDOR_DEPENDENT" + | "CONTROLLER_DEPENDENT" + | "PHY_DEPENDENT" + | "MISSING_SOURCE"; + +export type ProtocolChain = { + net?: string; + group?: string; + physical_interface_id?: string; + logical_protocol_id?: string; + constraint_id?: string; + document?: string; + section?: string; +}; + +export type ProtocolReportCheck = { + check: string; + level: string; + result: string; + mandatory?: string; + measured?: number | null; + limit?: number | null; + margin?: number | null; + unit?: string; + source?: string; + method?: string; + notes?: string; + skip_visible?: boolean; + rank?: number; + chain?: ProtocolChain; +}; + +const SKIP = new Set([ + "UNKNOWN", + "MISSING_SOURCE", + "VENDOR_DEPENDENT", + "CONTROLLER_DEPENDENT", + "PHY_DEPENDENT", +]); + +export function protocolResultRank(result: string): number { + if (result === "FAIL") return 0; + if (result === "WARNING" || SKIP.has(result)) return 1; + if (result === "NOT_APPLICABLE") return 2; + return 3; +} + +export function sortProtocolChecks( + rows: T[], +): T[] { + return [...rows].sort((a, b) => { + const d = protocolResultRank(a.result) - protocolResultRank(b.result); + if (d !== 0) return d; + return String(a.level).localeCompare(String(b.level)) || String(a.check).localeCompare(String(b.check)); + }); +} + +export function instanceWorstResult(results: string[]): string { + if (!results.length) return "UNKNOWN"; + return results.reduce((best, r) => + protocolResultRank(r) < protocolResultRank(best) ? r : best, + ); +} + +export function protocolResultToFindingStatus( + result: string, +): "ERROR" | "WARNING" | "INFO" { + if (result === "FAIL") return "ERROR"; + if (result === "WARNING" || SKIP.has(result)) return "WARNING"; + return "INFO"; +} + +export function formatProtocolChain(chain?: ProtocolChain | null): string { + if (!chain) return "—"; + const keys: (keyof ProtocolChain)[] = [ + "net", + "group", + "physical_interface_id", + "logical_protocol_id", + "constraint_id", + "document", + "section", + ]; + return keys.map((k) => chain[k] || "—").join(" → "); +} + +export function formatMeasured(row: ProtocolReportCheck): string { + if (row.measured == null) return "—"; + const unit = row.unit ? ` ${row.unit}` : ""; + return `${row.measured}${unit}`; +} + +export function formatLimit(row: ProtocolReportCheck): string { + if (row.limit == null) return "—"; + const unit = row.unit ? ` ${row.unit}` : ""; + return `${row.limit}${unit}`; +} + +export function formatMargin(row: ProtocolReportCheck): string { + if (row.margin == null) return "—"; + const unit = row.unit ? ` ${row.unit}` : ""; + return `${row.margin}${unit}`; +} diff --git a/tests/pcb/test_protocol_l0_m2.py b/tests/pcb/test_protocol_l0_m2.py index 8c31d26..1a97364 100644 --- a/tests/pcb/test_protocol_l0_m2.py +++ b/tests/pcb/test_protocol_l0_m2.py @@ -130,8 +130,8 @@ def test_axi_internal_l0_not_applicable_not_fail(): assert all(r.result != "FAIL" for r in rows) sec, findings = protocol_exam(_axi_graph()) assert sec.max_level_reached == "L2" - from backend.periscopex.protocol_l2 import PACK_MACROPHASE as L2_MACRO - assert sec.macrophase == L2_MACRO + from backend.periscopex.protocol_report import PACK_MACROPHASE as L5_MACRO + assert sec.macrophase == L5_MACRO assert all(f.status != "ERROR" for f in findings) assert all("AXI_AWVALID" not in (f.net or "") for f in findings if f.rule_id == "PE-PRT-L0-001") diff --git a/tests/pcb/test_protocol_l1_m3.py b/tests/pcb/test_protocol_l1_m3.py index e6cdc6b..747b7b6 100644 --- a/tests/pcb/test_protocol_l1_m3.py +++ b/tests/pcb/test_protocol_l1_m3.py @@ -166,7 +166,7 @@ def test_axi_internal_l1_not_applicable(): assert all(r.result != "FAIL" for r in rows) sec, exam_findings = protocol_exam(_axi_graph(), layout=LayoutGraph()) assert sec.max_level_reached == "L2" - assert sec.macrophase == "M4" + assert sec.macrophase == "M5" assert all(f.status != "ERROR" for f in exam_findings) blob = json.dumps([r.model_dump() for r in rows]) assert "electrical certification" not in blob.lower() or "not electrical" in blob.lower() diff --git a/tests/pcb/test_protocol_l2_m4.py b/tests/pcb/test_protocol_l2_m4.py index d012184..8ef0c7f 100644 --- a/tests/pcb/test_protocol_l2_m4.py +++ b/tests/pcb/test_protocol_l2_m4.py @@ -205,7 +205,7 @@ def test_axi_internal_l2_not_applicable(): assert all(r.result == "NOT_APPLICABLE" for r in rows) sec, findings = protocol_exam(_axi_graph()) assert sec.max_level_reached == "L2" - assert sec.macrophase == "M4" + assert sec.macrophase == "M5" assert all(f.status != "ERROR" for f in findings) blob = json.dumps([r.model_dump() for r in rows]) assert "L0/L1 are not electrical" in blob diff --git a/tests/pcb/test_protocol_report_m5.py b/tests/pcb/test_protocol_report_m5.py new file mode 100644 index 0000000..9df31d8 --- /dev/null +++ b/tests/pcb/test_protocol_report_m5.py @@ -0,0 +1,222 @@ +"""M5 protocol report: measured/limit/margin/source/method, chain, FAIL first, skips visible.""" + +from __future__ import annotations + +import json + +from backend.periscopex.models import ( + Component, + ComponentType, + DesignGraph, + Net, + NetType, + PinConnection, +) +from backend.periscopex.protocol_catalog import ProtocolCatalogError, parse_catalog +from backend.periscopex.protocol_l0 import protocol_exam +from backend.periscopex.protocol_report import ( + instance_worst_result, + result_rank, +) + + +def _ic(ref: str, pins: dict[str, str], *, mpn: str = "", value: str = "") -> Component: + return Component( + reference=ref, value=value or mpn, footprint="", + component_type=ComponentType.IC, mpn=mpn or None, pins=pins, + ) + + +def _net(name: str, *pairs: tuple[str, str]) -> Net: + return Net( + name=name, net_type=NetType.SIGNAL, + pins=[PinConnection(component_ref=r, pin_number=p) for r, p in pairs], + ) + + +def _usb_connected() -> DesignGraph: + return DesignGraph( + components={ + "U1": _ic("U1", {"1": "USB_D+", "2": "USB_D-"}, mpn="CH340E"), + "J2": Component( + reference="J2", value="USB2_TypeA", + footprint="USB_A", + component_type=ComponentType.CONNECTOR, + pins={"2": "USB_D+", "3": "USB_D-"}, + ), + }, + nets={ + "USB_D+": _net("USB_D+", ("U1", "1"), ("J2", "2")), + "USB_D-": _net("USB_D-", ("U1", "2"), ("J2", "3")), + }, + ) + + +def _usb_dangling_minus() -> DesignGraph: + return DesignGraph( + components={ + "U1": _ic("U1", {"1": "USB_D+"}, mpn="CH340E"), + "J2": Component( + reference="J2", value="USB2_TypeA", + footprint="USB_A", + component_type=ComponentType.CONNECTOR, + pins={"2": "USB_D+", "3": "USB_D-"}, + ), + }, + nets={ + "USB_D+": _net("USB_D+", ("U1", "1"), ("J2", "2")), + "USB_D-": _net("USB_D-", ("J2", "3")), + }, + ) + + +def _axi_graph() -> DesignGraph: + return DesignGraph( + components={ + "U3": _ic("U3", {"1": "AXI_AWVALID"}, mpn="XC7A100T", value="AXI4 interconnect"), + }, + nets={"AXI_AWVALID": _net("AXI_AWVALID", ("U3", "1"))}, + ) + + +def _ddr_no_grouping() -> DesignGraph: + pins = {str(i): f"MEM_DAT{i}" for i in range(8)} + return DesignGraph( + components={"U5": _ic("U5", pins, mpn="MT41K256M16TW", value="DDR4")}, + nets={n: _net(n, ("U5", p)) for p, n in pins.items()}, + ) + + +def test_result_rank_fail_before_warning(): + assert result_rank("FAIL") < result_rank("WARNING") + assert result_rank("FAIL") < result_rank("UNKNOWN") + assert instance_worst_result(["UNKNOWN", "FAIL", "PASS"]) == "FAIL" + + +def test_usb2_pair_l0_pass_l2_z_unknown_no_invented_ohm(): + sec, _ = protocol_exam(_usb_connected()) + assert sec.max_level_reached == "L2" + assert sec.macrophase == "M5" + usb = next( + r for r in sec.recognized_instances + if "usb" in str(r.get("logical_protocol_id")) + ) + checks = usb["report_checks"] + l0_topo = [c for c in checks if c["level"] == "L0" and c["check"] == "topology"] + assert l0_topo and l0_topo[0]["result"] == "PASS" + z = [c for c in checks if c["check"] == "differential_impedance"] + assert z + assert z[0]["result"] in {"MISSING_SOURCE", "UNKNOWN"} + assert z[0]["skip_visible"] is True + assert z[0]["measured"] is None + assert z[0]["limit"] is None + blob = json.dumps(z[0]) + assert "90" not in blob + chain = z[0]["chain"] + assert chain["physical_interface_id"] + assert chain["logical_protocol_id"] + assert chain["physical_interface_id"] != chain["logical_protocol_id"] + l3 = [c for c in checks if c["level"] == "L3"] + assert l3 and l3[0]["skip_visible"] + assert l3[0]["result"] != "PASS" + assert "OpenEMS" in l3[0]["notes"] + + +def test_fail_listed_before_warning_in_report_checks(): + sec, findings = protocol_exam(_usb_dangling_minus()) + usb = next( + r for r in sec.recognized_instances + if "usb" in str(r.get("logical_protocol_id")) + ) + assert usb["worst_result"] == "FAIL" + assert usb["fail_count"] >= 1 + results = [c["result"] for c in usb["report_checks"]] + assert "FAIL" in results + assert results.index("FAIL") < next( + i for i, r in enumerate(results) if r in {"UNKNOWN", "MISSING_SOURCE", "WARNING"} + ) + assert any(f.status == "ERROR" and f.rule_id == "PE-PRT-L0-001" for f in findings) + + +def test_axi_na_visible_not_fail(): + sec, _ = protocol_exam(_axi_graph()) + row = next( + r for r in sec.recognized_instances + if "axi" in str(r.get("logical_protocol_id")) + ) + checks = row["report_checks"] + assert any(c["result"] == "NOT_APPLICABLE" for c in checks) + assert row["worst_result"] != "FAIL" + assert all(c["result"] != "FAIL" for c in checks if c["level"] != "L3") + + +def test_ddr_grouping_skip_visible_not_pass(): + sec, findings = protocol_exam(_ddr_no_grouping()) + ddr = next( + r for r in sec.recognized_instances + if str(r.get("logical_protocol_id")).startswith("ddr") + ) + grouping = [ + c for c in ddr["report_checks"] + if c["check"] in {"byte_lane_mapping", "dq_to_dqs_skew", "byte_lane_skew"} + ] + assert grouping + assert all(c["result"] != "PASS" for c in grouping) + assert ddr["worst_result"] == "FAIL" or any(c["skip_visible"] for c in grouping) + notes = " ".join((c.get("notes") or "") for c in grouping).lower() + assert "byte-lane" in notes or "grouping" in notes or "geometria" in notes or "dq" in notes + assert any( + f.rule_id in {"PE-PRT-L0-001", "PE-PRT-L1-002", "PE-PRT-L1-001"} + for f in findings + ) + + +def test_report_check_has_measured_limit_margin_source_method_keys(): + sec, _ = protocol_exam(_usb_connected()) + usb = sec.recognized_instances[0] + row = usb["report_checks"][0] + for key in ( + "measured", "limit", "margin", "source", "method", + "chain", "skip_visible", "result", "level", + ): + assert key in row + chain = row["chain"] + for key in ( + "net", "group", "physical_interface_id", "logical_protocol_id", + "constraint_id", "document", "section", + ): + assert key in chain + + +def test_typical_still_rejected_as_standard(): + raw = { + "schema_version": "1.0.0", + "logical_protocols": [{ + "id": "usb2-hs", "name": "USB2 HS", "bus_type": "SERIAL", + }], + "physical_interfaces": [{ + "id": "usb2-hs-dpair", + "logical_protocol_id": "usb2-hs", + "bus_type": "SERIAL", + "physical_layer": "SERIAL_DIFFERENTIAL", + "pcb_relevant": "YES", + "required_checks": ["differential_impedance"], + "constraints": [{ + "id": "bad", + "parameter": "differential_impedance", + "value_kind": "NUMERIC", + "mandatory": "MANDATORY", + "source_type": "STANDARD", + "source_class": "NORMATIVE", + "value": 90, + "unit": "ohm", + "typical": True, + "source": {"document": "blog", "organization": "web"}, + }], + }], + } + try: + parse_catalog(raw) + raise AssertionError("typical-as-standard must be rejected") + except ProtocolCatalogError: + pass