event: add EventSource and InterruptSource.

Author: Jean-François Nguyen

nmigen_soc/event.py

@@ -0,0 +1,127 @@
+from nmigen import *
+from nmigen import tracer
+
+from nmigen_soc import csr
+
+
+__all__ = ["EventSource", "InterruptSource"]
+
+
+class EventSource:
+    """Event source.
+
+    Parameters
+    ----------
+    mode : ``"level"``, ``"rise"``, ``"fall"``
+        Trigger mode. If ``"level"``, a notification is raised when the ``stb`` signal is high.
+        If ``"rise"`` (or ``"fall"``) a notification is raised on a rising (or falling) edge
+        of ``stb``.
+    name : str
+        Name of the event. If ``None`` (default) the name is inferred from the variable
+        name this event source is assigned to.
+
+    Attributes
+    ----------
+    name : str
+        Name of the event source.
+    mode : ``"level"``, ``"rise"``, ``"fall"``
+        Trigger mode.
+    stb : Signal, in
+        Event strobe.
+    """
+    def __init__(self, *, mode="level", name=None, src_loc_at=0):
+        if name is not None and not isinstance(name, str):
+            raise TypeError("Name must be a string, not {!r}".format(name))
+        self.name = name or tracer.get_var_name(depth=2 + src_loc_at)
+
+        choices = ("level", "rise", "fall")
+        if mode not in choices:
+            raise ValueError("Invalid trigger mode {!r}; must be one of {}"
+                             .format(mode, ", ".join(choices)))
+        self.mode = mode
+
+        self.stb = Signal(name="{}_stb".format(self.name))
+
+
+class InterruptSource(Elaboratable):
+    """Interrupt source.
+
+    A mean of gathering multiple event sources into a single interrupt request line.
+
+    Parameters
+    ----------
+    events : iter(:class:`EventSource`)
+        Event sources.
+    name : str
+        Name of the interrupt source. If ``None`` (default) the name is inferred from the
+        variable name this interrupt source is assigned to.
+
+    Attributes
+    ----------
+    name : str
+        Name of the interrupt source.
+    status : :class:`csr.Element`, read-only
+        Event status register. Each bit displays the level of the strobe of an event source.
+        Events are ordered by position in the `events` parameter.
+    pending : :class:`csr.Element`, read/write
+        Event pending register. If a bit is 1, the associated event source has a pending
+        notification. Writing 1 to a bit clears it.
+        Events are ordered by position in the `events` parameter.
+    enable : :class:`csr.Element`, read/write
+        Event enable register. Writing 1 to a bit enables its associated event source.
+        Writing 0 disables it.
+        Events are ordered by position in the `events` parameter.
+    irq : Signal, out
+        Interrupt request. It is raised if any event source is enabled and has a pending
+        notification.
+    """
+    def __init__(self, events, *, name=None, src_loc_at=0):
+        if name is not None and not isinstance(name, str):
+            raise TypeError("Name must be a string, not {!r}".format(name))
+        self.name = name or tracer.get_var_name(depth=2 + src_loc_at)
+
+        for event in events:
+            if not isinstance(event, EventSource):
+                raise TypeError("Event source must be an instance of EventSource, not {!r}"
+                                .format(event))
+        self._events = list(events)
+
+        width = len(events)
+        self.status  = csr.Element(width, "r",  name="{}_status".format(self.name))
+        self.pending = csr.Element(width, "rw", name="{}_pending".format(self.name))
+        self.enable  = csr.Element(width, "rw", name="{}_enable".format(self.name))
+
+        self.irq = Signal(name="{}_irq".format(self.name))
+
+    def elaborate(self, platform):
+        m = Module()
+
+        with m.If(self.pending.w_stb):
+            m.d.sync += self.pending.r_data.eq(self.pending.r_data & ~self.pending.w_data)
+
+        with m.If(self.enable.w_stb):
+            m.d.sync += self.enable.r_data.eq(self.enable.w_data)
+
+        for i, event in enumerate(self._events):
+            m.d.sync += self.status.r_data[i].eq(event.stb)
+
+            if event.mode in ("rise", "fall"):
+                event_stb_r = Signal.like(event.stb, name_suffix="_r")
+                m.d.sync += event_stb_r.eq(event.stb)
+
+            event_trigger = Signal(name="{}_trigger".format(event.name))
+            if event.mode == "level":
+                m.d.comb += event_trigger.eq(event.stb)
+            elif event.mode == "rise":
+                m.d.comb += event_trigger.eq(~event_stb_r & event.stb)
+            elif event.mode == "fall":
+                m.d.comb += event_trigger.eq(event_stb_r & ~event.stb)
+            else:
+                assert False # :nocov:
+
+            with m.If(event_trigger):
+                m.d.sync += self.pending.r_data[i].eq(1)
+
+        m.d.comb += self.irq.eq((self.pending.r_data & self.enable.r_data).any())
+
+        return m

