wip

Author: robtaylor

chipflow_lib/platforms/_utils.py

@@ -65,7 +65,7 @@ class PowerDomain:
 ComponentName = str
 InterfaceName = str
 PowerDomainName = str
-
+InterfacePowerDomainName = str
 
 @dataclass
 class PowerConfig:
@@ -100,7 +100,7 @@ class PowerConfig:
 class IOModelOptions(TypedDict):
     invert: NotRequired[bool|Tuple[bool, ...]]
     all_have_oe: NotRequired[bool]
-    interface_power_domain: NotRequired[str]
+    interface_power_domains: NotRequired[List[str]]
     clock_domain: NotRequired[str]
     init: NotRequired[Annotated[Const, ConstSerializer, ConstSchema]]
 
@@ -120,7 +120,7 @@ class IOModel(IOModelOptions):
             the entire port. If the value is an iterable of :class:`bool`, the iterable must have the
             same length as the width of :py:`io`, and the inversion is specified for individual wires.
         allocate_power: Whether a io power domain should be set on this interface.  NB there is only one of these, so IO with multiple IO power domains must be split up.
-        interface_power_domain: the name of the power domain on the interface-side
+        interface_power_domains: the name of the available power domains on the interface
         clock_domain: the name of the `Amaranth.ClockDomain` for this port. NB there is only one of these, so IO with multiple input clocks must be split up.
         init: a :ref:`Const` value for the initial values of the port
     """
@@ -172,7 +172,8 @@ class Port(pydantic.BaseModel):
     pins: List[Pin] | None  # None implies must be allocated at end
     port_name: str
     iomodel: IOModel
-    port_power_domain: Optional[PowerDomainName] = None
+    power_map: Dict[InterfacePowerDomainName, PowerDomainName] = {}
+    power_domain: Optional[str] = None
 
     def model_post_init(self, __context):
         logger.debug(f"Instantiating port {self.port_name}: {self}")
@@ -199,12 +200,12 @@ def invert(self) -> Iterable[bool] | None:
             return None
 
     @property
-    def interface_power_domain(self) -> str | None:
-        if 'interface_power_domain' in self.iomodel:
-            assert isinstance(self.iomodel['interface_power_domain'], str)
-            return self.iomodel['interface_power_domain']
+    def interface_power_domains(self) -> List[str]:
+        if 'interface_power_domains' in self.iomodel:
+            assert isinstance(self.iomodel['interface_power_domains'], list)
+            return self.iomodel['interface_power_domains']
         else:
-            return None
+            return []
 
 
 def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Dict[str, Port]:
@@ -247,7 +248,7 @@ def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Dict[str,
 
 class PortMap(pydantic.BaseModel):
     ports: Dict[str, Component] = {}
-    port_power_domains: Dict[str, PowerDomain] = {}
+    port_power_domains: Dict[PowerDomainName, PowerDomain] = {}
 
     def get_ports(self, component: str, interface: str) -> Interface:
         "List the ports allocated in this PortMap for the given `Component` and `Interface`"
@@ -299,7 +300,7 @@ def populate(self, interfaces: dict, lockfile: Optional['LockFile'] = None):
                 if old_ports:
                     logger.debug(f"  {component}.{interface} found in pins.lock, reusing")
                     logger.debug(pformat(old_ports))
-                    old_width = sum([len(p.pins) for p in old_ports.values() if p.pins is not None])
+                    old_width = sum([len(p.pins) for p in old_ports.values() if p.pins is not None and p.type == 'io'])
                     if old_width != width:
                         raise ChipFlowError(
                             f"Interface {component}.{interface} has changed size. "
@@ -312,6 +313,26 @@ def populate(self, interfaces: dict, lockfile: Optional['LockFile'] = None):
                         self.ports[component] = {}
                     self.ports[component][interface] = create_ports(interface, v, interface)
 
+    def create_power_ports(self, component, interface):
+        power_map:Dict[InterfacePowerDomainName, PowerDomainName] = {}
+        for pn, p in self.ports[component][interface].items():
+            if p.power_map:
+                logger.debug(f"Allocating power pins for {component}.{interface}.{pn}")
+                power_map |= p.power_map
+        for ipd, ppd in power_map.items():
+            prefix = f"_power_{ipd}_{ppd}"
+            power_port = prefix + "_vdd"
+            power_domain = f"{component}.{interface}.{ipd}"
+            if power_port not in self.ports[component][interface]:
+                self.ports[component][interface][power_port] = Port(type='power', pins=None, port_name=power_port,
+                        power_domain=power_domain, iomodel=IOModel(width=1, direction=io.Direction.Input))
+
+            power_port = prefix + "_vss"
+            if power_port not in self.ports[component][interface]:
+                self.ports[component][interface][power_port] = Port(type='power', pins=None, port_name=power_port,
+                        power_domain=power_domain, iomodel=IOModel(width=1, direction=io.Direction.Input))
+
+
     def allocate_power(self, config: 'Config'):
         "Allocate power domains to top level ports"
         if not config.chipflow.power:
@@ -329,9 +350,9 @@ def allocate_power(self, config: 'Config'):
             return
 
         # special handling for core
-        for c, v in self.ports['_core'].items():
-            for i, p in v.items():
-                p.port_power_domain = '_core'
+        # for c, v in self.ports['_core'].items():
+        #     for i, p in v.items():
+        #         p.power= '_core'
 
         # convert nested dict structure into a mapping ic_power_domain:[component, interface, {port,} {ip power domain}]
         map_config = config.chipflow.power.map
@@ -340,19 +361,34 @@ def map_ports(c, i, *, port_power_domain, port_name=None, interface_power_domain
             # a bit verbose but easier to understand, I hope..
             if port_power_domain not in self.port_power_domains:
                 raise ChipFlowError(f"'{port_power_domain}' is not a known power domain in {c}.{i}{'.'+port_name if port_name else ''}{'.'+interface_power_domain if interface_power_domain else ''} = '{port_power_domain}')")
-            if not port_name and interface_power_domain:
-                for p in self.ports[c][i].values():
-                    if p.interface_power_domain == interface_power_domain:
+            if interface_power_domain:
+                if port_name:
+                    p = self.ports[c][i][port_name]
+                    if interface_power_domain not in p.interface_power_domains:
+                        raise ChipFlowError(f"{interface_power_domain} not found in {c}.{i}.{p} power domains ({p.interface_power_domains}")
                         p.port_power_domain = port_power_domain
-            elif not port_name:
-                for p in self.ports[c][i].values():
-                    p.port_power_domain = port_power_domain
-            elif port_power_domain and not interface_power_domain:
-                self.ports[c][i][port_name].port_power_domain = port_power_domain
+                        self.interface_power_map[interface_power_domain] = port_power_domain
+                else:
+                    for p in self.ports[c][i].values():
+                        if interface_power_domain in p.interface_power_domains:
+                            if interface_power_domain in p.power_map \
+                            and p.power_map[interface_power_domain] != port_power_domain:
+                                raise ChipFlowError(f"Clash in power domain mappings for {c}.{i}.{p}: {port_power_domain} differs from previously set {p.power_map[interface_power_domain]} for {interface_power_domain}")
+
+                            p.power_map[interface_power_domain] = port_power_domain
             else:
-                p = self.ports[c][i][port_name]
-                if p.interface_power_domain == interface_power_domain:
-                    p.port_power_domain = port_power_domain
+                if port_name:
+                    # first one considered default
+                    p = self.ports[c][i][port_name]
+                    if not p.interface_power_domains:
+                        raise ChipFlowError(f"No power domains available for {c}.{i}.{port_name}")
+                    interface_power_domain = p.interface_power_domains[0]
+                    p.power_map[interface_power_domain] = port_power_domain
+                else:
+                    for p in self.ports[c][i].values():
+                        if p.interface_power_domains:
+                            interface_power_domain = p.interface_power_domains[0]
+                            p.power_map[interface_power_domain] = port_power_domain
 
         # must be an easier way to do this...
         for c, v in map_config.items():
@@ -367,6 +403,7 @@ def map_ports(c, i, *, port_power_domain, port_name=None, interface_power_domain
                     map_ports(c, i, port_power_domain=v)
                 if isinstance(v, dict):
                     for x, v in v.items():
+                        print([p.interface_power_domains for p in self.ports[c][i].values() if p.interface_power_domains])
                         # is x a port name?
                         if x in self.ports[c][i]:
                             if isinstance(v, str):
@@ -375,46 +412,31 @@ def map_ports(c, i, *, port_power_domain, port_name=None, interface_power_domain
                                 for y, v in v.items():
                                     map_ports(c, i, port_name=x, interface_power_domain=y, port_power_domain=v)
                             else:
-                                raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c}.{i}.{x} = {v} ('{x}' is a port)")
+                                raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c}.{i}.{x} = {v} ('{v}' is not a valid power domain)")
                         # is x an interface-side power domain?
-                        elif any(x == p.interface_power_domain for p in self.ports[c][i].values()):
+                        elif any(x in p.interface_power_domains for p in self.ports[c][i].values() if p.interface_power_domains):
                             if not isinstance(v, str):
-                                raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c}.{i}.{x} = {v} ('{x}' is a power domaiwn)")
+                                raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c}.{i}.{x} = {v},  ('{x}' isnot a valid power domain)")
                             else:
                                 # map interface-side power domain x to port-side power domain v
                                 map_ports(c, i, interface_power_domain=x, port_power_domain=v)
                         else:
                             raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c}.{i}.{x} = {v} (unable to interpret '{x}')")
+
+                self.create_power_ports(c, i)
+
     def allocate_pins(self, config: 'Config', allocate: AllocateFunc, unallocated: Set[Pin]) -> None:
         logger.debug("Allocating pins")
         for c,v in self.ports.items():
             for i,v in v.items():
-                power_domains = set()
                 for n, port in v.items():
                     if not port.pins:
                         pins = allocate(unallocated, port.width)
                         if len(pins) == 0:  # TODO turn this into an exception?
                             raise ChipFlowError("No pins were allocated")
-                        logger.debug(f"allocated range: {pins}")
+                        # logger.debug(f"allocated range: {pins}")
                         unallocated = unallocated - set(pins)
                         port.pins = pins
-                    if port.interface_power_domain:
-                        power_domains.add(port.interface_power_domain)
-                for pd in power_domains:
-                    if pd == '_core':  # core power handled specially
-                        continue
-                    logger.debug(f"Allocating power pins for {c}.{i}, domain {pd}")
-                    powerpins = allocate(unallocated, 2)
-
-                    power_port = '_power_vdd_' + pd
-                    if power_port not in self.ports[c][i]:
-                        self.ports[c][i][power_port] = Port(type='vdd', pins=[powerpins[0]], port_name=power_port,
-                                iomodel=IOModel(width=1, direction=io.Direction.Input, interface_power_domain=pd))
-
-                    power_port = '_power_vss_' + pd
-                    if power_port not in self.ports[c][i]:
-                        self.ports[c][i][power_port] = Port(type='vss', pins=[powerpins[1]], port_name=power_port,
-                                iomodel=IOModel(width=1, direction=io.Direction.Input, interface_power_domain=pd))
 
 def construct_portmap(config: 'Config', interfaces: dict, lockfile: Optional['LockFile'], core: Component, allocate: AllocateFunc, unallocated: Set[Pin]) -> PortMap:
     portmap = PortMap()
@@ -459,6 +481,8 @@ class IOSignature(wiring.Signature):
     """
 
     def __init__(self, **kwargs: Unpack[IOModel]):
