wip

Author: robtaylor

chipflow_lib/platforms/_utils.py

@@ -26,11 +26,12 @@
 from amaranth.lib.wiring import In, Out
 from pydantic import (
         ConfigDict, TypeAdapter, PlainSerializer,
-        WithJsonSchema, WrapValidator
+        WithJsonSchema, WrapValidator, Field
         )
 
 
 from .. import ChipFlowError, _ensure_chipflow_root, _get_cls_by_reference
+from .._appresponse import AppResponseModel, OmitIfNone
 
 if TYPE_CHECKING:
     from .._config_models import Config
@@ -48,17 +49,15 @@ def chipflow_schema_uri(name: str, version: int) -> str:
           ]
 
 
-@dataclass
-class VoltageRange:
-    min: Optional[Voltage] = None
-    max: Optional[Voltage] = None
-    typical: Optional[Voltage] = None
+class VoltageRange(AppResponseModel):
+    min: Annotated[Optional[Voltage], OmitIfNone()] = None
+    max: Annotated[Optional[Voltage], OmitIfNone()] = None
+    typical: Annotated[Optional[Voltage], OmitIfNone()] = None
 
 
-@dataclass
-class PowerDomain:
+class PowerDomain(AppResponseModel):
     voltage: Voltage | VoltageRange
-    type: Optional[str] = None
+    type: Annotated[Optional[str], OmitIfNone()] = None
 
 
 # TODO: validation checks
@@ -67,12 +66,18 @@ class PowerDomain:
 PowerDomainName = str
 InterfacePowerDomainName = str
 
+
+@dataclass
+class PowerConfigDomains:
+    pads: Optional[Dict[PowerDomainName, PowerDomain]] = None
+
+
 @dataclass
 class PowerConfig:
-    domains: Optional[Dict[PowerDomainName, PowerDomain]] = None
-    map: Optional[Dict[ComponentName,
-                     Dict[InterfaceName, dict | str]
-                     ]
+    domains: PowerConfigDomains
+    allocation: Optional[Dict[ComponentName ,
+                            Dict[InterfaceName, dict | str]
+                      ]
                  ] = None
 
 
@@ -100,7 +105,7 @@ class PowerConfig:
 class IOModelOptions(TypedDict):
     invert: NotRequired[bool|Tuple[bool, ...]]
     all_have_oe: NotRequired[bool]
-    interface_power_domains: NotRequired[List[str]]
+    interface_power_domains: NotRequired[List[InterfacePowerDomainName]]
     clock_domain: NotRequired[str]
     init: NotRequired[Annotated[Const, ConstSerializer, ConstSchema]]
 
@@ -128,6 +133,7 @@ class IOModel(IOModelOptions):
     width: int
     direction: Annotated[io.Direction, PlainSerializer(lambda x: x.value)]
 
+
 Pin = Tuple[Any,...] | str | int
 PinSet = Set[Pin]
 PinList = List[Pin]
@@ -167,16 +173,22 @@ def _to_set(self) -> Set[Pin]:
                self.core_jtag._to_set()
 
 
-class Port(pydantic.BaseModel):
+class Port(AppResponseModel):
     type: str
     pins: List[Pin] | None  # None implies must be allocated at end
     port_name: str
     iomodel: IOModel
-    power_map: Dict[InterfacePowerDomainName, PowerDomainName] = {}
-    power_domain: Optional[str] = None
+    power_allocation: Dict[InterfacePowerDomainName, PowerDomainName] = {}
+    power_domain: Annotated[Optional[str], OmitIfNone()] = None
 
     def model_post_init(self, __context):
         logger.debug(f"Instantiating port {self.port_name}: {self}")
+        # every interface gets a default domain for the pad
+        if 'interface_power_domain' not in self.iomodel:
+            self.iomodel['interface_power_domains'] = ['default']
+        elif 'default' not in self.iomodel['interface_power_domains']:
+            self.iomodel['interface_power_domains'] = ['default'].extend(self.iomodel['interface_power_domain'])
+
         return super().model_post_init(__context)
 
     @property
@@ -202,15 +214,19 @@ def invert(self) -> Iterable[bool] | None:
     @property
     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 []
 
 
-def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Dict[str, Port]:
+Interface = OrderedDict[str, Port]
+Component = OrderedDict[str, Interface]
+AllocateFunc = Callable[[PinSet, int], PinList]
+
+
+def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Interface:
     "Allocate pins based of Amaranth member metadata"
-    pin_map: Dict[str, Port] = {}
+    pin_map = Interface()
 
     logger.debug(f"create_ports: name={name}")
     logger.debug(f"member={pformat(member)}")
@@ -232,7 +248,7 @@ def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Dict[str,
     elif member['type'] == 'port':
         logger.warning(f"Port '{name}' has no IOSignature, pin allocation likely to be wrong")
         width = member['width']
-        model = IOModel(width=width, direction=io.Direction(member['dir']))
+        model = IOModel(width=int(width), direction=io.Direction(member['dir']))
         pin_map[name] = Port(type='io', port_name=port_name, iomodel=model, pins=None)
         logger.debug(f"added '{name}':{pin_map[name]} to pin_map")
         return pin_map
@@ -241,14 +257,9 @@ def create_ports(name: str, member: Dict[str, Any], port_name: str) -> Dict[str,
         assert False
 
 
-
-Interface = Dict[str, Port]
-Component = Dict[str, Interface]
-AllocateFunc = Callable[[PinSet, int], PinList]
-
 class PortMap(pydantic.BaseModel):
-    ports: Dict[str, Component] = {}
-    port_power_domains: Dict[PowerDomainName, PowerDomain] = {}
+    ports: OrderedDict[str, Component] = OrderedDict()
+    pad_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`"
@@ -277,15 +288,15 @@ def get_resets(self) -> List[Port]:
     def add_port(self, component: str, interface: str, port_name: str, port: Port):
         "Internally used by a `PackageDef`"
         if component not in self.ports:
-            self.ports[component] = {}
+            self.ports[component] = Component()
         if interface not in self.ports[component]:
-            self.ports[component][interface] = {}
+            self.ports[component][interface] = Interface()
         self.ports[component][interface][port_name] = port
 
     def add_ports(self, component: str, interface: str, ports: Interface):
         "Internally used by a `PackageDef`"
         if component not in self.ports:
-            self.ports[component] = {}
+            self.ports[component] = Component()
         self.ports[component][interface] = ports
 
     def populate(self, interfaces: dict, lockfile: Optional['LockFile'] = None):
@@ -310,19 +321,19 @@ def populate(self, interfaces: dict, lockfile: Optional['LockFile'] = None):
                 else:
                     print(f"Creating ports for component: {component}, interface: {interface}")
                     if component not in self.ports:
-                        self.ports[component] = {}
+                        self.ports[component] = Component()
                     self.ports[component][interface] = create_ports(interface, v, interface)
 
     def create_power_ports(self, component, interface):
-        power_map:Dict[InterfacePowerDomainName, PowerDomainName] = {}
+        power_allocation: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():
+            if p.power_allocation:
+                logger.debug(f"Creating power ports for {component}.{interface}.{pn}")
+                power_allocation |= p.power_allocation
+        for ipd, ppd in power_allocation.items():
             prefix = f"_power_{ipd}_{ppd}"
             power_port = prefix + "_vdd"
-            power_domain = f"{component}.{interface}.{ipd}"
+            power_domain = f"{ppd}"
             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))
@@ -332,70 +343,73 @@ def create_power_ports(self, 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 check_core_power(self, core_domain, _type, voltage):
+        if core_domain not in self.pad_power_domains:
+            self.pad_power_domains[core_domain] = PowerDomain(type=_type, voltage=voltage)
+        else:
+            if self.pad_power_domains[core_domain].type != _type:
+                raise ChipFlowError("Default core_domain power domain must be type 'io")
 
     def allocate_power(self, config: 'Config'):
         "Allocate power domains to top level ports"
-        if not config.chipflow.power:
-            return
-
-        # TODO: allow for shell-defined power domains
-        if not config.chipflow.power.domains:
-            return
 
         # instantiate port-side power domains
-        self.port_power_domains = config.chipflow.power.domains
+        if config.chipflow.power and config.chipflow.power.domains and config.chipflow.power.domains.pads:
+            self.pad_power_domains = config.chipflow.power.domains.pads
+        else:
+            self.pad_power_domains = {}
 
-        if not config.chipflow.power.map:
-            logger.warning("[chipflow.power.domains] is defined, but no [chipflow.power.map] found")
-            return
+        self.check_core_power('_io', 'io', 3.3)
+        self.check_core_power('_core', 'core', 1.8)
 
-        # special handling for core
-        # for c, v in self.ports['_core'].items():
-        #     for i, p in v.items():
-        #         p.power= '_core'
+        # ensure default mappings exist
+        if config.chipflow.power and config.chipflow.power.allocation:
+            allocation_config = config.chipflow.power.allocation
+        else:
+            allocation_config = {}
 
         # convert nested dict structure into a mapping ic_power_domain:[component, interface, {port,} {ip power domain}]
-        map_config = config.chipflow.power.map
-
         def map_ports(c, i, *, port_power_domain, port_name=None, interface_power_domain=None):
             # a bit verbose but easier to understand, I hope..
-            if port_power_domain not in self.port_power_domains:
+            if port_power_domain not in self.pad_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 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
-                        self.interface_power_map[interface_power_domain] = port_power_domain
+                        self.interface_power_allocation[interface_power_domain] = port_power_domain
                 else:
                     for p in self.ports[c][i].values():
+                        if p.type != io:
+                            continue
                         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}")
+                            if interface_power_domain in p.power_allocation \
+                            and p.power_allocation[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_allocation[interface_power_domain]} for {interface_power_domain}")
 
-                            p.power_map[interface_power_domain] = port_power_domain
+                            p.power_allocation[interface_power_domain] = port_power_domain
             else:
                 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
+                    p.power_allocation[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
+                            p.power_allocation[interface_power_domain] = port_power_domain
 
         # must be an easier way to do this...
-        for c, v in map_config.items():
+        for c, v in allocation_config.items():
             if c not in self.ports:
-                raise ChipFlowError(f"In [chipflow.power.map], '{c}' is not a top level component")
+                raise ChipFlowError(f"In [chipflow.power.allocation], '{c}' is not a top level component")
             if not isinstance(v, dict):
-                raise ChipFlowError(f"Malformed [chipflow.power.map] section: {c} = {v}")
+                raise ChipFlowError(f"Malformed [chipflow.power.allocation] section: {c} = {v}")
             for i, v in v.items():
                 if i not in self.ports[c]:
                     raise ChipFlowError(f"In [chipflow.silicon.power], '{i}' is not an interface of {c}")
@@ -412,31 +426,53 @@ 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} ('{v}' is not a valid power domain)")
+                                raise ChipFlowError(f"Malformed [chipflow.power.allocation] section: {c}.{i}.{x} = {v} ('{v}' is not a valid power domain)")
                         # is x an interface-side power domain?
                         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}' isnot a valid power domain)")
+                                raise ChipFlowError(f"Malformed [chipflow.power.allocation] 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}')")
+                            raise ChipFlowError(f"Malformed [chipflow.power.allocation] section: {c}.{i}.{x} = {v} (unable to interpret '{x}')")
+
+                self.create_power_ports(c, i)
+        # apply default case
+        for c, v in self.ports.items():
+            for i, v in v.items():
+                for p in v.values():
+                    if not p.power_allocation:
+                        p.power_allocation = {'default': '_io'}
+                    elif 'default' not in p.power_allocation:
+                        p.power_allocation |= {'default': '_io'}
 
                 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():
-                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}")
-                        unallocated = unallocated - set(pins)
-                        port.pins = pins
+        def flatten_dict(d, parent_key='', sep='.'):
+            items = []
+            for k, v in d.items():
+                new_key = f"{parent_key}{sep}{k}" if parent_key else k
+                if isinstance(v, dict):
+                    items.extend(flatten_dict(v, new_key, sep=sep).items())
+                else:
+                    items.append((new_key, v))
+            return dict(items)
+        # group by pad power domains
+        ports = flatten_dict(self.ports)
+
+        for pd in self.pad_power_domains:
+            for name, port in ports.items():
+                if port.pad_power_domain == pd:
+                    logger.debug(f"Allocating pins for {name}: {port}")
+                    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}")
+                    unallocated = unallocated - set(pins)
+                    port.pins = pins
 
 def construct_portmap(config: 'Config', interfaces: dict, lockfile: Optional['LockFile'], core: Component, allocate: AllocateFunc, unallocated: Set[Pin]) -> PortMap:
     portmap = PortMap()
@@ -956,9 +992,9 @@ def _allocate_pins(self, config: 'Config', process: 'Process', lockfile: LockFil
 
     def _allocate(self, available: PinSet, width: int) -> List[Pin]:
         avail_n: List[Pin] = sorted(available)
-        logger.debug(f"QuadPackageDef.allocate {width} from {len(avail_n)} remaining: {available}")
+        # logger.debug(f"QuadPackageDef.allocate {width} from {len(avail_n)} remaining: {available}")
         ret = find_contiguous_sequence(self._ordered_pins, avail_n, width)
-        logger.debug(f"QuadPackageDef.returned {ret}")
+        # logger.debug(f"QuadPackageDef.returned {ret}")
         assert len(ret) == width
         return ret