Add protocol-certification M9 L3 channel certifier (2.71.0).

L3 CHANNEL structure for timing budget, insertion loss, return loss,
crosstalk, and complete channel. Numeric PASS/FAIL only from existing
channel FACT; otherwise a visible skip. No OpenEMS, no invented S-parameters,
no M6 packs.
This commit is contained in:
2026-09-22 12:58:18 +02:00
parent 14272d7446
commit 866c2e9242
14 changed files with 987 additions and 26 deletions
@@ -304,6 +304,10 @@ def _seed() -> None:
requirement="L2 electrical skip: Z/termination/levels need datasheet/stackup/fab/cited pack.")
_add("PE-PRT-L2-002", "MANDATORY", "RULE", domain="pcb",
requirement="L2 Z/termination vs cited pack or PHY datasheet window.")
_add("PE-PRT-L3-001", "TYPICAL", "REVIEW", domain="pcb",
requirement="L3 channel skip: missing existing IL/RL/crosstalk/budget FACT (no OpenEMS).")
_add("PE-PRT-L3-002", "MANDATORY", "RULE", domain="pcb",
requirement="L3 channel vs cited IL/RL/crosstalk/timing-budget FACT (no invented S-param).")
_seed()
@@ -300,11 +300,13 @@ def protocol_exam(
layout: LayoutGraph | None = None,
constraints_map: dict | None = None,
impedance_nets: list[dict] | dict | None = None,
channel_data: dict | None = None,
) -> tuple[ProtocolCertificationSection, list[Finding]]:
from backend.periscopex.protocol_l1 import certify_l1
from backend.periscopex.protocol_l2 import LEVEL as L2_LEVEL, certify_l2
from backend.periscopex.protocol_l2 import certify_l2
from backend.periscopex.protocol_l3 import LEVEL as L3_LEVEL, certify_l3
from backend.periscopex.protocol_report import (
PACK_MACROPHASE as L8_MACRO,
PACK_MACROPHASE as REPORT_MACRO,
attach_instance_report,
result_rank,
)
@@ -315,7 +317,7 @@ def protocol_exam(
instances = recognize_physical_buses(graph, cat)
if not instances:
sec = empty_protocol_section()
sec.macrophase = L8_MACRO
sec.macrophase = REPORT_MACRO
return sec, []
l0, findings = certify_l0(graph, instances, cat)
l1, l1_findings_out = certify_l1(graph, instances, layout, cat)
@@ -327,6 +329,10 @@ def protocol_exam(
impedance_nets=impedance_nets,
)
findings.extend(l2_findings_out)
l3, l3_findings_out = certify_l3(
instances, cat, channel_data=channel_data,
)
findings.extend(l3_findings_out)
ifaces = {p.id: p for p in cat.physical_interfaces}
dumps: list[dict[str, Any]] = []
for inst in instances:
@@ -334,20 +340,22 @@ def protocol_exam(
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["l3_checks"] = [c.model_dump() for c in l3.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, []),
l3.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=L8_MACRO,
macrophase=REPORT_MACRO,
recognized_instances=dumps,
message="",
max_level_reached=L2_LEVEL,
max_level_reached=L3_LEVEL,
)
return sec, findings
@@ -0,0 +1,601 @@
"""M9: L3 CHANNEL protocol certifier. Timing budget / IL / RL / crosstalk.
Numerical execution only when channel FACT already exists (extracted dB/ps).
Never invent S-parameters. Never OpenEMS/FEM. Never M6 pack numbers.
L0/L1 are not electrical; this module does not call L0/L1 electrical.
RECOMMENDED never FAIL. Internals (pcb_relevant=NO) are NOT_APPLICABLE.
Visible skip: «Interfaccia X non certificata a L3 per mancanza di …».
"""
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, Field
from backend.periscopex.finding_engine import complete_finding
from backend.periscopex.models import DesignGraph, Finding
from backend.periscopex.protocol_catalog import (
PhysicalInterface,
ProtocolCatalog,
ProtocolConstraint,
load_catalog,
map_protocol_outcome,
)
from backend.periscopex.protocol_l0 import _constraint_for, _mandatory
from backend.periscopex.protocol_recognize import PhysicalBusInstance
PACK_MACROPHASE = "M9"
SOURCE = "protocol_l3"
LEVEL = "L3"
L3_CHECKS = frozenset({
"timing_budget",
"insertion_loss",
"return_loss",
"crosstalk",
"channel",
"connector",
"via_transitions",
"via_transition",
"package",
})
# Always structured on PCB-relevant instances even if the catalog omits them.
L3_CORE = (
"timing_budget",
"insertion_loss",
"return_loss",
"crosstalk",
"channel",
)
_DB_UNITS = frozenset({"db", "dbm", "dbi"})
_TIME_UNITS = frozenset({"ps", "ns", "us", "µs", "ms", "s"})
_FEM_MARKERS = ("openems", "fem", "field_solver", "field-solver", "full-wave")
MandatoryClass = Literal["MANDATORY", "RECOMMENDED", "OPTIONAL", "INFORMATIONAL"]
L3Result = Literal[
"PASS", "FAIL", "WARNING", "UNKNOWN", "NOT_APPLICABLE",
"VENDOR_DEPENDENT", "CONTROLLER_DEPENDENT", "PHY_DEPENDENT", "MISSING_SOURCE",
]
class L3CheckResult(BaseModel):
check: str
level: Literal["L3"] = "L3"
result: L3Result
mandatory: MandatoryClass
nets: list[str] = Field(default_factory=list)
measured_db: float | None = None
limit_db: float | None = None
margin_db: float | None = None
measured_ps: float | None = None
limit_ps: float | None = None
margin_ps: float | None = None
total_budget_ps: float | None = None
used_budget_ps: float | None = None
remaining_margin_ps: float | None = None
value_kind: str = ""
notes: str = ""
finding_class: str = ""
status: str = ""
def _num(v: Any) -> float | None:
if v is None or isinstance(v, bool):
return None
try:
return float(v)
except (TypeError, ValueError):
return None
def _pack(
check: str,
result: str,
mandatory: MandatoryClass,
*,
nets: list[str] | None = None,
measured_db: float | None = None,
limit_db: float | None = None,
measured_ps: float | None = None,
limit_ps: float | None = None,
total_budget_ps: float | None = None,
used_budget_ps: float | None = None,
remaining_margin_ps: float | None = None,
value_kind: str = "",
notes: str = "",
) -> L3CheckResult:
if result == "FAIL" and mandatory != "MANDATORY":
result = "WARNING"
mapped = result
if result in {"CONTROLLER_DEPENDENT", "PHY_DEPENDENT"}:
mapped = "VENDOR_DEPENDENT"
cls, status, _ev = map_protocol_outcome(mapped, mandatory) # type: ignore[arg-type]
margin_db = None
if measured_db is not None and limit_db is not None:
margin_db = _db_margin(check, measured_db, limit_db)
margin_ps = remaining_margin_ps
if margin_ps is None and measured_ps is not None and limit_ps is not None:
margin_ps = measured_ps - limit_ps if limit_ps < 0 else limit_ps - measured_ps
return L3CheckResult(
check=check,
result=result, # type: ignore[arg-type]
mandatory=mandatory,
nets=nets or [],
measured_db=measured_db,
limit_db=limit_db,
margin_db=margin_db,
measured_ps=measured_ps,
limit_ps=limit_ps,
margin_ps=margin_ps,
total_budget_ps=total_budget_ps,
used_budget_ps=used_budget_ps,
remaining_margin_ps=remaining_margin_ps,
value_kind=value_kind,
notes=notes,
finding_class=cls,
status=status,
)
def _db_margin(check: str, measured: float, limit: float) -> float:
if check == "return_loss":
if limit < 0:
return limit - measured
return measured - limit
if limit < 0:
return measured - limit
return limit - measured
def _db_pass(check: str, measured: float, limit: float) -> bool:
"""IL/crosstalk: max magnitude (pos) or min S-param (neg). RL: min (pos) or max S11 (neg)."""
if check == "return_loss":
if limit < 0:
return measured <= limit
return measured >= limit
if limit < 0:
return measured >= limit
return measured <= limit
def _instance_nets(inst: PhysicalBusInstance) -> list[str]:
names: list[str] = []
for g in inst.groups:
names.extend(g.nets)
names.extend(inst.nets)
seen: set[str] = set()
out: list[str] = []
for n in names:
if n not in seen:
seen.add(n)
out.append(n)
return out
def _kind_skip(kind: str) -> str:
if kind in {
"MISSING_SOURCE", "PHY_DEPENDENT", "CONTROLLER_DEPENDENT",
"VENDOR_DEPENDENT", "UNKNOWN", "NOT_APPLICABLE",
}:
return kind
return "UNKNOWN"
def _blob_mentions_fem(obj: Any) -> bool:
text = str(obj).lower()
return any(m in text for m in _FEM_MARKERS)
def _looks_like_raw_sparam(facts: dict[str, Any]) -> bool:
"""Raw Touchstone / unevaluated S-matrix — we do not reduce it here."""
if facts.get("sparam_invented") is True:
return True
if facts.get("invent_sparam") is True:
return True
for key in ("touchstone", "sparam_file", "s2p", "s4p", "snp_path"):
if facts.get(key):
has_num = any(
_num(facts.get(k)) is not None
for k in (
"insertion_loss_db", "il_db", "s21_db",
"return_loss_db", "rl_db", "s11_db",
)
)
if not has_num:
return True
return False
def _as_fact_map(channel_data: Any) -> dict[str, Any]:
if channel_data is None:
return {}
if isinstance(channel_data, dict):
return channel_data
return {}
def lookup_channel_facts(
inst: PhysicalBusInstance,
channel_data: Any,
) -> dict[str, Any]:
"""Return FACT already on disk/in-memory. Never synthesize S-parameters."""
root = _as_fact_map(channel_data)
if not root:
return {}
candidates: list[Any] = []
for bag_key in ("instances", "physical_interfaces", "logical_protocols"):
bag = root.get(bag_key)
if isinstance(bag, dict):
if bag_key == "instances":
candidates.append(bag.get(inst.instance_id))
elif bag_key == "physical_interfaces":
candidates.append(bag.get(inst.physical_interface_id))
else:
candidates.append(bag.get(inst.logical_protocol_id))
candidates.append(root.get(inst.instance_id))
candidates.append(root.get(inst.physical_interface_id))
candidates.append(root.get(inst.logical_protocol_id))
default = root.get("default")
if isinstance(default, dict):
candidates.append(default)
records = root.get("records")
if isinstance(records, list):
for rec in records:
if not isinstance(rec, dict):
continue
if rec.get("instance_id") == inst.instance_id:
candidates.append(rec)
elif rec.get("physical_interface_id") == inst.physical_interface_id:
candidates.append(rec)
elif rec.get("logical_protocol_id") == inst.logical_protocol_id:
candidates.append(rec)
merged: dict[str, Any] = {}
for c in candidates:
if isinstance(c, dict):
merged.update(c)
return merged
def remaining_margin_ps(
total: float | None,
used: float | None,
remaining: float | None = None,
) -> float | None:
"""TOTAL USED = remaining margin. Prefer explicit remaining if given."""
if remaining is not None:
return remaining
if total is None or used is None:
return None
return total - used
def _used_from_parts(facts: dict[str, Any]) -> float | None:
keys = (
"package_delay_ps",
"pcb_trace_delay_ps",
"via_delay_ps",
"connector_delay_ps",
"cable_delay_ps",
"component_delay_ps",
)
parts = [_num(facts.get(k)) for k in keys]
present = [p for p in parts if p is not None]
if not present:
return _num(facts.get("used_budget_ps"))
extra = _num(facts.get("used_budget_ps"))
if extra is not None and extra >= sum(present):
return extra
return sum(present)
def _il_measured(facts: dict[str, Any]) -> float | None:
for k in ("insertion_loss_db", "il_db", "s21_db"):
n = _num(facts.get(k))
if n is not None:
return n
return None
def _rl_measured(facts: dict[str, Any]) -> float | None:
for k in ("return_loss_db", "rl_db", "s11_db"):
n = _num(facts.get(k))
if n is not None:
return n
return None
def _xt_measured(facts: dict[str, Any]) -> float | None:
for k in ("crosstalk_db", "xtalk_db", "next_db", "fext_db"):
n = _num(facts.get(k))
if n is not None:
return n
return None
def _limit_from_constraint(cons: ProtocolConstraint | None, units: set[str]) -> float | None:
if cons is None or cons.value_kind != "NUMERIC" or cons.value is None:
return None
unit = (cons.unit or "").strip().lower()
if unit in units or unit == "":
return cons.value
return None
def _skip(
check: str,
mand: MandatoryClass,
inst: PhysicalBusInstance,
kind: str,
missing: str,
*,
nets: list[str] | None = None,
) -> L3CheckResult:
return _pack(
check,
_kind_skip(kind),
mand,
nets=nets,
value_kind=kind,
notes=(
f"Interfaccia {inst.physical_interface_id} non certificata a L3 "
f"per mancanza di {missing}. OpenEMS/FEM out of product; S-parameters "
f"are not invented. L0/L1 are not electrical certification. Not PASS."
),
)
def _run_l3_check(
inst: PhysicalBusInstance,
iface: PhysicalInterface,
check: str,
facts: dict[str, Any],
) -> L3CheckResult:
mand = _mandatory(iface, check)
cons = _constraint_for(iface, check)
kind = cons.value_kind if cons else "UNKNOWN"
nets = _instance_nets(inst)
if facts and (_looks_like_raw_sparam(facts) or _blob_mentions_fem(facts.get("method") or facts.get("solver") or "")):
return _skip(
check, mand, inst, kind,
"dati canale ridotti a FACT (Touchstone/OpenEMS/FEM non eseguiti nel prodotto)",
nets=nets,
)
if check == "timing_budget":
total = _num(facts.get("total_budget_ps"))
used = _used_from_parts(facts)
remaining = remaining_margin_ps(total, used, _num(facts.get("remaining_margin_ps")))
limit = _limit_from_constraint(cons, _TIME_UNITS)
if remaining is None or total is None:
return _skip(
check, mand, inst, kind,
"timing budget FACT (total_budget_ps e used/remaining)",
nets=nets,
)
if cons is None or cons.value_kind != "NUMERIC":
# Measured budget exists but no cited min-margin — report numbers, skip cert.
return _pack(
check, _kind_skip(kind), mand, nets=nets,
total_budget_ps=total, used_budget_ps=used,
remaining_margin_ps=remaining, measured_ps=remaining,
value_kind=kind,
notes=(
f"Timing budget FACT total={total:g} ps used={used if used is not None else ''} "
f"remaining={remaining:g} ps. Interfaccia {inst.physical_interface_id} "
f"non certificata a L3 per mancanza di limite citato. Not invented. Not OpenEMS."
),
)
min_margin = limit if limit is not None else 0.0
verdict = "PASS" if remaining >= min_margin else "FAIL"
return _pack(
check, verdict, mand, nets=nets,
total_budget_ps=total, used_budget_ps=used,
remaining_margin_ps=remaining, measured_ps=remaining,
limit_ps=min_margin, value_kind="NUMERIC",
notes=(
f"L3 timing budget TOTALUSED=MARGIN ({total:g}{used if used is not None else 0:g}={remaining:g} ps). "
f"Cited min margin {min_margin:g} ps. Not L0/L1 electrical. Not OpenEMS."
),
)
if check in {"insertion_loss", "return_loss", "crosstalk"}:
if check == "insertion_loss":
measured = _il_measured(facts)
missing = "insertion loss FACT (dB) — S-parameters non inventati"
elif check == "return_loss":
measured = _rl_measured(facts)
missing = "return loss FACT (dB) — S-parameters non inventati"
else:
measured = _xt_measured(facts)
missing = "crosstalk FACT (dB) — S-parameters non inventati"
limit = _limit_from_constraint(cons, _DB_UNITS)
if measured is None:
return _skip(check, mand, inst, kind, missing, nets=nets)
if cons is None or cons.value_kind != "NUMERIC" or limit is None:
return _pack(
check, _kind_skip(kind), mand, nets=nets,
measured_db=measured, value_kind=kind,
notes=(
f"Channel FACT {check}={measured:g} dB. Interfaccia "
f"{inst.physical_interface_id} non certificata a L3 per mancanza di "
f"limite citato (pack NUMERIC). Not invented S-param. Not OpenEMS."
),
)
verdict = "PASS" if _db_pass(check, measured, limit) else "FAIL"
return _pack(
check, verdict, mand, nets=nets,
measured_db=measured, limit_db=limit, value_kind="NUMERIC",
notes=(
f"L3 {check} FACT {measured:g} dB vs cited {limit:g} dB. "
f"Not invented. Not OpenEMS. L0/L1 are not electrical."
),
)
if check == "channel":
if not facts:
return _skip(
check, mand, inst, kind,
"modello di canale completo (IL/RL/budget FACT già presenti)",
nets=nets,
)
if facts.get("channel_complete") is True:
il = _il_measured(facts)
rl = _rl_measured(facts)
total = _num(facts.get("total_budget_ps"))
used = _used_from_parts(facts)
remaining = remaining_margin_ps(total, used, _num(facts.get("remaining_margin_ps")))
if il is not None and rl is not None and remaining is not None:
il_c = _constraint_for(iface, "insertion_loss")
rl_c = _constraint_for(iface, "return_loss")
tb_c = _constraint_for(iface, "timing_budget")
il_lim = _limit_from_constraint(il_c, _DB_UNITS)
rl_lim = _limit_from_constraint(rl_c, _DB_UNITS)
tb_lim = _limit_from_constraint(tb_c, _TIME_UNITS)
if il_lim is None or rl_lim is None:
return _skip(
check, mand, inst, kind,
"limiti citati per IL/RL sul canale completo",
nets=nets,
)
ok = _db_pass("insertion_loss", il, il_lim) and _db_pass("return_loss", rl, rl_lim)
min_m = tb_lim if tb_lim is not None else 0.0
ok = ok and remaining >= min_m
return _pack(
check, "PASS" if ok else "FAIL", mand, nets=nets,
measured_db=il, remaining_margin_ps=remaining, value_kind="NUMERIC",
notes=(
"L3 complete channel from existing FACT (not OpenEMS). "
f"IL={il:g} dB RL={rl:g} dB margin={remaining:g} ps."
),
)
return _skip(
check, mand, inst, kind,
"canale completo (channel_complete + IL/RL/budget FACT)",
nets=nets,
)
# connector / via / package at L3 — FACT only, never FEM
if check in {"connector", "via_transitions", "via_transition", "package"}:
return _skip(
check, mand, inst, kind,
f"{check} channel FACT (no OpenEMS/FEM)",
nets=nets,
)
return _pack(check, "UNKNOWN", mand, value_kind=kind, notes="Not an L3 channel check.")
def certify_instance_l3(
inst: PhysicalBusInstance,
iface: PhysicalInterface,
channel_data: Any = None,
) -> list[L3CheckResult]:
if inst.pcb_relevant == "NO" or iface.pcb_relevant == "NO":
return [_pack(
"channel", "NOT_APPLICABLE", "INFORMATIONAL",
notes=(
"Internal interconnect — L3 channel N/A, not FAIL. "
"L0/L1 are not electrical. Not OpenEMS."
),
)]
checks: list[str] = []
for c in L3_CORE:
if c not in checks:
checks.append(c)
for c in iface.required_checks:
if c in L3_CHECKS and c not in checks:
checks.append(c)
facts = lookup_channel_facts(inst, channel_data)
return [_run_l3_check(inst, iface, c, facts) for c in checks]
def l3_findings(inst: PhysicalBusInstance, results: list[L3CheckResult]) -> list[Finding]:
out: list[Finding] = []
designator = inst.host_ref or inst.physical_interface_id
for r in results:
if r.result == "PASS":
continue
if r.result == "FAIL" and r.mandatory != "MANDATORY":
continue
skip_family = r.result in {
"UNKNOWN", "VENDOR_DEPENDENT", "CONTROLLER_DEPENDENT",
"PHY_DEPENDENT", "MISSING_SOURCE",
}
fail = r.result == "FAIL" and r.mandatory == "MANDATORY"
na = r.result == "NOT_APPLICABLE"
if not skip_family and not fail and not na:
continue
if fail:
rule_id = "PE-PRT-L3-002"
status = "ERROR"
finding = f"L3 channel {r.check}: {r.notes}"
elif skip_family:
rule_id = "PE-PRT-L3-001"
status = "WARNING"
finding = r.notes or (
f"Interfaccia {inst.physical_interface_id} non certificata a L3 "
f"per {r.check} ({r.result})."
)
else:
rule_id = "PE-PRT-L0-002"
status = "INFO"
finding = r.notes
f = Finding(
designator=designator,
mpn="",
aspect="protocol_l3",
finding=finding,
why=r.notes,
status=status,
source=SOURCE,
net=r.nets[0] if r.nets else None,
rule_id=rule_id,
facts=finding,
requirement="L3 channel only from existing FACT (dB/ps). No OpenEMS. No invented S-param.",
inference="L3 is channel. L0/L1 are not electrical. Timing budget = totalused=margin.",
provenance="MANDATORY" if fail else "TYPICAL",
finding_class="RULE" if fail else ("REVIEW" if skip_family else "INFO"),
evidence_status="INSUFFICIENT" if skip_family else "SUFFICIENT",
)
out.append(complete_finding(f))
return out
def certify_l3(
instances: list[PhysicalBusInstance],
catalog: ProtocolCatalog | None = None,
*,
channel_data: Any = None,
) -> tuple[dict[str, list[L3CheckResult]], list[Finding]]:
cat = catalog or load_catalog()
ifaces = {p.id: p for p in cat.physical_interfaces}
by_id: dict[str, list[L3CheckResult]] = {}
findings: list[Finding] = []
for inst in instances:
iface = ifaces.get(inst.physical_interface_id)
if iface is None:
continue
rows = certify_instance_l3(inst, iface, channel_data=channel_data)
by_id[inst.instance_id] = rows
findings.extend(l3_findings(inst, rows))
return by_id, findings
def load_channel_data_file(path: Any) -> dict[str, Any] | None:
"""Load existing channel_data.json. Does not parse Touchstone or run FEM."""
from pathlib import Path
import json
p = Path(path) if path is not None else None
if p is None or not p.is_file():
return None
try:
raw = json.loads(p.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
if not isinstance(raw, dict):
return None
return raw
@@ -1,6 +1,7 @@
"""M5: protocol report view — checks, chain, FAIL before WARNING, visible skips.
"""M5/M9: protocol report view — checks, chain, FAIL before WARNING, visible skips.
Does not invent Z. Does not run L3/OpenEMS. Does not add pack numbers.
Does not invent Z or S-parameters. L3 uses existing channel FACT only.
OpenEMS/FEM out of product. Does not add pack numbers.
L0/L1 are not electrical certification.
"""
@@ -12,7 +13,7 @@ from backend.periscopex.protocol_catalog import PhysicalInterface, ProtocolConst
from backend.periscopex.protocol_l0 import _constraint_for
from backend.periscopex.protocol_recognize import PhysicalBusInstance
PACK_MACROPHASE = "M8"
PACK_MACROPHASE = "M9"
_SKIP = frozenset({
"UNKNOWN", "MISSING_SOURCE", "VENDOR_DEPENDENT",
@@ -45,11 +46,19 @@ def _unit(dump: dict[str, Any]) -> str:
return "mm"
if dump.get("measured_ohm") is not None or dump.get("limit_ohm") is not None:
return "ohm"
if dump.get("measured_db") is not None or dump.get("limit_db") is not None:
return "dB"
if (
dump.get("measured_ps") is not None
or dump.get("limit_ps") is not None
or dump.get("remaining_margin_ps") is not None
):
return "ps"
return ""
def _measured(dump: dict[str, Any]) -> float | None:
for k in ("measured_ohm", "measured_mm"):
for k in ("measured_ohm", "measured_mm", "measured_db", "measured_ps", "remaining_margin_ps"):
v = dump.get(k)
if isinstance(v, (int, float)):
return float(v)
@@ -57,7 +66,7 @@ def _measured(dump: dict[str, Any]) -> float | None:
def _limit(dump: dict[str, Any]) -> float | None:
for k in ("limit_ohm", "limit_mm", "limit_ohm_max"):
for k in ("limit_ohm", "limit_mm", "limit_ohm_max", "limit_db", "limit_ps"):
v = dump.get(k)
if isinstance(v, (int, float)):
return float(v)
@@ -65,7 +74,7 @@ def _limit(dump: dict[str, Any]) -> float | None:
def _margin(dump: dict[str, Any]) -> float | None:
for k in ("margin_ohm", "margin_mm"):
for k in ("margin_ohm", "margin_mm", "margin_db", "margin_ps", "remaining_margin_ps"):
v = dump.get(k)
if isinstance(v, (int, float)):
return float(v)
@@ -178,17 +187,20 @@ def attach_instance_report(
l0: list[Any],
l1: list[Any],
l2: list[Any],
l3: list[Any] | None = None,
) -> 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]
+ [c.model_dump() if hasattr(c, "model_dump") else dict(c) for c in (l3 or [])]
)
rows = [check_to_report_row(d, inst, iface) for d in dumps]
rows.append(l3_not_run_row(inst))
if not any(r["level"] == "L3" for r in rows):
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
# L3 skip/UNKNOWN must not upgrade instance to PASS or hide FAIL
return {
"report_checks": rows,
"worst_result": worst,
@@ -111,6 +111,17 @@ def _load_constraints_map(
return result
def _load_channel_data(ws: PipelineWorkspace) -> dict | None:
"""Use existing channel_data.json only. Do not parse Touchstone or run OpenEMS."""
from backend.periscopex.protocol_l3 import load_channel_data_file
for rel in ("channel_data.json", "uploads/channel_data.json"):
loaded = load_channel_data_file(ws.local_path(rel))
if loaded is not None:
return loaded
return None
def si_extract_needed_skips(
graph: DesignGraph,
cmap: dict,
@@ -379,6 +390,7 @@ async def run_pcb_pipeline(
proto_section, proto_findings = protocol_exam(
graph, layout=layout,
constraints_map=cmap, impedance_nets=zrep,
channel_data=_load_channel_data(ws),
)
findings.extend(proto_findings)
assign_pcb_finding_ids(findings)
@@ -2,6 +2,13 @@
What's new in Periscope.
## 2.71.0 — 2026-09-22 — Protocol certification M9 (L3 channel + visible skip)
L3 **CHANNEL** certifier: structure for timing budget (TOTALUSED=MARGIN), insertion loss, return loss, crosstalk, and complete channel. Numerical PASS/FAIL only when **existing channel FACT** (dB/ps) is already present — never invented S-parameters, never OpenEMS/FEM. Otherwise a visible skip: «Interfaccia X non certificata a L3 per mancanza di …». Internals (AXI/HBM) stay NOT_APPLICABLE. L0/L1 are not electrical. No M6 numeric packs.
- [New] `protocol_l3.py`; `PE-PRT-L3-001` / `PE-PRT-L3-002`.
- [New] pytest `tests/pcb/test_protocol_l3_m9.py`.
## 2.70.0 — 2026-09-22 — Protocol certification M8 (DDR instance + L1 grouping)
DDR **physical instances** are controller-aware: one instance per memory device, DQ/DQS **byte lanes** separate from ADDRESS / COMMAND / CONTROL / CLOCK. Skew/length stay **CONTROLLER_DEPENDENT / PHY_DEPENDENT / UNKNOWN** — no universal mm/ps/ohm. HBM is package/interposer (`pcb_relevant: NO`); classic DDR PCB DQ rules are not applied. M6 numeric packs and L3 solver are not in this release.
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "periscope-web",
"version": "2.70.0",
"version": "2.71.0",
"private": true,
"scripts": {
"sync-version": "node scripts/sync-version.mjs",
@@ -19,7 +19,7 @@ function asChecks(row: Record<string, unknown>): ProtocolReportCheck[] {
return sortProtocolChecks(raw as ProtocolReportCheck[]);
}
const fallback: ProtocolReportCheck[] = [];
for (const key of ["l0_checks", "l1_checks", "l2_checks"] as const) {
for (const key of ["l0_checks", "l1_checks", "l2_checks", "l3_checks"] as const) {
const list = row[key];
if (!Array.isArray(list)) continue;
for (const c of list) {
@@ -53,12 +53,13 @@ export function ProtocolSection({
<p className="text-xs text-muted-foreground">
Livello massimo: <span className="font-mono text-foreground">{maxLevel}</span>
{" "}
(L0L3; L3 not run unless channel data exists)
(L0L3; L3 skip unless channel FACT exists no OpenEMS)
</p>
</div>
<p className="text-xs text-muted-foreground">
Logical protocol vs physical interface. L0 structural, L1 geometric,
L2 electrical (Z only from datasheet/stackup/fab or a cited pack).
L3 channel (timing budget / IL / RL / crosstalk) only from existing FACT.
L0/L1 are not electrical. Length is not delay. FAIL is listed first
it is not filed under Warning.
</p>
+1 -1
View File
@@ -141,7 +141,7 @@ def test_hbm_not_classic_ddr_dq_rules():
assert one.groups == []
assert one.physical_interface_id.endswith("-package")
sec, findings = protocol_exam(_hbm_graph())
assert sec.macrophase == "M8"
assert sec.macrophase == "M9"
assert all(f.status != "ERROR" for f in findings)
rows = [
c for r in sec.recognized_instances if "hbm" in str(r.get("logical_protocol_id"))
+1 -1
View File
@@ -129,7 +129,7 @@ def test_axi_internal_l0_not_applicable_not_fail():
assert all(r.result == "NOT_APPLICABLE" for r in rows)
assert all(r.result != "FAIL" for r in rows)
sec, findings = protocol_exam(_axi_graph())
assert sec.max_level_reached == "L2"
assert sec.max_level_reached == "L3"
from backend.periscopex.protocol_report import PACK_MACROPHASE as L8_MACRO
assert sec.macrophase == L8_MACRO
assert all(f.status != "ERROR" for f in findings)
+3 -3
View File
@@ -165,8 +165,8 @@ def test_axi_internal_l1_not_applicable():
assert all(r.result == "NOT_APPLICABLE" for r in rows)
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 == "M8"
assert sec.max_level_reached == "L3"
assert sec.macrophase == "M9"
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()
@@ -257,7 +257,7 @@ def test_l1_does_not_run_impedance():
def test_protocol_exam_attaches_l1_checks():
sec, _ = protocol_exam(_usb_connected(), layout=_usb_layout())
assert sec.max_level_reached == "L2"
assert sec.max_level_reached == "L3"
assert sec.recognized_instances
row = sec.recognized_instances[0]
assert row.get("l1_checks")
+3 -3
View File
@@ -204,8 +204,8 @@ def test_axi_internal_l2_not_applicable():
assert rows
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 == "M8"
assert sec.max_level_reached == "L3"
assert sec.macrophase == "M9"
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
@@ -253,7 +253,7 @@ def test_l2_does_not_run_l3_or_geometry():
def test_protocol_exam_attaches_l2_checks():
sec, _ = protocol_exam(_usb_connected())
assert sec.max_level_reached == "L2"
assert sec.max_level_reached == "L3"
row = sec.recognized_instances[0]
assert row.get("l2_checks")
assert all(c.get("level") == "L2" for c in row["l2_checks"])
+316
View File
@@ -0,0 +1,316 @@
"""M9 L3 CHANNEL certifier: structure + skip; numeric only with existing FACT."""
from __future__ import annotations
import json
from backend.periscopex.models import (
Component,
ComponentType,
DesignGraph,
Net,
NetType,
PinConnection,
)
from backend.periscopex.protocol_catalog import (
ConstraintSource,
PhysicalInterface,
ProtocolConstraint,
)
from backend.periscopex.protocol_l0 import protocol_exam
from backend.periscopex.protocol_l3 import (
_pack,
certify_instance_l3,
certify_l3,
remaining_margin_ps,
)
from backend.periscopex.protocol_recognize import (
PhysicalBusInstance,
recognize_physical_buses,
)
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 _hdmi_graph() -> DesignGraph:
pins = {"1": "HDMI_TX0_P", "2": "HDMI_TX0_N", "3": "HDMI_CLK_P", "4": "HDMI_CLK_N"}
return DesignGraph(
components={
"U8": _ic("U8", pins, mpn="ADV7511", value="HDMI transmitter"),
"J8": Component(
reference="J8", value="HDMI-A",
footprint="HDMI",
component_type=ComponentType.CONNECTOR,
pins=pins,
),
},
nets={n: _net(n, ("U8", p), ("J8", p)) for p, n in pins.items()},
schematic_fields={"J8": {"protocol": "hdmi-1.4-tmds-type-a"}},
)
def _pcie_graph() -> DesignGraph:
pins = {"1": "PCIE_TX_P", "2": "PCIE_TX_N", "3": "PCIE_RX_P", "4": "PCIE_RX_N"}
return DesignGraph(
components={
"U7": _ic("U7", pins, mpn="PI7C9X2G304GP", value="PCIe switch"),
"J7": Component(
reference="J7", value="PCIe x1",
footprint="PCIE",
component_type=ComponentType.CONNECTOR,
pins=pins,
),
},
nets={n: _net(n, ("U7", p), ("J7", p)) for p, n in pins.items()},
schematic_fields={"U7": {"protocol": "pcie-phy-pcb"}},
)
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 _cite() -> ConstraintSource:
return ConstraintSource(
document="Fixture channel pack", organization="TESTLAB", section="4.1",
)
def _iface_with_numeric(*, il_db: float = -8.0, rl_db: float = -10.0,
margin_ps: float = 0.0) -> PhysicalInterface:
return PhysicalInterface(
id="fixture-hs-dpair",
logical_protocol_id="fixture-hs",
bus_type="SERIAL",
physical_layer="SERIAL_DIFFERENTIAL",
pcb_relevant="YES",
required_checks=[
"insertion_loss", "return_loss", "crosstalk",
"timing_budget", "channel",
],
constraints=[
ProtocolConstraint(
id="fixture-il", parameter="insertion_loss",
value_kind="NUMERIC", mandatory="MANDATORY",
source_type="STANDARD", source_class="NORMATIVE",
value=il_db, unit="dB", source=_cite(),
),
ProtocolConstraint(
id="fixture-rl", parameter="return_loss",
value_kind="NUMERIC", mandatory="MANDATORY",
source_type="STANDARD", source_class="NORMATIVE",
value=rl_db, unit="dB", source=_cite(),
),
ProtocolConstraint(
id="fixture-xt", parameter="crosstalk",
value_kind="NUMERIC", mandatory="RECOMMENDED",
source_type="STANDARD", source_class="NORMATIVE",
value=-20.0, unit="dB", source=_cite(),
),
ProtocolConstraint(
id="fixture-tb", parameter="timing_budget",
value_kind="NUMERIC", mandatory="MANDATORY",
source_type="STANDARD", source_class="NORMATIVE",
value=margin_ps, unit="ps", source=_cite(),
),
],
)
def _inst() -> PhysicalBusInstance:
return PhysicalBusInstance(
instance_id="fixture-1",
logical_protocol_id="fixture-hs",
physical_interface_id="fixture-hs-dpair",
pcb_relevant="YES",
confidence=1.0,
evidence_kind="net_name",
recognition_status="RECOGNIZED",
nets=["CH_P", "CH_N"],
host_ref="U1",
)
def test_remaining_margin_is_total_minus_used():
assert remaining_margin_ps(200.0, 120.0) == 80.0
assert remaining_margin_ps(200.0, 120.0, 77.0) == 77.0
assert remaining_margin_ps(None, 10.0) is None
def test_usb_hdmi_pcie_skip_without_channel_data():
for graph, token in ((_usb_connected(), "usb"), (_hdmi_graph(), "hdmi"), (_pcie_graph(), "pcie")):
sec, findings = protocol_exam(graph)
assert sec.max_level_reached == "L3"
assert sec.macrophase == "M9"
row = next(
r for r in sec.recognized_instances
if token in str(r.get("logical_protocol_id")).lower()
or token in str(r.get("physical_interface_id")).lower()
)
l3 = row["l3_checks"]
assert l3
names = {c["check"] for c in l3}
assert {"timing_budget", "insertion_loss", "return_loss", "crosstalk", "channel"} <= names
assert all(c["result"] != "PASS" for c in l3)
notes = " ".join(c.get("notes") or "" for c in l3)
assert "non certificata a L3 per mancanza" in notes
assert "OpenEMS" in notes
blob = json.dumps(l3)
assert "90" not in blob
assert any(f.rule_id == "PE-PRT-L3-001" for f in findings)
assert "s21" not in blob.lower() or "invent" in notes.lower()
def test_axi_l3_not_applicable_not_fail():
sec, findings = protocol_exam(_axi_graph())
row = next(
r for r in sec.recognized_instances
if "axi" in str(r.get("logical_protocol_id"))
)
l3 = row["l3_checks"]
assert l3
assert all(c["result"] == "NOT_APPLICABLE" for c in l3)
assert all(c["result"] != "FAIL" for c in l3)
assert all(f.rule_id != "PE-PRT-L3-002" for f in findings)
assert all(f.status != "ERROR" or f.source != "protocol_l3" for f in findings)
def test_fixture_channel_pass_with_existing_fact():
iface = _iface_with_numeric()
inst = _inst()
data = {
"instances": {
inst.instance_id: {
"insertion_loss_db": -3.0,
"return_loss_db": -15.0,
"crosstalk_db": -28.0,
"total_budget_ps": 200.0,
"used_budget_ps": 120.0,
"channel_complete": True,
}
}
}
rows = certify_instance_l3(inst, iface, channel_data=data)
by_check = {r.check: r for r in rows}
assert by_check["insertion_loss"].result == "PASS"
assert by_check["return_loss"].result == "PASS"
assert by_check["timing_budget"].result == "PASS"
assert by_check["timing_budget"].remaining_margin_ps == 80.0
assert by_check["channel"].result == "PASS"
assert by_check["insertion_loss"].measured_db == -3.0
notes = " ".join(r.notes.lower() for r in rows)
assert "not invented" in notes
assert "not openems" in notes
def test_fixture_channel_fail_mandatory_il():
iface = _iface_with_numeric()
inst = _inst()
data = {
inst.instance_id: {
"s21_db": -12.0,
"s11_db": -15.0,
"total_budget_ps": 200.0,
"used_budget_ps": 50.0,
}
}
rows = certify_instance_l3(inst, iface, channel_data=data)
il = next(r for r in rows if r.check == "insertion_loss")
assert il.result == "FAIL"
findings = []
from backend.periscopex.protocol_l3 import l3_findings
findings = l3_findings(inst, rows)
assert any(f.rule_id == "PE-PRT-L3-002" and f.status == "ERROR" for f in findings)
def test_recommended_never_fail_l3():
rec = _pack("crosstalk", "FAIL", "RECOMMENDED", notes="xt")
assert rec.result != "FAIL"
assert rec.status != "ERROR"
def test_raw_touchstone_without_fact_is_skip_not_invented():
iface = _iface_with_numeric()
inst = _inst()
data = {inst.instance_id: {"sparam_file": "lane.s4p", "touchstone": "lane.s4p"}}
rows = certify_instance_l3(inst, iface, channel_data=data)
assert all(r.result != "PASS" for r in rows)
notes = " ".join(r.notes for r in rows)
assert "non certificata a L3" in notes
assert "invent" in notes.lower() or "non inventati" in notes.lower() or "not invented" in notes.lower()
blob = json.dumps([r.model_dump() for r in rows])
# no synthesized S21 numbers
assert '"measured_db": null' in blob or all(r.measured_db is None for r in rows)
def test_openems_method_is_explicit_skip():
iface = _iface_with_numeric()
inst = _inst()
data = {
inst.instance_id: {
"insertion_loss_db": -2.0,
"method": "OpenEMS",
"solver": "FEM",
}
}
rows = certify_instance_l3(inst, iface, channel_data=data)
notes = " ".join(r.notes for r in rows)
assert "non certificata a L3" in notes
assert all(r.result != "PASS" for r in rows)
def test_no_sparam_invention_on_empty_channel_data():
insts = recognize_physical_buses(_usb_connected())
by, findings = certify_l3(insts, channel_data=None)
rows = [r for lst in by.values() for r in lst]
blob = json.dumps([r.model_dump() for r in rows])
assert "s21" not in blob.lower()
assert all(r.measured_db is None for r in rows)
assert any("mancanza" in r.notes for r in rows)
assert any(f.rule_id == "PE-PRT-L3-001" for f in findings)
def test_protocol_exam_attaches_l3_and_does_not_call_l0_electrical():
sec, _ = protocol_exam(_usb_connected())
row = sec.recognized_instances[0]
assert row.get("l3_checks")
assert all(c.get("level") == "L3" for c in row["l3_checks"])
notes = " ".join(c.get("notes") or "" for c in row["l3_checks"])
assert "L0/L1 are not electrical" in notes
l0notes = json.dumps(row.get("l0_checks") or [])
assert "ohm" not in l0notes.lower() or "not" in l0notes.lower()
+2 -2
View File
@@ -95,8 +95,8 @@ def test_result_rank_fail_before_warning():
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 == "M8"
assert sec.max_level_reached == "L3"
assert sec.macrophase == "M9"
usb = next(
r for r in sec.recognized_instances
if "usb" in str(r.get("logical_protocol_id"))