lib.io: Implement *Port from RFC 55.

Author: wanda-phi

amaranth/build/res.py

@@ -14,17 +14,6 @@ class ResourceError(Exception):
     pass
 
 
-class SingleEndedPort:
-    def __init__(self, io):
-        self.io = io
-
-
-class DifferentialPort:
-    def __init__(self, p, n):
-        self.p = p
-        self.n = n
-
-
 class PortGroup:
     pass
 
@@ -144,20 +133,16 @@ def resolve(resource, dir, xdr, path, attrs):
                 if isinstance(phys, Pins):
                     phys_names = phys.names
                     io = IOPort(len(phys), name="__".join(path) + "__io")
-                    port = SingleEndedPort(io)
+                    port = SingleEndedPort(io, invert=phys.invert, direction=phys.dir)
                 if isinstance(phys, DiffPairs):
                     phys_names = []
+                    p = IOPort(len(phys), name="__".join(path) + "__p")
+                    n = IOPort(len(phys), name="__".join(path) + "__n")
                     if not self.should_skip_port_component(None, attrs, "p"):
-                        p = IOPort(len(phys), name="__".join(path) + "__p")
                         phys_names += phys.p.names
-                    else:
-                        p = None
                     if not self.should_skip_port_component(None, attrs, "n"):
-                        n = IOPort(len(phys), name="__".join(path) + "__n")
                         phys_names += phys.n.names
-                    else:
-                        n = None
-                    port = DifferentialPort(p, n)
+                    port = DifferentialPort(p, n, invert=phys.invert, direction=phys.dir)
                 if dir == "-":
                     pin = None
                 else:

amaranth/lib/io.py

@@ -1,10 +1,127 @@
-from .. import *
+import enum
+
+from ..hdl import *
 from ..lib import wiring
 from ..lib.wiring import In, Out
 from .. import tracer
 
 
-__all__ = ["Pin"]
+__all__ = ["Direction", "SingleEndedPort", "DifferentialPort", "Pin"]
+
+
+class Direction(enum.Enum):
+    Input  = "i"
+    Output = "o"
+    Bidir  = "io"
+
+    def __or__(self, other):
+        if not isinstance(other, Direction):
+            return NotImplemented
+        if self == other:
+            return self
+        else:
+            return Direction.Bidir
+
+
+class SingleEndedPort:
+    """TODO
+    """
+    def __init__(self, io, *, invert=False, direction=Direction.Bidir):
+        self._io = IOValue.cast(io)
+        if isinstance(invert, bool):
+            self._invert = (invert,) * len(self._io)
+        else:
+            self._invert = tuple(invert)
+            if len(self._invert) != len(self._io):
+                raise ValueError(f"Length of 'invert' ({len(self._invert)}) must match "
+                                 f"length of 'io' ({len(self._io)}).")
+            if not all(isinstance(item, bool) for item in self._invert):
+                raise ValueError(f"'invert' must be a bool or iterable of bool, not {invert!r}")
+        self._direction = Direction(direction)
+
+    @property
+    def io(self):
+        return self._io
+
+    @property
+    def invert(self):
+        return self._invert
+
+    @property
+    def direction(self):
+        return self._direction
+
+    def __len__(self):
+        return len(self._io)
+
+    def __invert__(self):
+        return SingleEndedPort(self._io, invert=tuple(~inv for inv in self._invert),
+                               direction=self._direction)
+
+    def __getitem__(self, index):
+        return SingleEndedPort(self._io[index], invert=self._invert[index],
+                               direction=self._direction)
+
+    def __add__(self, other):
+        if not isinstance(other, SingleEndedPort):
+            return NotImplemented
+        return SingleEndedPort(Cat(self._io, other._io), invert=self._invert + other._invert,
+                               direction=self._direction | other._direction)
+
+
+class DifferentialPort:
+    """TODO
+    """
+    def __init__(self, p, n, *, invert=False, direction=Direction.Bidir):
+        self._p = IOValue.cast(p)
+        self._n = IOValue.cast(n)
+        if len(self._p) != len(self._n):
+            raise ValueError(f"Length of 'p' ({len(self._p)}) must match length of 'n' "
+                             f"({len(self._n)})")
+        if isinstance(invert, bool):
+            self._invert = (invert,) * len(self._p)
+        else:
+            self._invert = tuple(invert)
+            if len(self._invert) != len(self._p):
+                raise ValueError(f"Length of 'invert' ({len(self._invert)}) must match "
+                                 f"length of 'p' ({len(self._p)}).")
+            if not all(isinstance(item, bool) for item in self._invert):
+                raise ValueError(f"'invert' must be a bool or iterable of bool, not {invert!r}")
+        self._direction = Direction(direction)
+
+    @property
+    def p(self):
+        return self._p
+
+    @property
+    def n(self):
+        return self._n
+
+    @property
+    def invert(self):
+        return self._invert
+
+    @property
+    def direction(self):
+        return self._direction
+
+    def __len__(self):
+        return len(self._p)
+
+    def __invert__(self):
+        return DifferentialPort(self._p, self._n, invert=tuple(~inv for inv in self._invert),
+                               direction=self._direction)
+
+    def __getitem__(self, index):
+        return DifferentialPort(self._p[index], self._n[index], invert=self._invert[index],
+                               direction=self._direction)
+
+    def __add__(self, other):
+        if not isinstance(other, DifferentialPort):
+            return NotImplemented
+        return DifferentialPort(Cat(self._p, other._p), Cat(self._n, other._n),
+                               invert=self._invert + other._invert,
+                               direction=self._direction | other._direction)
 
 
 class Pin(wiring.PureInterface):

docs/changes.rst

@@ -50,6 +50,7 @@ Implemented RFCs
 .. _RFC 46: https://amaranth-lang.org/rfcs/0046-shape-range-1.html
 .. _RFC 50: https://amaranth-lang.org/rfcs/0050-print.html
 .. _RFC 53: https://amaranth-lang.org/rfcs/0053-ioport.html
+.. _RFC 55: https://amaranth-lang.org/rfcs/0055-lib-io.html
 
 * `RFC 17`_: Remove ``log2_int``
 * `RFC 27`_: Testbench processes for the simulator
@@ -93,6 +94,7 @@ Standard library changes
 .. currentmodule:: amaranth.lib
 
 * Added: :mod:`amaranth.lib.memory`. (`RFC 45`_)
+* Added: :class:`amaranth.lib.io.SingleEndedPort`, :class:`amaranth.lib.io.DifferentialPort`. (`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`_)