wip

Author: robtaylor

chipflow_lib/pin_lock.py

@@ -1,15 +1,13 @@
 import inspect
 import json
 import logging
-import re
-import sys
 
 from pprint import pformat
 from pathlib import Path
 
-from amaranth import Shape
-from chipflow_lib import _parse_config, _get_cls_by_reference
-from chipflow_lib.platforms import PACKAGE_DEFINITIONS, PIN_ANNOTATION_SCHEMA
+from chipflow_lib import _parse_config
+from chipflow_lib.platforms import PACKAGE_DEFINITIONS, PIN_ANNOTATION_SCHEMA, top_interfaces
+from chipflow_lib.platforms.utils import LockFile, Package, Interface, Port
 
 # logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
 logger = logging.getLogger(__name__)
@@ -30,7 +28,7 @@ def count_pins(name, member):
 
 def count_old_pins(d):
     match d:
-        case {'type':t, 'pins':p}:
+        case {'type':_, 'pins':p}:
             return len(p)
         case _:
             return sum([count_old_pins(v) for v in d.values()])
@@ -52,15 +50,14 @@ def allocate_pins(name, member, pins:list):
                          'type':'io'}
         return pin_map, pins[width:]
     elif member['type'] == 'interface':
-        logger.warning("member without PinSignature, pin allocation likely to be wrong")
         for k, v in member['members'].items():
             n = '_'.join([name,k])
             _map, pins = allocate_pins(n, v, pins)
             pin_map |= _map
             logger.debug(f"{pin_map},{_map}")
         return pin_map, pins
     elif member['type'] == 'port':
-        logger.warning(f"Component `{name}` in this design is not using PinSignature")
+        logger.warning(f"Port '{name}' has no PinSignature, pin allocation likely to be wrong")
         name = name
         width = member['width']
         direction = member['dir']
@@ -87,93 +84,79 @@ def check_pins(name, member, old_map, new_map):
 def lock_pins():
     config = _parse_config()
     used_pins = set()
-    pin_map = {}
+    newlock = LockFile()
+    oldlock = None
     lockfile = Path('pins.lock')
     if lockfile.exists():
-        with open(lockfile) as f:
-            old_lock = json.load(f)
-            old_map = old_lock['map']
-    else:
-        old_map = {}
+        json_string = lockfile.read_text()
+        oldlock = LockFile.model_validate_json(json_string)
+
     print(f"Locking pins: {'using pins.lock' if lockfile.exists() else ''}")
     package_name = config["chipflow"]["silicon"]["pad_ring"]
 
     if package_name not in PACKAGE_DEFINITIONS:
-        logger.debug(f"Package '{package} is unknown")
+        logger.debug(f"Package '{package_name} is unknown")
     package = PACKAGE_DEFINITIONS[package_name]
 
+    newlock.package.package_name = package_name
     for d in ("pads", "power"):
         logger.debug(f"Checking [chipflow.silicon.{d}]:")
+        _map = {}
         for k, v in config["chipflow"]["silicon"][d].items():
             pin = str(v['loc'])
             used_pins.add(pin)
-            if k in old_map and old_map[k]['pins'] != [pin]:
+            if d in old_map and k in old_map[d] and old_map[d][k]['pins'] != [pin]:
                 print(f"chipflow.toml conflicts with pins.lock: "
                     f"{k} had pin {old_map[k]['pins']}, now {[pin]}.")
                 exit(1)
-            pin_map[k] = {
+            _map[k] = {
                 'pins': [pin],
-                'type': v['type'] if 'type' in v else None}
+                'type': v['type'] if 'type' in v else d}
+        pin_map[d] = _map
 
 
     logger.debug(f'Pins in use: {sorted(used_pins)}')
 
     unallocated = package.pins - used_pins
 
     logger.debug(f"unallocated pins = {sorted(unallocated)}")
-    interfaces = {}
-    top_components = config["chipflow"]["top"].items()
-    component_configs = {}
-    top = {}
-
-    for name, conf in top_components:
-        if '.' in name:
-            assert conf is dict
-            logger.debug("Config found for {name}")
-            component_configs[name.split('.')[0]] = conf
-
-    for name, ref in top_components:
-        cls = _get_cls_by_reference(ref, context=f"top component: {name}")
-        if name in component_configs:
-            top[name] = cls(component_configs[name])
-        else:
-            top[name] = cls()
-        logger.debug(pformat(top[name].metadata.origin.signature.members))
-        metadata = top[name].metadata.as_json()
-        interfaces |= metadata['interface']['members']
-
-    logger.debug(f"All interfaces: {interfaces.keys()}")
-    pin_map_interfaces = {}
+      
+    _, interfaces = top_interfaces(config)
+
+    logger.debug(f"All interfaces: {[v.keys() for i in interfaces]}")
 
     # we try to keep pins together for each interface
