build.{plat,res}: post-lib.io cleanup.

Author: wanda-phi

amaranth/build/plat.py

@@ -142,34 +142,14 @@ def prepare(self, elaboratable, name="top", **kwargs):
         def missing_domain_error(name):
             raise RuntimeError("Missing domain in pin fragment")
 
-        def add_pin_fragment(pin, pin_fragment):
-            pin_fragment = Fragment.get(pin_fragment, self)
-            pin_fragment._propagate_domains(missing_domain_error)
-            pin_fragment = DomainLowerer()(pin_fragment)
-            fragment.add_subfragment(pin_fragment, name=f"pin_{pin.name}")
-
-        for pin, port, attrs, invert in self.iter_single_ended_pins():
-            if pin.dir == "i":
-                add_pin_fragment(pin, self.get_input(pin, port, attrs, invert))
-            if pin.dir == "o":
-                add_pin_fragment(pin, self.get_output(pin, port, attrs, invert))
-            if pin.dir == "oe":
-                add_pin_fragment(pin, self.get_tristate(pin, port, attrs, invert))
-            if pin.dir == "io":
-                add_pin_fragment(pin, self.get_input_output(pin, port, attrs, invert))
-
-        for pin, port, attrs, invert in self.iter_differential_pins():
-            if pin.dir == "i":
-                add_pin_fragment(pin, self.get_diff_input(pin, port, attrs, invert))
-            if pin.dir == "o":
-                add_pin_fragment(pin, self.get_diff_output(pin, port, attrs, invert))
-            if pin.dir == "oe":
-                add_pin_fragment(pin, self.get_diff_tristate(pin, port, attrs, invert))
-            if pin.dir == "io":
-                add_pin_fragment(pin, self.get_diff_input_output(pin, port, attrs, invert))
-
-        fragment = Design(fragment, [], hierarchy=(name,))
-        return self.toolchain_prepare(fragment, name, **kwargs)
+        for pin, port, buffer in self.iter_pins():
+            buffer = Fragment.get(buffer, self)
+            buffer._propagate_domains(missing_domain_error)
+            buffer = DomainLowerer()(buffer)
+            fragment.add_subfragment(buffer, name=f"pin_{pin.name}")
+
+        design = Design(fragment, list(self.iter_ports()), hierarchy=(name,))
+        return self.toolchain_prepare(design, name, **kwargs)
 
     @abstractmethod
     def toolchain_prepare(self, fragment, name, **kwargs):
@@ -186,108 +166,6 @@ def toolchain_program(self, products, name, **kwargs):
         raise NotImplementedError("Platform '{}' does not support programming"
                                   .format(type(self).__name__))
 
