lib.io: Implement *Buffer from RFC 55.

Author: wanda-phi

amaranth/lib/io.py

@@ -1,13 +1,18 @@
 import enum
+import operator
 from collections.abc import Iterable
 
 from ..hdl import *
-from ..lib import wiring
+from ..lib import wiring, data
 from ..lib.wiring import In, Out
 from .. import tracer
 
 
-__all__ = ["Direction", "SingleEndedPort", "DifferentialPort", "Pin"]
+__all__ = [
+    "Direction", "SingleEndedPort", "DifferentialPort",
+    "Buffer", "FFBuffer", "DDRBuffer",
+    "Pin",
+]
 
 
 class Direction(enum.Enum):
@@ -253,6 +258,362 @@ def __repr__(self):
         return f"DifferentialPort({self._p!r}, {self._n!r}, invert={invert!r}, direction={self._direction})"
 
 
+class Buffer(wiring.Component):
+    """A combinatorial I/O buffer.
+
+    Parameters
+    ----------
+    direction : Direction
+    port : SingleEndedPort, DifferentialPort, or a custom platform-specific type
+
+    Attributes
+    ----------
+    signature : Buffer.Signature
+        Created based on constructor arguments.
+    """
+    class Signature(wiring.Signature):
+        """A signature of a combinatorial I/O buffer.
+
+        Parameters
+        ----------
+        direction : Direction
+        width : int
+
+        Attributes
+        ----------
+        i: :py:`unsigned(width)` (if :py:`direction in (Direction.Input, Direction.Bidir)`)
+        o: :py:`unsigned(width)` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        oe: :py:`unsigned(1)` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        """
+        def __init__(self, direction, width):
+            self._direction = Direction(direction)
+            self._width = operator.index(width)
+            members = {}
+            if self._direction is not Direction.Output:
+                members["i"] = wiring.Out(self._width)
+            if self._direction is not Direction.Input:
+                members["o"] = wiring.In(self._width)
+                members["oe"] = wiring.In(1, init=int(self._direction is Direction.Output))
+            super().__init__(members)
+
+        @property
+        def direction(self):
+            return self._direction
+
+        @property
+        def width(self):
+            return self._width
+
+        def __eq__(self, other):
+            return type(self) == type(other) and self.direction == other.direction and self.width == other.width
+
+        def __repr__(self):
+            return f"Buffer.Signature({self.direction}, {self.width})"
+
+    def __init__(self, direction, port):
+        if not (hasattr(port, "direction") and isinstance(port.direction, Direction)):
+            raise TypeError(f"'port' must be a 'SingleEndedPort', 'DifferentialPort', "
+                            f"or a platform-specific object, not {port!r}")
+        self._port = port
+        super().__init__(Buffer.Signature(direction, len(port)))
+        if port.direction is Direction.Input and self.direction is not Direction.Input:
+            raise ValueError(f"Input port cannot be used with {self.direction.name} buffer")
+        if port.direction is Direction.Output and self.direction is not Direction.Output:
+            raise ValueError(f"Output port cannot be used with {self.direction.name} buffer")
+
+    @property
+    def port(self):
+        return self._port
+
+    @property
+    def direction(self):
+        return self.signature.direction
+
+    def elaborate(self, platform):
+        if hasattr(platform, "get_io_buffer"):
+            res = platform.get_io_buffer(self)
+            if res is not None:
+                return res
+
+        m = Module()
+
+        invert = sum(bit << idx for idx, bit in enumerate(self._port.invert))
+        if self.direction is not Direction.Input:
+            if invert:
+                o_inv = Signal.like(self.o)
+                m.d.comb += o_inv.eq(self.o ^ invert)
+            else:
+                o_inv = self.o
+        if self.direction is not Direction.Output:
+            if invert:
+                i_inv = Signal.like(self.i)
+                m.d.comb += self.i.eq(i_inv ^ invert)
+            else:
+                i_inv = self.i
+
+        if isinstance(self._port, SingleEndedPort):
+            if self.direction is Direction.Input:
+                m.submodules.iob = IOBufferInstance(self._port.io, i=i_inv)
+            elif self.direction is Direction.Output:
+                m.submodules.iob = IOBufferInstance(self._port.io, o=o_inv, oe=self.oe)
+            else:
+                m.submodules.iob = IOBufferInstance(self._port.io, o=o_inv, oe=self.oe, i=i_inv)
+        elif isinstance(self._port, DifferentialPort):
+            if self.direction is Direction.Input:
+                m.submodules.iob_p = IOBufferInstance(self._port.p, i=i_inv)
+            elif self.direction is Direction.Output:
+                m.submodules.iob_p = IOBufferInstance(self._port.p, o=o_inv, oe=self.oe)
+                m.submodules.iob_n = IOBufferInstance(self._port.n, o=~o_inv, oe=self.oe)
+            else:
+                m.submodules.iob_p = IOBufferInstance(self._port.p, o=o_inv, oe=self.oe, i=i_inv)
+                m.submodules.iob_n = IOBufferInstance(self._port.n, o=~o_inv, oe=self.oe)
+        else:
+            raise TypeError("Cannot elaborate generic 'Buffer' with port {self._port!r}")
+
+        return m
+
+
+class FFBuffer(wiring.Component):
+    """A registered I/O buffer.
+
+    Equivalent to a plain :class:`Buffer` combined with reset-less registers on ``i``, ``o``, ``oe``.
+
+    Parameters
+    ----------
+    direction : Direction
+    port : SingleEndedPort, DifferentialPort, or a custom platform-specific type
+    i_domain : str
+        Domain for input register. Only used when :py:`direction in (Direction.Input, Direction.Bidir)`.
+        Defaults to :py:`"sync"`
+    o_domain : str
+        Domain for output and output enable registers. Only used when
+        :py:`direction in (Direction.Output, Direction.Bidir)`. Defaults to :py:`"sync"`
+
+    Attributes
+    ----------
+    signature : FFBuffer.Signature
+        Created based on constructor arguments.
+    """
+    class Signature(wiring.Signature):
+        """A signature of a registered I/O buffer.
+
+        Parameters
+        ----------
+        direction : Direction
+        width : int
+
+        Attributes
+        ----------
+        i: :py:`unsigned(width)` (if :py:`direction in (Direction.Input, Direction.Bidir)`)
+        o: :py:`unsigned(width)` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        oe: :py:`unsigned(1)` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        """
+        def __init__(self, direction, width):
+            self._direction = Direction(direction)
+            self._width = operator.index(width)
+            members = {}
+            if self._direction is not Direction.Output:
+                members["i"] = wiring.Out(self._width)
+            if self._direction is not Direction.Input:
+                members["o"] = wiring.In(self._width)
+                members["oe"] = wiring.In(1, init=int(self._direction is Direction.Output))
+            super().__init__(members)
+
+        @property
+        def direction(self):
+            return self._direction
+
+        @property
+        def width(self):
+            return self._width
+
+        def __eq__(self, other):
+            return type(self) == type(other) and self.direction == other.direction and self.width == other.width
+
+        def __repr__(self):
+            return f"FFBuffer.Signature({self.direction}, {self.width})"
+
+    def __init__(self, direction, port, *, i_domain=None, o_domain=None):
+        if not (hasattr(port, "direction") and isinstance(port.direction, Direction)):
+            raise TypeError(f"'port' must be a 'SingleEndedPort', 'DifferentialPort', "
+                            f"or a platform-specific object, not {port!r}")
+        self._port = port
+        super().__init__(FFBuffer.Signature(direction, len(port)))
+        if self.signature.direction is not Direction.Output:
+            self._i_domain = i_domain or "sync"
+        elif i_domain is not None:
+            raise ValueError("Output buffer doesn't have an input domain")
+        if self.signature.direction is not Direction.Input:
+            self._o_domain = o_domain or "sync"
+        elif o_domain is not None:
+            raise ValueError("Input buffer doesn't have an output domain")
+        if port.direction is Direction.Input and self.direction is not Direction.Input:
+            raise ValueError(f"Input port cannot be used with {self.direction.name} buffer")
+        if port.direction is Direction.Output and self.direction is not Direction.Output:
+            raise ValueError(f"Output port cannot be used with {self.direction.name} buffer")
+
+    @property
+    def port(self):
+        return self._port
+
+    @property
+    def direction(self):
+        return self.signature.direction
+
+    @property
+    def i_domain(self):
+        if self.direction is Direction.Output:
+            raise AttributeError("Output buffer doesn't have an input domain")
+        return self._i_domain
+
+    @property
+    def o_domain(self):
+        if self.direction is Direction.Input:
+            raise AttributeError("Input buffer doesn't have an output domain")
+        return self._o_domain
+
+    def elaborate(self, platform):
+        if hasattr(platform, "get_io_buffer"):
+            res = platform.get_io_buffer(self)
+            if res is not None:
+                return res
+
+        m = Module()
+
+        m.submodules.iob = iob = Buffer(self.direction, self.port)
+
+        if self.direction is not Direction.Output:
+            i_tmp = Signal.like(self.i, reset_less=True)
+            m.d[self.i_domain] += i_tmp.eq(iob.i)
+            m.d.comb += self.i.eq(i_tmp)
+
+        if self.direction is not Direction.Input:
+            o_tmp = Signal.like(self.o, reset_less=True)
+            oe_tmp = Signal.like(self.oe, reset_less=True)
+            m.d[self.o_domain] += o_tmp.eq(self.o)
+            m.d[self.o_domain] += oe_tmp.eq(self.oe)
+            m.d.comb += iob.o.eq(o_tmp)
+            m.d.comb += iob.oe.eq(oe_tmp)
+
+        return m
+
+
+class DDRBuffer(wiring.Component):
+    """A double data rate registered I/O buffer.
+
+    In the input direction, the port is sampled at both edges of the input clock domain.
+    The data sampled on the active clock edge of the domain appears on ``i[0]`` with a delay
+    of 1 clock cycle. The data sampled on the opposite clock edge appears on ``i[1]`` with a delay
+    of 0.5 clock cycle. Both ``i[0]`` and ``i[1]`` thus change on the active clock edge of the domain.
+
+    In the output direction, both ``o[0]`` and ``o[1]`` are sampled on the active clock edge
+    of the domain.  The value of ``o[0]`` immediately appears on the output port.  The value
+    of ``o[1]`` then appears on the output port on the opposite edge, with a delay of 0.5 clock cycle.
+
+    The degree of support for this component varies by platform.
+
+    Parameters
+    ----------
+    direction : Direction
+    port : SingleEndedPort, DifferentialPort, or a custom platform-specific type
+    i_domain : str
+        Domain for input register. Only used when :py:`direction in (Direction.Input, Direction.Bidir)`.
+    o_domain : str
+        Domain for output and output enable registers. Only used when
+        :py:`direction in (Direction.Output, Direction.Bidir)`.
+
+    Attributes
+    ----------
+    signature : DDRBuffer.Signature
+        Created based on constructor arguments.
+    """
+    class Signature(wiring.Signature):
+        """A signature of a double data rate registered I/O buffer.
+
+        Parameters
+        ----------
+        direction : Direction
+        width : int
+
+        Attributes
+        ----------
+        i: :py:`unsigned(ArrayLayout(width, 2))` (if :py:`direction in (Direction.Input, Direction.Bidir)`)
+        o: :py:`unsigned(ArrayLayout(width, 2))` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        oe: :py:`unsigned(1)` (if :py:`direction in (Direction.Output, Direction.Bidir)`)
+        """
+        def __init__(self, direction, width):
+            self._direction = Direction(direction)
+            self._width = operator.index(width)
+            members = {}
+            if self._direction is not Direction.Output:
+                members["i"] = wiring.Out(data.ArrayLayout(self._width, 2))
+            if self._direction is not Direction.Input:
+                members["o"] = wiring.In(data.ArrayLayout(self._width, 2))
+                members["oe"] = wiring.In(1, init=int(self._direction is Direction.Output))
+            super().__init__(members)
+
+        @property
+        def direction(self):
+            return self._direction
+
+        @property
+        def width(self):
+            return self._width
+
+        def __eq__(self, other):
+            return type(self) == type(other) and self.direction == other.direction and self.width == other.width
+
+        def __repr__(self):
+            return f"DDRBuffer.Signature({self.direction}, {self.width})"
+
+    def __init__(self, direction, port, *, i_domain=None, o_domain=None):
+        if not (hasattr(port, "direction") and isinstance(port.direction, Direction)):
+            raise TypeError(f"'port' must be a 'SingleEndedPort', 'DifferentialPort', "
+                            f"or a platform-specific object, not {port!r}")
+        self._port = port
+        super().__init__(DDRBuffer.Signature(direction, len(port)))
+        if self.signature.direction is not Direction.Output:
+            self._i_domain = i_domain or "sync"
+        elif i_domain is not None:
+            raise ValueError("Output buffer doesn't have an input domain")
+        if self.signature.direction is not Direction.Input:
+            self._o_domain = o_domain or "sync"
+        elif o_domain is not None:
+            raise ValueError("Input buffer doesn't have an output domain")
+        if port.direction is Direction.Input and self.direction is not Direction.Input:
+            raise ValueError(f"Input port cannot be used with {self.direction.name} buffer")
+        if port.direction is Direction.Output and self.direction is not Direction.Output:
+            raise ValueError(f"Output port cannot be used with {self.direction.name} buffer")
+
+    @property
+    def port(self):
+        return self._port
+
+    @property
+    def direction(self):
+        return self.signature.direction
+
+    @property
+    def i_domain(self):
+        if self.direction is Direction.Output:
+            raise AttributeError("Output buffer doesn't have an input domain")
+        return self._i_domain
+
+    @property
+    def o_domain(self):
+        if self.direction is Direction.Input:
+            raise AttributeError("Input buffer doesn't have an output domain")
+        return self._o_domain
+
+    def elaborate(self, platform):
+        if hasattr(platform, "get_io_buffer"):
+            res = platform.get_io_buffer(self)
+            if res is not None:
+                return res
+
+        raise NotImplementedError("DDR buffers cannot be elaborated without a supported platform")
+
+
 class Pin(wiring.PureInterface):
     """
     An interface to an I/O buffer or a group of them that provides uniform access to input, output,

docs/changes.rst

@@ -95,6 +95,7 @@ Standard library changes
 
 * Added: :mod:`amaranth.lib.memory`. (`RFC 45`_)
 * Added: :class:`amaranth.lib.io.SingleEndedPort`, :class:`amaranth.lib.io.DifferentialPort`. (`RFC 55`_)
+* Added: :class:`amaranth.lib.io.Buffer`, :class:`amaranth.lib.io.FFBuffer`, :class:`amaranth.lib.io.DDRBuffer`. (`RFC 55`_)
 * Removed: (deprecated in 0.4) :mod:`amaranth.lib.scheduler`. (`RFC 19`_)
 * Removed: (deprecated in 0.4) :class:`amaranth.lib.fifo.FIFOInterface` with ``fwft=False``. (`RFC 20`_)
 * Removed: (deprecated in 0.4) :class:`amaranth.lib.fifo.SyncFIFO` with ``fwft=False``. (`RFC 20`_)