Fix: actually wire up ports. NB this also changes the port_map layout

Author: robtaylor

chipflow_lib/pin_lock.py

@@ -32,40 +32,42 @@ def count_member_pins(name: str, member: Dict[str, Any]) -> int:
         return member['width']
 
 
-def allocate_pins(name: str, member: Dict[str, Any], pins: List[str]) -> Tuple[Dict[str, Port], List[str]]:
+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"
 
     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 PinSignature {sig}")
-        name = name
         sig = member['annotations'][PIN_ANNOTATION_SCHEMA]
         width = sig['width']
         options = sig['options']
         pin_map[name] = {'pins': pins[0:width],
-                        'direction': sig['direction'],
-                        'type': 'io',
-                        'options': options}
+                         'direction': sig['direction'],
+                         'type': 'io',
+                         'port_name': port_name,
+                         'options': options}
         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():
-            n = '_'.join([name, k])
-            _map, pins = allocate_pins(n, v, pins)
+            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 PinSignature, pin allocation likely to be wrong")
-        name = name
         width = member['width']
         pin_map[name] = {'pins': pins[0:width],
-                        'direction': member['dir'],
-                        'type': 'io'}
+                              '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:

chipflow_lib/platforms/silicon.py

@@ -9,7 +9,7 @@
 
 from amaranth.lib import wiring, io
 from amaranth.lib.cdc import FFSynchronizer
-from amaranth.lib.wiring import Component, In
+from amaranth.lib.wiring import Component, In, PureInterface, flipped, connect
 
 from amaranth.back import rtlil
 from amaranth.hdl import Fragment
@@ -72,21 +72,29 @@ def __init__(self,
         self._invert = invert
         self._options = port.options
 
-        self._i = self._o = self._oe = Signal(1)
         if self._direction in (io.Direction.Input, io.Direction.Bidir):
             self._i = Signal(port.width, name=f"{component}_{name}__i")
         if self._direction in (io.Direction.Output, io.Direction.Bidir):
             self._o = Signal(port.width, name=f"{component}_{name}__o")
         if self._direction is io.Direction.Bidir:
             if "all_have_oe" in self._options and self._options["all_have_oe"]:
-                self._oe = Signal(port.width, name=f"{component}_{name}__oe")
+                self._oe = Signal(port.width, name=f"{component}_{name}__oe", init=-1)
             else:
-                self._oe = Signal(1, name=f"{component}_{name}__oe")
+                self._oe = Signal(1, name=f"{component}_{name}__oe", init=-1)
 
         self._pins = port.pins
         logger.debug(f"Created SiliconPlatformPort {name}, width={len(port.pins)},dir{self._direction}")
 
+    def wire(self, m: Module, interface: PureInterface):
+        assert self._direction == interface.signature.direction
+        if hasattr(interface, 'i'):
+            m.d.comb += interface.i.eq(self.i)
+        for d in ['o', 'oe']:
+            if hasattr(interface, d):
+                m.d.comb += getattr(self, d).eq(getattr(interface, d))
+
     @property
+
     def i(self):
         if self._i is None:
             raise AttributeError("SiliconPlatformPort with output direction does not have an "
@@ -229,6 +237,10 @@ def __init__(self, config):
         self._ports = {}
         self._files = {}
 
+    @property
+    def ports(self):
+        return self._ports
+
     def instantiate_ports(self, m: Module):
         if hasattr(self, "pinlock"):
             return
@@ -237,7 +249,7 @@ def instantiate_ports(self, m: Module):
         for component, iface in pinlock.port_map.items():
             for k, v in iface.items():
                 for name, port in v.items():
-                    self._ports[name] = SiliconPlatformPort(component, name, port)
+                    self._ports[port.port_name] = SiliconPlatformPort(component, name, port)
 
         for clock, name in self._config["chipflow"]["clocks"].items():
             if name not in pinlock.package.clocks:
@@ -246,6 +258,7 @@ def instantiate_ports(self, m: Module):
             port_data = pinlock.package.clocks[name]
             port = SiliconPlatformPort(component, name, port_data, invert=True)
             self._ports[name] = port
+
             if clock == 'default':
                 clock = 'sync'
             setattr(m.domains, clock, ClockDomain(name=clock))
@@ -311,13 +324,14 @@ def _prepare(self, elaboratable, name="top"):
         # Check that only a single clock domain is used.
         self._check_clock_domains(fragment)
 
-        # Prepare toplevel ports according to chipflow.toml.
+        # Prepare toplevel ports according to pinlock
         ports = []
         for port_name, port in self._ports.items():
             if port.direction in (io.Direction.Input, io.Direction.Bidir):
                 ports.append((f"io${port_name}$i", port.i, PortDirection.Input))
             if port.direction in (io.Direction.Output, io.Direction.Bidir):
                 ports.append((f"io${port_name}$o", port.o, PortDirection.Output))
+            if port.direction is io.Direction.Bidir:
                 ports.append((f"io${port_name}$oe", port.oe, PortDirection.Output))
 
         # Prepare design for RTLIL conversion.

chipflow_lib/platforms/utils.py

@@ -98,6 +98,16 @@ def __init__(self, direction: io.Direction, width: int = 1, all_have_oe: bool =
 
         super().__init__(sig)
 
+    @property
+    def direction(self):
+        return self._direction
+
+    def width(self):
+        return self._width
+
+    def options(self):
+        return self._options
+
     def annotations(self, *args):
         annotations = wiring.Signature.annotations(self, *args)
         pin_annotation = _PinAnnotation(direction=self._direction, width=self._width, options=self._options)
@@ -307,6 +317,7 @@ def sortpins(self, pins: Union[List[str], Set[str]]) -> List[str]:
 class Port(pydantic.BaseModel):
     type: str
     pins: List[str]
+    port_name: str
     direction: Optional[str] = None
     options: Optional[dict] = None
 
@@ -337,13 +348,13 @@ def check_pad(self, name: str, defn: dict):
     def add_pad(self, name: str, defn: dict):
         match defn:
             case {"type": "clock", "loc": loc}:
-                self.clocks[name] = Port(type="clock", pins=[loc], direction=io.Direction.Input)
+                self.clocks[name] = Port(type="clock", pins=[loc], direction=io.Direction.Input, port_name=name)
             case {"type": "reset", "loc": loc}:
-                self.resets[name] = Port(type="reset", pins=[loc], direction=io.Direction.Input)
+                self.resets[name] = Port(type="reset", pins=[loc], direction=io.Direction.Input, port_name=name)
             case {"type": "power", "loc": loc}:
-                self.power[name] = Port(type="power", pins=[loc])
+                self.power[name] = Port(type="power", pins=[loc], port_name=name)
             case {"type": "ground", "loc": loc}:
-                self.power[name] = Port(type="ground", pins=[loc])
+                self.power[name] = Port(type="ground", pins=[loc], port_name=name)
             case _:
                 pass
 

chipflow_lib/steps/silicon.py

@@ -12,7 +12,9 @@
 import time
 
 import dotenv
+from pprint import pprint
 from amaranth import *
+from amaranth.lib.wiring import connect, flipped
 
 from .. import ChipFlowError
 from ..platforms import SiliconPlatform, top_interfaces
@@ -43,6 +45,13 @@ def elaborate(self, platform: SiliconPlatform):
         for n, t in top.items():
             setattr(m.submodules, n, t)
 
+        for component, iface in platform.pinlock.port_map.items():
+            for iface_name, member, in iface.items():
+                for name, port in member.items():
+                    platform.ports[port.port_name].wire(
+                            m,
+                            getattr(getattr(top[component], iface_name), name)
+                            )
         return m