diff --git a/CHANGELOG.md b/CHANGELOG.md index c2121a5..1854f82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,6 +80,28 @@ All notable changes to the KiCAD MCP Server project are documented here. ### Tool Enhancements +- `autoroute`: best-of-N support. New optional parameters `attempts`, + `targetNets`, and `passSchedule`. When `attempts > 1`, Freerouting is + invoked multiple times with varied `--max-passes` values, each result + is scored by `(nets_routed * 1000) + segments` plus a 50,000-point + bonus when every `targetNets` entry is routed, and the winning SES is + imported into the board. Single-attempt behaviour is unchanged when + `attempts` is omitted, so existing callers don't need updates. + + Motivation: on dense boards a single Freerouting run routinely leaves + 1–7 nets unrouted. Cycling through a few `-mp` values typically drives + the unrouted count to zero. Empirically, 3 attempts is usually enough + for 4-layer designs; 5–8 for stubborn cases. + + The scoring approach and the default `passSchedule` are ported from + [morningfire-pcb-automation](https://github.com/NiNjA-CodE/morningfire-pcb-automation) + (`scripts/routing/freeroute_runner.py`). The MCP version adds: + cleaner per-attempt result reporting, automatic single-thread + optimisation (`-mt 1`) during scored attempts so the multi-threaded + optimiser's known clearance-violation bug doesn't distort the + comparison, and graceful degradation when one attempt errors out + (the run continues and the best of the remainder wins). + - `edit_schematic_component`: extended with two new optional parameters that promote arbitrary custom properties to first-class citizens: - **`properties`** — map of property name to either a string value or a full diff --git a/python/commands/freerouting.py b/python/commands/freerouting.py index 8dff349..e450e8c 100644 --- a/python/commands/freerouting.py +++ b/python/commands/freerouting.py @@ -11,11 +11,12 @@ Supports two execution modes: import logging import os +import re import shutil import subprocess import time from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterable, List, Optional logger = logging.getLogger("kicad_interface") @@ -27,6 +28,12 @@ DEFAULT_FREEROUTING_JAR = os.environ.get( DOCKER_IMAGE = "eclipse-temurin:21-jre" +# Default schedule of `-mp` (max passes) values used when ``attempts`` > 1. +# Cycles through a range that empirically produces enough variation between +# runs to surface a better result than any single fixed value. Ported from +# morningfire-pcb-automation/scripts/routing/freeroute_runner.py. +DEFAULT_PASS_SCHEDULE = [50, 60, 65, 70, 75, 80, 85, 90, 55, 95] + def _find_java() -> Optional[str]: """Find java executable on the system.""" @@ -91,8 +98,17 @@ def _build_freerouting_cmd( ses_path: str, passes: int, use_docker: bool, + single_thread: bool = False, ) -> List[str]: - """Build the command to run Freerouting.""" + """Build the command to run Freerouting. + + ``single_thread`` forces ``-mt 1`` (single-threaded optimisation). + Freerouting 2.x's multi-threaded optimiser is documented to produce + clearance violations in some cases (the runtime even prints a warning); + best-of-N callers should pass this so each attempt's score reflects a + valid routed board, not an artefact of MT optimisation. + """ + extra = ["-mt", "1"] if single_thread else [] if use_docker: docker_exe = _find_docker() if docker_exe is None: @@ -119,6 +135,7 @@ def _build_freerouting_cmd( f"/work/{ses_name}", "-mp", str(passes), + *extra, ] else: java_exe = _find_java() @@ -134,9 +151,64 @@ def _build_freerouting_cmd( ses_path, "-mp", str(passes), + *extra, ] +# --------------------------------------------------------------------------- +# Best-of-N scoring helpers (ported from morningfire-pcb-automation) +# --------------------------------------------------------------------------- +# +# Approach lifted from +# https://github.com/NiNjA-CodE/morningfire-pcb-automation +# scripts/routing/freeroute_runner.py::score_ses +# +# Single-shot Freerouting on dense boards routinely leaves 1–7 nets +# unrouted. Re-running with varied --max-passes values surfaces a better +# solution most of the time; the scoring function below picks the best +# SES across attempts. +# --------------------------------------------------------------------------- + +_SES_NET_RE = re.compile(r'\(net\s+(\S+)\s*\n\s*\(wire') + + +def _score_ses(ses_text: str, target_nets: Iterable[str]) -> Dict[str, Any]: + """Score a Specctra SES file by routing completeness. + + Score = (nets_routed * 1000) + segments + 50000_if_all_targets_routed + + The ``nets_routed * 1000`` term dominates segment count so an attempt + that routes one more net always beats an attempt with marginally more + segments. The target-net bonus is huge so any attempt that routes all + critical nets wins, regardless of segment count. + + Returns: ``{"score": int, "nets": int, "segments": int, "vias": int, + "targets_found": [...], "targets_missing": [...]}`` + """ + nets = set(_SES_NET_RE.findall(ses_text)) + # Strip wrapping quotes if Freerouting emits them. + clean_nets = {n.strip('"') for n in nets} + segments = len(re.findall(r'\(wire', ses_text)) + vias = len(re.findall(r'\(via ', ses_text)) + + targets = set(target_nets) if target_nets else set() + found = sorted(targets & clean_nets) + missing = sorted(targets - clean_nets) + + score = len(clean_nets) * 1000 + segments + if targets and not missing: + score += 50_000 + + return { + "score": score, + "nets": len(clean_nets), + "segments": segments, + "vias": vias, + "targets_found": found, + "targets_missing": missing, + } + + class FreeroutingCommands: """Handles Freerouting autoroute operations.""" @@ -174,11 +246,30 @@ class FreeroutingCommands: def autoroute(self, params: Dict[str, Any]) -> Dict[str, Any]: """Run Freerouting autorouter on the current board. - Flow: - 1. Export board to Specctra DSN - 2. Run Freerouting CLI on DSN -> SES - 3. Import SES back into the board - 4. Save the board + Single-attempt flow (default): + 1. Export board to Specctra DSN + 2. Run Freerouting CLI on DSN -> SES (one pass with ``maxPasses``) + 3. Import SES back into the board + 4. Save the board + + Best-of-N flow (``attempts > 1``): + 1. Export DSN once + 2. Run Freerouting ``attempts`` times, varying ``--max-passes`` + per the ``passSchedule`` (defaults to a built-in schedule + of 10 spread-out values). + 3. Score each SES by (nets_routed * 1000) + segments, plus a + 50,000-point bonus when every ``targetNets`` entry routed. + 4. Keep the highest-scoring SES; import that one into the board. + + Single-attempt behaviour is unchanged when ``attempts`` is omitted + or set to 1, so existing callers do not need updates. + + The best-of-N scoring approach is ported from + morningfire-pcb-automation + (https://github.com/NiNjA-CodE/morningfire-pcb-automation, + scripts/routing/freeroute_runner.py). On dense boards a single + run regularly leaves 1–7 nets unrouted; cycling through a few + ``-mp`` values typically gets the count to zero. """ try: import pcbnew @@ -211,6 +302,27 @@ class FreeroutingCommands: timeout = params.get("timeout", 300) passes = params.get("maxPasses", 20) + # Best-of-N parameters + attempts_raw = params.get("attempts", 1) + try: + attempts = int(attempts_raw) if attempts_raw is not None else 1 + except (TypeError, ValueError): + return { + "success": False, + "message": "Invalid attempts value", + "errorDetails": f"attempts must be a positive integer; got {attempts_raw!r}", + } + if attempts < 1: + return { + "success": False, + "message": "Invalid attempts value", + "errorDetails": "attempts must be >= 1", + } + target_nets = list(params.get("targetNets") or []) + pass_schedule = list(params.get("passSchedule") or DEFAULT_PASS_SCHEDULE) + if not pass_schedule: + pass_schedule = [passes] + # Validate Freerouting JAR if not os.path.isfile(jar_path): return { @@ -239,8 +351,9 @@ class FreeroutingCommands: board_stem = Path(board_path).stem dsn_path = os.path.join(board_dir, f"{board_stem}.dsn") ses_path = os.path.join(board_dir, f"{board_stem}.ses") + best_ses_path = os.path.join(board_dir, f"{board_stem}_best.ses") - # Step 1: Export DSN + # Step 1: Export DSN (once, regardless of attempt count) logger.info(f"Exporting DSN to {dsn_path}") try: result = pcbnew.ExportSpecctraDSN(self.board, dsn_path) @@ -267,57 +380,149 @@ class FreeroutingCommands: dsn_size = os.path.getsize(dsn_path) logger.info(f"DSN exported: {dsn_size} bytes") - # Step 2: Run Freerouting - cmd = _build_freerouting_cmd(jar_path, dsn_path, ses_path, passes, use_docker) - + # Step 2: Run Freerouting (single or multiple attempts) mode_label = "docker" if use_docker else "direct" - logger.info(f"Running Freerouting ({mode_label}): {' '.join(cmd)}") - start_time = time.time() + total_start = time.time() + attempt_results: List[Dict[str, Any]] = [] + best_score = -1 + best_attempt_idx = -1 + best_proc_stdout = "" - try: - proc = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout, - cwd=board_dir, + # If only one attempt, use the legacy maxPasses value (preserves + # exact backward-compatible behaviour). Otherwise cycle through + # passSchedule. Always run single-threaded when scoring multiple + # attempts so the optimiser doesn't introduce clearance violations + # that would distort the comparison. + for idx in range(attempts): + if attempts == 1: + attempt_passes = passes + single_thread = False + else: + attempt_passes = pass_schedule[idx % len(pass_schedule)] + single_thread = True + + cmd = _build_freerouting_cmd( + jar_path, dsn_path, ses_path, attempt_passes, use_docker, + single_thread=single_thread, + ) + logger.info( + f"Freerouting attempt {idx + 1}/{attempts} " + f"(mp={attempt_passes}, mode={mode_label})" ) - elapsed = round(time.time() - start_time, 1) - if proc.returncode != 0: + attempt_start = time.time() + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + cwd=board_dir, + ) + attempt_elapsed = round(time.time() - attempt_start, 1) + except subprocess.TimeoutExpired: return { "success": False, - "message": (f"Freerouting exited with code " f"{proc.returncode}"), - "errorDetails": proc.stderr or proc.stdout, - "elapsed_seconds": elapsed, - "mode": mode_label, + "message": f"Freerouting timed out after {timeout}s", + "errorDetails": "Increase timeout or reduce board complexity", + "attempts_completed": idx, } - except subprocess.TimeoutExpired: + except Exception as e: + return { + "success": False, + "message": "Failed to run Freerouting", + "errorDetails": str(e), + "attempts_completed": idx, + } + + if proc.returncode != 0: + # Don't abort the whole best-of-N just because one attempt + # exits nonzero — record it and move on. + attempt_results.append({ + "attempt": idx + 1, + "max_passes": attempt_passes, + "elapsed_seconds": attempt_elapsed, + "ok": False, + "exit_code": proc.returncode, + "stderr": (proc.stderr or "")[:200], + }) + if attempts == 1: + return { + "success": False, + "message": f"Freerouting exited with code {proc.returncode}", + "errorDetails": proc.stderr or proc.stdout, + "elapsed_seconds": attempt_elapsed, + "mode": mode_label, + } + continue + + if not os.path.isfile(ses_path): + attempt_results.append({ + "attempt": idx + 1, + "max_passes": attempt_passes, + "elapsed_seconds": attempt_elapsed, + "ok": False, + "error": "no SES produced", + }) + if attempts == 1: + return { + "success": False, + "message": "Freerouting did not produce SES output", + "errorDetails": ( + f"Expected at: {ses_path}. Stdout: {proc.stdout[:500]}" + ), + "elapsed_seconds": attempt_elapsed, + } + continue + + # Score this attempt + with open(ses_path, "r", encoding="utf-8", errors="replace") as fh: + ses_text = fh.read() + score_info = _score_ses(ses_text, target_nets) + score = score_info["score"] + attempt_results.append({ + "attempt": idx + 1, + "max_passes": attempt_passes, + "elapsed_seconds": attempt_elapsed, + "ok": True, + **score_info, + }) + logger.info( + f" attempt {idx + 1}: score={score} " + f"({score_info['nets']} nets, {score_info['segments']} segs, " + f"{score_info['vias']} vias)" + ) + + if score > best_score: + best_score = score + best_attempt_idx = idx + best_proc_stdout = proc.stdout or "" + # Snapshot the SES that produced this score so later + # attempts (which overwrite ses_path) don't clobber it. + shutil.copy2(ses_path, best_ses_path) + + elapsed = round(time.time() - total_start, 1) + + if best_attempt_idx == -1: return { "success": False, - "message": (f"Freerouting timed out after {timeout}s"), - "errorDetails": ("Increase timeout or reduce board complexity"), - } - except Exception as e: - return { - "success": False, - "message": "Failed to run Freerouting", - "errorDetails": str(e), + "message": "All Freerouting attempts failed", + "errorDetails": "No attempt produced a usable SES file", + "elapsed_seconds": elapsed, + "attempts": attempt_results, } - # Check SES output - if not os.path.isfile(ses_path): - return { - "success": False, - "message": "Freerouting did not produce SES output", - "errorDetails": (f"Expected at: {ses_path}. " f"Stdout: {proc.stdout[:500]}"), - "elapsed_seconds": elapsed, - } + # Restore the winning SES as the canonical output file + if attempts > 1: + shutil.copy2(best_ses_path, ses_path) ses_size = os.path.getsize(ses_path) - logger.info(f"SES produced: {ses_size} bytes in {elapsed}s") + logger.info( + f"Best SES: attempt {best_attempt_idx + 1}, score={best_score}, " + f"{ses_size} bytes (total {elapsed}s)" + ) - # Step 3: Import SES + # Step 3: Import the winning SES logger.info(f"Importing SES from {ses_path}") try: result = pcbnew.ImportSpecctraSES(self.board, ses_path) @@ -325,8 +530,9 @@ class FreeroutingCommands: return { "success": False, "message": "SES import failed", - "errorDetails": (f"ImportSpecctraSES returned: {result}"), + "errorDetails": f"ImportSpecctraSES returned: {result}", "elapsed_seconds": elapsed, + "attempts": attempt_results, } except Exception as e: return { @@ -334,6 +540,7 @@ class FreeroutingCommands: "message": "SES import failed", "errorDetails": str(e), "elapsed_seconds": elapsed, + "attempts": attempt_results, } # Step 4: Save board @@ -352,7 +559,7 @@ class FreeroutingCommands: else: track_count += 1 - return { + response: Dict[str, Any] = { "success": True, "message": f"Autoroute completed in {elapsed}s", "mode": mode_label, @@ -363,8 +570,14 @@ class FreeroutingCommands: "tracks": track_count, "vias": via_count, }, - "freerouting_stdout": (proc.stdout[:1000] if proc.stdout else ""), + "freerouting_stdout": best_proc_stdout[:1000], } + if attempts > 1: + response["attempts"] = attempt_results + response["best_attempt"] = best_attempt_idx + 1 + response["best_score"] = best_score + response["best_ses_path"] = best_ses_path + return response def export_dsn(self, params: Dict[str, Any]) -> Dict[str, Any]: """Export the board to Specctra DSN format only.""" diff --git a/src/tools/freerouting.ts b/src/tools/freerouting.ts index c1924ec..dcd89a1 100644 --- a/src/tools/freerouting.ts +++ b/src/tools/freerouting.ts @@ -9,9 +9,17 @@ import { z } from "zod"; export function registerFreeroutingTools(server: McpServer, callKicadScript: Function) { // Full autoroute: export DSN -> run Freerouting -> import SES + // + // Best-of-N support (the `attempts` / `targetNets` / `passSchedule` + // parameters) is ported from morningfire-pcb-automation: + // https://github.com/NiNjA-CodE/morningfire-pcb-automation + // (scripts/routing/freeroute_runner.py — `score_ses` + run loop) + // On dense boards a single attempt regularly leaves 1–7 nets unrouted; + // cycling through a few `--max-passes` values typically drives the + // unrouted count to zero. server.tool( "autoroute", - "Run Freerouting autorouter on the current PCB. Exports to Specctra DSN, runs Freerouting CLI, and imports the routed SES result. Requires Java 11+ and freerouting.jar (see check_freerouting).", + "Run Freerouting autorouter on the current PCB. Exports to Specctra DSN, runs Freerouting CLI, and imports the routed SES result. Requires Java 11+ and freerouting.jar (see check_freerouting). Set `attempts` > 1 to run best-of-N: Freerouting is invoked multiple times with varied `--max-passes`, each result is scored by (nets_routed * 1000 + segments, +50000 bonus when every `targetNets` entry routed), and the winning SES is imported. Single-attempt behaviour is unchanged when `attempts` is omitted.", { boardPath: z.string().optional().describe("Path to .kicad_pcb file (default: current board)"), freeroutingJar: z @@ -20,8 +28,30 @@ export function registerFreeroutingTools(server: McpServer, callKicadScript: Fun .describe( "Path to freerouting.jar (default: ~/.kicad-mcp/freerouting.jar or FREEROUTING_JAR env)", ), - maxPasses: z.number().optional().describe("Maximum routing passes (default: 20)"), - timeout: z.number().optional().describe("Timeout in seconds (default: 300)"), + maxPasses: z.number().optional().describe( + "Maximum routing passes for single-attempt mode (default: 20). Ignored when `attempts` > 1; use `passSchedule` instead.", + ), + timeout: z.number().optional().describe("Per-attempt timeout in seconds (default: 300)"), + attempts: z + .number() + .int() + .min(1) + .optional() + .describe( + "Number of Freerouting runs to try (default: 1 — backward-compatible). When > 1, runs best-of-N: scores each attempt by routing completeness and keeps the SES with the highest score. Recommended: 3–5 for dense boards.", + ), + targetNets: z + .array(z.string()) + .optional() + .describe( + "Optional list of critical net names. An attempt that routes all of them earns a 50,000-point scoring bonus, breaking ties in favour of designs that include the must-have nets.", + ), + passSchedule: z + .array(z.number()) + .optional() + .describe( + "Per-attempt `--max-passes` values to cycle through (default: [50, 60, 65, 70, 75, 80, 85, 90, 55, 95]). The list wraps if `attempts` exceeds its length.", + ), }, async (args: any) => { const result = await callKicadScript("autoroute", args); diff --git a/tests/test_autoroute_best_of_n.py b/tests/test_autoroute_best_of_n.py new file mode 100644 index 0000000..99a14d2 --- /dev/null +++ b/tests/test_autoroute_best_of_n.py @@ -0,0 +1,295 @@ +"""Integration tests for autoroute's best-of-N loop. + +Mocks subprocess + pcbnew so the test verifies the orchestration without +actually invoking Freerouting or KiCad. Specifically: + + - attempts=1 preserves single-attempt behaviour (no `attempts`/ + `passSchedule` in response, no per-attempt `_best.ses` snapshotting). + - attempts>1 runs N times and imports the highest-scoring SES. + - One failing attempt doesn't abort the whole best-of-N run. + - The pass schedule wraps when attempts > len(schedule). +""" +import sys +import types +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +PYTHON_DIR = Path(__file__).parent.parent / "python" +sys.path.insert(0, str(PYTHON_DIR)) + +from commands import freerouting as fr_mod # noqa: E402 +from commands.freerouting import FreeroutingCommands # noqa: E402 + + +# Synthetic SES text: a `(net ...` section with N (wire ...) inside -> +# scores as N nets, N segments (matching the scorer's expectations). +def _make_ses(num_nets, segs_per_net=2): + chunks = [] + for i in range(num_nets): + chunks.append(f'(net "N{i}"\n') + for _ in range(segs_per_net): + chunks.append(" (wire (path F.Cu 200 0 0 1 1))\n") + chunks.append(")\n") + return "".join(chunks) + + +def _stub_pcbnew(dsn_to_create=None): + """A minimal pcbnew stub for autoroute's needs. + + ExportSpecctraDSN normally writes a file as a side effect; tests pass + a path so the stub creates an empty placeholder file to match the + real-world contract (the `if not os.path.isfile(dsn_path):` guard + inside autoroute). + """ + pcb = MagicMock(name="pcbnew_module") + + def _fake_export(_board, dsn_path, *_args, **_kw): + # Write a placeholder DSN so the existence check passes + with open(dsn_path, "w") as f: + f.write("(pcb \"stub\")\n") + return True + + pcb.ExportSpecctraDSN.side_effect = _fake_export + pcb.ImportSpecctraSES.return_value = True + return pcb + + +def _make_cmds(board_path, dsn_exists=True): + """FreeroutingCommands with a mocked board pointing at board_path.""" + cc = FreeroutingCommands.__new__(FreeroutingCommands) + board = MagicMock(name="board") + board.GetFileName.return_value = str(board_path) + board.Save.return_value = None + tracks = MagicMock() + tracks.__iter__ = lambda s: iter([]) + board.GetTracks.return_value = tracks + cc.board = board + return cc + + +@pytest.fixture() +def workdir(tmp_path): + # Pretend we have an existing board file + (tmp_path / "test.kicad_pcb").write_text("(kicad_pcb)\n") + return tmp_path + + +@pytest.fixture() +def fake_jar(tmp_path): + p = tmp_path / "freerouting.jar" + p.write_text("not a real jar") + return p + + +def _patch_exec_mode(cc): + """Force direct (non-Docker) mode so we don't go down the docker branch.""" + cc._resolve_execution_mode = MagicMock(return_value={ + "mode": "direct", "use_docker": False, + }) + + +@pytest.mark.unit +def test_single_attempt_default_keeps_legacy_response_shape(workdir, fake_jar): + """attempts=1 (or omitted): no `attempts` list in response.""" + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + ses_path = workdir / "test.ses" + + def fake_run(cmd, **kw): + # Write a one-net SES on each subprocess call + ses_path.write_text(_make_ses(num_nets=5)) + proc = types.SimpleNamespace(returncode=0, stdout="ok", stderr="") + return proc + + fake_pcb = _stub_pcbnew() + + with patch.object(fr_mod, "subprocess") as sp, \ + patch.dict(sys.modules, {"pcbnew": fake_pcb}): + sp.run.side_effect = fake_run + sp.TimeoutExpired = TimeoutError + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + }) + + assert out["success"], out + assert "attempts" not in out, "single-attempt response must not include attempts list" + assert "best_attempt" not in out + assert sp.run.call_count == 1 + + +@pytest.mark.unit +def test_best_of_three_picks_highest_scoring_ses(workdir, fake_jar): + """attempts=3 with varying SES content: best one wins.""" + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + ses_path = workdir / "test.ses" + best_path = workdir / "test_best.ses" + + # Per-attempt SES net counts: 5, 8, 7. Best should be attempt 2 (8 nets). + counts = iter([5, 8, 7]) + + def fake_run(cmd, **kw): + n = next(counts) + ses_path.write_text(_make_ses(num_nets=n)) + return types.SimpleNamespace(returncode=0, stdout=f"n={n}", stderr="") + + fake_pcb = _stub_pcbnew() + + with patch.object(fr_mod, "subprocess") as sp, \ + patch.dict(sys.modules, {"pcbnew": fake_pcb}): + sp.run.side_effect = fake_run + sp.TimeoutExpired = TimeoutError + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + "attempts": 3, + }) + + assert out["success"], out + assert "attempts" in out + assert len(out["attempts"]) == 3 + assert out["best_attempt"] == 2, f"attempt 2 has 8 nets, should win; got {out['best_attempt']}" + assert out["best_score"] == 8 * 1000 + (8 * 2) # 8 nets, 16 segments + # The final ses_path content should match the winning attempt (8 nets) + final_ses = ses_path.read_text() + assert final_ses.count("(net ") == 8 + # The _best.ses snapshot must also exist + assert best_path.exists() + + +@pytest.mark.unit +def test_one_failing_attempt_does_not_abort_best_of_n(workdir, fake_jar): + """Attempt 2 exits nonzero; best-of-N keeps going and picks among the rest.""" + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + ses_path = workdir / "test.ses" + + call_idx = {"n": 0} + + def fake_run(cmd, **kw): + call_idx["n"] += 1 + if call_idx["n"] == 2: + # Fail the middle attempt + return types.SimpleNamespace(returncode=99, stdout="", stderr="boom") + # First and third produce 5 and 9 nets respectively + nets = 5 if call_idx["n"] == 1 else 9 + ses_path.write_text(_make_ses(num_nets=nets)) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + fake_pcb = _stub_pcbnew() + + with patch.object(fr_mod, "subprocess") as sp, \ + patch.dict(sys.modules, {"pcbnew": fake_pcb}): + sp.run.side_effect = fake_run + sp.TimeoutExpired = TimeoutError + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + "attempts": 3, + }) + + assert out["success"], out + assert sp.run.call_count == 3 + assert out["best_attempt"] == 3, "attempt 3 (9 nets) must win over attempt 1 (5 nets)" + # The middle attempt is recorded but marked not-ok + middle = next(a for a in out["attempts"] if a["attempt"] == 2) + assert middle["ok"] is False + assert middle["exit_code"] == 99 + + +@pytest.mark.unit +def test_pass_schedule_wraps_when_attempts_exceeds_schedule_length(workdir, fake_jar): + """Custom 2-entry schedule + 5 attempts → schedule cycles [a, b, a, b, a].""" + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + ses_path = workdir / "test.ses" + + seen_mp = [] + + def fake_run(cmd, **kw): + # extract `-mp N` from the command + mp = int(cmd[cmd.index("-mp") + 1]) + seen_mp.append(mp) + ses_path.write_text(_make_ses(num_nets=3)) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + fake_pcb = _stub_pcbnew() + + with patch.object(fr_mod, "subprocess") as sp, \ + patch.dict(sys.modules, {"pcbnew": fake_pcb}): + sp.run.side_effect = fake_run + sp.TimeoutExpired = TimeoutError + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + "attempts": 5, + "passSchedule": [42, 99], + }) + + assert out["success"], out + assert seen_mp == [42, 99, 42, 99, 42] + + +@pytest.mark.unit +def test_target_nets_bonus_wins_against_more_nets_without_targets(workdir, fake_jar): + """Attempt with all targets beats higher-net attempt missing one target.""" + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + ses_path = workdir / "test.ses" + + # Attempt 1: 8 nets, none named "CRIT" — misses target + # Attempt 2: 4 nets including "CRIT" — hits target, gets 50k bonus + def fake_run(cmd, **kw): + attempt = sum(1 for c in cmd if c == "-mp") # rough counter — patched below + ses_path.write_text(fake_run.next_ses) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + sess = [ + _make_ses(num_nets=8), + # 4 nets but one named CRIT + '(net "N0"\n (wire (path F.Cu 200 0 0 1 1))\n)\n' + '(net "N1"\n (wire (path F.Cu 200 0 0 1 1))\n)\n' + '(net "N2"\n (wire (path F.Cu 200 0 0 1 1))\n)\n' + '(net "CRIT"\n (wire (path F.Cu 200 0 0 1 1))\n)\n', + ] + seq = iter(sess) + + def fake_run(cmd, **kw): + fake_run.next_ses = next(seq) + ses_path.write_text(fake_run.next_ses) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + fake_pcb = _stub_pcbnew() + + with patch.object(fr_mod, "subprocess") as sp, \ + patch.dict(sys.modules, {"pcbnew": fake_pcb}): + sp.run.side_effect = fake_run + sp.TimeoutExpired = TimeoutError + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + "attempts": 2, + "targetNets": ["CRIT"], + }) + + assert out["success"], out + assert out["best_attempt"] == 2, ( + "attempt 2 routed the critical target — its 50k bonus must beat " + "attempt 1's higher raw net count" + ) + + +@pytest.mark.unit +def test_invalid_attempts_rejected_cleanly(workdir, fake_jar): + cc = _make_cmds(workdir / "test.kicad_pcb") + _patch_exec_mode(cc) + out = cc.autoroute({ + "boardPath": str(workdir / "test.kicad_pcb"), + "freeroutingJar": str(fake_jar), + "attempts": 0, + }) + assert out["success"] is False + assert "Invalid attempts" in out["message"] diff --git a/tests/test_autoroute_score.py b/tests/test_autoroute_score.py new file mode 100644 index 0000000..921ff5b --- /dev/null +++ b/tests/test_autoroute_score.py @@ -0,0 +1,122 @@ +"""Tests for the best-of-N scoring in autoroute (`_score_ses`). + +Best-of-N support is ported from morningfire-pcb-automation +(https://github.com/NiNjA-CodE/morningfire-pcb-automation, +scripts/routing/freeroute_runner.py::score_ses). These tests pin the +scoring contract so future changes don't silently shift the ranking. +""" +import sys +from pathlib import Path + +import pytest + +PYTHON_DIR = Path(__file__).parent.parent / "python" +sys.path.insert(0, str(PYTHON_DIR)) + +from commands.freerouting import _score_ses # noqa: E402 + + +def _ses(nets, segments, vias=0): + """Build a minimal SES text with the given (net, segment_count) shape. + + The parser only looks at `(net NAME\n (wire` occurrences and `(wire` + / `(via ` substrings, so a synthetic fixture is enough. + """ + chunks = [] + for net, seg_count in nets: + chunks.append(f'(net "{net}"\n') + for _ in range(seg_count): + chunks.append(" (wire (path F.Cu 200 0 0 1 1))\n") + chunks.append(")\n") + # Tack on extra wires not tied to a (net ...) header — counted as + # segments but not as nets. + for _ in range(segments - sum(s for _, s in nets)): + chunks.append(" (wire (path F.Cu 200 0 0 1 1))\n") + for _ in range(vias): + chunks.append(' (via "Via[0-1]_600:300_um" 100 200)\n') + return "".join(chunks) + + +@pytest.mark.unit +def test_empty_ses_scores_zero(): + r = _score_ses("", []) + assert r["score"] == 0 + assert r["nets"] == 0 + assert r["segments"] == 0 + + +@pytest.mark.unit +def test_more_nets_always_beats_more_segments(): + """A single extra net (1000 pts) must beat any reasonable seg count delta.""" + a = _score_ses(_ses([("N1", 50), ("N2", 50), ("N3", 50)], segments=150), []) + b = _score_ses(_ses([("N1", 50), ("N2", 50)], segments=999), []) + assert a["nets"] == 3 and b["nets"] == 2 + assert a["score"] > b["score"], "+1 net (1000 pts) must dominate segment delta" + + +@pytest.mark.unit +def test_segments_break_ties_when_net_counts_equal(): + a = _score_ses(_ses([("N1", 10), ("N2", 10)], segments=20), []) + b = _score_ses(_ses([("N1", 30), ("N2", 30)], segments=60), []) + assert a["nets"] == b["nets"] == 2 + assert b["score"] > a["score"], "more segments breaks tie when net counts equal" + + +@pytest.mark.unit +def test_target_bonus_dominates_when_all_targets_routed(): + """The 50,000-point bonus must outweigh marginal nets/segments differences.""" + # `with_targets` has the targets but fewer overall nets. + with_targets = _score_ses( + _ses([("CRITICAL_A", 10), ("CRITICAL_B", 10)], segments=20), + target_nets=["CRITICAL_A", "CRITICAL_B"], + ) + # `without_targets` has more nets but is missing one target. + without_targets = _score_ses( + _ses([ + ("CRITICAL_A", 10), + ("OTHER1", 10), ("OTHER2", 10), ("OTHER3", 10), + ("OTHER4", 10), ("OTHER5", 10), ("OTHER6", 10), + ], segments=70), + target_nets=["CRITICAL_A", "CRITICAL_B"], + ) + assert without_targets["targets_missing"] == ["CRITICAL_B"] + assert with_targets["targets_missing"] == [] + assert with_targets["score"] > without_targets["score"], ( + "all-targets bonus must beat marginal net-count gain" + ) + + +@pytest.mark.unit +def test_target_bonus_inactive_when_targets_unspecified(): + """No targets configured -> score == nets*1000 + segments (no bonus).""" + r = _score_ses(_ses([("X", 5)], segments=5), []) + assert r["score"] == 1000 + 5 + assert r["targets_found"] == [] and r["targets_missing"] == [] + + +@pytest.mark.unit +def test_quoted_net_names_in_ses_are_normalised(): + """Freerouting writes nets as `(net "X"` — surrounding quotes must be stripped.""" + text = '(net "MY_NET"\n (wire (path F.Cu 200 0 0 1 1))\n)\n' + r = _score_ses(text, target_nets=["MY_NET"]) + assert r["targets_missing"] == [], "quoted net name should match unquoted target" + assert r["targets_found"] == ["MY_NET"] + + +@pytest.mark.unit +def test_targets_missing_and_found_are_sorted_for_stable_output(): + text = _ses( + [("Z_NET", 1), ("A_NET", 1), ("M_NET", 1)], + segments=3, + ) + r = _score_ses(text, target_nets=["Z_NET", "A_NET", "MISSING_X", "MISSING_A"]) + assert r["targets_found"] == ["A_NET", "Z_NET"] + assert r["targets_missing"] == ["MISSING_A", "MISSING_X"] + + +@pytest.mark.unit +def test_via_count_is_reported_independently_of_score(): + r = _score_ses(_ses([("X", 2)], segments=2, vias=7), []) + assert r["vias"] == 7 + # Score formula does NOT include vias by design. + assert r["score"] == 1 * 1000 + 2