feat: Enable configuring of package pin allocation from the package definition

Also enables grid array and quad format packages

Author: robtaylor

chipflow_lib/__init__.py

@@ -8,7 +8,10 @@
 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")
 
@@ -44,12 +47,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 +68,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/cli.py

@@ -33,14 +33,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,12 +1,11 @@
 # SPDX-License-Identifier: BSD-2-Clause
 import re
-from typing import Dict, Optional, Literal, Any
+from typing import Dict, Optional, Literal, Any, List
 
 from pydantic import BaseModel, model_validator, ValidationInfo, field_validator
 
 from .platforms.utils import Process
 
-
 class PadConfig(BaseModel):
     """Configuration for a pad in chipflow.toml."""
     type: Literal["io", "i", "o", "oe", "clock", "reset", "power", "ground"]
@@ -36,13 +35,16 @@ def validate_pad_dict(cls, v: dict, info: ValidationInfo):
         return v
 
 
+Voltage = float
+
 class SiliconConfig(BaseModel):
     """Configuration for silicon in chipflow.toml."""
-    process: Process
+    process: 'Process'
     package: Literal["caravel", "cf20", "pga144"]
-    pads: Dict[str, PadConfig] = {}
-    power: Dict[str, PadConfig] = {}
+    power: Dict[str, Voltage] = {}
     debug: Optional[Dict[str, bool]] = None
+    # This is still kept around to allow forcing pad locations.
+    pads: Optional[Dict[str, PadConfig]] = {}
 
     @field_validator('pads', 'power', mode='before')
     @classmethod
@@ -64,8 +66,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):

chipflow_lib/pin_lock.py

@@ -2,184 +2,46 @@
 import inspect
 import logging
 
-from pprint import pformat
 from pathlib import Path
-from typing import Any, List, Dict, Tuple
-
-from chipflow_lib import _parse_config, _ensure_chipflow_root, ChipFlowError
-from chipflow_lib.platforms import (
-        PACKAGE_DEFINITIONS,
-        PIN_ANNOTATION_SCHEMA,
-        top_interfaces,
-        LockFile,
-        Package,
-        PortMap,
-        Port
-)
-from chipflow_lib.config_models import Config
+from pprint import pformat
+
+from . import _parse_config, _ensure_chipflow_root, ChipFlowError
+from .platforms import top_components, LockFile, PACKAGE_DEFINITIONS
 
 # logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
 logger = logging.getLogger(__name__)
 
 
-def count_member_pins(name: str, member: Dict[str, Any]) -> int:
-    "Counts the pins from amaranth metadata"
-    logger.debug(
-        f"count_pins {name} {member['type']} "
-        f"{member['annotations'] if 'annotations' in member else 'no annotations'}"
-    )
-    if member['type'] == 'interface' and 'annotations' in member \
-       and PIN_ANNOTATION_SCHEMA in member['annotations']:
-        return member['annotations'][PIN_ANNOTATION_SCHEMA]['width']
-    elif member['type'] == 'interface':
-        width = 0
-        for n, v in member['members'].items():
-            width += count_member_pins('_'.join([name, n]), v)
-        return width
-    elif member['type'] == 'port':
-        return member['width']
-
-
-def allocate_pins(name: str, member: Dict[str, Any], pins: List[str], port_name: str = None) -> Tuple[Dict[str, Port], List[str]]:
-    "Allocate pins based of Amaranth member metadata"
-
-    if port_name is None:
-        port_name = name
-
-    pin_map = {}
-
-    logger.debug(f"allocate_pins: name={name}, pins={pins}")
-    logger.debug(f"member={pformat(member)}")
-
-    if member['type'] == 'interface' and 'annotations' in member \
-       and PIN_ANNOTATION_SCHEMA in member['annotations']:
-        logger.debug("matched IOSignature {sig}")
-        sig = member['annotations'][PIN_ANNOTATION_SCHEMA]
-        width = sig['width']
-        options = sig['options']
-        pin_map[name] = {'pins': pins[0:width],
-                         'direction': sig['direction'],
-                         'type': 'io',
-                         'port_name': port_name,
-                         'options': options}
-        if 'invert' in sig and sig['invert']:
-            pin_map[name]['invert'] = sig['invert']
-
-        logger.debug(f"added '{name}':{pin_map[name]} to pin_map")
-        return pin_map, pins[width:]
-    elif member['type'] == 'interface':
-        for k, v in member['members'].items():
-            port_name = '_'.join([name, k])
-            _map, pins = allocate_pins(k, v, pins, port_name=port_name)
-            pin_map |= _map
-            logger.debug(f"{pin_map},{_map}")
-        return pin_map, pins
-    elif member['type'] == 'port':
-        logger.warning(f"Port '{name}' has no IOSignature, pin allocation likely to be wrong")
-        width = member['width']
-        pin_map[name] = {'pins': pins[0:width],
-                              'direction': member['dir'],
-                              'type': 'io',
-                              'port_name': port_name
-                              }
-        logger.debug(f"added '{name}':{pin_map[name]} to pin_map")
-        return pin_map, pins[width:]
-    else:
-        logging.debug(f"Shouldnt get here. member = {member}")
-        assert False
-
-
 def lock_pins() -> None:
-    # Get the config as dict for backward compatibility with top_interfaces
-    config_dict = _parse_config()
+    config = _parse_config()
 
     # Parse with Pydantic for type checking and strong typing
-    config_model = Config.model_validate(config_dict)
-
-    used_pins = set()
-    oldlock = None
 
     chipflow_root = _ensure_chipflow_root()
     lockfile = Path(chipflow_root, 'pins.lock')
+    oldlock = None
+
     if lockfile.exists():
-        json_string = lockfile.read_text()
-        oldlock = LockFile.model_validate_json(json_string)
-
-    print(f"Locking pins: {'using pins.lock' if lockfile.exists() else ''}")
-
-    process = config_model.chipflow.silicon.process
-    package_name = config_model.chipflow.silicon.package
-
-    if package_name not in PACKAGE_DEFINITIONS:
-        logger.debug(f"Package '{package_name} is unknown")
-    package_type = PACKAGE_DEFINITIONS[package_name]
-
-    package = Package(package_type=package_type)
-
-    # Process pads and power configurations using Pydantic models
-    for d in ("pads", "power"):
-        logger.debug(f"Checking [chipflow.silicon.{d}]:")
-        silicon_config = getattr(config_model.chipflow.silicon, d, {})
-        for k, v in silicon_config.items():
-            pin = str(v.loc)
-            used_pins.add(pin)
-
-            # Convert Pydantic model to dict for backward compatibility
-            v_dict = {"type": v.type, "loc": v.loc}
-            port = oldlock.package.check_pad(k, v_dict) if oldlock else None
-
-            if port and port.pins != [pin]:
-                raise ChipFlowError(
-                    f"chipflow.toml conflicts with pins.lock: "
-                    f"{k} had pin {port.pins}, now {[pin]}."
-                )
-
-            # Add pad to package
-            package.add_pad(k, v_dict)
-
-    logger.debug(f'Pins in use: {package_type.sortpins(used_pins)}')
-
-    unallocated = package_type.pins - used_pins
-
-    logger.debug(f"unallocated pins = {package_type.sortpins(unallocated)}")
-
-    # Use the raw dict for top_interfaces since it expects the legacy format
-    _, interfaces = top_interfaces(config_dict)
-
-    logger.debug(f"All interfaces:\n{pformat(interfaces)}")
-
-    port_map = PortMap({})
-    # we try to keep pins together for each interface
-    for component, iface in interfaces.items():
-        for k, v in iface['interface']['members'].items():
-            logger.debug(f"Interface {component}.{k}:")
-            logger.debug(pformat(v))
-            width = count_member_pins(k, v)
-            logger.debug(f"  {k}: total {width} pins")
-            old_ports = oldlock.port_map.get_ports(component, k) if oldlock else None
-            if old_ports:
-                logger.debug(f"  {component}.{k} found in pins.lock, reusing")
-                logger.debug(pformat(old_ports))
-                old_width = sum([len(p.pins) for p in old_ports.values()])
-                if old_width != width:
-                    raise ChipFlowError(
-                        f"top level interface has changed size. "
-                        f"Old size = {old_width}, new size = {width}"
-                    )
-                port_map.add_ports(component, k, old_ports)
-            else:
-                pins = package_type.allocate(unallocated, width)
-                if len(pins) == 0:
-                    raise ChipFlowError("No pins were allocated by {package}")
-                logger.debug(f"allocated range: {pins}")
-                unallocated = unallocated - set(pins)
-                _map, _ = allocate_pins(k, v, pins)
-                port_map.add_ports(component, k, _map)
-
-    newlock = LockFile(process=process,
-                       package=package,
-                       port_map=port_map,
-                       metadata=interfaces)
+        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))
@@ -190,9 +52,10 @@ 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])
+            "lock", help=inspect.getdoc(self.lock).splitlines()[0])  # type: ignore
 
     def run_cli(self, args):
         logger.debug(f"command {args}")