event: migrate to lib.wiring interfaces.

Author: jfng

amaranth_soc/event.py

@@ -1,75 +1,131 @@
-import enum
-from collections import OrderedDict
-
 from amaranth import *
+from amaranth.lib import enum, wiring
+from amaranth.lib.wiring import In, Out
 
 
 __all__ = ["Source", "EventMap", "Monitor"]
 
 
-class Source(Record):
+class Source(wiring.Interface):
     class Trigger(enum.Enum):
         """Event trigger mode."""
         LEVEL = "level"
         RISE  = "rise"
         FALL  = "fall"
 
+    class Signature(wiring.Signature):
+        """Event source signature.
+
+        Parameters
+        ----------
+        trigger : :class:`Source.Trigger`
+            Trigger mode. An event can be edge- or level-triggered by the input line.
+
+        Interface attributes
+        --------------------
+        i : Signal()
+            Input line. Sampled in order to detect an event.
+        trg : Signal()
+            Event trigger. Asserted when an event occurs, according to the trigger mode.
+
+        Raises
+        ------
+        See :meth:`Source.Signature.check_parameters`.
+        """
+        def __init__(self, *, trigger="level"):
+            self.check_parameters(trigger=trigger)
+
+            self._trigger = Source.Trigger(trigger)
+
+            members = {
+                "i":   In(1),
+                "trg": Out(1),
+            }
+            super().__init__(members)
+
+        @property
+        def trigger(self):
+            return self._trigger
+
+        def check_parameters(cls, *, trigger):
+            """Validate signature parameters.
+
+            Raises
+            ------
+            :exc:`ValueError`
+                If ``trigger`` is not a member of :class:`Source.Trigger`.
+            """
+            # TODO(py3.9): Remove this. Python 3.8 and below use cls.__name__ in the error message
+            # instead of cls.__qualname__.
+            # Source.Trigger(trigger)
+            try:
+                Source.Trigger(trigger)
+            except ValueError as e:
+                raise ValueError(f"{trigger!r} is not a valid Source.Trigger") from e
+
+        def create(self, *, path=()):
+            """Create a compatible interface.
+
+            See :meth:`wiring.Signature.create` for details.
+
+            Returns
+            -------
+            A :class:`Source` object using this signature.
+            """
+            return Source(trigger=self.trigger, path=path)
+
+        def __eq__(self, other):
+            """Compare signatures.
+
+            Two signatures are equal if they have the same trigger mode.
+            """
+            return isinstance(other, Source.Signature) and self.trigger == other.trigger
+
+        def __repr__(self):
+            return f"event.Source.Signature({self.members!r})"
+
     """Event source interface.
 
     Parameters
     ----------
-    trigger : :class:`Trigger`
+    trigger : :class:`Source.Trigger`
         Trigger mode. An event can be edge- or level-triggered by the input line.
-    name: str
-        Name of the underlying record.
+    path : iter(:class:`str`)
+        Path to this event source interface. Optional. See :class:`wiring.Interface`.
 
     Attributes
     ----------
-    i : Signal()
-        Input line. Sampled in order to detect an event.
-    trg : Signal()
-        Event trigger. Asserted when an event occurs, according to the trigger mode.
+    event_map : :class:`EventMap`
+        A collection of event sources.
+
+    Raises
+    ------
+    See :meth:`Source.Signature.check_parameters`.
     """
-    def __init__(self, *, trigger="level", name=None, src_loc_at=0):
-        choices = ("level", "rise", "fall")
-        if not isinstance(trigger, Source.Trigger) and trigger not in choices:
-            raise ValueError("Invalid trigger mode {!r}; must be one of {}"
-                             .format(trigger, ", ".join(choices)))
-        self.trigger = Source.Trigger(trigger)
-        self._map    = None
-
-        super().__init__([
-            ("i",   1),
-            ("trg", 1),
-        ], name=name, src_loc_at=1 + src_loc_at)
+    def __init__(self, *, trigger="level", path=()):
+        super().__init__(Source.Signature(trigger=trigger), path=path)
 
-    @property
-    def event_map(self):
-        """Event map.
+        self._map = None
 
-        Return value
-        ------------
-        A :class:`EventMap` describing subordinate event sources.
+    @property
+    def trigger(self):
+        return self.signature.trigger
 
-        Exceptions
-        ----------
-        Raises :exn:`NotImplementedError` if the source does not have an event map.
-        """
+    @property
+    def event_map(self):
         if self._map is None:
-            raise NotImplementedError("Event source {!r} does not have an event map"
-                                      .format(self))
+            raise NotImplementedError(f"{self!r} does not have an event map")
         return self._map
 
     @event_map.setter
     def event_map(self, event_map):
         if not isinstance(event_map, EventMap):
-            raise TypeError("Event map must be an instance of EventMap, not {!r}"
-                            .format(event_map))
+            raise TypeError(f"Event map must be an instance of EventMap, not {event_map!r}")
         event_map.freeze()
         self._map = event_map
 
-    # FIXME: get rid of this
-    __hash__ = object.__hash__
+    def __repr__(self):
+        return f"event.Source({self.signature!r})"
 
 
 class EventMap:
@@ -80,7 +136,7 @@ class EventMap:
     increment, starting at 0.
     """
     def __init__(self):
-        self._sources = OrderedDict()
+        self._sources = dict()
         self._frozen  = False
 
     @property
@@ -117,8 +173,8 @@ def add(self, src):
         if not isinstance(src, Source):
             raise TypeError("Event source must be an instance of event.Source, not {!r}"
                             .format(src))
-        if src not in self._sources:
-            self._sources[src] = self.size
+        if id(src) not in self._sources:
+            self._sources[id(src)] = src, self.size
 
     def index(self, src):
         """Get the index corresponding to an event source.
@@ -139,7 +195,8 @@ def index(self, src):
         if not isinstance(src, Source):
             raise TypeError("Event source must be an instance of event.Source, not {!r}"
                             .format(src))
-        return self._sources[src]
+        _, index = self._sources[id(src)]
+        return index
 
     def sources(self):
         """Iterate event sources.
@@ -148,8 +205,7 @@ def sources(self):
         ------------
         A tuple ``src, index`` corresponding to an event source and its index.
         """
-        for src, index in self._sources.items():
-            yield src, index
+        yield from self._sources.values()
 
 
 class Monitor(Elaboratable):
@@ -160,7 +216,7 @@ class Monitor(Elaboratable):
     Parameters
     ----------
     event_map : :class:`EventMap`
-        Event map.
+        A collection of event sources.
     trigger : :class:`Source.Trigger`
         Trigger mode. See :class:`Source`.
 
@@ -176,7 +232,7 @@ class Monitor(Elaboratable):
         Clear selected pending events.
     """
     def __init__(self, event_map, *, trigger="level"):
-        self.src = Source(trigger=trigger)
+        self.src = Source(trigger=trigger, path=("src",))
         self.src.event_map = event_map
 
         self.enable  = Signal(event_map.size)

tests/test_csr_event.py

@@ -30,39 +30,39 @@ def test_trigger_wrong(self):
 
     def test_add(self):
         monitor = EventMonitor(data_width=8)
-        sub = event.Source()
+        sub = event.Source.Signature().create()
         monitor.add(sub)
         self.assertEqual(monitor.src.event_map.size, 1)
         self.assertEqual(monitor.src.event_map.index(sub), 0)
 
     def test_freeze(self):
         monitor = EventMonitor(data_width=8)
         monitor.freeze()
-        sub = event.Source()
+        sub = event.Source.Signature().create()
         with self.assertRaisesRegex(ValueError,
                 r"Event map has been frozen. Cannot add source."):
             monitor.add(sub)
 
     def test_src_freeze(self):
         monitor = EventMonitor(data_width=8)
         monitor.src
-        sub = event.Source()
+        sub = event.Source.Signature().create()
         with self.assertRaisesRegex(ValueError,
                 r"Event map has been frozen. Cannot add source."):
             monitor.add(sub)
 
     def test_bus_freeze(self):
         monitor = EventMonitor(data_width=8)
         monitor.bus
-        sub = event.Source()
+        sub = event.Source.Signature().create()
         with self.assertRaisesRegex(ValueError,
                 r"Event map has been frozen. Cannot add source."):
             monitor.add(sub)
 
     def test_csr_regs(self):
         monitor = EventMonitor(data_width=8)
-        sub_0 = event.Source()
-        sub_1 = event.Source()
+        sub_0 = event.Source.Signature().create()
+        sub_1 = event.Source.Signature().create()
         monitor.add(sub_0)
         monitor.add(sub_1)
         resources = list(monitor.bus.memory_map.resources())
@@ -90,7 +90,7 @@ def test_freeze_idempotent(self):
 class EventMonitorSimulationTestCase(unittest.TestCase):
     def test_simple(self):
         dut = EventMonitor(data_width=8)
-        sub = event.Source()
+        sub = event.Source.Signature().create(path=("sub",))
         dut.add(sub)
 
         addr_enable  = 0x0