-    for k,v in interfaces.items():
-        logger.debug(f"Interface {k}:")
-        logger.debug(pformat(v))
-        width = count_pins(k, v)
-        print(f"  {k}: total {width} pins")
-        if k in old_map:
-            print(f"  {k} found in pins.lock, reusing")
-            logger.debug(pformat(old_map[k]))
-            old_width = count_old_pins(old_map[k])
-            if old_width != width:
-                # TODO: option to allocate new pins nonconsecutively
-                print(f"top level interface has changed size. Old size = {old_width}, new size = {width}")
-                exit(1)
-            pin_map[k] = old_map[k]
-        else:
-            pins = package.allocate(unallocated, width)
-            if len(pins) == 0:
-                print("ERROR: No pins were allocation by {package}")
-                exit(1)
-            logger.debug(f"allocated range: {pins}")
-            unallocated = unallocated - set(pins)
-            _map, _ = allocate_pins(k, v, pins)
-            pin_map[k] = _map
+    for component, iface in interfaces.items():
+        pin_map[component] = {}
+        for k,v in iface.items():
+            logger.debug(f"Interface {component}.{k}:")
+            logger.debug(pformat(v))
+            width = count_pins(k, v)
+            print(f"  {k}: total {width} pins")
+            if component in old_map and k in old_map[component]:
+                print(f"  {component}.{k} found in pins.lock, reusing")
+                logger.debug(pformat(old_map[component][k]))
+                old_width = count_old_pins(old_map[k])
+                if old_width != width:
+                    # TODO: option to allocate new pins nonconsecutively
+                    print(f"top level interface has changed size. Old size = {old_width}, new size = {width}")
+                    exit(1)
+                pin_map[component][k] = old_map[component][k]
+            else:
+                pins = package.allocate(unallocated, width)
+                if len(pins) == 0:
+                    print("ERROR: No pins were allocation by {package}")
+                    exit(1)
+                logger.debug(f"allocated range: {pins}")
+                unallocated = unallocated - set(pins)
+                _map, _ = allocate_pins(k, v, pins)
+                pin_map[component][k] = _map
 
     with open('pins.lock', 'w') as f:
         newlock = {'map': pin_map,
-                   'metadata': metadata}
+                   'metadata': interfaces}
 
+        assert _LockFile.model_validate(newlock)
         json.dump(newlock, f, indent=2, sort_keys=True)
 
 class PinCommand:
@@ -182,7 +165,7 @@ def __init__(self, config):
 
     def build_cli_parser(self, parser):
         action_argument = parser.add_subparsers(dest="action")
