Use chipflow fork of amaranth-soc as need the unmerged docs changes

.github/workflows/test-examples.yml

@@ -89,6 +89,7 @@ jobs:
       - name: Submit build ${{ env.is_dry }}
         working-directory: ${{ env.test_repo_path }}/${{ matrix.repo.design }}
         run: |
+          set -o pipefail
           pdm run chipflow silicon submit --wait $DRY | cat
         env:
           CHIPFLOW_API_KEY: ${{ secrets.CHIPFLOW_API_KEY}}

chipflow_lib/__init__.py

@@ -8,10 +8,14 @@
 import sys
 import tomli
 from pathlib import Path
-from pydantic import ValidationError
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .config_models import Config
 
 __version__ = importlib.metadata.version("chipflow_lib")
 
+
 logger = logging.getLogger(__name__)
 
 class ChipFlowError(Exception):
@@ -44,12 +48,19 @@ def _ensure_chipflow_root():
 
     if os.environ["CHIPFLOW_ROOT"] not in sys.path:
         sys.path.append(os.environ["CHIPFLOW_ROOT"])
-    _ensure_chipflow_root.root = Path(os.environ["CHIPFLOW_ROOT"]).absolute()
-    return _ensure_chipflow_root.root
+    _ensure_chipflow_root.root = Path(os.environ["CHIPFLOW_ROOT"]).absolute()  #type: ignore
+    return _ensure_chipflow_root.root  #type: ignore
+
+
+def _get_src_loc(src_loc_at=0):
+      frame = sys._getframe(1 + src_loc_at)
+      return (frame.f_code.co_filename, frame.f_lineno)
+
 
 
-def _parse_config():
+def _parse_config() -> 'Config':
     """Parse the chipflow.toml configuration file."""
+    from .config import _parse_config_file
     chipflow_root = _ensure_chipflow_root()
     config_file = Path(chipflow_root) / "chipflow.toml"
     try:
@@ -58,26 +69,3 @@ def _parse_config():
        raise ChipFlowError(f"Config file not found. I expected to find it at {config_file}")
     except tomli.TOMLDecodeError as e:
         raise ChipFlowError(f"TOML Error found when loading {config_file}: {e.msg} at line {e.lineno}, column {e.colno}")
-
-
-def _parse_config_file(config_file):
-    """Parse a specific chipflow.toml configuration file."""
-    from .config_models import Config
-
-    with open(config_file, "rb") as f:
-        config_dict = tomli.load(f)
-
-    try:
-        # Validate with Pydantic
-        Config.model_validate(config_dict)  # Just validate the config_dict
-        return config_dict  # Return the original dict for backward compatibility
-    except ValidationError as e:
-        # Format Pydantic validation errors in a user-friendly way
-        error_messages = []
-        for error in e.errors():
-            location = ".".join(str(loc) for loc in error["loc"])
-            message = error["msg"]
-            error_messages.append(f"Error at '{location}': {message}")
-
-        error_str = "\n".join(error_messages)
-        raise ChipFlowError(f"Validation error in chipflow.toml:\n{error_str}")

chipflow_lib/_appresponse.py

@@ -0,0 +1,41 @@
+# SPDX-License-Identifier: BSD-2-Clause
+
+from dataclasses import dataclass
+
+from pydantic import BaseModel, PlainSerializer, model_serializer
+
+@dataclass
+class OmitIfNone:
+    pass
+
+class AppResponseModel(BaseModel):
+    @model_serializer
+    def _serialize(self):
+        skip_if_none = set()
+        serialize_aliases = dict()
+
+        # Gather fields that should omit if None
+        for name, field_info in self.model_fields.items():
+            if any(
+                isinstance(metadata, OmitIfNone) for metadata in field_info.metadata
+            ):
+                skip_if_none.add(name)
+            elif field_info.serialization_alias:
+                serialize_aliases[name] = field_info.serialization_alias
+
+        serialized = dict()
+
+        for name, value in self:
+            # Skip serializing None if it was marked with "OmitIfNone"
+            if value is None and name in skip_if_none:
+                continue
+            serialize_key = serialize_aliases.get(name, name)
+
+            # Run Annotated PlainSerializer
+            for metadata in self.model_fields[name].metadata:
+                if isinstance(metadata, PlainSerializer):
+                    value = metadata.func(value)  # type: ignore
+
+            serialized[serialize_key] = value
+
+        return serialized

