csr.periph: add Peripheral base class.

Author: Jean-François Nguyen

nmigen_soc/csr/__init__.py

@@ -1 +1,2 @@
 from .bus import *
+from .periph import *

nmigen_soc/csr/periph.py

@@ -0,0 +1,328 @@
+from nmigen import *
+from nmigen import tracer
+
+from .bus import *
+
+
+__all__ = ["Peripheral", "IRQLine", "Event", "PeripheralBridge"]
+
+
+class Peripheral:
+    """CSR-capable peripheral.
+
+    A helper class to reduce the boilerplate needed to control a peripheral with a CSR interface.
+    It provides facilities for instantiating CSR registers and sending interrupt requests to the
+    CPU.
+
+    The ``Peripheral`` class is not meant to be instantiated as-is, but rather as a base class for
+    actual peripherals.
+
+    Usage example
+    -------------
+
+    ```
+    class ExamplePeripheral(csr.Peripheral, Elaboratable):
+        def __init__(self):
+            super().__init__()
+            self._data   = self.csr(8, "w")
+            self._rdy    = self.event(mode="rise")
+
+            self._bridge = self.csr_bridge()
+            self.csr_bus = self._bridge.bus
+            self.irq     = self._bridge.irq
+
+        def elaborate(self, platform):
+            m = Module()
+            m.submodules.bridge = self._bridge
+            # ...
+            return m
+    ```
+
+    Parameters
+    ----------
+    name : str
+        Name of this peripheral. If ``None`` (default) the name is inferred from the variable
+        name this peripheral is assigned to.
+
+    Attributes
+    ----------
+    name : str
+        Name of the peripheral.
+    """
+    def __init__(self, name=None, src_loc_at=1):
+        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)
+
+        self._csr_regs = []
+        self._csr_bus  = None
+
+        self._events   = []
+        self._irq      = None
+
+    @property
+    def csr_bus(self):
+        """CSR bus providing access to registers.
+
+        Return value
+        ------------
+        An instance of :class:`Interface`.
+
+        Exceptions
+        ----------
+        Raises :exn:`NotImplementedError` if the peripheral does not have a CSR bus.
+        """
+        if self._csr_bus is None:
+            raise NotImplementedError("Peripheral {!r} does not have a CSR bus interface"
+                                      .format(self))
+        return self._csr_bus
+
+    @csr_bus.setter
+    def csr_bus(self, csr_bus):
+        if not isinstance(csr_bus, Interface):
+            raise TypeError("CSR bus interface must be an instance of csr.Interface, not {!r}"
+                            .format(csr_bus))
+        self._csr_bus = csr_bus
+
+    @property
+    def irq(self):
+        """Interrupt request line.
+
+        Return value
+        ------------
+        An instance of :class:`IRQLine`.
+
+        Exceptions
+        ----------
+        Raises :exn:`NotImplementedError` if the peripheral does not have an IRQ line.
+        """
+        if self._irq is None:
+            raise NotImplementedError("Peripheral {!r} does not have an IRQ line"
+                                      .format(self))
+        return self._irq
+
+    @irq.setter
+    def irq(self, irq):
+        if not isinstance(irq, IRQLine):
+            raise TypeError("IRQ line must be an instance of IRQLine, not {!r}"
+                            .format(irq))
+        self._irq = irq
+
+    def csr(self, width, access, *, addr=None, alignment=0, name=None, src_loc_at=0):
+        """Request a CSR register.
+
+        Parameters
+        ----------
+        width : int
+            Width of the register. See :class:`Element`.
+        access : :class:`Access`
+            Register access mode. See :class:`Element`.
+        addr : int
+            Address of the register. See :meth:`Multiplexer.add`.
+        alignment : int
+            Register alignment. See :class:`Multiplexer`.
+        name : str
+            Name of the register. If ``None`` (default) the name is inferred from the variable
+            name this register is assigned to.
+
+        Return value
+        ------------
+        An instance of :class:`Element`.
+        """
+        if name is not None and not isinstance(name, str):
+            raise TypeError("Name must be a string, not {!r}".format(name))
+        elem_name = name or tracer.get_var_name(depth=2 + src_loc_at)
+
+        elem = Element(width, access, name="{}_{}".format(self.name, elem_name))
+        self._csr_regs.append((elem, addr, alignment))
+        return elem
+
+    def event(self, *, mode="level", name=None, src_loc_at=0):
+        """Request an event source.
+
+        See :class:`Event` for details.
+
+        Return value
+        ------------
+        An instance of :class:`Event`.
+        """
+        event = Event(mode=mode, name=name, src_loc_at=1 + src_loc_at)
+        self._events.append(event)
+        return event
+
+    def csr_bridge(self, *, data_width=8, alignment=0):
+        """Request a bridge to the resources of the peripheral.
+
+        See :class:`PeripheralBridge` for details.
+
+        Return value
+        ------------
+        An instance of :class:`PeripheralBridge` providing access to the registers
+        of the peripheral and managing its event sources.
+        """
+        return PeripheralBridge(self, data_width=data_width, alignment=alignment)
+
+    def csr_registers(self):
+        """Iterate requested CSR registers and their parameters.
+
+        Yield values
+        ------------
+        A tuple ``elem, addr, alignment`` describing the register and its parameters.
+        """
+        for elem, addr, alignment in self._csr_regs:
+            yield elem, addr, alignment
+
+    def events(self):
+        """Iterate requested event sources.
+
+        Event sources are ordered by request order.
+
+        Yield values
+        ------------
+        An instance of :class:`Event`.
+        """
+        for event in self._events:
+            yield event
+
+
+class IRQLine(Signal):
+    """Interrupt request line."""
+    def __init__(self, *, name=None, src_loc_at=0):
+        super().__init__(name=name, src_loc_at=1 + src_loc_at)
+
+
+class Event:
+    """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
+    mode : ``"level"``, ``"rise"``, ``"fall"``
+        Trigger mode.
+    stb : Signal, in
+        Event strobe.
+    """
+    def __init__(self, *, mode, 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))
+
+        choices = ("level", "rise", "fall")
+        if mode not in choices:
+            raise ValueError("Invalid trigger mode {!r}; must be one of {}"
+                             .format(mode, ", ".join(choices)))
+
+        self.name = name or tracer.get_var_name(depth=2 + src_loc_at)
+        self.mode = mode
+        self.stb  = Signal(name="{}_stb".format(self.name))
+
+
+class PeripheralBridge(Elaboratable):
+    """Peripheral bridge.
+
+    A bridge providing access to the registers of a peripheral, and support for interrupt
+    requests (IRQs) from its event sources.
+
+    CSR registers
+    -------------
+    ev_status : read-only
+        Event status. Each bit displays the value of the ``stb`` signal of an event source.
+        The register width is ``len(list(periph.events())`` bits. Event sources are ordered by
+        request order.
+    ev_pending : read/write
+        Event pending. Each bit displays whether an event source has a pending notification.
+        Writing 1 to a bit clears the notification.
+        The register width is ``len(list(periph.events())`` bits. Event sources are ordered by
+        request order.
+    ev_enable : read/write
+        Event enable. Writing 1 to a bit enables an event source. Writing 0 disables it.
+        The register width is ``len(list(periph.events())`` bits. Event sources are ordered by
+        request order.
+
+    Parameters
+    ----------
+    periph : :class:`Peripheral`
+        The peripheral whose resources are exposed by this bridge.
+    data_width : int
+        Data width of the CSR bus. See :class:`Multiplexer`.
+    alignment : int
+        Register alignment. See :class:`Multiplexer`.
+
+    Attributes
+    ----------
+    bus : :class:`Interface`
+        CSR bus providing access to the registers of the peripheral.
+    irq : :class:`IRQLine` or None
+        IRQ line providing notifications from local events to the CPU. It is raised if any
+        event source is both enabled and has a pending notification. If the peripheral has
+        no event sources, it is set to ``None``.
+    """
+    def __init__(self, periph, *, data_width, alignment):
+        if not isinstance(periph, Peripheral):
+            raise TypeError("Peripheral must be an instance of Peripheral, not {!r}"
+                            .format(periph))
+
+        self._mux = Multiplexer(addr_width=1, data_width=data_width, alignment=alignment)
+        for elem, elem_addr, elem_alignment in periph.csr_registers():
+            self._mux.add(elem, addr=elem_addr, alignment=elem_alignment, extend=True)
+
+        self._events = list(periph.events())
+        if len(self._events) > 0:
+            width = len(self._events)
+            self._ev_status  = Element(width, "r",  name="{}_ev_status".format(periph.name))
+            self._ev_pending = Element(width, "rw", name="{}_ev_pending".format(periph.name))
+            self._ev_enable  = Element(width, "rw", name="{}_ev_enable".format(periph.name))
+            self._mux.add(self._ev_status,  extend=True)
+            self._mux.add(self._ev_pending, extend=True)
+            self._mux.add(self._ev_enable,  extend=True)
+            self.irq = IRQLine(name="{}_irq".format(periph.name))
+        else:
+            self.irq = None
+
+        self.bus = self._mux.bus
+
+    def elaborate(self, platform):
+        m = Module()
+
+        m.submodules.mux = self._mux
+
+        if self.irq is not None:
+            with m.If(self._ev_pending.w_stb):
+                m.d.sync += self._ev_pending.r_data.eq(   self._ev_pending.r_data
+                                                       & ~self._ev_pending.w_data)
+            with m.If(self._ev_enable.w_stb):
+                m.d.sync += self._ev_enable.r_data.eq(self._ev_enable.w_data)
+
+            for i, ev in enumerate(self._events):
+                m.d.sync += self._ev_status.r_data[i].eq(ev.stb)
+
+                if ev.mode in ("rise", "fall"):
+                    ev_stb_r = Signal.like(ev.stb, name_suffix="_r")
+                    m.d.sync += ev_stb_r.eq(ev.stb)
+
+                ev_trigger = Signal(name="{}_trigger".format(ev.name))
+                if ev.mode == "level":
+                    m.d.comb += ev_trigger.eq(ev.stb)
+                elif ev.mode == "rise":
+                    m.d.comb += ev_trigger.eq(~ev_stb_r &  ev.stb)
+                elif ev.mode == "fall":
+                    m.d.comb += ev_trigger.eq( ev_stb_r & ~ev.stb)
+                else:
+                    assert False # :nocov:
+
+                with m.If(ev_trigger):
+                    m.d.sync += self._ev_pending.r_data[i].eq(1)
+
+            m.d.comb += self.irq.eq((self._ev_pending.r_data & self._ev_enable.r_data).any())
+
+        return m