+        # runtime check..
+        assert set(kwargs.keys()).issubset(set(IOModel.__annotations__.keys()))
         model = IOModel(**kwargs)
         assert 'width' in model
         assert 'direction' in model
@@ -602,17 +626,17 @@ def group_consecutive_items(ordering: PinList, lst: PinList) -> OrderedDict[int,
     last = lst[0]
     current_group = [last]
 
-    logger.debug(f"_group_consecutive_items starting with {current_group}")
+    # logger.debug(f"_group_consecutive_items starting with {current_group}")
 
     for item in lst[1:]:
         idx = ordering.index(last)
         next = ordering[idx + 1] if idx < len(ordering) - 1 else None
-        logger.debug(f"inspecting {item}, index {idx}, next {next}")
+        # logger.debug(f"inspecting {item}, index {idx}, next {next}")
         if item == next:
             current_group.append(item)
-            logger.debug("found consecutive, adding to current group")
+            # logger.debug("found consecutive, adding to current group")
         else:
-            logger.debug("found nonconsecutive, creating new group")
+            # logger.debug("found nonconsecutive, creating new group")
             grouped.append(current_group)
             current_group = [item]
         last = item
@@ -755,13 +779,13 @@ def _allocate_bringup(self, config: 'Config') -> Component:
                                           pins=[self.bringup_pins.core_clock],
                                           port_name='sync-clk',
                                           iomodel=IOModel(width=1, direction=io.Direction.Input,
-                                                          clock_domain="sync", interface_power_domain='_core')
+                                                          clock_domain="sync")
                                       ),
                           'sync-rst': Port(type='reset',
                                            pins=[self.bringup_pins.core_reset],
                                            port_name='sync-rst',
                                            iomodel=IOModel(width=1, direction=io.Direction.Input,
-                                                          clock_domain="sync", interface_power_domain='_core')
+                                                          clock_domain="sync")
                                       )
                        }
         vdd_pins = []
