From 866c2e9242c0af8f7177c8e8ac9afe609c3225b3 Mon Sep 17 00:00:00 2001 From: Michele Bigi Date: Tue, 22 Sep 2026 12:58:18 +0200 Subject: [PATCH] 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. --- .../src/backend/periscopex/finding_engine.py | 4 + .../src/backend/periscopex/protocol_l0.py | 18 +- .../src/backend/periscopex/protocol_l3.py | 601 ++++++++++++++++++ .../src/backend/periscopex/protocol_report.py | 28 +- .../src/backend/services/pcb_pipeline.py | 12 + periscope/src/frontend/content/changelog.md | 7 + periscope/src/frontend/package.json | 2 +- .../components/report/protocol-section.tsx | 5 +- tests/pcb/test_protocol_ddr_m8.py | 2 +- tests/pcb/test_protocol_l0_m2.py | 2 +- tests/pcb/test_protocol_l1_m3.py | 6 +- tests/pcb/test_protocol_l2_m4.py | 6 +- tests/pcb/test_protocol_l3_m9.py | 316 +++++++++ tests/pcb/test_protocol_report_m5.py | 4 +- 14 files changed, 987 insertions(+), 26 deletions(-) create mode 100644 periscope/src/backend/periscopex/protocol_l3.py create mode 100644 tests/pcb/test_protocol_l3_m9.py diff --git a/periscope/src/backend/periscopex/finding_engine.py b/periscope/src/backend/periscopex/finding_engine.py index a7c2077..1f5b3fa 100644 --- a/periscope/src/backend/periscopex/finding_engine.py +++ b/periscope/src/backend/periscopex/finding_engine.py @@ -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() diff --git a/periscope/src/backend/periscopex/protocol_l0.py b/periscope/src/backend/periscopex/protocol_l0.py index 3626c9d..e4f67bf 100644 --- a/periscope/src/backend/periscopex/protocol_l0.py +++ b/periscope/src/backend/periscopex/protocol_l0.py @@ -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 diff --git a/periscope/src/backend/periscopex/protocol_l3.py b/periscope/src/backend/periscopex/protocol_l3.py new file mode 100644 index 0000000..96b9a5e --- /dev/null +++ b/periscope/src/backend/periscopex/protocol_l3.py @@ -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 TOTAL−USED=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 = total−used=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 diff --git a/periscope/src/backend/periscopex/protocol_report.py b/periscope/src/backend/periscopex/protocol_report.py index 2d89a47..fc7524b 100644 --- a/periscope/src/backend/periscopex/protocol_report.py +++ b/periscope/src/backend/periscopex/protocol_report.py @@ -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, diff --git a/periscope/src/backend/services/pcb_pipeline.py b/periscope/src/backend/services/pcb_pipeline.py index 5ed77ab..98fca49 100644 --- a/periscope/src/backend/services/pcb_pipeline.py +++ b/periscope/src/backend/services/pcb_pipeline.py @@ -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) diff --git a/periscope/src/frontend/content/changelog.md b/periscope/src/frontend/content/changelog.md index ee6ab9c..aa23738 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.71.0 — 2026-09-22 — Protocol certification M9 (L3 channel + visible skip) + +L3 **CHANNEL** certifier: structure for timing budget (TOTAL−USED=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. diff --git a/periscope/src/frontend/package.json b/periscope/src/frontend/package.json index a3662a1..2222420 100644 --- a/periscope/src/frontend/package.json +++ b/periscope/src/frontend/package.json @@ -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", diff --git a/periscope/src/frontend/src/components/report/protocol-section.tsx b/periscope/src/frontend/src/components/report/protocol-section.tsx index 1afffa0..d50abe4 100644 --- a/periscope/src/frontend/src/components/report/protocol-section.tsx +++ b/periscope/src/frontend/src/components/report/protocol-section.tsx @@ -19,7 +19,7 @@ function asChecks(row: Record): 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({

Livello massimo: {maxLevel} {" "} - (L0–L3; L3 not run unless channel data exists) + (L0–L3; L3 skip unless channel FACT exists — no OpenEMS)

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.

diff --git a/tests/pcb/test_protocol_ddr_m8.py b/tests/pcb/test_protocol_ddr_m8.py index e8158df..c9ba3ac 100644 --- a/tests/pcb/test_protocol_ddr_m8.py +++ b/tests/pcb/test_protocol_ddr_m8.py @@ -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")) diff --git a/tests/pcb/test_protocol_l0_m2.py b/tests/pcb/test_protocol_l0_m2.py index 15c0245..a6ddfb3 100644 --- a/tests/pcb/test_protocol_l0_m2.py +++ b/tests/pcb/test_protocol_l0_m2.py @@ -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) diff --git a/tests/pcb/test_protocol_l1_m3.py b/tests/pcb/test_protocol_l1_m3.py index 76fd6d1..0448f0c 100644 --- a/tests/pcb/test_protocol_l1_m3.py +++ b/tests/pcb/test_protocol_l1_m3.py @@ -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") diff --git a/tests/pcb/test_protocol_l2_m4.py b/tests/pcb/test_protocol_l2_m4.py index 4209245..75f7324 100644 --- a/tests/pcb/test_protocol_l2_m4.py +++ b/tests/pcb/test_protocol_l2_m4.py @@ -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"]) diff --git a/tests/pcb/test_protocol_l3_m9.py b/tests/pcb/test_protocol_l3_m9.py new file mode 100644 index 0000000..281ccae --- /dev/null +++ b/tests/pcb/test_protocol_l3_m9.py @@ -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() diff --git a/tests/pcb/test_protocol_report_m5.py b/tests/pcb/test_protocol_report_m5.py index 7d8376d..5531f26 100644 --- a/tests/pcb/test_protocol_report_m5.py +++ b/tests/pcb/test_protocol_report_m5.py @@ -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"))