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