michele
bfb781f94f
Apri progetto KiCad picks the folder that contains .kicad_pro and loads sibling hierarchical sheets and the matching .kicad_pcb. Connectivity and MPN/PNM/Value come from the schematic; a leftover PADS .asc is ignored. A lone .kicad_pro upload is rejected unless that path exists on disk.
271 lines
8.5 KiB
Python
271 lines
8.5 KiB
Python
"""KiCad project folder ingest: *.kicad_pro + siblings, no netlist."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.periscopex.kicad_project import load_kicad_project, sniff_kicad_pro
|
|
from backend.periscopex.netlist_bundle import materialize_netlist_upload, sniff_netlist_kind
|
|
from backend.periscopex.parsers import parse_netlist_any
|
|
|
|
HUBAUDIO = Path("/Users/michelebigi/Development/HubAudio/hardware/kicad/HubAudio")
|
|
|
|
_LIB_R = """
|
|
(lib_symbols
|
|
(symbol "Device:R"
|
|
(pin passive (at 0 3.81 90) (length 2.54)
|
|
(name "~" (effects (font (size 1.27 1.27))))
|
|
(number "1" (effects (font (size 1.27 1.27))))
|
|
)
|
|
(pin passive (at 0 -3.81 90) (length 2.54)
|
|
(name "~" (effects (font (size 1.27 1.27))))
|
|
(number "2" (effects (font (size 1.27 1.27))))
|
|
)
|
|
)
|
|
(symbol "Device:U"
|
|
(pin passive (at 0 3.81 90) (length 2.54)
|
|
(name "~" (effects (font (size 1.27 1.27))))
|
|
(number "1" (effects (font (size 1.27 1.27))))
|
|
)
|
|
)
|
|
)
|
|
"""
|
|
|
|
|
|
def _sch(*body: str) -> str:
|
|
return (
|
|
"(kicad_sch (version 20250114) (uuid \"11111111-1111-1111-1111-111111111111\")"
|
|
+ _LIB_R
|
|
+ "".join(body)
|
|
+ "\n)\n"
|
|
)
|
|
|
|
|
|
def _resistor(ref: str, value: str) -> str:
|
|
uid = "aaaaaaaa-aaaa-aaaa-aaaa-" + ref.encode().hex()[:12].ljust(12, "0")
|
|
return f"""
|
|
(symbol
|
|
(lib_id "Device:R")
|
|
(at 0 0 0)
|
|
(unit 1)
|
|
(uuid "{uid}")
|
|
(property "Reference" "{ref}" (at 0 0 0) (effects (font (size 1.27 1.27))))
|
|
(property "Value" "{value}" (at 0 0 0) (effects (font (size 1.27 1.27))))
|
|
(pin "1" (uuid "p1"))
|
|
(pin "2" (uuid "p2"))
|
|
)
|
|
"""
|
|
|
|
|
|
def _ic(ref: str, value: str, pnm: str) -> str:
|
|
uid = "bbbbbbbb-bbbb-bbbb-bbbb-" + ref.encode().hex()[:12].ljust(12, "0")
|
|
return f"""
|
|
(symbol
|
|
(lib_id "Device:U")
|
|
(at 0 0 0)
|
|
(unit 1)
|
|
(uuid "{uid}")
|
|
(property "Reference" "{ref}" (at 0 0 0) (effects (font (size 1.27 1.27))))
|
|
(property "Value" "{value}" (at 0 0 0) (effects (font (size 1.27 1.27))))
|
|
(property "PNM" "{pnm}" (at 0 0 0) (effects (font (size 1.27 1.27))))
|
|
(pin "1" (uuid "u1"))
|
|
)
|
|
"""
|
|
|
|
|
|
def _pro(root_name: str) -> str:
|
|
return json.dumps({
|
|
"meta": {"filename": root_name.replace(".kicad_sch", ".kicad_pro")},
|
|
"schematic": {
|
|
"top_level_sheets": [{"filename": root_name, "name": "root"}],
|
|
},
|
|
})
|
|
|
|
|
|
def _write_project(folder: Path) -> None:
|
|
child = _sch(
|
|
_ic("U9", "TPD2E007DCKR", "TPD2E007DCKR"),
|
|
"""
|
|
(global_label "GND" (at 0 3.81 0) (uuid "cccccccccccccccccccccccccccccccccccc"))
|
|
""",
|
|
)
|
|
root = _sch(
|
|
_resistor("R1", "10k"),
|
|
"""
|
|
(global_label "GND" (at 0 3.81 0) (uuid "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"))
|
|
(sheet
|
|
(at 50 0)
|
|
(size 20 20)
|
|
(property "Sheetname" "Codec" (at 50 0 0) (effects (font (size 1.27 1.27))))
|
|
(property "Sheetfile" "Codec.kicad_sch" (at 50 0 0) (effects (font (size 1.27 1.27))))
|
|
)
|
|
""",
|
|
)
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
(folder / "HubAudio.kicad_pro").write_text(_pro("HubAudio.kicad_sch"))
|
|
(folder / "HubAudio.kicad_sch").write_text(root)
|
|
(folder / "Codec.kicad_sch").write_text(child)
|
|
(folder / "HubAudio.kicad_pcb").write_text(
|
|
'(kicad_pcb (version 20240108) (generator pcbnew)\n (net 0 "")\n)\n'
|
|
)
|
|
(folder / "netlist.asc").write_text(
|
|
"*PADS-PCB*\n*PART*\nU13 SOT23\nR1 0603\n*NET*\n"
|
|
"*SIGNAL* GND\nU13.1 R1.1\n*END*\n"
|
|
)
|
|
|
|
|
|
def test_sniff_kicad_pro_json():
|
|
body = _pro("HubAudio.kicad_sch").encode()
|
|
assert sniff_kicad_pro(body, "HubAudio.kicad_pro")
|
|
assert sniff_netlist_kind(body, "HubAudio.kicad_pro") == "kicad_pro"
|
|
|
|
|
|
def test_lone_kicad_pro_bytes_are_rejected(tmp_path: Path):
|
|
with pytest.raises(ValueError, match="da solo non basta"):
|
|
materialize_netlist_upload(
|
|
[("HubAudio.kicad_pro", _pro("HubAudio.kicad_sch").encode())],
|
|
tmp_path / "work",
|
|
)
|
|
|
|
|
|
def test_folder_next_to_pro_loads_sheets_and_pcb(tmp_path: Path):
|
|
src = tmp_path / "HubAudio"
|
|
_write_project(src)
|
|
files = [
|
|
(str(p.relative_to(tmp_path)), p.read_bytes())
|
|
for p in src.iterdir()
|
|
if p.is_file()
|
|
]
|
|
parsed = materialize_netlist_upload(files, tmp_path / "work")
|
|
parts, _nets, fmt = parse_netlist_any(parsed.root)
|
|
assert fmt == "kicad_sch"
|
|
assert parsed.pcb is not None and parsed.pcb.name == "HubAudio.kicad_pcb"
|
|
assert "R1" in parts and "U9" in parts
|
|
assert "U13" not in parts
|
|
assert {p.name for p in parsed.extra_sch} == {"Codec.kicad_sch"}
|
|
|
|
|
|
def test_stale_pads_asc_ignored_when_pro_present(tmp_path: Path):
|
|
src = tmp_path / "HubAudio"
|
|
_write_project(src)
|
|
loaded = load_kicad_project(src)
|
|
assert loaded.root_sch.name == "HubAudio.kicad_sch"
|
|
parts, _nets, fmt = parse_netlist_any(loaded.root_sch)
|
|
assert fmt == "kicad_sch"
|
|
assert "U13" not in parts
|
|
|
|
|
|
def test_disk_path_to_pro_copies_siblings(tmp_path: Path):
|
|
src = tmp_path / "HubAudio"
|
|
_write_project(src)
|
|
parsed = materialize_netlist_upload(
|
|
[(str(src / "HubAudio.kicad_pro"), (src / "HubAudio.kicad_pro").read_bytes())],
|
|
tmp_path / "work",
|
|
)
|
|
parts, _nets, _fmt = parse_netlist_any(parsed.root)
|
|
assert "U9" in parts and "U13" not in parts
|
|
assert parsed.pcb is not None
|
|
|
|
|
|
def test_graph_without_bom_csv_uses_schematic_pnm(tmp_path: Path):
|
|
from backend.periscopex.graph import build_graph
|
|
|
|
src = tmp_path / "HubAudio"
|
|
_write_project(src)
|
|
missing = tmp_path / "no-bom.csv"
|
|
g = build_graph(
|
|
src / "HubAudio.kicad_sch",
|
|
missing,
|
|
tmp_path / "ex",
|
|
tmp_path / "pat",
|
|
tmp_path / "mod",
|
|
)
|
|
assert g.components["U9"].mpn == "TPD2E007DCKR"
|
|
assert "U13" not in g.components
|
|
|
|
|
|
def test_empty_child_pnm_not_invented(tmp_path: Path):
|
|
from backend.periscopex.graph import build_graph
|
|
|
|
folder = tmp_path / "proj"
|
|
folder.mkdir()
|
|
child = _sch(
|
|
_ic("U15", "", ""),
|
|
"""
|
|
(global_label "GND" (at 0 3.81 0) (uuid "cccccccccccccccccccccccccccccccccccc"))
|
|
""",
|
|
)
|
|
root = _sch(
|
|
_resistor("R1", "10k"),
|
|
"""
|
|
(global_label "GND" (at 0 3.81 0) (uuid "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"))
|
|
(sheet
|
|
(at 50 0)
|
|
(size 20 20)
|
|
(property "Sheetname" "Codec" (at 50 0 0) (effects (font (size 1.27 1.27))))
|
|
(property "Sheetfile" "Codec.kicad_sch" (at 50 0 0) (effects (font (size 1.27 1.27))))
|
|
)
|
|
""",
|
|
)
|
|
(folder / "p.kicad_pro").write_text(_pro("p.kicad_sch"))
|
|
(folder / "p.kicad_sch").write_text(root)
|
|
(folder / "Codec.kicad_sch").write_text(child)
|
|
g = build_graph(
|
|
folder / "p.kicad_sch",
|
|
tmp_path / "missing.csv",
|
|
tmp_path / "ex",
|
|
tmp_path / "pat",
|
|
tmp_path / "mod",
|
|
)
|
|
assert not (g.components["U15"].mpn or "").strip()
|
|
|
|
|
|
@pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk")
|
|
def test_hubaudio_folder_siblings_only():
|
|
loaded = load_kicad_project(HUBAUDIO)
|
|
assert loaded.pro.name == "HubAudio.kicad_pro"
|
|
assert loaded.root_sch.name == "HubAudio.kicad_sch"
|
|
names = {p.name for p in loaded.sheets}
|
|
assert "Codec.kicad_sch" in names
|
|
assert "POWER.kicad_sch" in names
|
|
assert loaded.pcb is not None and loaded.pcb.name == "HubAudio.kicad_pcb"
|
|
assert loaded.folder == HUBAUDIO.resolve()
|
|
parts, _nets, fmt = parse_netlist_any(loaded.root_sch)
|
|
assert fmt == "kicad_sch"
|
|
assert "U13" not in parts
|
|
assert "U9" in parts
|
|
|
|
|
|
def test_api_folder_upload_no_csv(tmp_path: Path):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.main import app
|
|
from backend.services.storage import LocalStorageBackend
|
|
|
|
app.state.storage = LocalStorageBackend(tmp_path)
|
|
client = TestClient(app)
|
|
pid = client.post("/api/projects", json={"name": "ha"}).json()["id"]
|
|
src = tmp_path / "src"
|
|
_write_project(src)
|
|
files = [
|
|
("files", (p.name, p.read_bytes(), "application/octet-stream"))
|
|
for p in src.iterdir()
|
|
if p.suffix in {".kicad_pro", ".kicad_sch", ".kicad_pcb"}
|
|
]
|
|
paths = [p.name for p in src.iterdir() if p.suffix in {".kicad_pro", ".kicad_sch", ".kicad_pcb"}]
|
|
resp = client.post(
|
|
f"/api/projects/{pid}/upload/netlist",
|
|
files=files,
|
|
data={"paths": json.dumps(paths)},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["format"] == "kicad_sch"
|
|
assert body["pcb_saved"] is True
|
|
assert body["bom_saved"] is True
|
|
assert "U13" not in str(body)
|
|
assert body["parts"] >= 2
|