examples/basic/uart.py

   1 from nmigen import *
   2
   3
   4 class UART(Elaboratable):
   5     """
   6     Parameters
   7     ----------
   8     divisor : int
   9         Set to ``round(clk-rate / baud-rate)``.
  10         E.g. ``12e6 / 115200`` = ``104``.
  11     """
  12     def __init__(self, divisor, data_bits=8):
  13         assert divisor >= 4
  14
  15         self.data_bits = data_bits
  16         self.divisor   = divisor
  17
  18         self.tx_o    = Signal()
  19         self.rx_i    = Signal()
  20
  21         self.tx_data = Signal(data_bits)
  22         self.tx_rdy  = Signal()
  23         self.tx_ack  = Signal()
  24
  25         self.rx_data = Signal(data_bits)
  26         self.rx_err  = Signal()
  27         self.rx_ovf  = Signal()
  28         self.rx_rdy  = Signal()
  29         self.rx_ack  = Signal()
  30
  31     def elaborate(self, platform):
  32         m = Module()
  33
  34         tx_phase = Signal(range(self.divisor))
  35         tx_shreg = Signal(1 + self.data_bits + 1, reset=-1)
  36         tx_count = Signal(range(len(tx_shreg) + 1))
  37
  38         m.d.comb += self.tx_o.eq(tx_shreg[0])
  39         with m.If(tx_count == 0):
  40             m.d.comb += self.tx_ack.eq(1)
  41             with m.If(self.tx_rdy):
  42                 m.d.sync += [
  43                     tx_shreg.eq(Cat(C(0, 1), self.tx_data, C(1, 1))),
  44                     tx_count.eq(len(tx_shreg)),
  45                     tx_phase.eq(self.divisor - 1),
  46                 ]
  47         with m.Else():
  48             with m.If(tx_phase != 0):
  49                 m.d.sync += tx_phase.eq(tx_phase - 1)
  50             with m.Else():
  51                 m.d.sync += [
  52                     tx_shreg.eq(Cat(tx_shreg[1:], C(1, 1))),
  53                     tx_count.eq(tx_count - 1),
  54                     tx_phase.eq(self.divisor - 1),
  55                 ]
  56
  57         rx_phase = Signal(range(self.divisor))
  58         rx_shreg = Signal(1 + self.data_bits + 1, reset=-1)
  59         rx_count = Signal(range(len(rx_shreg) + 1))
  60
  61         m.d.comb += self.rx_data.eq(rx_shreg[1:-1])
  62         with m.If(rx_count == 0):
  63             m.d.comb += self.rx_err.eq(~(~rx_shreg[0] & rx_shreg[-1]))
  64             with m.If(~self.rx_i):
  65                 with m.If(self.rx_ack | ~self.rx_rdy):
  66                     m.d.sync += [
  67                         self.rx_rdy.eq(0),
  68                         self.rx_ovf.eq(0),
  69                         rx_count.eq(len(rx_shreg)),
  70                         rx_phase.eq(self.divisor // 2),
  71                     ]
  72                 with m.Else():
  73                     m.d.sync += self.rx_ovf.eq(1)
  74         with m.Else():
  75             with m.If(rx_phase != 0):
  76                 m.d.sync += rx_phase.eq(rx_phase - 1)
  77             with m.Else():
  78                 m.d.sync += [
  79                     rx_shreg.eq(Cat(rx_shreg[1:], self.rx_i)),
  80                     rx_count.eq(rx_count - 1),
  81                     rx_phase.eq(self.divisor - 1),
  82                 ]
  83                 with m.If(rx_count == 1):
  84                     m.d.sync += self.rx_rdy.eq(1)
  85
  86         return m
  87
  88
  89 if __name__ == "__main__":
  90     uart = UART(divisor=5)
  91     ports = [
  92         uart.tx_o, uart.rx_i,
  93         uart.tx_data, uart.tx_rdy, uart.tx_ack,
  94         uart.rx_data, uart.rx_rdy, uart.rx_err, uart.rx_ovf, uart.rx_ack
  95     ]
  96
  97     import argparse
  98
  99     parser = argparse.ArgumentParser()
 100     p_action = parser.add_subparsers(dest="action")
 101     p_action.add_parser("simulate")
 102     p_action.add_parser("generate")
 103
 104     args = parser.parse_args()
 105     if args.action == "simulate":
 106         from nmigen.back.pysim import Simulator, Passive
 107
 108         sim = Simulator(uart)
 109         sim.add_clock(1e-6)
 110
 111         def loopback_proc():
 112             yield Passive()
 113             while True:
 114                 yield uart.rx_i.eq((yield uart.tx_o))
 115                 yield
 116         sim.add_sync_process(loopback_proc)
 117
 118         def transmit_proc():
 119             assert (yield uart.tx_ack)
 120             assert not (yield uart.rx_rdy)
 121
 122             yield uart.tx_data.eq(0x5A)
 123             yield uart.tx_rdy.eq(1)
 124             yield
 125             yield uart.tx_rdy.eq(0)
 126             yield
 127             assert not (yield uart.tx_ack)
 128
 129             for _ in range(uart.divisor * 12): yield
 130
 131             assert (yield uart.tx_ack)
 132             assert (yield uart.rx_rdy)
 133             assert not (yield uart.rx_err)
 134             assert (yield uart.rx_data) == 0x5A
 135
 136             yield uart.rx_ack.eq(1)
 137             yield
 138         sim.add_sync_process(transmit_proc)
 139
 140         with sim.write_vcd("uart.vcd", "uart.gtkw"):
 141             sim.run()
 142
 143     if args.action == "generate":
 144         from nmigen.back import verilog
 145
 146         print(verilog.convert(uart, ports=ports))