chipflow_lib/_pin_lock.py

@@ -0,0 +1,70 @@
+# SPDX-License-Identifier: BSD-2-Clause
+import inspect
+import logging
+
+from pathlib import Path
+from pprint import pformat
+
+from . import _parse_config, _ensure_chipflow_root, ChipFlowError
+from .platforms._internal import top_components, LockFile, PACKAGE_DEFINITIONS
+
+# logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+def lock_pins() -> None:
+    config = _parse_config()
+
+    # Parse with Pydantic for type checking and strong typing
+
+    chipflow_root = _ensure_chipflow_root()
+    lockfile = Path(chipflow_root, 'pins.lock')
+    oldlock = None
+
+    if lockfile.exists():
+        print("Reusing current pin allocation from `pins.lock`")
+        oldlock = LockFile.model_validate_json(lockfile.read_text())
+    logger.debug(f"Old Lock =\n{pformat(oldlock)}")
+    logger.debug(f"Locking pins: {'using pins.lock' if lockfile.exists() else ''}")
+
+    if not config.chipflow.silicon:
+        raise ChipFlowError("no [chipflow.silicon] section found in chipflow.toml")
+
+    # Get package definition from dict instead of Pydantic model
+    package_name = config.chipflow.silicon.package
+    package_def = PACKAGE_DEFINITIONS[package_name]
+    process = config.chipflow.silicon.process
+
+    top = top_components(config)
+
+    # Use the PackageDef to allocate the pins:
+    for name, component in top.items():
+        package_def.register_component(name, component)
+
+    newlock = package_def.allocate_pins(config, process, oldlock)
+
+    with open(lockfile, 'w') as f:
+        f.write(newlock.model_dump_json(indent=2, serialize_as_any=True))
+
+
+class PinCommand:
+    def __init__(self, config):
+        self.config = config
+
+    def build_cli_parser(self, parser):
+        assert inspect.getdoc(self.lock) is not None
+        action_argument = parser.add_subparsers(dest="action")
+        action_argument.add_parser(
+            "lock", help=inspect.getdoc(self.lock).splitlines()[0])  # type: ignore
+
+    def run_cli(self, args):
+        logger.debug(f"command {args}")
+        if args.action == "lock":
+            self.lock()
+
+    def lock(self):
+        """Lock the pin map for the design.
+
+        Will attempt to reuse previous pin positions.
+        """
+        lock_pins()

chipflow_lib/cli.py

@@ -1,4 +1,5 @@
 # SPDX-License-Identifier: BSD-2-Clause
+
 import argparse
 import inspect
 import sys
@@ -13,7 +14,7 @@
     _get_cls_by_reference,
     _parse_config,
 )
-from .pin_lock import PinCommand
+from ._pin_lock import PinCommand
 
 class UnexpectedError(ChipFlowError):
     pass
@@ -33,14 +34,15 @@ def run(argv=sys.argv[1:]):
     commands = {}
     commands["pin"] = PinCommand(config)
 
