feat(mpw): consume native RIBP generator

scripts/generate_mpw.py

@@ -1,240 +1,86 @@
 #!/usr/bin/env python3
+"""Generate native RIBP MPW sources from the locked mini-ver-mpw manifest."""
 
 from __future__ import annotations
 
 import argparse
-import errno
 import fcntl
-import hashlib
-import os
-import re
-import shutil
+import json
 import subprocess
-import tempfile
-import time
-from contextlib import contextmanager
+import sys
 from pathlib import Path
-from typing import Iterator, TextIO
 
 
-def tree_digest(root: Path) -> str:
-    digest = hashlib.sha256()
-    if not root.exists():
-        return ""
-    for path in sorted(
-        item for item in root.rglob("*") if item.is_file() and item.name != ".stamp"
-    ):
-        digest.update(path.relative_to(root).as_posix().encode("utf-8"))
-        with path.open("rb") as stream:
-            for chunk in iter(lambda: stream.read(1024 * 1024), b""):
-                digest.update(chunk)
-    return digest.hexdigest()
-
-
-def run(root: Path, log: TextIO, script: Path, *args: str) -> None:
-    command = ["python3", str(script), *args]
-    log.write("+ " + " ".join(command) + "\n")
-    log.flush()
-    subprocess.run(
-        command,
-        cwd=root,
-        check=True,
-        stdout=log,
-        stderr=subprocess.STDOUT,
-        text=True,
-    )
-
-
-def rewrite_filelists(root: Path, replacements: tuple[tuple[Path, Path], ...]) -> None:
-    for filelist in root.rglob("*.fl"):
-        content = filelist.read_text(encoding="utf-8")
-        for old_prefix, new_prefix in replacements:
-            content = content.replace(str(old_prefix.resolve()), str(new_prefix.resolve()))
-        filelist.write_text(content, encoding="utf-8")
-
-
-def migrate_user_extension_includes(root: Path) -> list[Path]:
-    """Migrate legacy user-extension includes in isolated MPW sources."""
-    legacy_include = re.compile(r'`include\s+"mdd_config\.svh"')
-    migrated: list[Path] = []
-    for source in sorted(root.rglob("*.sv")):
-        content = source.read_text(encoding="utf-8")
-        content, replacements = legacy_include.subn('`include "user_extensions.svh"', content)
-        if replacements == 0:
-            continue
-        source.write_text(content, encoding="utf-8")
-        migrated.append(source)
-    return migrated
-
-
-def migrate_user_gpio_interfaces(root: Path) -> list[Path]:
-    """Migrate legacy MPW user-IP GPIO ports in isolated build sources."""
-    legacy_port = re.compile(r"\bgpio_if\.dut\s+gpio\b")
-    legacy_pad_control = re.compile(
-        r"^[ \t]*assign[ \t]+gpio\.gpio_(?:cs|pu|pd)[ \t]*=[^;]*;[^\n]*\n",
-        flags=re.MULTILINE,
-    )
-    migrated: list[Path] = []
-    for source in sorted(root.rglob("*.sv")):
-        content = source.read_text(encoding="utf-8")
-        if legacy_port.search(content) is None:
-            continue
-
-        content, port_count = legacy_port.subn("user_gpio_if.user_ip gpio", content)
-        if port_count != 1:
-            raise ValueError(f"expected one legacy GPIO port in {source}")
-        content = legacy_pad_control.sub("", content)
-        for legacy_name, interface_name in (
-            ("gpio.gpio_oe", "gpio.oe_o"),
-            ("gpio.gpio_out", "gpio.do_o"),
-            ("gpio.gpio_in", "gpio.di_i"),
-        ):
-            content = content.replace(legacy_name, interface_name)
-        if "gpio.gpio_" in content:
-            raise ValueError(f"unsupported legacy GPIO signal in {source}")
-        source.write_text(content, encoding="utf-8")
-        migrated.append(source)
-    return migrated
-
-
-def migrate_user_core_ribp_interfaces(root: Path) -> list[Path]:
-    """Migrate isolated user-core sources from the retired RIB spelling."""
-    migrated: list[Path] = []
-    for source in sorted(
-        path for path in root.rglob("*") if path.is_file() and path.suffix in {".sv", ".fl"}
-    ):
-        content = source.read_text(encoding="utf-8")
-        updated = content.replace("rib_if", "ribp_if")
-        updated = re.sub(
-            r"\bribp_if\.(master|slave)\s+rib\b", r"ribp_if.\1 ribp", updated
-        )
-        updated = re.sub(r"\brib\.", "ribp.", updated)
-        updated = re.sub(r"\.rib(\s*\()", r".ribp\1", updated)
-        updated = re.sub(r"\bahbl2rib\b", "ahbl2ribp", updated)
-        updated = re.sub(r"\brib\b", "ribp", updated)
-        if updated == content:
-            continue
-        source.write_text(updated, encoding="utf-8")
-        migrated.append(source)
-    return migrated
-
-
-def remove_legacy_picorv32_entry(filelist: Path, management_core_dir: Path) -> None:
-    """Remove the PicoRV32 line emitted by the legacy user-core generator."""
-    excluded = {
-        str((management_core_dir / "picorv32.v").resolve()),
-        str((management_core_dir / "picorv32_ver.v").resolve()),
-    }
-    content = filelist.read_text(encoding="utf-8")
-    filtered = [line for line in content.splitlines() if line.strip() not in excluded]
-    filelist.write_text("\n".join(filtered) + "\n", encoding="utf-8")
-
-
-def remove_tree(path: Path) -> None:
-    """Remove an MPW workspace, tolerating transient NFS directory entries."""
-    for attempt in range(3):
-        try:
-            shutil.rmtree(path)
-            return
-        except OSError as error:
-            if error.errno != errno.ENOTEMPTY or attempt == 2:
-                raise
-            time.sleep(0.1)
-
-
-@contextmanager
-def temporary_workspace(parent: Path, prefix: str) -> Iterator[Path]:
-    workspace = Path(tempfile.mkdtemp(prefix=prefix, dir=parent))
-    try:
-        yield workspace
-    finally:
-        if workspace.exists():
-            remove_tree(workspace)
-
-
-def prepare_legacy_mpw_workspace(root: Path, source: Path, workspace: Path) -> Path:
-    legacy_mpw = workspace / "rtl" / "mini" / "mpw"
-    legacy_mpw.mkdir(parents=True, exist_ok=True)
-    for name in ("common.py", "core.py", "info.py", "ip.py", "user_design_info.h"):
-        shutil.copy2(source / name, legacy_mpw / name)
-    for name in ("core", "ip"):
-        shutil.copytree(source / name, legacy_mpw / name)
-    (legacy_mpw / ".build").mkdir()
-
-    legacy_core = workspace / "rtl" / "mini" / "core"
-    legacy_core.mkdir(parents=True, exist_ok=True)
-    picorv32 = root / "rtl" / "managed" / "picorv32" / "rtl"
-    shutil.copy2(picorv32 / "picorv32.v", legacy_core / "picorv32.v")
-    shutil.copy2(picorv32 / "picorv32_ver.v", legacy_core / "picorv32_ver.v")
-    return legacy_mpw
-
-
-def generate(args: argparse.Namespace) -> None:
-    root = args.root.resolve()
+def generate(root: Path, output: Path) -> None:
     managed_mpw = root / "rtl/managed/mpw"
