Accept a KiCad project drop so hierarchical sheets and the board upload in one step.
A single .kicad_sch was parsed in isolation, so child sheets looked like an empty netlist. Report 404s now surface the real API error instead of a generic fetch failure. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
@@ -0,0 +1,233 @@
|
||||
"""Unpack a netlist upload: one file, several KiCad sheets, or a zip.
|
||||
|
||||
The hierarchical ``.kicad_sch`` parser needs sibling files on disk. A single
|
||||
temp file named ``tmpXXXX.kicad_sch`` cannot see ``Sheetfile`` children.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import zipfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from backend.pinscopex.parsers import detect_netlist_format
|
||||
|
||||
MAX_BUNDLE_BYTES = 30 * 1024 * 1024
|
||||
_MAX_ZIP_MEMBERS = 400
|
||||
|
||||
_KIND = str # pads | edif | kicad_* | zip | kicad_pcb | unknown
|
||||
|
||||
|
||||
@dataclass
|
||||
class NetlistUpload:
|
||||
root: Path
|
||||
work_dir: Path
|
||||
pcb: Path | None
|
||||
extra_sch: list[Path]
|
||||
|
||||
|
||||
def sniff_netlist_kind(content: bytes) -> str:
|
||||
if content[:2] == b"PK":
|
||||
return "zip"
|
||||
head = content[:2048].decode("utf-8", errors="replace").lstrip("\ufeff").lstrip()
|
||||
low = head[:40].lower()
|
||||
if low.startswith("(kicad_pcb"):
|
||||
return "kicad_pcb"
|
||||
if low.startswith("(kicad_sch"):
|
||||
return "kicad_sch"
|
||||
if low.startswith("(edif"):
|
||||
return "edif"
|
||||
if low.startswith("(export"):
|
||||
return "kicad_sexp"
|
||||
if low.startswith("<?xml") or low.startswith("<export"):
|
||||
return "kicad_xml"
|
||||
if "*PADS-PCB*" in head.upper() or head.lstrip().startswith("*PART*"):
|
||||
return "pads"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _safe_rel(name: str) -> str:
|
||||
rel = name.replace("\\", "/").strip()
|
||||
if not rel or rel.startswith("/") or rel.startswith("\\"):
|
||||
raise ValueError(f"Rejected path: {name}")
|
||||
parts = Path(rel).parts
|
||||
if ".." in parts or (parts and parts[0] == ".."):
|
||||
raise ValueError(f"Rejected path: {name}")
|
||||
return rel
|
||||
|
||||
|
||||
_KEEP_SUFFIX = {
|
||||
".kicad_sch",
|
||||
".kicad_pcb",
|
||||
".kicad_pro",
|
||||
".kicad_net",
|
||||
".xml",
|
||||
".edn",
|
||||
".edif",
|
||||
".edf",
|
||||
".asc",
|
||||
".net",
|
||||
}
|
||||
|
||||
|
||||
def _keep_zip_member(rel: str) -> bool:
|
||||
parts = Path(rel).parts
|
||||
if any(
|
||||
p.endswith("-backups") or p.endswith(".pretty") or p.lower() in {"3dmodels", "__macosx"}
|
||||
for p in parts
|
||||
):
|
||||
return False
|
||||
return Path(rel).suffix.lower() in _KEEP_SUFFIX
|
||||
|
||||
|
||||
def _extract_zip(data: bytes, dest: Path) -> None:
|
||||
dest.mkdir(parents=True, exist_ok=True)
|
||||
total = 0
|
||||
kept = 0
|
||||
with zipfile.ZipFile(io.BytesIO(data)) as zf:
|
||||
for info in zf.infolist():
|
||||
if info.is_dir():
|
||||
continue
|
||||
rel = _safe_rel(info.filename)
|
||||
if not _keep_zip_member(rel):
|
||||
continue
|
||||
kept += 1
|
||||
if kept > _MAX_ZIP_MEMBERS:
|
||||
raise ValueError("Zip has too many schematic files")
|
||||
total += max(info.file_size, 0)
|
||||
if total > MAX_BUNDLE_BYTES:
|
||||
raise ValueError("Zip is too large")
|
||||
out = dest / rel
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
with zf.open(info) as src:
|
||||
payload = src.read()
|
||||
if len(payload) > MAX_BUNDLE_BYTES:
|
||||
raise ValueError("Zip member is too large")
|
||||
out.write_bytes(payload)
|
||||
|
||||
|
||||
def _write_named(name: str, data: bytes, dest: Path) -> None:
|
||||
rel = _safe_rel(name)
|
||||
kind = sniff_netlist_kind(data)
|
||||
if kind == "zip":
|
||||
_extract_zip(data, dest)
|
||||
return
|
||||
out = dest / Path(rel).name
|
||||
# Keep a single subdirectory when the client sent webkitRelativePath.
|
||||
if "/" in rel:
|
||||
out = dest / rel
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
else:
|
||||
dest.mkdir(parents=True, exist_ok=True)
|
||||
out.write_bytes(data)
|
||||
|
||||
|
||||
def find_kicad_pcb(work: Path) -> Path | None:
|
||||
hits = sorted(p for p in work.rglob("*.kicad_pcb") if p.is_file())
|
||||
return hits[0] if hits else None
|
||||
|
||||
|
||||
def _sheetfiles_of(path: Path) -> list[str]:
|
||||
from backend.pinscopex.parsers_kicad import _parse_sexp, _sheetfiles, _tag
|
||||
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
tree = _parse_sexp(text)
|
||||
if _tag(tree) != "kicad_sch":
|
||||
return []
|
||||
return _sheetfiles(tree)
|
||||
|
||||
|
||||
def _pick_root(work: Path) -> Path:
|
||||
schs: list[Path] = []
|
||||
exported: list[Path] = []
|
||||
for p in work.rglob("*"):
|
||||
if not p.is_file():
|
||||
continue
|
||||
kind = sniff_netlist_kind(p.read_bytes()[:2048])
|
||||
if kind == "kicad_sch":
|
||||
schs.append(p)
|
||||
elif kind in ("kicad_xml", "kicad_sexp", "edif", "pads"):
|
||||
exported.append(p)
|
||||
|
||||
if exported:
|
||||
pref = [
|
||||
p for p in exported
|
||||
if sniff_netlist_kind(p.read_bytes()[:2048]) in (
|
||||
"kicad_xml", "kicad_sexp", "edif",
|
||||
)
|
||||
]
|
||||
return (pref or exported)[0]
|
||||
|
||||
if not schs:
|
||||
raise ValueError(
|
||||
"No schematic found. Drop the KiCad project folder or a netlist."
|
||||
)
|
||||
|
||||
referenced: set[Path] = set()
|
||||
for p in schs:
|
||||
for rel in _sheetfiles_of(p):
|
||||
try:
|
||||
child = (p.parent / rel.replace("\\", "/")).resolve()
|
||||
except ValueError:
|
||||
continue
|
||||
referenced.add(child)
|
||||
roots = [p for p in schs if p.resolve() not in referenced]
|
||||
if not roots:
|
||||
raise ValueError("Cyclic sheet includes — upload an exported KiCad netlist instead.")
|
||||
|
||||
pro = list(work.rglob("*.kicad_pro"))
|
||||
if len(roots) > 1 and pro:
|
||||
stems = {p.stem for p in pro}
|
||||
matched = [r for r in roots if r.stem in stems]
|
||||
if len(matched) == 1:
|
||||
return matched[0]
|
||||
if len(roots) > 1:
|
||||
names = ", ".join(sorted(r.name for r in roots))
|
||||
raise ValueError(
|
||||
f"Multiple root sheets ({names}). Upload a zip of the project, "
|
||||
"or the top-level .kicad_sch together with every Sheetfile child."
|
||||
)
|
||||
return roots[0]
|
||||
|
||||
|
||||
def materialize_netlist_upload(
|
||||
files: list[tuple[str, bytes]],
|
||||
dest: Path,
|
||||
) -> NetlistUpload:
|
||||
"""Write uploaded bytes into ``dest`` and return the file to parse."""
|
||||
if not files:
|
||||
raise ValueError("No netlist file uploaded")
|
||||
dest.mkdir(parents=True, exist_ok=True)
|
||||
total = sum(len(b) for _n, b in files)
|
||||
if total > MAX_BUNDLE_BYTES:
|
||||
raise ValueError("Upload is too large")
|
||||
|
||||
if len(files) == 1:
|
||||
name, data = files[0]
|
||||
kind = sniff_netlist_kind(data)
|
||||
if kind == "kicad_pcb":
|
||||
raise ValueError(
|
||||
"This is a board file. Drop the KiCad project folder, or put "
|
||||
"the .kicad_pcb on the optional board step."
|
||||
)
|
||||
if kind == "unknown" and not name.lower().endswith(".zip"):
|
||||
raise ValueError(
|
||||
"Not a netlist. Drop the KiCad project folder, a zip, or a "
|
||||
"PADS / EDIF / KiCad netlist."
|
||||
)
|
||||
|
||||
for name, data in files:
|
||||
_write_named(name, data, dest)
|
||||
|
||||
root = _pick_root(dest)
|
||||
pcb = find_kicad_pcb(dest)
|
||||
extras = [
|
||||
p for p in work_sch_files(dest)
|
||||
if p.resolve() != root.resolve()
|
||||
]
|
||||
return NetlistUpload(root=root, work_dir=dest, pcb=pcb, extra_sch=extras)
|
||||
|
||||
|
||||
def work_sch_files(dest: Path) -> list[Path]:
|
||||
return sorted(p for p in dest.rglob("*.kicad_sch") if p.is_file())
|
||||
@@ -556,7 +556,10 @@ def parse_kicad_sch_project(
|
||||
if path in seen:
|
||||
raise ValueError(f"Cyclic sheet include: {path.name}")
|
||||
if not path.is_file():
|
||||
raise ValueError(f"Missing sheet file: {path.name}")
|
||||
raise ValueError(
|
||||
f"Missing sheet file: {path.name}. Drop the whole KiCad "
|
||||
"project folder, not a single sheet."
|
||||
)
|
||||
seen.add(path)
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
tree = _parse_sexp(text)
|
||||
|
||||
+67
-35
@@ -1,7 +1,10 @@
|
||||
"""Project CRUD and file upload endpoints."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, HTTPException, Request, UploadFile
|
||||
from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
from pydantic import BaseModel
|
||||
|
||||
@@ -462,51 +465,80 @@ async def upload_bom(
|
||||
|
||||
|
||||
@router.post("/projects/{project_id}/upload/netlist")
|
||||
async def upload_netlist(project_id: str, file: UploadFile, request: Request):
|
||||
async def upload_netlist(
|
||||
project_id: str,
|
||||
request: Request,
|
||||
file: UploadFile | None = File(default=None),
|
||||
files: list[UploadFile] | None = File(default=None),
|
||||
paths: str | None = Form(default=None),
|
||||
):
|
||||
storage = get_storage(request)
|
||||
result = proj_svc.resolve_project_access(storage, get_user_id(request), project_id)
|
||||
if not result:
|
||||
raise HTTPException(404, "Project not found")
|
||||
user_id = result[0] # owner_user_id for storage paths
|
||||
data = await file.read()
|
||||
if len(data) > MAX_UPLOAD_BYTES:
|
||||
raise HTTPException(413, f"File too large (max {MAX_UPLOAD_BYTES // 1024 // 1024} MB)")
|
||||
|
||||
# Auto-detect PADS vs EDIF from the file's first bytes — users don't pick
|
||||
# a format, the wizard accepts either.
|
||||
from backend.pinscopex.parsers import (
|
||||
detect_netlist_format, parse_netlist_any, validate_netlist,
|
||||
)
|
||||
from backend.pinscopex.netlist_bundle import materialize_netlist_upload
|
||||
from backend.pinscopex.parsers import parse_netlist_any, validate_netlist
|
||||
from backend.pinscopex.parsers_edif import list_edif_subdesigns
|
||||
import tempfile, os
|
||||
import tempfile
|
||||
|
||||
blobs: list[tuple[str, bytes]] = []
|
||||
seen: set[tuple[str, int]] = set()
|
||||
uploads = list(files or []) if files else ([file] if file is not None else [])
|
||||
rels: list[str] | None = None
|
||||
if paths:
|
||||
try:
|
||||
parsed_paths = json.loads(paths)
|
||||
except json.JSONDecodeError:
|
||||
parsed_paths = None
|
||||
if isinstance(parsed_paths, list) and all(isinstance(x, str) for x in parsed_paths):
|
||||
rels = parsed_paths
|
||||
for i, uf in enumerate(uploads):
|
||||
data = await uf.read()
|
||||
if len(data) > MAX_UPLOAD_BYTES:
|
||||
raise HTTPException(
|
||||
413,
|
||||
f"File too large (max {MAX_UPLOAD_BYTES // 1024 // 1024} MB)",
|
||||
)
|
||||
name = (
|
||||
rels[i]
|
||||
if rels is not None and i < len(rels)
|
||||
else (uf.filename or "netlist")
|
||||
)
|
||||
mark = (name, len(data))
|
||||
if mark in seen:
|
||||
continue
|
||||
seen.add(mark)
|
||||
blobs.append((name, data))
|
||||
if not blobs:
|
||||
raise HTTPException(400, "No netlist file uploaded")
|
||||
|
||||
fmt = detect_netlist_format(data)
|
||||
suffix = {
|
||||
"edif": ".edn",
|
||||
"kicad_xml": ".xml",
|
||||
"kicad_sexp": ".kicad_net",
|
||||
"kicad_sch": ".kicad_sch",
|
||||
}.get(fmt, ".asc")
|
||||
sub_designs: list[dict] = []
|
||||
try:
|
||||
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
|
||||
tmp.write(data)
|
||||
tmp.close()
|
||||
parts, nets, _ = parse_netlist_any(tmp.name)
|
||||
# For EDIF, also surface the sub-design layout so the wizard can
|
||||
# decide whether to prompt the user. Cheap second parse — same file.
|
||||
if fmt == "edif":
|
||||
sub_designs = list_edif_subdesigns(tmp.name)
|
||||
os.unlink(tmp.name)
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
parsed = materialize_netlist_upload(blobs, Path(tmp) / "work")
|
||||
parts, nets, fmt = parse_netlist_any(parsed.root)
|
||||
if fmt == "edif":
|
||||
sub_designs = list_edif_subdesigns(parsed.root)
|
||||
issues = validate_netlist(parts, nets)
|
||||
if issues:
|
||||
raise ValueError("; ".join(issues))
|
||||
root_bytes = parsed.root.read_bytes()
|
||||
key = proj_svc.save_netlist(storage, user_id, project_id, root_bytes, fmt=fmt)
|
||||
if fmt == "kicad_sch":
|
||||
proj_svc.save_companion_sheets(
|
||||
storage, user_id, project_id, parsed.root, parsed.extra_sch,
|
||||
)
|
||||
else:
|
||||
proj_svc.clear_companion_sheets(storage, user_id, project_id)
|
||||
if parsed.pcb is not None:
|
||||
proj_svc.save_pcb(storage, user_id, project_id, parsed.pcb.read_bytes())
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(400, f"Invalid netlist: {e}")
|
||||
issues = validate_netlist(parts, nets)
|
||||
if issues:
|
||||
raise HTTPException(400, f"Netlist failed sanity check: {'; '.join(issues)}")
|
||||
key = proj_svc.save_netlist(storage, user_id, project_id, data, fmt=fmt)
|
||||
# EDIF: emit a designator→pins preview matching the PADS browser-side
|
||||
# shape, so the wizard's power-sources step can render its dropdowns
|
||||
# without re-parsing the (s-expression-heavy) file in the browser.
|
||||
raise HTTPException(400, f"Netlist failed sanity check: {e}") from e
|
||||
|
||||
designator_pins: list[dict] = []
|
||||
if fmt != "pads":
|
||||
designator_pins = _build_designator_pins(parts, nets)
|
||||
|
||||
@@ -26,6 +26,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -663,6 +664,34 @@ def save_netlist(
|
||||
return key
|
||||
|
||||
|
||||
def clear_companion_sheets(
|
||||
storage: StorageBackend, user_id: str, project_id: str,
|
||||
) -> None:
|
||||
prefix = f"{_project_prefix(user_id, project_id)}/uploads/"
|
||||
for key in storage.list_recursive(prefix):
|
||||
rel = key[len(prefix):]
|
||||
if rel.endswith(".kicad_sch") and rel != "netlist.kicad_sch":
|
||||
storage.delete_key(key)
|
||||
|
||||
|
||||
def save_companion_sheets(
|
||||
storage: StorageBackend,
|
||||
user_id: str,
|
||||
project_id: str,
|
||||
root: Path,
|
||||
extras: list[Path],
|
||||
) -> None:
|
||||
"""Keep Sheetfile children next to ``uploads/netlist.kicad_sch``."""
|
||||
clear_companion_sheets(storage, user_id, project_id)
|
||||
parent = root.parent
|
||||
prefix = f"{_project_prefix(user_id, project_id)}/uploads/"
|
||||
for extra in extras:
|
||||
rel = extra.relative_to(parent).as_posix()
|
||||
if rel == "netlist.kicad_sch":
|
||||
continue
|
||||
storage.write_bytes(prefix + rel, extra.read_bytes())
|
||||
|
||||
|
||||
def save_pcb(
|
||||
storage: StorageBackend, user_id: str, project_id: str, data: bytes
|
||||
) -> str:
|
||||
|
||||
Reference in New Issue
Block a user