nmigen_soc/test/test_event.py

@@ -0,0 +1,138 @@
+# nmigen: UnusedElaboratable=no
+
+import unittest
+from nmigen import *
+from nmigen.back.pysim import *
+
+from ..event import *
+
+
+def simulation_test(dut, process):
+    with Simulator(dut, vcd_file=open("test.vcd", "w")) as sim:
+        sim.add_clock(1e-6)
+        sim.add_sync_process(process)
+        sim.run()
+
+
+class EventSourceTestCase(unittest.TestCase):
+    def test_simple(self):
+        ev = EventSource()
+        self.assertEqual(ev.name, "ev")
+        self.assertEqual(ev.mode, "level")
+
+    def test_name_wrong(self):
+        with self.assertRaisesRegex(TypeError,
+                r"Name must be a string, not 2"):
+            EventSource(name=2)
+
+    def test_mode_wrong(self):
+        with self.assertRaisesRegex(ValueError,
+                r"Invalid trigger mode 'foo'; must be one of level, rise, fall"):
+            ev = EventSource(mode="foo")
+
+
+class InterruptSourceTestCase(unittest.TestCase):
+    def test_simple(self):
+        ev_0 = EventSource()
+        ev_1 = EventSource()
+        dut = InterruptSource((ev_0, ev_1))
+        self.assertEqual(dut.name, "dut")
+        self.assertEqual(dut.status.width, 2)
+        self.assertEqual(dut.pending.width, 2)
+        self.assertEqual(dut.enable.width, 2)
+
+    def test_name_wrong(self):
+        with self.assertRaisesRegex(TypeError,
+                r"Name must be a string, not 2"):
+            InterruptSource((), name=2)
+
+    def test_event_wrong(self):
+        with self.assertRaisesRegex(TypeError,
+                r"Event source must be an instance of EventSource, not 'foo'"):
+            dut = InterruptSource(("foo",))
+
+    def test_events(self):
+        ev_0 = EventSource(mode="level")
+        ev_1 = EventSource(mode="rise")
+        ev_2 = EventSource(mode="fall")
+        dut  = InterruptSource((ev_0, ev_1, ev_2))
+
+        def process():
+            yield ev_0.stb.eq(1)
+            yield ev_1.stb.eq(0)
+            yield ev_2.stb.eq(1)
+            yield
+            self.assertEqual((yield dut.irq), 0)
+
+            yield dut.status.r_stb.eq(1)
+            yield
+            yield dut.status.r_stb.eq(0)
+            yield
+            self.assertEqual((yield dut.status.r_data), 0b101)
+            yield
+
+            yield dut.enable.w_stb.eq(1)
+            yield dut.enable.w_data.eq(0b111)
+            yield
+            yield dut.enable.w_stb.eq(0)
+            yield
+            yield
+            self.assertEqual((yield dut.irq), 1)
+
+            yield dut.pending.w_stb.eq(1)
+            yield dut.pending.w_data.eq(0b001)
+            yield
+            yield dut.pending.w_stb.eq(0)
+            yield
+
+            yield dut.pending.r_stb.eq(1)
+            yield
+            yield dut.pending.r_stb.eq(0)
+            yield
+            self.assertEqual((yield dut.pending.r_data), 0b001)
+            self.assertEqual((yield dut.irq), 1)
+            yield
+
+            yield ev_0.stb.eq(0)
+            yield dut.pending.w_stb.eq(1)
+            yield dut.pending.w_data.eq(0b001)
+            yield
+            yield dut.pending.w_stb.eq(0)
+            yield
+            yield
+            self.assertEqual((yield dut.irq), 0)
+
+            yield ev_1.stb.eq(1)
+            yield dut.pending.r_stb.eq(1)
+            yield
+            yield dut.pending.r_stb.eq(0)
+            yield
+            self.assertEqual((yield dut.pending.r_data), 0b010)
+            self.assertEqual((yield dut.irq), 1)
+
+            yield dut.pending.w_stb.eq(1)
+            yield dut.pending.w_data.eq(0b010)
+            yield
+            yield dut.pending.w_stb.eq(0)
+            yield
+            yield
+            self.assertEqual((yield dut.irq), 0)
+
+            yield ev_2.stb.eq(0)
+            yield
+            yield dut.pending.r_stb.eq(1)
+            yield
+            yield dut.pending.r_stb.eq(0)
+            yield
+            self.assertEqual((yield dut.pending.r_data), 0b100)
+            self.assertEqual((yield dut.irq), 1)
+
+            yield dut.pending.w_stb.eq(1)
+            yield dut.pending.w_data.eq(0b100)
+            yield
+            yield dut.pending.w_stb.eq(0)
+            yield
+            yield
+            self.assertEqual((yield dut.irq), 0)
+
+        simulation_test(dut, process)