-    def _check_feature(self, feature, pin, attrs, valid_xdrs, valid_attrs):
-        if len(valid_xdrs) == 0:
-            raise NotImplementedError("Platform '{}' does not support {}"
-                                      .format(type(self).__name__, feature))
-        elif pin.xdr not in valid_xdrs:
-            raise NotImplementedError("Platform '{}' does not support {} for XDR {}"
-                                      .format(type(self).__name__, feature, pin.xdr))
-
-        if not valid_attrs and attrs:
-            raise NotImplementedError("Platform '{}' does not support attributes for {}"
-                                      .format(type(self).__name__, feature))
-
-    @staticmethod
-    def _invert_if(invert, value):
-        if invert:
-            return ~value
-        else:
-            return value
-
-    def get_input(self, pin, port, attrs, invert):
-        self._check_feature("input", pin, attrs,
-                            valid_xdrs=(0, 1, 2), valid_attrs=True)
-
-        m = Module()
-        if pin.xdr == 0:
-            m.submodules.buf = buf = io.Buffer(io.Direction.Input, port)
-            m.d.comb += pin.i.eq(buf.i)
-        elif pin.xdr == 1:
-            m.domains.input = cd_input = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.FFBuffer(io.Direction.Input, port, i_domain="input")
-            m.d.comb += pin.i.eq(buf.i)
-            m.d.comb += cd_input.clk.eq(pin.i_clk)
-        elif pin.xdr == 2:
-            m.domains.input = cd_input = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.DDRBuffer(io.Direction.Input, port, i_domain="input")
-            m.d.comb += pin.i0.eq(buf.i[0])
-            m.d.comb += pin.i1.eq(buf.i[1])
-            m.d.comb += cd_input.clk.eq(pin.i_clk)
-        return m
-
-    def get_output(self, pin, port, attrs, invert):
-        self._check_feature("output", pin, attrs,
-                            valid_xdrs=(0, 1, 2), valid_attrs=True)
-
-        m = Module()
-        if pin.xdr == 0:
-            m.submodules.buf = buf = io.Buffer(io.Direction.Output, port)
-            m.d.comb += buf.o.eq(pin.o)
-        elif pin.xdr == 1:
-            m.domains.output = cd_output = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.FFBuffer(io.Direction.Output, port, o_domain="output")
-            m.d.comb += buf.o.eq(pin.o)
-            m.d.comb += cd_output.clk.eq(pin.o_clk)
-        elif pin.xdr == 2:
-            m.domains.output = cd_output = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.DDRBuffer(io.Direction.Output, port, o_domain="output")
-            m.d.comb += buf.o[0].eq(pin.o0)
-            m.d.comb += buf.o[1].eq(pin.o1)
-            m.d.comb += cd_output.clk.eq(pin.o_clk)
-        if pin.dir == "oe":
-            m.d.comb += buf.oe.eq(pin.oe)
-        return m
-
-    get_tristate = get_output
-
-    def get_input_output(self, pin, port, attrs, invert):
-        self._check_feature("single-ended input/output", pin, attrs,
-                            valid_xdrs=(0, 1, 2), valid_attrs=True)
-
-        m = Module()
-        if pin.xdr == 0:
-            m.submodules.buf = buf = io.Buffer(io.Direction.Bidir, port)
-            m.d.comb += pin.i.eq(buf.i)
-            m.d.comb += buf.o.eq(pin.o)
-            m.d.comb += buf.oe.eq(pin.oe)
-        elif pin.xdr == 1:
-            m.domains.input = cd_input = ClockDomain(reset_less=True)
-            m.domains.output = cd_output = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.FFBuffer(io.Direction.Bidir, port, i_domain="input", o_domain="output")
-            m.d.comb += pin.i.eq(buf.i)
-            m.d.comb += buf.o.eq(pin.o)
-            m.d.comb += buf.oe.eq(pin.oe)
-            m.d.comb += cd_input.clk.eq(pin.i_clk)
-            m.d.comb += cd_output.clk.eq(pin.o_clk)
-        elif pin.xdr == 2:
-            m.domains.input = cd_input = ClockDomain(reset_less=True)
-            m.domains.output = cd_output = ClockDomain(reset_less=True)
-            m.submodules.buf = buf = io.DDRBuffer(io.Direction.Bidir, port, i_domain="input", o_domain="output")
-            m.d.comb += pin.i0.eq(buf.i[0])
-            m.d.comb += pin.i1.eq(buf.i[1])
-            m.d.comb += buf.o[0].eq(pin.o0)
-            m.d.comb += buf.o[1].eq(pin.o1)
-            m.d.comb += buf.oe.eq(pin.oe)
-            m.d.comb += cd_input.clk.eq(pin.i_clk)
-            m.d.comb += cd_output.clk.eq(pin.o_clk)
-        return m
-
-    get_diff_input = get_input
-    get_diff_output = get_output
-    get_diff_tristate = get_tristate
-    get_diff_input_output = get_input_output
-
 
 class TemplatedPlatform(Platform):
     toolchain         = property(abstractmethod(lambda: None))

amaranth/build/res.py

@@ -1,8 +1,8 @@
 from collections import OrderedDict
 
-from ..hdl._ast import *
-from ..lib.io import *
-from ..lib import wiring
+from ..hdl import *
+from ..hdl._ast import SignalDict
+from ..lib import wiring, io
 
 from .dsl import *
 
@@ -24,6 +24,78 @@ def __init__(self, name, attrs):
         self.attrs = attrs
 
 
