michele
027546b0bd
Apri progetto walked every nested file (HubAudio .history ~2000) so onFilesChange never ran and Next stayed disabled. Read only kicad_pro, sch, and pcb in the project dir; timeout to an error; enable Next when .kicad_pro is in.
409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""KiCad project folder ingest: *.kicad_pro + siblings, no netlist."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
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
|
|
|
|
|
|
def _looks_like_kicad_pro(name: str, relative: str = "") -> bool:
|
|
"""Same rule as frontend looksLikeKicadPro (name and/or webkitRelativePath)."""
|
|
return any(".kicad_pro" in s.lower() for s in (name, relative, Path(relative or name).name))
|
|
|
|
|
|
def test_safari_stem_name_is_still_hubaudio_kicad_pro():
|
|
"""Live 2.60.6 checked only File.name.endsWith('.kicad_pro') and missed Safari."""
|
|
assert not "HubAudio".lower().endswith(".kicad_pro")
|
|
assert _looks_like_kicad_pro("HubAudio", "HubAudio/HubAudio.kicad_pro")
|
|
assert _looks_like_kicad_pro("HubAudio.kicad_pro", "")
|
|
assert sniff_netlist_kind(
|
|
_pro("HubAudio.kicad_sch").encode(), "HubAudio",
|
|
) == "kicad_pro"
|
|
|
|
|
|
def test_safari_stem_upload_writes_kicad_pro_suffix(tmp_path: Path):
|
|
src = tmp_path / "HubAudio"
|
|
_write_project(src)
|
|
pro = (src / "HubAudio.kicad_pro").read_bytes()
|
|
files = [
|
|
("HubAudio", pro),
|
|
("HubAudio.kicad_sch", (src / "HubAudio.kicad_sch").read_bytes()),
|
|
("Codec.kicad_sch", (src / "Codec.kicad_sch").read_bytes()),
|
|
("HubAudio.kicad_pcb", (src / "HubAudio.kicad_pcb").read_bytes()),
|
|
]
|
|
parsed = materialize_netlist_upload(files, tmp_path / "work")
|
|
assert parsed.root.name == "HubAudio.kicad_sch"
|
|
assert any(p.name == "HubAudio.kicad_pro" for p in (tmp_path / "work").rglob("*"))
|
|
parts, _nets, fmt = parse_netlist_any(parsed.root)
|
|
assert fmt == "kicad_sch"
|
|
assert "U9" in parts and "U13" not in parts
|
|
|
|
|
|
@pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk")
|
|
def test_real_hubaudio_kicad_pro_filename_and_folder_ingest(tmp_path: Path):
|
|
pro_path = HUBAUDIO / "HubAudio.kicad_pro"
|
|
assert pro_path.name == "HubAudio.kicad_pro"
|
|
data = pro_path.read_bytes()
|
|
assert sniff_kicad_pro(data, "HubAudio.kicad_pro")
|
|
assert sniff_netlist_kind(data, "HubAudio") == "kicad_pro"
|
|
files: list[tuple[str, bytes]] = []
|
|
for path in HUBAUDIO.iterdir():
|
|
if path.suffix.lower() in {".kicad_pro", ".kicad_sch", ".kicad_pcb"}:
|
|
files.append((f"HubAudio/{path.name}", path.read_bytes()))
|
|
assert any(name.endswith("HubAudio.kicad_pro") for name, _ in files)
|
|
parsed = materialize_netlist_upload(files, tmp_path / "work")
|
|
assert parsed.root.name == "HubAudio.kicad_sch"
|
|
assert parsed.pcb is not None
|
|
parts, _nets, fmt = parse_netlist_any(parsed.root)
|
|
assert fmt == "kicad_sch"
|
|
assert "U13" not in parts
|
|
|
|
|
|
# Mirrors periscope/src/frontend/src/lib/kicad-project-files.ts (folder walk + Next).
|
|
def _skip_walk_dir_name(name: str) -> bool:
|
|
n = name.rstrip("/").lower()
|
|
if not n or n.startswith("."):
|
|
return True
|
|
return bool(re.search(r"(-backups|\.pretty|3dmodels|__macosx|node_modules)$", n))
|
|
|
|
|
|
def _is_kicad_project_file_name(name: str) -> bool:
|
|
base = name.replace("\\", "/").rsplit("/", 1)[-1].lower()
|
|
return base.endswith((".kicad_pro", ".kicad_sch", ".kicad_pcb"))
|
|
|
|
|
|
def _skip_rel(rel: str) -> bool:
|
|
parts = [p for p in rel.replace("\\", "/").lower().split("/") if p]
|
|
return any(_skip_walk_dir_name(p) for p in parts)
|
|
|
|
|
|
def _details_step_ready(project_name: str, files: list[dict[str, str]]) -> bool:
|
|
if not project_name.strip():
|
|
return False
|
|
return any(_looks_like_kicad_pro(f.get("name", ""), f.get("webkitRelativePath", "")) for f in files)
|
|
|
|
|
|
def test_folder_walk_skips_history_and_keeps_project_dir_only():
|
|
"""Live 2.60.8 recursed handle.entries() into HubAudio/.history (~2000 files) so Next never enabled."""
|
|
names = [
|
|
"HubAudio.kicad_pro",
|
|
"HubAudio.kicad_sch",
|
|
"Codec.kicad_sch",
|
|
"HubAudio.kicad_pcb",
|
|
"HubAudio.kicad_prl",
|
|
"~HubAudio.kicad_pcb.lck",
|
|
".history",
|
|
".git",
|
|
"fp-lib-table",
|
|
]
|
|
kept = [n for n in names if _is_kicad_project_file_name(n)]
|
|
skipped_dirs = [n for n in names if _skip_walk_dir_name(n)]
|
|
assert kept == [
|
|
"HubAudio.kicad_pro",
|
|
"HubAudio.kicad_sch",
|
|
"Codec.kicad_sch",
|
|
"HubAudio.kicad_pcb",
|
|
]
|
|
assert ".history" in skipped_dirs and ".git" in skipped_dirs
|
|
assert _skip_rel("HubAudio/.history/objects/aa")
|
|
assert not _skip_rel("HubAudio/HubAudio.kicad_pro")
|
|
|
|
|
|
def test_details_next_enables_when_kicad_pro_is_in():
|
|
empty: list[dict[str, str]] = []
|
|
assert not _details_step_ready("HubAudio", empty)
|
|
assert not _details_step_ready(
|
|
"",
|
|
[{"name": "HubAudio.kicad_pro", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}],
|
|
)
|
|
assert _details_step_ready(
|
|
"HubAudio",
|
|
[{"name": "HubAudio.kicad_pro", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}],
|
|
)
|
|
assert _details_step_ready(
|
|
"HubAudio",
|
|
[{"name": "HubAudio", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}],
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk")
|
|
def test_real_hubaudio_recursive_tree_is_huge_project_dir_is_small():
|
|
tree = list(HUBAUDIO.rglob("*"))
|
|
project_dir = [
|
|
p
|
|
for p in HUBAUDIO.iterdir()
|
|
if p.is_file() and _is_kicad_project_file_name(p.name)
|
|
]
|
|
assert len(tree) > 100
|
|
assert (HUBAUDIO / ".history").is_dir()
|
|
assert any(p.name == "HubAudio.kicad_pro" for p in project_dir)
|
|
assert any(p.name.endswith(".kicad_sch") for p in project_dir)
|
|
assert any(p.name.endswith(".kicad_pcb") for p in project_dir)
|
|
assert all(".history" not in str(p) for p in project_dir)
|
|
assert len(project_dir) < 40
|