Add gated Placement F2 pack skeleton for satellite xy.

Propose positions only from PCB footprints plus numeric decoupling max_distance_mm; otherwise skip with an explicit reason.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-09-13 17:50:47 +02:00
co-authored by Cursor
parent 54b0b07917
commit 86121e6547
10 changed files with 465 additions and 10 deletions
+188
View File
@@ -0,0 +1,188 @@
"""Layout F2 skeleton — propose satellite xy from PCB anchors + numeric rules.
No millimetres are invented. Packing runs only when a LayoutGraph has
footprints and at least one ``decoupling_proximity`` rule carries a numeric
``max_distance_mm``. Otherwise the report is ``skipped`` with an explicit reason.
"""
from __future__ import annotations
import math
from typing import Literal
from pydantic import BaseModel
from backend.pinscopex.functional_groups import FunctionalGroupsReport, PlacementIcGroup
from backend.pinscopex.models import DesignGraph, LayoutGraph, LayoutPad
SkipReason = Literal[
"no_pcb_footprints",
"no_numeric_layout_rules",
"no_packable_satellites",
]
class PlacementProposal(BaseModel):
ref: str
anchor_ref: str
rule_kind: str
max_distance_mm: float
proposed_x: float
proposed_y: float
layer: str = ""
basis: str = "ic_pad+rule"
class PlacementPackReport(BaseModel):
objective: Literal["routing"] = "routing"
status: Literal["packed", "skipped"] = "skipped"
skip_reason: SkipReason | None = None
placements: list[PlacementProposal] = []
def build_placement_pack(
plan: FunctionalGroupsReport,
layout: LayoutGraph | None,
graph: DesignGraph | None = None,
) -> PlacementPackReport:
"""Propose satellite positions within extracted proximity limits."""
if layout is None or not layout.footprints:
return PlacementPackReport(status="skipped", skip_reason="no_pcb_footprints")
if not _has_numeric_proximity(plan):
return PlacementPackReport(
status="skipped",
skip_reason="no_numeric_layout_rules",
)
placements: list[PlacementProposal] = []
used_refs: set[str] = set()
for group in plan.groups:
placements.extend(
_pack_group(group, layout, graph, used_refs),
)
if not placements:
return PlacementPackReport(
status="skipped",
skip_reason="no_packable_satellites",
)
return PlacementPackReport(status="packed", placements=placements)
def _has_numeric_proximity(plan: FunctionalGroupsReport) -> bool:
for g in plan.groups:
for rule in g.layout_rules:
if rule.get("kind") != "decoupling_proximity":
continue
if _num(rule.get("max_distance_mm")) is not None:
return True
return False
def _num(raw) -> float | None:
if raw is None or isinstance(raw, bool):
return None
try:
v = float(raw)
except (TypeError, ValueError):
return None
return v if v > 0 else None
def _pack_group(
group: PlacementIcGroup,
layout: LayoutGraph,
graph: DesignGraph | None,
used_refs: set[str],
) -> list[PlacementProposal]:
ic_fp = layout.footprints.get(group.ref)
if not ic_fp:
return []
candidates = [
s for s in group.satellites
if s.role_hint in ("decoupling", "bulk") and s.ref not in used_refs
]
if not candidates:
return []
out: list[PlacementProposal] = []
for rule in group.layout_rules:
if rule.get("kind") != "decoupling_proximity":
continue
limit = _num(rule.get("max_distance_mm"))
if limit is None:
continue
pad = _anchor_pad(group, layout, graph, str(rule.get("pin") or ""))
if pad is None:
# Fall back to footprint origin when pin is unknown but rule is numeric.
pad = LayoutPad(number="", x=ic_fp.x, y=ic_fp.y, net="")
basis = "ic_origin+rule"
else:
basis = "ic_pad+rule"
net = pad.net or None
matched = [
s for s in candidates
if s.ref not in used_refs and (not net or net in (s.nets or []))
]
if not matched:
matched = [s for s in candidates if s.ref not in used_refs]
if not matched:
continue
for i, sat in enumerate(matched):
angle = (2.0 * math.pi * i) / max(len(matched), 8)
radius = limit * 0.5
px = pad.x + radius * math.cos(angle)
py = pad.y + radius * math.sin(angle)
layer = ic_fp.layer or ""
out.append(PlacementProposal(
ref=sat.ref,
anchor_ref=group.ref,
rule_kind="decoupling_proximity",
max_distance_mm=limit,
proposed_x=round(px, 4),
proposed_y=round(py, 4),
layer=layer,
basis=basis,
))
used_refs.add(sat.ref)
# One numeric rule per IC is enough for the skeleton.
break
return out
def _anchor_pad(
group: PlacementIcGroup,
layout: LayoutGraph,
graph: DesignGraph | None,
pin_token: str,
) -> LayoutPad | None:
fp = layout.footprints.get(group.ref)
if not fp or not fp.pads:
return None
want = (pin_token or "").strip()
if want:
for pad in fp.pads:
if pad.number == want:
return pad
if graph is not None:
comp = graph.components.get(group.ref)
if comp:
for pin_num, net in comp.pins.items():
if str(pin_num) == want:
for pad in fp.pads:
if pad.number == str(pin_num):
return pad
# No matching pad number — pick any pad on that net.
for pad in fp.pads:
if pad.net and pad.net == net:
return pad
# Prefer a pad on a power-looking net shared with decoupling sats.
for pad in fp.pads:
if pad.net:
return pad
return fp.pads[0]
+11
View File
@@ -630,6 +630,17 @@ async def get_placement_plan(project_id: str, request: Request):
return storage.read_json(key)
@router.get("/pipeline/{project_id}/placement/pack")
async def get_placement_pack(project_id: str, request: Request):
"""Return ``placement_pack.json`` (F2 — skipped without PCB + numeric rules)."""
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
key = f"{proj_svc.project_prefix(owner_user_id, project_id)}/placement_pack.json"
if not storage.exists(key):
raise HTTPException(404, "Placement pack not found — run placement first")
return storage.read_json(key)
@router.get("/pipeline/{project_id}/placement/events")
async def placement_events(project_id: str, request: Request):
"""SSE stream for Placement pipeline progress (watches placement_* only)."""
+55 -4
View File
@@ -1,7 +1,8 @@
"""Placement pipeline — parallel to analysis, topology only (no LLM / no mm).
"""Placement pipeline — parallel to analysis (no LLM).
Stages: ensure_graph → classify → write_plan.
Writes ``placement_plan.json`` (+ refreshes ``functional_groups.json``).
Stages: ensure_graph → classify → write_plan → pack (F2 gated).
Writes ``placement_plan.json`` (+ ``functional_groups.json``) and
``placement_pack.json`` (mm only when PCB + numeric layout_rules exist).
Uses ``placement_status`` so analysis ``status`` is untouched.
"""
@@ -12,7 +13,8 @@ from pathlib import Path
from backend.pinscopex.functional_groups import build_placement_plan
from backend.pinscopex.graph import build_graph
from backend.pinscopex.models import ComponentConstraints, DesignGraph
from backend.pinscopex.models import ComponentConstraints, DesignGraph, LayoutGraph
from backend.pinscopex.placement_pack import build_placement_pack
from backend.services import projects as proj_svc
from backend.services.pipeline import PipelineWorkspace, broker
from backend.services.storage import StorageBackend
@@ -112,18 +114,41 @@ async def run_placement_pipeline(
ws._upload_file("functional_groups.json")
_step(project_id, "write_plan", "complete", "placement_plan.json")
if _cancelled(storage, user_id, project_id):
_finish_cancelled(storage, user_id, project_id)
return
_step(project_id, "pack", "running", "F2 gated pack")
layout = _load_layout(ws)
pack = build_placement_pack(plan, layout, graph)
pack_path = ws.local_path("placement_pack.json")
pack_path.write_text(pack.model_dump_json(indent=2) + "\n")
ws._upload_file("placement_pack.json")
pack_detail = (
f"{len(pack.placements)} proposals"
if pack.status == "packed"
else f"skipped:{pack.skip_reason}"
)
_step(project_id, "pack", "complete", pack_detail)
proj_svc.update_project(
storage, user_id, project_id,
placement_status="complete",
placement_state={
"domains": len(plan.domains),
"groups": len(plan.groups),
"pack_status": pack.status,
"pack_count": len(pack.placements),
"pack_skip_reason": pack.skip_reason,
},
placement_cancel_requested=False,
)
_publish(project_id, "placement_complete", {
"domains": len(plan.domains),
"groups": len(plan.groups),
"pack_status": pack.status,
"pack_count": len(pack.placements),
"pack_skip_reason": pack.skip_reason,
})
except Exception as e:
logger.exception("placement pipeline failed for %s", project_id)
@@ -178,6 +203,32 @@ async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignG
return graph
def _load_layout(ws: PipelineWorkspace) -> LayoutGraph | None:
"""Reuse layout_graph.json, or parse uploads/pcb.kicad_pcb once."""
cached = ws.local_path("layout_graph.json")
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.pinscopex.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
def _cancelled(storage: StorageBackend, user_id: str, project_id: str) -> bool:
meta = proj_svc.get_project(storage, user_id, project_id)
return bool(meta and meta.placement_cancel_requested)