Add cf preview for macro scan, strategy, and remote synth

chipfoundry_cli/__init__.py

@@ -1,2 +1,2 @@
 """ChipFoundry CLI package: Automate project submission to SFTP."""
-__version__ = "2.5.4"
+__version__ = "2.6.0"

chipfoundry_cli/librelane_run.py

@@ -0,0 +1,294 @@
+"""Shared LibreLane invocation helpers for `cf harden` and `cf preview`."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional, Sequence, Tuple
+
+CONFIG_EXTS = ("json", "yaml", "yml", "tcl")
+OPENLANE_DIR_NAMES = ("openlane", "librelane")
+SYNTH_TO_STEP = "Yosys.Synthesis"
+
+
+def find_flow_root(project_root: Path) -> Optional[Path]:
+    """Return the first existing openlane/ or librelane/ directory."""
+    for name in OPENLANE_DIR_NAMES:
+        candidate = project_root / name
+        if candidate.is_dir():
+            return candidate
+    return None
+
+
+def find_config_file(macro_dir: Path) -> Optional[Path]:
+    for ext in CONFIG_EXTS:
+        path = macro_dir / f"config.{ext}"
+        if path.is_file():
+            return path
+    return None
+
+
+def list_macro_configs(project_root: Path) -> List[Dict[str, str]]:
+    """Discover LibreLane macros as ``{name, config_path, flow_root}``."""
+    found: List[Dict[str, str]] = []
+    seen = set()
+    for dir_name in OPENLANE_DIR_NAMES:
+        flow_root = project_root / dir_name
+        if not flow_root.is_dir():
+            continue
+        for child in sorted(flow_root.iterdir()):
+            if not child.is_dir() or child.name.startswith("."):
+                continue
+            config = find_config_file(child)
+            if config is None:
+                continue
+            rel = str(config.relative_to(project_root))
+            if child.name in seen:
+                continue
+            seen.add(child.name)
+            found.append(
+                {
+                    "name": child.name,
+                    "config_path": rel,
+                    "flow_root": dir_name,
+                }
+            )
+    return found
+
+
+def detect_pdk(project_root: Path, pdk_override: Optional[str] = None) -> str:
+    if pdk_override:
+        return pdk_override
+    project_json = project_root / ".cf" / "project.json"
+    if project_json.is_file():
+        data = json.loads(project_json.read_text())
+        pdk = data.get("pdk")
+        if isinstance(pdk, str) and pdk.strip():
+            return pdk.strip()
+    return "sky130A"
+
+
+def pdk_root_dir(project_root: Path) -> Path:
+    return project_root / "dependencies" / "pdks"
+
+
+def librelane_venv(flow_root: Path) -> Path:
+    return flow_root / ".venv"
+
+
+def resolve_execution_backend(
+    *,
+    openlane_version: str,
+    force_nix: bool = False,
+    force_docker: bool = False,
+) -> Tuple[str, Optional[str]]:
+    """Return ``(\"nix\"|\"docker\", error)``."""
+    if force_nix and force_docker:
+        return "", "Cannot use both --use-nix and --use-docker"
+
+    use_nix = False
+    if force_nix or not force_docker:
+        if shutil.which("nix") is not None:
+            try:
+                result = subprocess.run(
+                    [
+                        "nix",
+                        "flake",
+                        "metadata",
+                        f"github:chipfoundry/openlane-2/{openlane_version}",
+                        "--json",
+                    ],
+                    capture_output=True,
+                    timeout=5,
+                )
+                use_nix = result.returncode == 0
+            except (OSError, subprocess.TimeoutExpired):
+                use_nix = False
+        if force_nix and not use_nix:
+            return "", "Nix not available or cannot access LibreLane flake"
+
+    if not use_nix:
+        use_docker = False
+        try:
+            result = subprocess.run(
+                ["docker", "info"],
+                capture_output=True,
+                timeout=5,
+            )
+            use_docker = result.returncode == 0
+        except (OSError, subprocess.TimeoutExpired):
+            use_docker = False
+        if force_docker and not use_docker:
+            return "", "Docker not available"
+        if use_docker:
+            return "docker", None
+        if not force_nix:
+            return "", "Neither Nix nor Docker is available"
+
+    return "nix", None
+
+
+def build_librelane_command(
+    *,
+    project_root: Path,
+    flow_root: Path,
+    config_file: Path,
+    pdk: str,
+    pdk_root: Path,
+    tag: str,
+    openlane_version: str,
+    backend: str,
+    from_step: Optional[str] = None,
+    to_step: Optional[str] = None,
+    overwrite: bool = True,
+) -> Tuple[List[str], Dict[str, str]]:
+    """Build a Nix or Docker LibreLane command. Does not include GUI flows."""
+    env = os.environ.copy()
+    env.update(
+        {
+            "PROJECT_ROOT": str(project_root),
+            "PDK_ROOT": str(pdk_root),
+            "PDK": pdk,
+            "LIBRELANE_RUN_TAG": tag,
+        }
+    )
+
+    if backend == "nix":
+        cmd = [
+            "nix",
+            "run",
+            f"github:chipfoundry/openlane-2/{openlane_version}",
+            "--",
+            "--manual-pdk",
+            "--pdk-root",
+            str(pdk_root),
+            "--pdk",
+            pdk,
+            "--ef-save-views-to",
+            str(project_root),
+            "--run-tag",
+            tag,
+        ]
+    else:
+        venv_bin = librelane_venv(flow_root) / "bin"
+        env["PATH"] = f"{venv_bin}:{env.get('PATH', '')}"
+        py_major = sys.version_info.major
+        py_minor = sys.version_info.minor
+        env["PYTHONPATH"] = str(
+            librelane_venv(flow_root)
+            / "lib"
+            / f"python{py_major}.{py_minor}"
+            / "site-packages"
+        )
+        cmd = [
+            str(venv_bin / "python3"),
+            "-m",
+            "librelane",
+            "-m",
+            str(project_root),
+            "-m",
+            str(pdk_root),
+        ]
+        try:
+            if not sys.stdin.isatty():
+                cmd.append("--docker-no-tty")
+        except Exception:
+            cmd.append("--docker-no-tty")
+        cmd.append("--dockerized")
+        cmd.extend(
+            [
+                "--manual-pdk",
+                "--pdk-root",
+                str(pdk_root),
+                "--pdk",
+                pdk,
+                "--ef-save-views-to",
+                str(project_root),
+                "--run-tag",
+                tag,
+            ]
+        )
+
+    if overwrite and not from_step:
+        cmd.append("--overwrite")
+    if from_step:
+        cmd.extend(["--from", from_step])
+    if to_step:
+        cmd.extend(["--to", to_step])
+    cmd.append(str(config_file))
+    return cmd, env
+
+
+def new_run_tag() -> str:
+    return datetime.now().strftime("%y_%m_%d_%H_%M")
+
+
+def find_synth_stat_json(flow_root: Path, macro: str, run_tag: str) -> Optional[Path]:
+    """Locate Yosys stat.json written by a synth-only or full LibreLane run."""
+    run_dir = flow_root / macro / "runs" / run_tag
+    candidates = [
+        run_dir / "Yosys.Synthesis" / "reports" / "stat.json",
+        run_dir / "05-yosys-synthesis" / "reports" / "stat.json",
+        run_dir / "final" / "metrics.json",
+    ]
+    if run_dir.is_dir():
+        for path in sorted(run_dir.rglob("stat.json")):
+            candidates.append(path)
+        for path in sorted(run_dir.rglob("metrics.json")):
+            candidates.append(path)
+    for path in candidates:
+        if path.is_file():
+            return path
+    return None
+
+
+def parse_synth_metrics(stat_path: Path) -> Dict[str, object]:
+    """Parse Yosys stat.json or LibreLane metrics.json into gate/area fields.
+
+    Raises ValueError if the file cannot be parsed or has no cell count.
+    """
+    raw = json.loads(stat_path.read_text())
+    if not isinstance(raw, dict):
+        raise ValueError(f"Synth metrics at {stat_path} must be a JSON object")
+
+    gate_count: Optional[int] = None
+    area_um2: Optional[float] = None
+
+    design = raw.get("design")
+    if isinstance(design, dict):
+        cells = design.get("num_cells")
+        if cells is not None:
+            gate_count = int(cells)
+        area = design.get("area")
+        if area not in (None, 0, 0.0, "0"):
+            area_um2 = float(area)
+
+    if gate_count is None and "design__instance__count" in raw:
+        gate_count = int(float(raw["design__instance__count"]))
+    if area_um2 is None and raw.get("design__instance__area") not in (None, "", 0, 0.0):
+        area_um2 = float(raw["design__instance__area"])
+
+    if gate_count is None:
+        raise ValueError(f"No cell count in synth metrics at {stat_path}")
+
+    area_mm2 = (area_um2 / 1_000_000.0) if area_um2 is not None else None
+    return {
+        "gate_count": gate_count,
+        "estimated_area_mm2": area_mm2,
+        "metrics_path": str(stat_path),
+    }
+
+
+def run_librelane(
+    cmd: Sequence[str],
+    *,
+    cwd: Path,
+    env: Dict[str, str],
+) -> int:
+    process = subprocess.Popen(list(cmd), cwd=str(cwd), env=env)
+    return process.wait()

chipfoundry_cli/main.py

@@ -3801,16 +3801,20 @@ def _queue_and_maybe_poll_remote_job(
     poll: bool,
     wait_timeout: int,
     label: str,
-) -> None:
-    """POST a remote platform job and optionally poll until terminal status."""
+):
+    """POST a remote platform job and optionally poll until terminal status.
+
+    Returns the create-response job dict, or the last GET payload when polling
+    completes. Returns None for dry-run.
+    """
     import time
     from urllib.parse import urlencode
 
     import httpx as httpx_remote
 
     if dry_run:
         console.print(f"[cyan]Would POST[/cyan] {create_path}?" + urlencode(params))