-        prepare_subparser = action_argument.add_parser(
+        action_argument.add_parser(
             "lock", help=inspect.getdoc(self.lock).splitlines()[0])
 
     def run_cli(self, args):

chipflow_lib/platforms/__init__.py

@@ -1,186 +1,3 @@
-import logging
-
-from enum import Enum
-from dataclasses import dataclass
-from typing import Set, List, TypeVar, Dict
-
-from amaranth import Shape
-from amaranth.lib import wiring, io, meta
-from amaranth.lib.wiring import In, Out
-from pydantic import BaseModel, ConfigDict
-from pprint import pformat
-
-from .. import ChipFlowError
 from .silicon import *
 from .sim import *
-
-logger = logging.getLogger(__name__)
-
-def chipflow_schema_uri(name: str, version:int) -> str:
-    return f"https://api.chipflow.com/schemas/{version}/{name}"
-
-class PortGroup:
-    """Group of Amaranth library I/O ports.
-
-    This object is a stand-in for the object returned by the Amaranth :py:`platform.request()`
-    function, as expected by the I/O cores.
-    """
-
-    def __init__(self, **kwargs):
-        for name, port in kwargs.items():
-            setattr(self, name, port)
-
-    def __getitem__(self, key):
-        return self.__dict__[key]
-
-    def __setattr__(self, name, value):
-        if not name.startswith("_"):
-            assert value is None or isinstance(value, (io.PortLike, PortGroup))
-        object.__setattr__(self, name, value)
-
-    def __repr__(self):
-        attrs = []
-        for name, value in self.__dict__.items():
-            if not name.startswith('_'):
-                attrs.append(f"{name}={value!r}")
-        return f"{self.__class__.__name__}({', '.join(attrs)})"
-
-class PinAnnotationModel(BaseModel):
-    model_config = ConfigDict(use_enum_values=True)
-    direction: io.Direction
-    width: int
-
-    @classmethod
-    def _annotation_schema(cls):
-            schema = PinAnnotationModel.model_json_schema()
-            schema['$schema'] = "https://json-schema.org/draft/2020-12/schema"
-            schema['$id'] = chipflow_schema_uri("pin-annotation",0)
-            return schema 
-
-    def __init__(self, **kwargs):
-        kwargs['url'] = chipflow_schema_uri("pin-annotation",0)
-        super().__init__(**kwargs)
-
-class PinAnnotation(meta.Annotation):
-    schema = PinAnnotationModel._annotation_schema()
-
-    def __init__(self, **kwargs):
-        self.model = PinAnnotationModel(**kwargs)
-
-    @property
-    def origin(self): # type: ignore
-        return self.model
-
-    def as_json(self): # type: ignore
-        return self.model.model_dump()
-
-class Schemas(Enum):
-    PIN_ANNOTATION = str(chipflow_schema_uri("pin-annotation",0))
-PIN_ANNOTATION_SCHEMA = str(chipflow_schema_uri("pin-annotation",0))
-
-class PinSignature(wiring.Signature):
-    def __init__(self, direction, width=1, init=None):
-        self._direction = direction
-        self._width = width
-        self._init = init
-        match direction:
-            case io.Direction.Bidir:
-                sig = {
-                    "o": Out(width),
-                    "oe": Out(1),
-                    "i": In(width)
-                }
-            case io.Direction.Input:
-                sig = {"i": In(width)}
-            case io.Direction.Output:
-                sig = {"o": Out(width)}
-            case _:
-                assert False
-
-        super().__init__(sig)
-
-    def annotations(self, *args):
-        return wiring.Signature.annotations(self, *args) + (PinAnnotation(direction=self._direction, width=self._width),)
-
-    def __repr__(self):
-        return f"PinSignature({self._direction}, {self._width})"
-    
-def OutputPinSignature(width, **kwargs):
-    return PinSignature(io.Direction.Output, width=width, **kwargs)
-
-def InputPinSignature(width, **kwargs):
-    return PinSignature(io.Direction.Input, width=width, **kwargs)
-
-def BidirPinSignature(width, **kwargs):
-    return PinSignature(io.Direction.Bidir, width=width, **kwargs)
-
-def group_consecutive_items(lst: List[int]) -> Dict[int, List[List[int]]]:
-    if not lst:
-        return {}
-
-    grouped = []
-    current_group = [lst[0]]
-
-    for item in lst[1:]:
-        if item == current_group[-1] + 1:
-            current_group.append(item)
-        else:
-            grouped.append(current_group)
-            current_group = [item]
-
-    grouped.append(current_group)
-    d = {}
-    for g in grouped:
-        d.setdefault(len(g), []).append(g)
-    return d
-
-def find_contiguous_sequence(lst: List[int], total: int) -> List[int]:
-    """Find the next sequence of n consecutive numbers in a sorted list
-    
-    Args:
-        lst: Sorted list of numbers
-        n: Length of consecutive sequence to find
-        
-    Returns:
-        A slice indexing the first sequence of n consecutive numbers found within the given list
-        if unable to find a consecutive list, allocate as contigously as possible
-    """
-    if not lst or len(lst) < total:
-        raise ChipFlowError("Invalid request to find_contiguous_argument")
-
-    grouped = group_consecutive_items(lst)
-
-    ret = []
-    n = total
-    for k in sorted(grouped.keys(), reverse=True):
-        for g in grouped[k]:
-            assert (n+len(ret)==total)
-            if k >= n:
-                ret += g[0:min(n,k)]
-                return ret
-            else:
-                n = n - k
-                ret += g[0:k]
-
-    return ret
-
-
-@dataclass
-class PGAPackageDef:
-    size: int
-
-    @property
-    def pins(self) -> Set[str]:
-        return set([str(i) for i in range(self.size-1)])
-
-    def allocate(self, available: Set[str], width:int) -> List[str]:
-        avail_n = sorted([int(i) for i in available])
-        logger.debug(f"PGAPackageDef.allocate {width} from {len(avail_n)} remaining")
-        ret = find_contiguous_sequence(avail_n, width)
-        logger.debug(f"PGAPackageDef.returned {ret}")
-        assert(len(ret)==width)
-        return [str(i) for i in ret]
-        
-PACKAGE_DEFINITIONS = {
-    "pga144": PGAPackageDef(144)
-}
+from .utils import *