periscope/tests/pcb/test_protocol_l1_m3.py

"""M3 L1 GEOMETRIC certifier: mm vs NUMERIC/DESIGN only; no invented ps/ohm."""

from __future__ import annotations

import json

from backend.periscopex.models import (
    Component,
    ComponentType,
    DesignGraph,
    LayoutGraph,
    LayoutSegment,
    LayoutVia,
    Net,
    NetType,
    PinConnection,
)
from backend.periscopex.protocol_catalog import (
    ConstraintSource,
    PhysicalInterface,
    ProtocolConstraint,
)
from backend.periscopex.protocol_l0 import protocol_exam
from backend.periscopex.protocol_l1 import (
    _pack,
    certify_instance_l1,
    certify_l1,
    geometric_verdict,
)
from backend.periscopex.protocol_recognize import 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 _usb_layout(*, length_p: float = 10.0, length_m: float = 10.1) -> LayoutGraph:
    return LayoutGraph(
        segments=[
            LayoutSegment(start=(0, 0), end=(length_p, 0), width=0.2, layer="F.Cu", net="USB_D+"),
            LayoutSegment(start=(0, 0.2), end=(length_m, 0.2), width=0.2, layer="F.Cu", net="USB_D-"),
        ],
        vias=[LayoutVia(x=1.0, y=0.0, net="USB_D+", drill=0.3, layers=("F.Cu", "B.Cu"))],
    )


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_with_lanes() -> DesignGraph:
    pins = {str(i): f"DDR4_DQ{i}" for i in range(8)}
    pins.update({"8": "DDR4_DQS0_P", "9": "DDR4_DQS0_N"})
    u_mem = _ic("U5", pins, mpn="MT41K256M16TW", value="DDR4")
    u_cpu = _ic("U4", {str(i): pins[str(i)] for i in pins}, mpn="STM32", value="MCU")
    nets = {
        n: _net(n, ("U5", p), ("U4", p)) for p, n in pins.items()
    }
    return DesignGraph(components={"U5": u_mem, "U4": u_cpu}, nets=nets)


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 _cite() -> ConstraintSource:
    return ConstraintSource(document="project-netclass.txt", organization="design")


def _design_mm(parameter: str, value: float) -> ProtocolConstraint:
    return ProtocolConstraint(
        id=f"design-{parameter}",
        parameter=parameter,
        value_kind="NUMERIC",
        mandatory="MANDATORY",
        source_type="PCB",
        source_class="IMPLEMENTATION",
        value=value,
        unit="mm",
        source=_cite(),
        limit_kind="DESIGN",
    )


def _l1_for(graph: DesignGraph, logical_sub: str, layout: LayoutGraph | None):
    insts = recognize_physical_buses(graph)
    by, findings = certify_l1(graph, insts, layout)
    rows = []
    for inst in insts:
        if logical_sub in inst.logical_protocol_id:
            rows.extend(by.get(inst.instance_id, []))
    return rows, findings, insts


def test_usb_skew_without_numeric_is_unknown_not_invented_ps():
    rows, _, _ = _l1_for(_usb_connected(), "usb2", _usb_layout())
    skew = [r for r in rows if r.check == "intra_pair_skew"]
    assert skew
    assert skew[0].result in {"UNKNOWN", "MISSING_SOURCE", "VENDOR_DEPENDENT"}
    assert skew[0].result != "PASS"
    blob = json.dumps(skew[0].model_dump())
    assert "ohm" not in blob.lower()
    assert "90" not in blob
    assert "ps" not in blob.lower() or skew[0].result != "PASS"


def test_usb_length_unknown_without_numeric_limit():
    rows, _, _ = _l1_for(_usb_connected(), "usb2", _usb_layout())
    length = [r for r in rows if r.check == "length"]
    assert length
    assert length[0].measured_mm is not None
    assert length[0].result in {"UNKNOWN", "MISSING_SOURCE"}
    assert length[0].limit_mm is None


def test_usb_no_layout_is_visible_skip_not_pass():
    rows, findings, _ = _l1_for(_usb_connected(), "usb2", None)
    assert rows
    assert all(r.result != "PASS" or r.check == "never" for r in rows)
    assert any(r.result == "UNKNOWN" for r in rows)
    assert any(f.rule_id == "PE-PRT-L1-001" for f in findings)
    assert any("non certificata a L1" in (f.finding or "") for f in findings)


