Add protocol-certification M5 Protocolli report UI (2.69.0).
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.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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 "",
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<string, unknown>): 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<string, unknown>;
|
||||
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 (
|
||||
<section className="rounded-lg border border-border bg-card p-4 space-y-2">
|
||||
<h2 className="text-sm font-semibold">Protocolli</h2>
|
||||
<section className="rounded-lg border border-border bg-card p-4 space-y-3">
|
||||
<div className="flex flex-wrap items-baseline justify-between gap-2">
|
||||
<h2 className="text-sm font-semibold">Protocolli</h2>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Livello massimo: <span className="font-mono text-foreground">{maxLevel}</span>
|
||||
{" "}
|
||||
(L0–L3; L3 not run unless channel data exists)
|
||||
</p>
|
||||
</div>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
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.
|
||||
</p>
|
||||
{instances.length === 0 ? (
|
||||
<p className="text-sm">{message}</p>
|
||||
) : (
|
||||
<ul className="text-sm space-y-2">
|
||||
<ul className="space-y-4">
|
||||
{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<string, unknown>;
|
||||
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 (
|
||||
<li key={String(row.instance_id ?? i)} className="space-y-0.5">
|
||||
<p className="font-mono text-xs">
|
||||
{log} → {phys}
|
||||
{status ? ` · ${status}` : ""}
|
||||
{typeof conf === "number" ? ` · ${conf.toFixed(2)}` : ""}
|
||||
</p>
|
||||
{groups.length > 0 ? (
|
||||
<p className="text-xs text-muted-foreground">
|
||||
{groups
|
||||
.map((g) => {
|
||||
const rec = g as Record<string, unknown>;
|
||||
return String(rec.kind ?? rec.id ?? "group");
|
||||
})
|
||||
.join(", ")}
|
||||
<li
|
||||
key={String(rec.instance_id ?? i)}
|
||||
className={cn(
|
||||
"rounded-md border p-3 space-y-2",
|
||||
fail
|
||||
? "border-rose-500/50 bg-rose-500/5"
|
||||
: "border-border",
|
||||
)}
|
||||
>
|
||||
<div className="flex flex-wrap items-center gap-2">
|
||||
<p className="font-mono text-xs">
|
||||
{log} → {phys}
|
||||
{status ? ` · ${status}` : ""}
|
||||
</p>
|
||||
{worst ? (
|
||||
<StatusBadge status={protocolResultToFindingStatus(worst)} />
|
||||
) : null}
|
||||
{fail ? (
|
||||
<span className="text-xs font-medium text-rose-600 dark:text-rose-400">
|
||||
FAIL visible — not under Warning
|
||||
</span>
|
||||
) : null}
|
||||
</div>
|
||||
<div className="overflow-x-auto">
|
||||
<table className="w-full text-xs">
|
||||
<thead>
|
||||
<tr className="text-left text-muted-foreground">
|
||||
<th className="pr-2 py-1 font-medium">Lv</th>
|
||||
<th className="pr-2 py-1 font-medium">Check</th>
|
||||
<th className="pr-2 py-1 font-medium">Result</th>
|
||||
<th className="pr-2 py-1 font-medium">Measured</th>
|
||||
<th className="pr-2 py-1 font-medium">Limit</th>
|
||||
<th className="pr-2 py-1 font-medium">Margin</th>
|
||||
<th className="pr-2 py-1 font-medium">Source</th>
|
||||
<th className="py-1 font-medium">Method</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{checks.map((c, j) => {
|
||||
const isFail = c.result === "FAIL";
|
||||
const skip = Boolean(c.skip_visible);
|
||||
return (
|
||||
<tr
|
||||
key={`${c.level}-${c.check}-${j}`}
|
||||
className={cn(
|
||||
isFail && "text-rose-700 dark:text-rose-400 font-medium",
|
||||
skip && !isFail && "text-amber-800 dark:text-amber-300",
|
||||
)}
|
||||
>
|
||||
<td className="pr-2 py-1 font-mono">{c.level}</td>
|
||||
<td className="pr-2 py-1">{c.check}</td>
|
||||
<td className="pr-2 py-1">
|
||||
{c.result}
|
||||
{skip ? " · skip" : ""}
|
||||
</td>
|
||||
<td className="pr-2 py-1 font-mono">{formatMeasured(c)}</td>
|
||||
<td className="pr-2 py-1 font-mono">{formatLimit(c)}</td>
|
||||
<td className="pr-2 py-1 font-mono">{formatMargin(c)}</td>
|
||||
<td className="pr-2 py-1">{c.source || "—"}</td>
|
||||
<td className="py-1">{c.method || "—"}</td>
|
||||
</tr>
|
||||
);
|
||||
})}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
{String(rec.chain_example || "") ? (
|
||||
<p className="text-[11px] text-muted-foreground break-all">
|
||||
Catena: {String(rec.chain_example)}
|
||||
</p>
|
||||
) : null}
|
||||
{Array.isArray(row.l0_checks) && row.l0_checks.length > 0 ? (
|
||||
<p className="text-xs text-muted-foreground">
|
||||
L0{" "}
|
||||
{(row.l0_checks as Record<string, unknown>[])
|
||||
.map((c) => `${String(c.check)}=${String(c.result)}`)
|
||||
.join("; ")}
|
||||
</p>
|
||||
) : null}
|
||||
{Array.isArray(row.l1_checks) && row.l1_checks.length > 0 ? (
|
||||
<p className="text-xs text-muted-foreground">
|
||||
L1{" "}
|
||||
{(row.l1_checks as Record<string, unknown>[])
|
||||
.map((c) => `${String(c.check)}=${String(c.result)}`)
|
||||
.join("; ")}
|
||||
</p>
|
||||
) : null}
|
||||
{Array.isArray(row.l2_checks) && row.l2_checks.length > 0 ? (
|
||||
<p className="text-xs text-muted-foreground">
|
||||
L2{" "}
|
||||
{(row.l2_checks as Record<string, unknown>[])
|
||||
.map((c) => `${String(c.check)}=${String(c.result)}`)
|
||||
.join("; ")}
|
||||
</p>
|
||||
{checks
|
||||
.filter((c) => c.result === "FAIL" || c.skip_visible)
|
||||
.map((c, j) =>
|
||||
c.chain ? (
|
||||
<p
|
||||
key={`chain-${j}`}
|
||||
className="text-[11px] text-muted-foreground break-all"
|
||||
>
|
||||
{c.level}/{c.check}: {formatProtocolChain(c.chain)}
|
||||
</p>
|
||||
) : null,
|
||||
)}
|
||||
{checks.some((c) => c.skip_visible && c.notes) ? (
|
||||
<ul className="text-xs text-amber-800 dark:text-amber-300 space-y-0.5">
|
||||
{checks
|
||||
.filter((c) => c.skip_visible && c.notes)
|
||||
.map((c, j) => (
|
||||
<li key={`skip-${j}`}>{c.notes}</li>
|
||||
))}
|
||||
</ul>
|
||||
) : null}
|
||||
</li>
|
||||
);
|
||||
|
||||
@@ -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<Finding> & Pick<Finding, "designator" | "status">): 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"));
|
||||
});
|
||||
@@ -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<T extends { result: string; level?: string; check?: string }>(
|
||||
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}`;
|
||||
}
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user