gram/compat.py

   1 # This file is Copyright (c) 2020 LambdaConcept <contact@lambdaconcept.com>
   2
   3 import unittest
   4
   5 from nmigen import *
   6 from nmigen import tracer
   7 from nmigen.compat import Case
   8 from nmigen.back.pysim import *
   9
  10 __ALL__ = ["delayed_enter", "Timeline", "CSRPrefixProxy"]
  11
  12
  13 def delayed_enter(m, src, dst, delay):
  14     if not isinstance(m, Module):
  15         raise ValueError("m must be a module object, not {!r}".format(m))
  16     if not isinstance(delay, int):
  17         raise ValueError("Delay must be an integer, not {!r}".format(delay))
  18     if delay < 1:
  19         raise ValueError("Delay must be at least one cycle, not {!r}".format(delay))
  20
  21     for i in range(delay):
  22         if i == 0:
  23             statename = src
  24         else:
  25             statename = "{}-{}".format(src, i)
  26
  27         if i == delay-1:
  28             deststate = dst
  29         else:
  30             deststate = "{}-{}".format(src, i+1)
  31
  32         with m.State(statename):
  33             m.next = deststate
  34
  35
  36 class Timeline(Elaboratable):
  37     def __init__(self, events):
  38         self.trigger = Signal()
  39         self._events = events
  40
  41     def elaborate(self, platform):
  42         m = Module()
  43
  44         lastevent = max([e[0] for e in self._events])
  45         counter = Signal(range(lastevent+1))
  46
  47         # Counter incrementation
  48         # (with overflow handling)
  49         if (lastevent & (lastevent + 1)) != 0:
  50             with m.If(counter == lastevent):
  51                 m.d.sync += counter.eq(0)
  52             with m.Else():
  53                 with m.If(counter != 0):
  54                     m.d.sync += counter.eq(counter+1)
  55                 with m.Elif(self.trigger):
  56                     m.d.sync += counter.eq(1)
  57         else:
  58             with m.If(counter != 0):
  59                 m.d.sync += counter.eq(counter+1)
  60             with m.Elif(self.trigger):
  61                 m.d.sync += counter.eq(1)
  62
  63         for e in self._events:
  64             if e[0] == 0:
  65                 with m.If(self.trigger & (counter == 0)):
  66                     m.d.sync += e[1]
  67             else:
  68                 with m.If(counter == e[0]):
  69                     m.d.sync += e[1]
  70
  71         return m
  72
  73
  74 class CSRPrefixProxy:
  75     def __init__(self, bank, prefix):
  76         self._bank = bank
  77         self._prefix = prefix
  78
  79     def csr(self, width, access, *, addr=None, alignment=None, name=None,
  80             src_loc_at=0):
  81         if name is not None and not isinstance(name, str):
  82             raise TypeError("Name must be a string, not {!r}".format(name))
  83         name = name or tracer.get_var_name(depth=2 + src_loc_at).lstrip("_")
  84
  85         prefixed_name = "{}_{}".format(self._prefix, name)
  86         return self._bank.csr(width=width, access=access, addr=addr,
  87                               alignment=alignment, name=prefixed_name)