def test_axi_internal_l1_not_applicable():
    rows, findings, _ = _l1_for(_axi_graph(), "axi4", LayoutGraph())
    assert rows
    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 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()


def test_ddr_missing_grouping_visible_skip_not_pass():
    rows, findings, insts = _l1_for(_ddr_no_grouping(), "ddr4", LayoutGraph())
    assert insts
    mapping = [r for r in rows if r.check in {"byte_lane_mapping", "dq_to_dqs_skew", "byte_lane_skew"}]
    assert mapping
    assert all(r.result != "PASS" for r in mapping)
    assert any(r.result == "UNKNOWN" for r in mapping)
    assert any(f.rule_id == "PE-PRT-L1-002" for f in findings)
    assert any("grouping" in (f.finding or "").lower() for f in findings)


def test_ddr_reconstructed_lane_skew_stays_phy_dependent():
    layout = LayoutGraph(segments=[
        LayoutSegment(start=(0, 0), end=(20, 0), layer="F.Cu", net="DDR4_DQ0"),
        LayoutSegment(start=(0, 0), end=(21, 0), layer="F.Cu", net="DDR4_DQS0_P"),
    ])
    rows, _, insts = _l1_for(_ddr_with_lanes(), "ddr4", layout)
    assert any(g.kind == "BYTE_LANE" for i in insts for g in i.groups)
    mapping = [r for r in rows if r.check == "byte_lane_mapping"]
    assert mapping and mapping[0].result == "PASS"
    skew = [r for r in rows if r.check == "dq_to_dqs_skew"]
    assert skew
    assert skew[0].result in {"PHY_DEPENDENT", "CONTROLLER_DEPENDENT", "VENDOR_DEPENDENT", "UNKNOWN"}
    assert skew[0].result != "PASS"
    assert skew[0].limit_mm is None


def test_numeric_mm_design_limit_pass_and_fail():
    pass_v = geometric_verdict(_design_mm("length", 50.0), 10.0)
    fail_v = geometric_verdict(_design_mm("length", 50.0), 80.0)
    assert pass_v == "PASS"
    assert fail_v == "FAIL"
    graph = _usb_connected()
    insts = recognize_physical_buses(graph)
    usb = [i for i in insts if "usb" in i.logical_protocol_id][0]
    iface = PhysicalInterface(
        id="usb2-hs-dpair",
        logical_protocol_id="usb2-hs",
        bus_type="SERIAL",
        physical_layer="SERIAL_DIFFERENTIAL",
        pcb_relevant="YES",
        required_checks=["length"],
        constraints=[_design_mm("length", 50.0)],
    )
    short = certify_instance_l1(graph, usb, iface, _usb_layout(length_p=10, length_m=10))
    assert short[0].result == "PASS"
    assert short[0].measured_mm == 10.0
    assert short[0].limit_mm == 50.0
    long = certify_instance_l1(graph, usb, iface, _usb_layout(length_p=80, length_m=80))
    assert long[0].result == "FAIL"
    assert long[0].mandatory == "MANDATORY"


def test_time_unit_numeric_is_unknown_not_converted_ps():
    cons = ProtocolConstraint(
        id="skew-ps",
        parameter="intra_pair_skew",
        value_kind="NUMERIC",
        mandatory="MANDATORY",
        source_type="STANDARD",
        source_class="NORMATIVE",
        value=15.0,
        unit="ps",
        source=ConstraintSource(document="USB-IF", organization="USB-IF"),
        limit_kind="STANDARD",
    )
    assert geometric_verdict(cons, 0.05) == "UNKNOWN"


def test_recommended_never_fail_l1():
    rec = _pack("length", "FAIL", "RECOMMENDED", notes="over")
    assert rec.result != "FAIL"
    assert rec.status != "ERROR"


def test_l1_does_not_run_impedance():
    rows, _, _ = _l1_for(_usb_connected(), "usb2", _usb_layout())
    assert not any(r.check in {"differential_impedance", "impedance", "termination"} for r in rows)
    blob = json.dumps([r.model_dump() for r in rows])
    assert "90" not in blob
    assert "electrical Z" in blob or "not electrical" in blob.lower()


def test_protocol_exam_attaches_l1_checks():
    sec, _ = protocol_exam(_usb_connected(), layout=_usb_layout())
    assert sec.max_level_reached == "L2"
    assert sec.recognized_instances
    row = sec.recognized_instances[0]
    assert row.get("l1_checks")
    assert all(c.get("level") == "L1" for c in row["l1_checks"])