Re-parse .kicad_pcb on PCB exam instead of stale layout_graph.

Cached layout_graph still listed stitching via-pads as component pads,
so PE-BOM-011 kept firing after the ingest fix. PCB/placement jobs now
parse the board file and refresh the cache.
This commit is contained in:
2026-09-20 16:16:46 +02:00
parent 09c49f8529
commit 34a9b30a35
3 changed files with 73 additions and 31 deletions
+19 -15
View File
@@ -417,28 +417,32 @@ async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignG
def _load_layout(ws: PipelineWorkspace) -> LayoutGraph | None:
"""Parse ``uploads/pcb.kicad_pcb`` with the current ingest.
``layout_graph.json`` is a cache written by this parser (or schematic
ingest). A stale cache still lists footprint vias as pads, so PCB exam
always re-reads the board when the file is present.
"""
pcb = ws.local_path("uploads/pcb.kicad_pcb")
cached = ws.local_path("layout_graph.json")
if pcb.is_file():
try:
from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb
layout = parse_kicad_pcb(pcb)
cached.write_text(layout.model_dump_json(indent=2) + "\n")
ws._upload_file("layout_graph.json")
return layout
except Exception:
logger.exception("kicad_pcb parse failed during PCB review")
if cached.is_file():
try:
return LayoutGraph.model_validate_json(
cached.read_text(encoding="utf-8"),
)
except Exception:
logger.exception("bad layout_graph.json — trying pcb parse")
pcb = ws.local_path("uploads/pcb.kicad_pcb")
if not pcb.is_file():
return None
try:
from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb
layout = parse_kicad_pcb(pcb)
cached.write_text(layout.model_dump_json(indent=2) + "\n")
ws._upload_file("layout_graph.json")
return layout
except Exception:
logger.exception("kicad_pcb parse failed during PCB review")
return None
logger.exception("bad layout_graph.json")
return None
def _cancelled(storage: StorageBackend, user_id: str, project_id: str) -> bool:
@@ -204,29 +204,27 @@ async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignG
def _load_layout(ws: PipelineWorkspace) -> LayoutGraph | None:
"""Reuse layout_graph.json, or parse uploads/pcb.kicad_pcb once."""
"""Parse the board when present so via/pad classification stays current."""
pcb = ws.local_path("uploads/pcb.kicad_pcb")
cached = ws.local_path("layout_graph.json")
if pcb.is_file():
try:
from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb
layout = parse_kicad_pcb(pcb)
cached.write_text(layout.model_dump_json(indent=2) + "\n")
ws._upload_file("layout_graph.json")
return layout
except Exception:
logger.exception("kicad_pcb parse failed during placement pack")
if cached.is_file():
try:
return LayoutGraph.model_validate_json(
cached.read_text(encoding="utf-8"),
)
except Exception:
logger.exception("bad layout_graph.json — trying pcb parse")
pcb = ws.local_path("uploads/pcb.kicad_pcb")
if not pcb.is_file():
return None
try:
from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb
layout = parse_kicad_pcb(pcb)
cached.write_text(layout.model_dump_json(indent=2) + "\n")
ws._upload_file("layout_graph.json")
return layout
except Exception:
logger.exception("kicad_pcb parse failed during placement pack")
return None
logger.exception("bad layout_graph.json")
return None
def _cancelled(storage: StorageBackend, user_id: str, project_id: str) -> bool:
+40
View File
@@ -269,3 +269,43 @@ def test_tht_solder_pads_stay_component_pads(tmp_path: Path):
g = parse_kicad_pcb(p)
assert [p.number for p in g.footprints["R1"].pads] == ["1", "2"]
assert g.vias == []
class _Ws:
def __init__(self, root: Path):
self.root = root
def local_path(self, rel: str) -> Path:
return self.root / rel
def _upload_file(self, rel: str) -> None:
return None
def test_pcb_pipeline_reparses_stale_layout_graph(tmp_path: Path):
from backend.periscopex.models import LayoutFootprint, LayoutGraph, LayoutPad
from backend.services.pcb_pipeline import _load_layout
uploads = tmp_path / "uploads"
uploads.mkdir()
pcb = _qfn24_pcb(
tmp_path,
n_pads=24,
stitch_via_pads=[(26, 0.0, 0.0, "GND")],
)
(uploads / "pcb.kicad_pcb").write_text(pcb.read_text())
stale = LayoutGraph(
footprints={
"U1": LayoutFootprint(
reference="U1",
footprint="QFN-24",
x=0, y=0, layer="F.Cu",
pads=[LayoutPad(number=str(i), x=0, y=0, net="GND") for i in range(1, 35)],
),
},
)
(tmp_path / "layout_graph.json").write_text(stale.model_dump_json())
layout = _load_layout(_Ws(tmp_path))
assert layout is not None
assert len(layout.footprints["U1"].pads) == 24
assert len(layout.vias) == 1