diff --git a/backend/pinscopex/functional_groups.py b/backend/pinscopex/functional_groups.py index 870ecb7..9a8a1b9 100644 --- a/backend/pinscopex/functional_groups.py +++ b/backend/pinscopex/functional_groups.py @@ -132,6 +132,12 @@ def build_functional_groups( return FunctionalGroupsReport(objective="routing", domains=domains, groups=groups) +# Alias used by the dedicated Placement pipeline (same topology artifact). +build_placement_plan = build_functional_groups +PlacementPlan = FunctionalGroupsReport + + + def load_capacitance_farads(comp: Component) -> float | None: """Crystal CL from SimpleComponentSpecs.values, if present.""" specs = comp.specs diff --git a/backend/pipeline_worker.py b/backend/pipeline_worker.py index c6573d2..a23837e 100644 --- a/backend/pipeline_worker.py +++ b/backend/pipeline_worker.py @@ -101,8 +101,11 @@ async def _run() -> None: await pipeline_svc.run_regen_pipeline( storage, user_id, project_id, stages, ) + elif mode == "placement": + from backend.services import placement_pipeline as placement_svc + await placement_svc.run_placement_pipeline(storage, user_id, project_id) else: - raise SystemExit(f"unknown MODE={mode!r}; expected 'run' or 'regen'") + raise SystemExit(f"unknown MODE={mode!r}; expected 'run', 'regen', or 'placement'") def main() -> None: diff --git a/backend/routers/pipeline.py b/backend/routers/pipeline.py index 0377ac7..4bef76d 100644 --- a/backend/routers/pipeline.py +++ b/backend/routers/pipeline.py @@ -492,9 +492,204 @@ async def status(project_id: str, request: Request): "summary": meta.summary, "pipeline_state": meta.pipeline_state, "running": meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED), + "placement_status": meta.placement_status, + "placement_state": meta.placement_state, + "placement_running": (meta.placement_status or "draft") in ("queued", "running"), } +# --------------------------------------------------------------------------- +# Placement pipeline (parallel — topology only, no LLM / no credits) +# --------------------------------------------------------------------------- + + +_PLACEMENT_START_OK = frozenset({"draft", "complete", "error", "cancelled"}) +_PLACEMENT_SSE_TERMINAL = frozenset({ + "placement_complete", + "placement_error", + "placement_cancelled", +}) + + +@router.post("/pipeline/{project_id}/placement/start", status_code=202) +async def start_placement(project_id: str, request: Request): + """Enqueue the Placement topology pipeline (free, no analysis status change).""" + from backend.services.placement_pipeline import analysis_busy, placement_busy + + storage = get_storage(request) + owner_user_id, meta = await resolve_or_404(request, project_id) + if not meta.has_bom or not meta.has_netlist: + raise HTTPException(400, "Upload BOM and netlist before starting placement") + if analysis_busy(meta): + raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first") + if placement_busy(meta): + raise HTTPException(409, "Placement pipeline already running or queued") + if (meta.placement_status or "draft") not in _PLACEMENT_START_OK: + raise HTTPException( + 409, + f"Cannot start placement from placement_status={meta.placement_status}", + ) + + proj_svc.update_project( + storage, owner_user_id, project_id, + placement_status="queued", + placement_cancel_requested=False, + placement_state=None, + placement_execution_name=None, + ) + # Clear before enqueue so the placement SSE client never stops on a + # leftover analysis ``pipeline_complete`` in the shared event log. + try: + event_bridge.GCSEventBroker(storage, owner_user_id).clear_history(project_id) + except Exception: + logger.exception("failed to clear events before placement start for %s", project_id) + + try: + execution_name = job_runner.enqueue_placement_pipeline( + project_id, owner_user_id, + ) + except Exception: + logger.exception("enqueue_placement_pipeline failed for %s", project_id) + proj_svc.update_project( + storage, owner_user_id, project_id, + placement_status="error", + placement_state={"error": "Failed to enqueue placement worker"}, + ) + raise HTTPException(503, "Failed to enqueue placement worker; please retry") + + proj_svc.update_project( + storage, owner_user_id, project_id, + placement_execution_name=execution_name, + ) + return {"status": "started", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/placement/cancel") +async def cancel_placement(project_id: str, request: Request): + """Soft-cancel the Placement pipeline via ``placement_cancel_requested``.""" + from backend.services.placement_pipeline import placement_busy + + storage = get_storage(request) + owner_user_id, meta = await resolve_or_404(request, project_id) + if not placement_busy(meta): + raise HTTPException( + 409, + f"Placement is not running (placement_status={meta.placement_status})", + ) + proj_svc.update_project( + storage, owner_user_id, project_id, + placement_cancel_requested=True, + ) + return {"status": "cancel_requested", "project_id": project_id} + + +@router.get("/pipeline/{project_id}/placement/plan") +async def get_placement_plan(project_id: str, request: Request): + """Return ``placement_plan.json`` (F1 topology — no coordinates).""" + 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_plan.json" + if not storage.exists(key): + # Fallback for plans written only as functional_groups during analysis. + key = f"{proj_svc.project_prefix(owner_user_id, project_id)}/functional_groups.json" + if not storage.exists(key): + raise HTTPException(404, "Placement plan 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).""" + owner_user_id, meta = await resolve_or_404(request, project_id) + storage = get_storage(request) + + async def event_generator(): + execution_name = meta.placement_execution_name + crash_detected: dict[str, str | None] = {"reason": None} + + async def watch_status() -> None: + poll_interval = 2.0 + saw_active = (meta.placement_status or "draft") in ("queued", "running") + while True: + await asyncio.sleep(poll_interval) + try: + cur = proj_svc.get_project(storage, owner_user_id, project_id) + except Exception: + continue + if cur is None: + continue + pst = cur.placement_status or "draft" + if pst in ("queued", "running"): + saw_active = True + elif saw_active and pst in ("complete", "error", "cancelled"): + # Worker wrote terminal status; if SSE missed the event, + # surface a synthetic terminal after a short grace. + crash_detected["reason"] = f"placement_status={pst} (terminal)" + return + if execution_name: + try: + state = job_runner.get_execution_state(execution_name) + except Exception: + state = "unknown" + if state in _EXEC_TERMINAL and ( + saw_active or pst in ("queued", "running") + ): + crash_detected["reason"] = f"execution state={state}" + return + + watcher = asyncio.create_task(watch_status()) + try: + async for msg in event_bridge.tail_events( + storage, owner_user_id, project_id, + terminal_events=_PLACEMENT_SSE_TERMINAL, + ): + if crash_detected["reason"] is not None: + break + ev = msg["event"] + # Skip leftover analysis events if the log was not cleared yet. + if not ( + ev.startswith("placement_") + or ev == "heartbeat" + ): + continue + yield { + "event": ev, + "data": json.dumps(msg.get("data", {})), + } + if ev in _PLACEMENT_SSE_TERMINAL: + return + + if crash_detected["reason"] is not None: + cur = proj_svc.get_project(storage, owner_user_id, project_id) + err = None + if cur and cur.placement_state: + err = cur.placement_state.get("error") + yield { + "event": "placement_error", + "data": json.dumps({ + "error": err or crash_detected["reason"] + or "placement worker terminated without a terminal event", + "synthetic": True, + }), + } + finally: + watcher.cancel() + try: + await watcher + except (asyncio.CancelledError, Exception): + pass + + return EventSourceResponse( + event_generator(), + ping=15, + headers={ + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "Connection": "keep-alive", + }, + ) + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- diff --git a/backend/services/event_bridge.py b/backend/services/event_bridge.py index a51cb3b..f3a99eb 100644 --- a/backend/services/event_bridge.py +++ b/backend/services/event_bridge.py @@ -46,6 +46,9 @@ TERMINAL_EVENTS = frozenset({ "pipeline_error", "pipeline_cancelled", "pipeline_paused", + "placement_complete", + "placement_error", + "placement_cancelled", }) @@ -131,19 +134,19 @@ async def tail_events( *, poll_interval: float = 0.5, heartbeat_interval: float = 15.0, + terminal_events: frozenset[str] | None = None, ) -> AsyncIterator[dict]: """Yield events from the GCS-backed event log in order. - Stops yielding after a terminal event (``pipeline_complete``, - ``pipeline_error``, ``pipeline_cancelled``). Emits a - ``{"event": "heartbeat", "data": {}}`` synthetic event roughly every - ``heartbeat_interval`` seconds when no real events arrive, matching - the behaviour of the in-memory broker's SSE loop. + Stops yielding after a terminal event (default ``TERMINAL_EVENTS``). + Emits a ``{"event": "heartbeat", "data": {}}`` synthetic event roughly + every ``heartbeat_interval`` seconds when no real events arrive. The caller is expected to handle disconnects/cancellations and secondary terminal-detection (``meta.status``, Cloud Run execution state) on top of this iterator. """ + stop_on = terminal_events if terminal_events is not None else TERMINAL_EVENTS prefix = _events_prefix(user_id, project_id) last_seen_key: str | None = None last_emit_ts = 0.0 @@ -166,7 +169,7 @@ async def tail_events( emitted_any = True last_seen_key = key last_emit_ts = asyncio.get_event_loop().time() - if msg.get("event") in TERMINAL_EVENTS: + if msg.get("event") in stop_on: return now = asyncio.get_event_loop().time() diff --git a/backend/services/job_runner.py b/backend/services/job_runner.py index 655cd36..63e405c 100644 --- a/backend/services/job_runner.py +++ b/backend/services/job_runner.py @@ -58,8 +58,11 @@ def _spawn_local_subprocess( free: bool, mode: str = "run", regen_stages: list[str] | None = None, + proc_key: str | None = None, + execution_name: str | None = None, ) -> str: - name = _local_execution_name(project_id) + key = proc_key or project_id + name = execution_name or _local_execution_name(project_id) env = os.environ.copy() env["PROJECT_ID"] = project_id env["USER_ID"] = user_id @@ -74,17 +77,20 @@ def _spawn_local_subprocess( env=env, stdin=subprocess.DEVNULL, ) - _write_pid(project_id, proc.pid) + _write_pid(key, proc.pid) with _local_procs_lock: - # Reap any old proc for the same project before tracking the new one. - prior = _local_procs.pop(project_id, None) + # Reap any old proc for the same key before tracking the new one. + prior = _local_procs.pop(key, None) if prior is not None: try: prior.terminate() except Exception: pass - _local_procs[project_id] = proc - logger.info("dev: spawned worker subprocess pid=%s for %s", proc.pid, project_id) + _local_procs[key] = proc + logger.info( + "dev: spawned worker subprocess pid=%s for %s mode=%s", + proc.pid, project_id, mode, + ) return name @@ -341,6 +347,9 @@ def get_execution_state(execution_name: str | None) -> ExecutionState: if execution_name.startswith("local/projects/"): project_id = execution_name.split("/", 2)[-1] return _local_state(project_id) + if execution_name.startswith("local/placement/"): + project_id = execution_name.split("/", 2)[-1] + return _local_state(f"placement:{project_id}") return _cloud_run_state(execution_name) @@ -356,4 +365,21 @@ def cancel_execution(execution_name: str | None) -> None: project_id = execution_name.split("/", 2)[-1] _local_cancel(project_id) return + if execution_name.startswith("local/placement/"): + project_id = execution_name.split("/", 2)[-1] + _local_cancel(f"placement:{project_id}") + return _cloud_run_cancel(execution_name) + + +def enqueue_placement_pipeline(project_id: str, user_id: str) -> str: + """Dispatch the parallel Placement pipeline (topology plan, no LLM).""" + if use_cloud_run_jobs(): + return _enqueue_cloud_run_job( + project_id, user_id, resume=False, free=True, mode="placement", + ) + return _spawn_local_subprocess( + project_id, user_id, resume=False, free=True, mode="placement", + proc_key=f"placement:{project_id}", + execution_name=f"local/placement/{project_id}", + ) diff --git a/backend/services/pipeline.py b/backend/services/pipeline.py index da3a9fc..c1d5206 100644 --- a/backend/services/pipeline.py +++ b/backend/services/pipeline.py @@ -259,6 +259,8 @@ class PipelineWorkspace: self._upload_file("design_graph.json") self._upload_file("layout_graph.json") self._upload_file("impedance_nets.json") + self._upload_file("functional_groups.json") + self._upload_file("placement_plan.json") self._upload_file("bom_summary.json") self._upload_file("derating.json") self._upload_file("report.json") diff --git a/backend/services/placement_pipeline.py b/backend/services/placement_pipeline.py new file mode 100644 index 0000000..c52f15f --- /dev/null +++ b/backend/services/placement_pipeline.py @@ -0,0 +1,200 @@ +"""Placement pipeline — parallel to analysis, topology only (no LLM / no mm). + +Stages: ensure_graph → classify → write_plan. +Writes ``placement_plan.json`` (+ refreshes ``functional_groups.json``). +Uses ``placement_status`` so analysis ``status`` is untouched. +""" + +from __future__ import annotations + +import logging +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.services import projects as proj_svc +from backend.services.pipeline import PipelineWorkspace, broker +from backend.services.storage import StorageBackend + +logger = logging.getLogger(__name__) + +_PLACEMENT_ACTIVE = frozenset({"queued", "running"}) +_ANALYSIS_BUSY = frozenset({ + proj_svc.STATUS_QUEUED, + proj_svc.STATUS_RUNNING, +}) + + +def _load_constraints_map(extracted_dir: Path) -> dict[str, ComponentConstraints]: + """Load per-MPN extractions without importing the Anthropic review path.""" + result: dict[str, ComponentConstraints] = {} + if not extracted_dir.is_dir(): + return result + for f in extracted_dir.glob("*.json"): + try: + c = ComponentConstraints.model_validate_json( + f.read_text(encoding="utf-8"), + ) + except Exception: + logger.exception("skipping bad extraction %s", f) + continue + result[c.mpn] = c + return result + + +def _publish(project_id: str, event: str, data: dict) -> None: + broker.publish(project_id, event, data) + + +def _step(project_id: str, stage: str, status: str, detail: str = "") -> None: + payload: dict = {"stage": stage, "status": status} + if detail: + payload["detail"] = detail + _publish(project_id, "placement_step_update", payload) + + +async def run_placement_pipeline( + storage: StorageBackend, user_id: str, project_id: str, +) -> None: + """Run the placement topology pipeline (no extraction / review).""" + meta = proj_svc.get_project(storage, user_id, project_id) + if not meta: + raise ValueError(f"Project {project_id} not found") + + # Boot: queued → running on placement_status only. + if meta.placement_status not in _PLACEMENT_ACTIVE: + logger.warning( + "placement worker booted with placement_status=%s for %s; exiting", + meta.placement_status, project_id, + ) + return + + proj_svc.update_project( + storage, user_id, project_id, + placement_status="running", + placement_cancel_requested=False, + placement_state=None, + ) + + try: + async with PipelineWorkspace(storage, user_id, project_id) as ws: + if _cancelled(storage, user_id, project_id): + _finish_cancelled(storage, user_id, project_id) + return + + graph = await _ensure_graph(ws, meta, project_id) + if _cancelled(storage, user_id, project_id): + _finish_cancelled(storage, user_id, project_id) + return + + _step(project_id, "classify", "running", "domains and satellite roles") + extracted_dir = ws.local_path("extracted") + cmap = _load_constraints_map(extracted_dir) + plan = build_placement_plan(graph, cmap) + _step( + project_id, "classify", "complete", + f"{len(plan.domains)} domains, {len(plan.groups)} IC groups", + ) + + if _cancelled(storage, user_id, project_id): + _finish_cancelled(storage, user_id, project_id) + return + + _step(project_id, "write_plan", "running") + plan_path = ws.local_path("placement_plan.json") + plan_json = plan.model_dump_json(indent=2) + "\n" + plan_path.write_text(plan_json) + # Keep functional_groups.json in sync for consumers that already read it. + fg_path = ws.local_path("functional_groups.json") + fg_path.write_text(plan_json) + ws._upload_file("placement_plan.json") + ws._upload_file("functional_groups.json") + _step(project_id, "write_plan", "complete", "placement_plan.json") + + proj_svc.update_project( + storage, user_id, project_id, + placement_status="complete", + placement_state={ + "domains": len(plan.domains), + "groups": len(plan.groups), + }, + placement_cancel_requested=False, + ) + _publish(project_id, "placement_complete", { + "domains": len(plan.domains), + "groups": len(plan.groups), + }) + except Exception as e: + logger.exception("placement pipeline failed for %s", project_id) + proj_svc.update_project( + storage, user_id, project_id, + placement_status="error", + placement_state={"error": str(e)}, + ) + _publish(project_id, "placement_error", {"error": str(e)}) + + +async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignGraph: + """Reuse design_graph.json when present; otherwise graph_build only.""" + graph_path = ws.local_path("design_graph.json") + if graph_path.is_file(): + _step(project_id, "ensure_graph", "running", "reusing design_graph.json") + graph = DesignGraph.model_validate_json(graph_path.read_text(encoding="utf-8")) + _step( + project_id, "ensure_graph", "complete", + f"{len(graph.components)} components (cached)", + ) + return graph + + _step(project_id, "ensure_graph", "running", "building design graph") + bom_path = ws.local_path("uploads/bom.csv") + netlist_path = ws.netlist_local_path() + if not bom_path.is_file() or not Path(netlist_path).is_file(): + raise FileNotFoundError("Missing BOM or netlist for placement graph_build") + + col_map = meta.bom_columns or {} + graph = build_graph( + str(netlist_path), + str(bom_path), + str(ws.local_path("extracted")), + str(ws.local_path("patterns")), + str(ws.local_path("models")), + reference_col=col_map.get("reference", "Reference"), + mpn_col=col_map.get("mpn", "Manufacturer Part Number"), + include_subdesigns=( + set(meta.netlist_subdesigns) + if meta.netlist_subdesigns is not None + else None + ), + pcb_path=ws.local_path("uploads/pcb.kicad_pcb"), + ) + graph_path.write_text(graph.model_dump_json(indent=2) + "\n") + ws._upload_file("design_graph.json") + _step( + project_id, "ensure_graph", "complete", + f"{len(graph.components)} components, {len(graph.nets)} nets", + ) + return graph + + +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) + + +def _finish_cancelled(storage: StorageBackend, user_id: str, project_id: str) -> None: + proj_svc.update_project( + storage, user_id, project_id, + placement_status="cancelled", + placement_cancel_requested=False, + ) + _publish(project_id, "placement_cancelled", {}) + + +def placement_busy(meta: proj_svc.ProjectMeta) -> bool: + return (meta.placement_status or "draft") in _PLACEMENT_ACTIVE + + +def analysis_busy(meta: proj_svc.ProjectMeta) -> bool: + return meta.status in _ANALYSIS_BUSY diff --git a/backend/services/projects.py b/backend/services/projects.py index 03a1bcf..9d01093 100644 --- a/backend/services/projects.py +++ b/backend/services/projects.py @@ -132,6 +132,13 @@ class ProjectMeta(BaseModel): # gate (inside _charge_for_logs) and exits cleanly. cancel_requested: bool = False + # Placement pipeline (parallel to analysis — does not overwrite status). + # draft | queued | running | complete | error | cancelled + placement_status: str = "draft" + placement_state: dict[str, Any] | None = None + placement_execution_name: str | None = None + placement_cancel_requested: bool = False + def completed_review_refs_for_retry( storage: StorageBackend, user_id: str, project_id: str, diff --git a/docs/piano-implementazione.md b/docs/piano-implementazione.md index 7ffdc35..e780624 100644 --- a/docs/piano-implementazione.md +++ b/docs/piano-implementazione.md @@ -23,9 +23,10 @@ Fonte originale: canvas *Pinscope: crescita e DeepSeek*. Qui lo stato operativo. | P2 | Crystal CL + NC pin | **Done** — check deterministici (numeri solo se presenti) | | P3 | Plugin CI / chat report | Todo | | Layout F1 | Domini / gruppi / satelliti | **Done** — `functional_groups.json` (no mm) | +| Layout F1b | Pipeline Placement parallela | **Done** — API/UI `placement_*`, `placement_plan.json` | | Layout F2 | Placement IC packing mm | **Dopo** — gated `.kicad_pcb` + `layout_rules` numerici | -**Done when (prossimo pacchetto):** smoke `--live` verde; hit_ratio visibile in UI logs; pipeline Placement parallela (API dedicata). +**Done when (prossimo pacchetto):** smoke `--live` verde; hit_ratio visibile in UI logs; packing mm (F2) gated. --- @@ -42,7 +43,7 @@ Obiettivo unico: **routing migliore** (loop corti, meno crossing, canali liberi) | 5 | `assemble_order` dominio → chip → gruppi (contratto packer) | F1 metadato | | 6 | Packing mm / zone PCB / export | **F2** | -Output F1: `functional_groups.json` scritto in `graph_build`. Verifica PCB esistente resta `placement_check` (PS-PLC*) — non confondere con packing. +Output F1: `functional_groups.json` scritto in `graph_build`. Pipeline parallela Placement (`POST …/placement/start`) riscrive anche `placement_plan.json` senza toccare lo `status` di analisi. Verifica PCB esistente resta `placement_check` (PS-PLC*) — non confondere con packing. --- diff --git a/frontend/content/changelog.md b/frontend/content/changelog.md index fa31b28..4f3500a 100644 --- a/frontend/content/changelog.md +++ b/frontend/content/changelog.md @@ -2,6 +2,14 @@ What's new in Pinscope. +## 2.28.1 — 2026-09-12 — Placement pipeline (parallel) + +Dedicated Placement job builds the routing-first topology plan without touching the analysis pipeline status or spending credits. No millimetres — domains, IC groups, satellites only. + +- [New] `POST /api/pipeline/{id}/placement/start` (+ cancel, events SSE, get plan). +- [New] Project page **Build placement plan** → `/project/{id}/placement` progress + topology viewer. +- [New] Worker `MODE=placement` writes `placement_plan.json` (and refreshes `functional_groups.json`). + ## 2.28.0 — 2026-09-12 — Layout F1 topology + crystal/NC checks Routing-first floorplan foundation without inventing millimetres: domains and satellite role hints after graph build, plus deterministic crystal CL and NC-pin checks. diff --git a/frontend/src/app/(app)/project/[id]/page.tsx b/frontend/src/app/(app)/project/[id]/page.tsx index ce9de64..871a781 100644 --- a/frontend/src/app/(app)/project/[id]/page.tsx +++ b/frontend/src/app/(app)/project/[id]/page.tsx @@ -16,6 +16,7 @@ import { removeCollaborator, makeCollaboratorOwner, startPipeline, + startPlacementPipeline, reprocessPipeline, resumePipeline, fetchPipelineEstimate, @@ -38,6 +39,7 @@ import { Copy, Check, Upload, + LayoutGrid, } from "lucide-react"; import { useOptionalUser } from "@/hooks/use-optional-auth"; import { ImpedancePanel } from "@/components/project/impedance-panel"; @@ -106,9 +108,20 @@ export default function ProjectDetailPage({ } }, [project?.status, id, router]); + // Placement progress page when placement is active + useEffect(() => { + if ( + project?.placementStatus === "running" || + project?.placementStatus === "queued" + ) { + router.replace(`/project/${id}/placement`); + } + }, [project?.placementStatus, id, router]); + const searchParams = useSearchParams(); const tab = searchParams.get("tab") ?? "bom"; const [starting, setStarting] = useState(false); + const [startingPlacement, setStartingPlacement] = useState(false); const [estimate, setEstimate] = useState(null); const [rerunProject, setRerunProject] = useState(null); @@ -183,6 +196,26 @@ export default function ProjectDetailPage({ } }; + const analysisBusy = + project?.status === "running" || project?.status === "queued"; + const placementBusy = + project?.placementStatus === "running" || + project?.placementStatus === "queued"; + const canStartPlacement = + Boolean(canRun) && !analysisBusy && !placementBusy; + + const handlePlacement = async () => { + if (!canStartPlacement) return; + setStartingPlacement(true); + try { + await startPlacementPipeline(id); + router.push(`/project/${id}/placement`); + } catch (e) { + setStartingPlacement(false); + alert(e instanceof Error ? e.message : "Failed to start placement"); + } + }; + const hasFailedReviews = Boolean(hasSkipped); return ( @@ -271,6 +304,26 @@ export default function ProjectDetailPage({ + + {project.placementStatus === "complete" && ( + + + + )} ) : isPaused ? ( @@ -305,6 +358,19 @@ export default function ProjectDetailPage({ {starting ? "Starting..." : "Run Pipeline"} )} + {!canRun && ( Upload BOM and netlist to enable diff --git a/frontend/src/app/(app)/project/[id]/placement/page.tsx b/frontend/src/app/(app)/project/[id]/placement/page.tsx new file mode 100644 index 0000000..17622eb --- /dev/null +++ b/frontend/src/app/(app)/project/[id]/placement/page.tsx @@ -0,0 +1,201 @@ +"use client"; + +import { use, useEffect, useState } from "react"; +import Link from "next/link"; +import { useRouter } from "next/navigation"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { PipelineStepper } from "@/components/progress/pipeline-stepper"; +import { usePlacementProgress } from "@/hooks/use-placement-progress"; +import { + cancelPlacementPipeline, + fetchPlacementPlan, + fetchProject, +} from "@/lib/api"; +import { + ArrowLeft, + CheckCircle2, + Loader2, + OctagonX, + Ban, +} from "lucide-react"; + +type Plan = Awaited>; + +export default function PlacementPage({ + params, +}: { + params: Promise<{ id: string }>; +}) { + const { id } = use(params); + const router = useRouter(); + const [projectName, setProjectName] = useState(""); + const [placementStatus, setPlacementStatus] = useState("draft"); + const [plan, setPlan] = useState(null); + const [cancelling, setCancelling] = useState(false); + + const alreadyDone = placementStatus === "complete"; + const { steps, done, cancelled, error, summary, started } = usePlacementProgress( + id, + !alreadyDone && placementStatus !== "draft", + ); + + useEffect(() => { + fetchProject(id) + .then((p) => { + setProjectName(p.name); + setPlacementStatus(p.placementStatus ?? "draft"); + }) + .catch(() => {}); + }, [id]); + + useEffect(() => { + if (!alreadyDone && !done) return; + fetchPlacementPlan(id) + .then(setPlan) + .catch(() => setPlan(null)); + }, [id, alreadyDone, done]); + + const handleCancel = async () => { + setCancelling(true); + try { + await cancelPlacementPipeline(id); + } catch { + // may already be finished + } finally { + setCancelling(false); + } + }; + + const finished = alreadyDone || done; + const isRunning = !finished && !cancelled && !error; + const isQueued = isRunning && !started && !alreadyDone; + + return ( +
+
+
+

Placement plan

+

+ {projectName ? `${projectName} · ` : ""} + Routing-first topology (no millimetres) +

+
+ + + +
+ + {isQueued && ( +
+ +

Queued — starting placement worker…

+
+ )} + + {isRunning && !isQueued && ( + + + Progress + + + +
+ +
+ + + )} + + {cancelled && ( +
+ + Placement cancelled +
+ )} + + {error && ( +
+

Placement failed

+

{error}

+ +
+ )} + + {finished && !error && !cancelled && ( +
+
+ + Placement plan ready + {summary && ( + + · {summary.domains ?? plan?.domains.length ?? "?"} domains,{" "} + {summary.groups ?? plan?.groups.length ?? "?"} IC groups + + )} +
+ + {plan && ( + + + Topology + + + {plan.domains.map((d) => ( +
+

{d.domain_id}

+

+ Power: {d.power_nets.join(", ") || "—"} +

+

+ Assemble: {d.assemble_order.join(" → ") || "—"} +

+
+ ))} + {plan.groups.length > 0 && ( +
+

IC groups

+ {plan.groups.map((g) => ( +
+ {g.ref} + {g.mpn ? ` · ${g.mpn}` : ""} + {g.component_subtype ? ` · ${g.component_subtype}` : ""} + {g.satellites.length > 0 && ( + + {" "} + — satellites:{" "} + {g.satellites + .map((s) => `${s.ref}(${s.role_hint ?? "other"})`) + .join(", ")} + + )} +
+ ))} +
+ )} + + + )} +
+ )} +
+ ); +} diff --git a/frontend/src/hooks/use-placement-progress.ts b/frontend/src/hooks/use-placement-progress.ts new file mode 100644 index 0000000..02f6870 --- /dev/null +++ b/frontend/src/hooks/use-placement-progress.ts @@ -0,0 +1,173 @@ +"use client"; + +import { useState, useEffect, useRef, useCallback } from "react"; +import type { PipelineStep } from "@/lib/types"; +import { placementEventsUrl } from "@/lib/api"; +import { useOptionalAuth } from "@/hooks/use-optional-auth"; + +const PLACEMENT_STAGES = [ + { + id: "ensure_graph", + title: "Ensure design graph", + description: "Reuse or build design_graph.json", + }, + { + id: "classify", + title: "Classify topology", + description: "Domains, IC groups, satellite roles", + }, + { + id: "write_plan", + title: "Write placement plan", + description: "Save placement_plan.json (no millimetres)", + }, +] as const; + +const STAGE_INDEX: Record = Object.fromEntries( + PLACEMENT_STAGES.map((s, i) => [s.id, i]), +); + +function createInitialSteps(): PipelineStep[] { + return PLACEMENT_STAGES.map((s) => ({ + title: s.title, + description: s.description, + status: "pending" as const, + substeps: [], + })); +} + +export function usePlacementProgress(projectId: string | null, enabled = true) { + const [steps, setSteps] = useState(createInitialSteps); + const [done, setDone] = useState(false); + const [cancelled, setCancelled] = useState(false); + const [error, setError] = useState(null); + const [summary, setSummary] = useState<{ domains?: number; groups?: number } | null>(null); + const [started, setStarted] = useState(false); + const esRef = useRef(null); + const terminalRef = useRef(false); + const { getToken } = useOptionalAuth(); + + const handleEvent = useCallback((event: MessageEvent) => { + const eventType = event.type || "message"; + if (eventType === "heartbeat") return; + + let data: Record; + try { + data = JSON.parse(event.data); + } catch { + return; + } + + if (eventType === "placement_complete") { + setSummary({ + domains: Number(data.domains) || 0, + groups: Number(data.groups) || 0, + }); + setDone(true); + terminalRef.current = true; + esRef.current?.close(); + return; + } + + if (eventType === "placement_cancelled") { + setCancelled(true); + setDone(true); + terminalRef.current = true; + esRef.current?.close(); + return; + } + + if (eventType === "placement_error") { + setError((data.error as string) || "Placement failed"); + setDone(true); + terminalRef.current = true; + esRef.current?.close(); + return; + } + + if (eventType !== "placement_step_update") return; + + setStarted(true); + const stage = data.stage as string; + const status = data.status as "pending" | "running" | "complete" | "failed"; + const detail = data.detail as string | undefined; + + setSteps((prev) => { + const next = prev.map((s) => ({ ...s, substeps: [...s.substeps] })); + const idx = STAGE_INDEX[stage]; + if (idx === undefined) return next; + const step = next[idx]; + if (status === "running") { + step.status = "running"; + if (detail) step.description = detail; + } else if (status === "complete") { + step.status = "complete"; + if (detail) step.description = detail; + } + return next; + }); + }, []); + + useEffect(() => { + if (!projectId || !enabled) return; + + let es: EventSource | null = null; + let retries = 0; + const MAX_RETRIES = 50; + let reconnectTimer: ReturnType | null = null; + let closed = false; + + async function connect() { + if (closed) return; + if (es) { + es.close(); + es = null; + } + + const token = await getToken(); + const baseUrl = placementEventsUrl(projectId!); + const url = token ? `${baseUrl}?token=${token}` : baseUrl; + + es = new EventSource(url); + esRef.current = es; + + for (const eventName of [ + "placement_step_update", + "placement_complete", + "placement_error", + "placement_cancelled", + "heartbeat", + ]) { + es.addEventListener(eventName, (event: MessageEvent) => { + retries = 0; + handleEvent(event); + }); + } + + es.onerror = () => { + if (closed || terminalRef.current) return; + es?.close(); + es = null; + esRef.current = null; + if (retries >= MAX_RETRIES) { + setError("Lost connection to placement pipeline. Refresh to reconnect."); + setDone(true); + return; + } + retries++; + reconnectTimer = setTimeout(connect, 30_000); + }; + } + + connect(); + + return () => { + closed = true; + if (reconnectTimer) clearTimeout(reconnectTimer); + es?.close(); + esRef.current = null; + }; + }, [projectId, enabled, getToken, handleEvent]); + + return { steps, done, cancelled, error, summary, started }; +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index a9705f6..3a03a18 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -102,6 +102,8 @@ function mapProject(p: Record): Project { pinscopeVersion: (p.pinscope_version as string | null | undefined) ?? null, netlistFormat: (p.netlist_format as Project["netlistFormat"]) ?? null, netlistSubdesigns: (p.netlist_subdesigns as string[] | null) ?? null, + placementStatus: (p.placement_status as Project["placementStatus"]) ?? "draft", + placementState: (p.placement_state as Record | null) ?? null, }; } @@ -578,6 +580,60 @@ export async function fetchPipelineStatus(projectId: string) { return res.json(); } +export async function startPlacementPipeline(projectId: string) { + const res = await authFetch( + `${BASE}/api/pipeline/${projectId}/placement/start`, + { method: "POST" }, + ); + if (!res.ok) { + const err = await res.json().catch(() => ({ detail: "Failed to start placement" })); + throw new Error(err.detail || "Failed to start placement"); + } + return res.json(); +} + +export async function cancelPlacementPipeline(projectId: string) { + const res = await authFetch( + `${BASE}/api/pipeline/${projectId}/placement/cancel`, + { method: "POST" }, + ); + if (!res.ok) { + const err = await res.json().catch(() => ({ detail: "Failed to cancel placement" })); + throw new Error(err.detail || "Failed to cancel placement"); + } + return res.json(); +} + +export function placementEventsUrl(projectId: string): string { + return `${BASE}/api/pipeline/${projectId}/placement/events`; +} + +export async function fetchPlacementPlan(projectId: string): Promise<{ + objective?: string; + domains: Array<{ + domain_id: string; + power_nets: string[]; + ic_refs: string[]; + assemble_order: string[]; + }>; + groups: Array<{ + ref: string; + mpn?: string | null; + component_subtype?: string | null; + rank?: number; + satellites: Array<{ ref: string; role_hint?: string; hop?: number }>; + layout_rules?: unknown[]; + assemble_order?: string[]; + }>; +}> { + const res = await authFetch(`${BASE}/api/pipeline/${projectId}/placement/plan`); + if (!res.ok) { + const err = await res.json().catch(() => ({ detail: "Placement plan not found" })); + throw new Error(err.detail || "Placement plan not found"); + } + return res.json(); +} + // --- Logs --- export async function fetchProjectLogs( diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 4986892..dac4439 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -280,6 +280,9 @@ export interface Project { // null means "include every sub-design found in the file" — the default // for single-sub-design EDIFs and all PADS netlists. netlistSubdesigns?: string[] | null; + // Placement pipeline (parallel to analysis — topology only). + placementStatus?: "draft" | "queued" | "running" | "complete" | "error" | "cancelled"; + placementState?: Record | null; } // One entry per EDIF sub-design (`&NNNN` ID prefix). Returned by the upload diff --git a/tests/test_functional_groups.py b/tests/test_functional_groups.py index 0b21b19..f8d7e87 100644 --- a/tests/test_functional_groups.py +++ b/tests/test_functional_groups.py @@ -56,3 +56,11 @@ def test_domains_cover_all_ics(): covered = {r for d in report.domains for r in d.ic_refs} assert covered == {"U1", "U2", "U3"} assert all(d.assemble_order for d in report.domains) + + +def test_build_placement_plan_alias(): + from backend.pinscopex.functional_groups import build_placement_plan + report = build_placement_plan(_graph()) + assert report.objective == "routing" + assert {g.ref for g in report.groups} >= {"U1", "U2", "U3"} + diff --git a/tests/test_placement_pipeline.py b/tests/test_placement_pipeline.py new file mode 100644 index 0000000..40bf6a2 --- /dev/null +++ b/tests/test_placement_pipeline.py @@ -0,0 +1,52 @@ +"""Placement pipeline smoke — topology only, no LLM.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from backend.pinscopex.functional_groups import ( + FunctionalGroupsReport, + build_placement_plan, +) +from backend.pinscopex.models import DesignGraph + + +ROOT = Path(__file__).resolve().parents[1] +SIMPLE = ROOT / "simple_project" + + +@pytest.fixture +def graph() -> DesignGraph: + path = SIMPLE / "design_graph.json" + return DesignGraph.model_validate_json(path.read_text(encoding="utf-8")) + + +def test_placement_plan_writes_domains_and_groups(graph: DesignGraph, tmp_path: Path): + plan = build_placement_plan(graph) + assert plan.objective == "routing" + assert plan.domains + assert plan.groups + out = tmp_path / "placement_plan.json" + out.write_text(plan.model_dump_json(indent=2) + "\n") + loaded = FunctionalGroupsReport.model_validate_json(out.read_text()) + assert len(loaded.groups) == len(plan.groups) + + +def test_placement_busy_helpers(): + from backend.services.projects import ProjectMeta, STATUS_QUEUED, STATUS_RUNNING + + # Mirror placement_pipeline helpers without importing the worker stack + # (that pulls Anthropic via services.pipeline in lean test envs). + active = frozenset({"queued", "running"}) + analysis = frozenset({STATUS_QUEUED, STATUS_RUNNING}) + + draft = ProjectMeta(id="p", name="t", created="2026-01-01", user_id="u") + assert (draft.placement_status or "draft") not in active + assert draft.status not in analysis + + draft.placement_status = "queued" + assert (draft.placement_status or "draft") in active + draft.status = STATUS_RUNNING + assert draft.status in analysis