-    output = args.output.resolve()
-    log_path = output.parent / f"{output.name}-generation.log"
-
-    output.parent.mkdir(parents=True, exist_ok=True)
-    with (
-        log_path.open("w", encoding="utf-8") as log,
-        temporary_workspace(output.parent, f".{output.name}.") as workspace,
-    ):
-        log.write(f"simulator={args.simu} user-extensions=enabled\n")
-        candidate = workspace / "output"
-        candidate.mkdir()
-        mpw = prepare_legacy_mpw_workspace(root, managed_mpw, workspace)
-        shared = mpw / ".build"
-        (shared / "user_design_info.h").unlink(missing_ok=True)
-
-        run(workspace, log, mpw / "ip.py")
-        run(workspace, log, mpw / "info.py", "IP")
-        shutil.copytree(shared / "ip", candidate / "ip")
-        migrated_includes = migrate_user_extension_includes(candidate / "ip")
-        migrated_gpio = migrate_user_gpio_interfaces(candidate / "ip")
-        log.write(f"migrated user-extension includes: {len(migrated_includes)}\n")
-        log.write(f"migrated user GPIO interfaces: {len(migrated_gpio)}\n")
-        run(workspace, log, mpw / "core.py", args.simu)
-        run(workspace, log, mpw / "info.py", "CORE")
-        shutil.copytree(shared / "core", candidate / "core")
-        migrated_core_ribp = migrate_user_core_ribp_interfaces(candidate / "core")
-        log.write(f"migrated user-core RIBP interfaces: {len(migrated_core_ribp)}\n")
-        info = shared / "user_design_info.h"
-        if info.is_file():
-            shutil.copy2(info, candidate / info.name)
-
-        rewrite_filelists(
-            candidate,
-            (
-                (shared, output),
-                (workspace / "rtl" / "mini" / "core", root / "rtl/managed/picorv32/rtl"),
-            ),
-        )
-        remove_legacy_picorv32_entry(candidate / "core/core.fl", root / "rtl/managed/picorv32/rtl")
-        if tree_digest(candidate) == tree_digest(output):
-            print(f"[mpw] generated output unchanged: {output} (log: {log_path})")
-            return
-        if output.exists():
-            shutil.rmtree(output)
-        os.replace(candidate, output)
-        print(f"[mpw] generated isolated output: {output} (log: {log_path})")
+    manifest = managed_mpw / "mpw.toml"
+    if not manifest.is_file():
+        raise FileNotFoundError(f"MPW v2 manifest is missing: {manifest}")
+    command = [
+        sys.executable,
+        "-m",
+        "mpwgen",
+        "generate",
+        "--manifest",
+        str(manifest),
+        "--output",
+        str(output),
+    ]
+    subprocess.run(command, cwd=managed_mpw, check=True)
+    validate_extension_bindings(root, output)
 
 
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Generate isolated MPW build sources")
-    parser.add_argument("--root", type=Path, required=True)
-    parser.add_argument("--output", type=Path, required=True)
-    parser.add_argument("--simu", choices=("VCS", "VERILATOR", "IVERILOG"), required=True)
-    parser.add_argument("--lock-file", type=Path, required=True)
-    return parser.parse_args()
+def validate_extension_bindings(root: Path, output: Path) -> None:
+    """Require SoC slot bindings to match the generated MPW v2 manifest."""
+    extensions_path = root / "rtl/mini/integration/user_extensions.json"
+    manifest_path = output / "manifest.json"
+    extensions = json.loads(extensions_path.read_text(encoding="utf-8"))
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    designs = manifest.get("designs")
+    if not isinstance(designs, list):
+        raise ValueError("generated MPW manifest has no design list")
+    for kind, target_key in (("core", "core_targets"), ("ip", "ip_targets")):
+        expected: dict[object, object] = {}
+        for design in designs:
+            if not isinstance(design, dict) or design.get("kind") != kind:
+                continue
+            slot = design.get("slot")
+            if slot in expected:
+                raise ValueError(f"generated MPW manifest duplicates {kind} slot {slot}")
+            expected[slot] = design
+        targets = extensions.get(target_key)
+        if not isinstance(targets, list):
+            raise ValueError(f"{extensions_path}: {target_key} must be a list")
+        actual: dict[object, object] = {}
+        for target in targets:
+            if not isinstance(target, dict):
+                raise ValueError(f"{extensions_path}: {target_key} contains a non-object entry")
+            slot = target.get("slot")
+            if slot in actual:
+                raise ValueError(f"{extensions_path}: {target_key} duplicates slot {slot}")
+            actual[slot] = target
+        if set(actual) != set(expected):
+            raise ValueError(f"{extensions_path}: {target_key} slots do not match MPW v2")
+        for slot, design in expected.items():
+            target = actual[slot]
+            if target.get("module") != design.get("top"):
+                raise ValueError(
+                    f"{extensions_path}: {kind} slot {slot} does not match {design.get('top')}"
+                )
+            if kind == "core" and target.get("reset") != design.get("reset"):
+                raise ValueError(f"{extensions_path}: core slot {slot} reset type does not match MPW v2")
 
 
 def main() -> int:
-    args = parse_args()
-    args.lock_file.parent.mkdir(parents=True, exist_ok=True)
-    with args.lock_file.open("a+", encoding="utf-8") as lock:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--lock-file", required=True, type=Path)
+    arguments = parser.parse_args()
+    arguments.lock_file.parent.mkdir(parents=True, exist_ok=True)
+    with arguments.lock_file.open("a+", encoding="utf-8") as lock:
         fcntl.flock(lock, fcntl.LOCK_EX)
-        generate(args)
+        generate(arguments.root.resolve(), arguments.output.resolve())
     return 0
 
 

scripts/prepare_mpw.py

@@ -10,6 +10,11 @@ import sys
 import tarfile
 from pathlib import Path
 
+try:
+    import tomllib
+except ModuleNotFoundError:  # Python 3.10
+    import tomli as tomllib
+
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT))
 from scripts.dependency_lock import archive, source  # noqa: E402
@@ -53,6 +58,18 @@ def extract_serv() -> Path:
     return extracted
 
 
+def manifest_design_dir(identifier: str) -> Path:
+    manifest = MPW_DIR / "mpw.toml"
+    document = tomllib.loads(manifest.read_text(encoding="utf-8"))
+    for design in document.get("design", []):
+        if design.get("id") == identifier:
+            source_dir = Path(design["source_dir"])
+            if source_dir.is_absolute() or ".." in source_dir.parts:
+                raise RuntimeError(f"unsafe source directory for MPW design {identifier}")
+            return MPW_DIR / source_dir
+    raise RuntimeError(f"MPW manifest does not define {identifier}")
+
+
 def prepare(update: bool) -> None:
     if not (MPW_DIR / ".git").is_dir():
         raise SystemExit("MPW generator is missing; run 'python3 setup.py' first")
@@ -62,11 +79,6 @@ def prepare(update: bool) -> None:
         hazard3["url"], ROOT / hazard3["destination"], hazard3["revision"],
         update=update,
     )
-    ibex = source("ibex")
-    ensure_git_repo(
-        ibex["url"], ROOT / ibex["destination"], ibex["revision"],
-        update=update,
-    )
     serv_archive = archive("serv")
     download_file(
         serv_archive["url"], ROOT / serv_archive["destination"],
@@ -74,10 +86,8 @@ def prepare(update: bool) -> None:
     )
     serv = extract_serv()
 
-    sync_tree(MPW_DIR / "Hazard3/hdl", MPW_DIR / "core/username3/Hazard3")
-    sync_tree(serv / "rtl", MPW_DIR / "core/username4/serv")
-    sync_tree(MPW_DIR / "ibex/rtl", MPW_DIR / "core/username7/ibex")
-    (MPW_DIR / ".build").mkdir(exist_ok=True)
+    sync_tree(MPW_DIR / "Hazard3/hdl", manifest_design_dir("hazard3") / "Hazard3")
+    sync_tree(serv / "rtl", manifest_design_dir("serv") / "serv")
     print("pinned MPW core sources are ready")