+class PinBuffer(Elaboratable):
+    def __init__(self, pin, port):
+        if pin.xdr not in (0, 1, 2):
+            raise ValueError(f"Unsupported 'xdr' value {pin.xdr}")
+        self.pin = pin
+        self.port = port
+
+    def elaborate(self, platform):
+        m = Module()
+
+        if self.pin.dir == "i":
+            if self.pin.xdr == 0:
+                m.submodules.buf = buf = io.Buffer(io.Direction.Input, self.port)
+                m.d.comb += self.pin.i.eq(buf.i)
+            elif self.pin.xdr == 1:
+                m.domains.input = cd_input = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.FFBuffer(io.Direction.Input, self.port, i_domain="input")
+                m.d.comb += self.pin.i.eq(buf.i)
+                m.d.comb += cd_input.clk.eq(self.pin.i_clk)
+            elif self.pin.xdr == 2:
+                m.domains.input = cd_input = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.DDRBuffer(io.Direction.Input, self.port, i_domain="input")
+                m.d.comb += self.pin.i0.eq(buf.i[0])
+                m.d.comb += self.pin.i1.eq(buf.i[1])
+                m.d.comb += cd_input.clk.eq(self.pin.i_clk)
+        if self.pin.dir in ("o", "oe"):
+            if self.pin.xdr == 0:
+                m.submodules.buf = buf = io.Buffer(io.Direction.Output, self.port)
+                m.d.comb += buf.o.eq(self.pin.o)
+            elif self.pin.xdr == 1:
+                m.domains.output = cd_output = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.FFBuffer(io.Direction.Output, self.port, o_domain="output")
+                m.d.comb += buf.o.eq(self.pin.o)
+                m.d.comb += cd_output.clk.eq(self.pin.o_clk)
+            elif self.pin.xdr == 2:
+                m.domains.output = cd_output = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.DDRBuffer(io.Direction.Output, self.port, o_domain="output")
+                m.d.comb += buf.o[0].eq(self.pin.o0)
+                m.d.comb += buf.o[1].eq(self.pin.o1)
+                m.d.comb += cd_output.clk.eq(self.pin.o_clk)
+            if self.pin.dir == "oe":
+                m.d.comb += buf.oe.eq(self.pin.oe)
+        if self.pin.dir == "io":
+            if self.pin.xdr == 0:
+                m.submodules.buf = buf = io.Buffer(io.Direction.Bidir, self.port)
+                m.d.comb += self.pin.i.eq(buf.i)
+                m.d.comb += buf.o.eq(self.pin.o)
+                m.d.comb += buf.oe.eq(self.pin.oe)
+            elif self.pin.xdr == 1:
+                m.domains.input = cd_input = ClockDomain(reset_less=True)
+                m.domains.output = cd_output = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.FFBuffer(io.Direction.Bidir, self.port, i_domain="input", o_domain="output")
+                m.d.comb += self.pin.i.eq(buf.i)
+                m.d.comb += buf.o.eq(self.pin.o)
+                m.d.comb += buf.oe.eq(self.pin.oe)
+                m.d.comb += cd_input.clk.eq(self.pin.i_clk)
+                m.d.comb += cd_output.clk.eq(self.pin.o_clk)
+            elif self.pin.xdr == 2:
+                m.domains.input = cd_input = ClockDomain(reset_less=True)
+                m.domains.output = cd_output = ClockDomain(reset_less=True)
+                m.submodules.buf = buf = io.DDRBuffer(io.Direction.Bidir, self.port, i_domain="input", o_domain="output")
+                m.d.comb += self.pin.i0.eq(buf.i[0])
+                m.d.comb += self.pin.i1.eq(buf.i[1])
+                m.d.comb += buf.o[0].eq(self.pin.o0)
+                m.d.comb += buf.o[1].eq(self.pin.o1)
+                m.d.comb += buf.oe.eq(self.pin.oe)
+                m.d.comb += cd_input.clk.eq(self.pin.i_clk)
+                m.d.comb += cd_output.clk.eq(self.pin.o_clk)
+
+        return m
+
+
 class ResourceManager:
     def __init__(self, resources, connectors):
         self.resources  = OrderedDict()
@@ -33,8 +105,11 @@ def __init__(self, resources, connectors):
         self.connectors = OrderedDict()
         self._conn_pins = OrderedDict()
 
-        # Constraint lists
+        # List of all IOPort instances created
         self._ports     = []
+        # List of (pin, port, buffer) pairs for non-dir="-" requests.
+        self._pins      = []
+        # Constraint list
         self._clocks    = SignalDict()
 
         self.add_resources(resources)
@@ -139,11 +214,12 @@ def resolve(resource, dir, xdr, path, attrs):
                     direction = phys.dir
                 if isinstance(phys, Pins):
                     phys_names = phys.map_names(self._conn_pins, resource)
-                    io = IOPort(len(phys), name="__".join(path) + "__io", metadata=[
+                    iop = IOPort(len(phys), name="__".join(path) + "__io", metadata=[
                         PortMetadata(name, attrs)
                         for name in phys_names
                     ])
-                    port = SingleEndedPort(io, invert=phys.invert, direction=direction)
+                    self._ports.append(iop)
+                    port = io.SingleEndedPort(iop, invert=phys.invert, direction=direction)
                 if isinstance(phys, DiffPairs):
                     phys_names_p = phys.p.map_names(self._conn_pins, resource)
                     phys_names_n = phys.n.map_names(self._conn_pins, resource)
@@ -156,11 +232,8 @@ def resolve(resource, dir, xdr, path, attrs):
                         PortMetadata(name, attrs)
                         for name in phys_names_n
                     ])