@@ -770,14 +794,16 @@ def _allocate_bringup(self, config: 'Config') -> Component:
             vdd_pins.append(pp.ground)
             vss_pins.append(pp.power)
 
-        d |= {'vdd' : Port(type='vdd',
-                            pins=vdd_pins,
-                            port_name="vdd-core",
-                            iomodel=IOModel(width=len(vdd_pins), direction=io.Direction.Input, interface_power_domain='_core')),
-              'vss' : Port(type='vss',
+        d |= {'vdd' : Port(type='power',
+                           pins=vdd_pins,
+                           port_name="vdd-core",
+                           power_domain="_core",
+                           iomodel=IOModel(width=len(vdd_pins), direction=io.Direction.Input)),
+              'vss' : Port(type='power',
                            pins=vss_pins,
                            port_name="vss-core",
-                           iomodel=IOModel(width=len(vss_pins), direction=io.Direction.Input, interface_power_domain='_core'))
+                           power_domain="_core",
+                           iomodel=IOModel(width=len(vss_pins), direction=io.Direction.Input))
               }
 
 
@@ -787,7 +813,7 @@ def _allocate_bringup(self, config: 'Config') -> Component:
             d['heartbeat'] = Port(type='heartbeat',
                                   pins=[self.bringup_pins.core_heartbeat],
                                   port_name='heartbeat',
-                                  iomodel=IOModel(width=1, direction=io.Direction.Output, clock_domain="sync", interface_power_domain='_core')
+                                  iomodel=IOModel(width=1, direction=io.Direction.Output, clock_domain="sync")
                               )
         #TODO: JTAG
         return {'bringup_pins': d}
@@ -984,7 +1010,7 @@ def _jtag(self) -> JTAGPins:
         """
         # Default JTAG pin allocations
         # Use consecutive pins at the start of the package
-        start_pin = 2
+        start_pin = 1
         return JTAGPins(
             trst=start_pin,
             tck=start_pin + 1,