-        return
+        return None
     if poll and wait_timeout < 0:
         console.print(
             "[red]✗[/red] --wait-timeout must be >= 0 (0 means no limit while polling)."
@@ -3858,13 +3862,13 @@ def _queue_and_maybe_poll_remote_job(
             console.print(f"[green]✓[/green] {label} completed")
             if job.get("github_pr_url"):
                 console.print(f"  Pull request: {job['github_pr_url']}")
-            return
+            return job
         if not poll:
             console.print(
                 f"[dim]Not waiting: use [bold]--remote --poll[/bold] to stream progress "
                 f"([bold]--wait-timeout 0[/bold] = no time limit while polling).[/dim]"
             )
-            return
+            return job
         deadline = None if wait_timeout == 0 else time.monotonic() + wait_timeout
         if wait_timeout == 0:
             console.print("[dim]Polling until the job completes (no timeout).[/dim]")
@@ -3928,6 +3932,7 @@ def _queue_and_maybe_poll_remote_job(
             console.print(f"[green]✓[/green] {label} completed")
             if github_pr_url:
                 console.print(f"  Pull request: {github_pr_url}")
+            return j2
         elif terminal == "failed":
             console.print(f"[red]✗[/red] {label} failed: {fail_message}")
             raise SystemExit(1)
@@ -5782,5 +5787,10 @@ def whoami_cmd():
         raise SystemExit(1)
 
 
+from chipfoundry_cli.preview import register_preview
+
+register_preview(main)
+
+
 if __name__ == "__main__":
     main()