-                    port = DifferentialPort(p, n, invert=phys.invert, direction=direction)
-                if dir == "-":
-                    pin = None
-                else:
-                    pin = wiring.flipped(Pin(len(phys), dir, xdr=xdr, path=path))
+                    self._ports += [p, n]
+                    port = io.DifferentialPort(p, n, invert=phys.invert, direction=direction)
 
                 for phys_name in phys_names:
                     if phys_name in self._phys_reqd:
@@ -171,12 +244,16 @@ def resolve(resource, dir, xdr, path, attrs):
                                                     ".".join(self._phys_reqd[phys_name])))
                     self._phys_reqd[phys_name] = path
 
-                self._ports.append((resource, pin, port, attrs))
-
-                if pin is not None and resource.clock is not None:
-                    self.add_clock_constraint(pin.i, resource.clock.frequency)
+                if dir == "-":
+                    return port
+                else:
+                    pin = wiring.flipped(io.Pin(len(phys), dir, xdr=xdr, path=path))
+                    buffer = PinBuffer(pin, port)
+                    self._pins.append((pin, port, buffer))
 
-                return pin if pin is not None else port
+                    if resource.clock is not None:
+                        self.add_clock_constraint(pin.i, resource.clock.frequency)
+                    return pin
 
             else:
                 assert False # :nocov:
@@ -188,56 +265,19 @@ def resolve(resource, dir, xdr, path, attrs):
         self._requested[resource.name, resource.number] = value
         return value
 
-    def iter_single_ended_pins(self):
-        for res, pin, port, attrs in self._ports:
-            if pin is None:
-                continue
-            if isinstance(res.ios[0], Pins):
-                yield pin, port, attrs, res.ios[0].invert
-
-    def iter_differential_pins(self):
-        for res, pin, port, attrs in self._ports:
-            if pin is None:
-                continue
-            if isinstance(res.ios[0], DiffPairs):
-                yield pin, port, attrs, res.ios[0].invert
-
-    def should_skip_port_component(self, port, attrs, component):
-        return False
+    def iter_pins(self):
+        yield from self._pins
 
     def iter_ports(self):
-        for res, pin, port, attrs in self._ports:
-            if isinstance(res.ios[0], Pins):
-                if not self.should_skip_port_component(port, attrs, "io"):
-                    yield port.io
-            elif isinstance(res.ios[0], DiffPairs):
-                if not self.should_skip_port_component(port, attrs, "p"):
-                    yield port.p
-                if not self.should_skip_port_component(port, attrs, "n"):
-                    yield port.n
-            else:
-                assert False
-
-    def iter_port_constraints(self):
-        for res, pin, port, attrs in self._ports:
-            if isinstance(res.ios[0], Pins):
-                if not self.should_skip_port_component(port, attrs, "io"):
-                    yield port.io.name, res.ios[0].map_names(self._conn_pins, res), attrs
-            elif isinstance(res.ios[0], DiffPairs):
-                if not self.should_skip_port_component(port, attrs, "p"):
-                    yield port.p.name, res.ios[0].p.map_names(self._conn_pins, res), attrs
-                if not self.should_skip_port_component(port, attrs, "n"):
-                    yield port.n.name, res.ios[0].n.map_names(self._conn_pins, res), attrs
-            else:
-                assert False
+        yield from self._ports
 
     def iter_port_constraints_bits(self):
-        for port_name, pin_names, attrs in self.iter_port_constraints():
-            if len(pin_names) == 1:
-                yield port_name, pin_names[0], attrs
+        for port in self._ports:
+            if len(port) == 1:
+                yield port.name, port.metadata[0].name, port.metadata[0].attrs
             else:
-                for bit, pin_name in enumerate(pin_names):
-                    yield f"{port_name}[{bit}]", pin_name, attrs
+                for bit, meta in enumerate(port.metadata):
+                    yield f"{port.name}[{bit}]", meta.name, meta.attrs
 
     def add_clock_constraint(self, clock, frequency):
         if isinstance(clock, ClockSignal):
@@ -267,11 +307,11 @@ def iter_clock_constraints(self):
         # Constraints on nets with no corresponding input pin (e.g. PLL or SERDES outputs) are not
         # affected.
         pin_i_to_port = SignalDict()
-        for res, pin, port, attrs in self._ports:
+        for pin, port, _fragment in self._pins:
             if hasattr(pin, "i"):
-                if isinstance(res.ios[0], Pins):
+                if isinstance(port, io.SingleEndedPort):
                     pin_i_to_port[pin.i] = port.io
-                elif isinstance(res.ios[0], DiffPairs):
+                elif isinstance(port, io.DifferentialPort):
                     pin_i_to_port[pin.i] = port.p
                 else:
                     assert False