-    steps = DEFAULT_STEPS | config["chipflow"]["steps"]
-    for step_name, step_reference in steps.items():
-        step_cls = _get_cls_by_reference(step_reference, context=f"step `{step_name}`")
-        try:
-            commands[step_name] = step_cls(config)
-        except Exception:
-            raise ChipFlowError(f"Encountered error while initializing step `{step_name}` "
-                              f"using `{step_reference}`")
+    if config.chipflow.steps:
+        steps = DEFAULT_STEPS |config.chipflow.steps
+        for step_name, step_reference in steps.items():
+            step_cls = _get_cls_by_reference(step_reference, context=f"step `{step_name}`")
+            try:
+                commands[step_name] = step_cls(config)
+            except Exception:
+                raise ChipFlowError(f"Encountered error while initializing step `{step_name}` "
+                                f"using `{step_reference}`")
 
     parser = argparse.ArgumentParser(
         prog="chipflow",

chipflow_lib/config.py

@@ -2,9 +2,38 @@
 import os
 
 
+import tomli
+from pydantic import ValidationError
+
+from . import ChipFlowError
+from .config_models import Config
+
 def get_dir_models():
     return os.path.dirname(__file__) + "/models"
 
 
 def get_dir_software():
     return os.path.dirname(__file__) + "/software"
+
+
+def _parse_config_file(config_file) -> 'Config':
+    """Parse a specific chipflow.toml configuration file."""
+
+    with open(config_file, "rb") as f:
+        config_dict = tomli.load(f)
+
+    try:
+        # Validate with Pydantic
+        return Config.model_validate(config_dict)  # Just validate the config_dict
+    except ValidationError as e:
+        # Format Pydantic validation errors in a user-friendly way
+        error_messages = []
+        for error in e.errors():
+            location = ".".join(str(loc) for loc in error["loc"])
+            message = error["msg"]
+            error_messages.append(f"Error at '{location}': {message}")
+
+        error_str = "\n".join(error_messages)
+        raise ChipFlowError(f"Validation error in chipflow.toml:\n{error_str}")
+
+

chipflow_lib/config_models.py

@@ -1,61 +1,23 @@
 # SPDX-License-Identifier: BSD-2-Clause
-import re
-from typing import Dict, Optional, Literal, Any
+from typing import Dict, Optional, Any, List
 
-from pydantic import BaseModel, model_validator, ValidationInfo, field_validator
+from pydantic import BaseModel
 
-from .platforms.utils import Process
+from .platforms._internal import PACKAGE_DEFINITIONS, Process, Voltage
 
 
-class PadConfig(BaseModel):
-    """Configuration for a pad in chipflow.toml."""
-    type: Literal["io", "i", "o", "oe", "clock", "reset", "power", "ground"]
-    loc: str
-
-    @model_validator(mode="after")
-    def validate_loc_format(self):
-        """Validate that the location is in the correct format."""
-        if not re.match(r"^[NSWE]?[0-9]+$", self.loc):
-            raise ValueError(f"Invalid location format: {self.loc}, expected format: [NSWE]?[0-9]+")
-        return self
-
-    @classmethod
-    def validate_pad_dict(cls, v: dict, info: ValidationInfo):
-        """Custom validation for pad dicts from TOML that may not have all fields."""
-        if isinstance(v, dict):
-            # Handle legacy format - if 'type' is missing but should be inferred from context
-            if 'loc' in v and 'type' not in v:
-                if info.field_name == 'power':
-                    v['type'] = 'power'
-
-            # Map legacy 'clk' type to 'clock' to match our enum
-            if 'type' in v and v['type'] == 'clk':
-                v['type'] = 'clock'
-
-            return v
-        return v
+def known_package(package: str):
+    if package not in PACKAGE_DEFINITIONS.keys():
+        raise ValueError(f"{package} is not a valid package type. Valid package types are {PACKAGE_DEFINITIONS.keys()}")
 
 
 class SiliconConfig(BaseModel):
     """Configuration for silicon in chipflow.toml."""
-    process: Process
-    package: Literal["caravel", "cf20", "pga144"]
-    pads: Dict[str, PadConfig] = {}
-    power: Dict[str, PadConfig] = {}
+    process: 'Process'
+    package: str
+    power: Dict[str, Voltage] = {}
     debug: Optional[Dict[str, bool]] = None
-
-    @field_validator('pads', 'power', mode='before')
-    @classmethod
-    def validate_pad_dicts(cls, v, info: ValidationInfo):
-        """Pre-process pad dictionaries to handle legacy format."""
-        if isinstance(v, dict):
-            result = {}
-            for key, pad_dict in v.items():
-                # Apply the pad validator with context about which field we're in
-                validated_pad = PadConfig.validate_pad_dict(pad_dict, info)
-                result[key] = validated_pad
-            return result
-        return v
+    # This is still kept around to allow forcing pad locations.
 
 
 class ChipFlowConfig(BaseModel):
@@ -64,8 +26,7 @@ class ChipFlowConfig(BaseModel):
     top: Dict[str, Any] = {}
     steps: Optional[Dict[str, str]] = None
     silicon: Optional[SiliconConfig] = None
-    clocks: Optional[Dict[str, str]] = None
-    resets: Optional[Dict[str, str]] = None
+    clock_domains: Optional[List[str]] = None
 
 
 class Config(BaseModel):