liteeth/test/udpip_tb.py

   1 from migen.fhdl.std import *
   2 from migen.bus import wishbone
   3 from migen.bus.transactions import *
   4 from migen.sim.generic import run_simulation
   5
   6 from liteeth.common import *
   7 from liteeth.core import LiteEthUDPIPCore
   8
   9 from liteeth.test.common import *
  10 from liteeth.test.model import phy, mac, arp, ip, udp
  11
  12 ip_address = 0x12345678
  13 mac_address = 0x12345678abcd
  14
  15 class TB(Module):
  16         def __init__(self):
  17                 self.submodules.phy_model = phy.PHY(8, debug=False)
  18                 self.submodules.mac_model = mac.MAC(self.phy_model, debug=False, loopback=False)
  19                 self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address, debug=False)
  20                 self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address, debug=False, loopback=False)
  21                 self.submodules.udp_model = udp.UDP(self.ip_model, ip_address, debug=False, loopback=True)
  22
  23                 self.submodules.udp_ip = LiteEthUDPIPCore(self.phy_model, mac_address, ip_address, 100000)
  24                 self.submodules.streamer = PacketStreamer(eth_udp_user_description(8))
  25                 self.submodules.logger = PacketLogger(eth_udp_user_description(8))
  26                 self.comb += [
  27                         Record.connect(self.streamer.source, self.udp_ip.sink),
  28                         self.udp_ip.sink.ip_address.eq(0x12345678),
  29                         self.udp_ip.sink.src_port.eq(0x1234),
  30                         self.udp_ip.sink.dst_port.eq(0x5678),
  31                         self.udp_ip.sink.length.eq(64),
  32                         Record.connect(self.udp_ip.source, self.logger.sink)
  33                 ]
  34
  35                 # use sys_clk for each clock_domain
  36                 self.clock_domains.cd_eth_rx = ClockDomain()
  37                 self.clock_domains.cd_eth_tx = ClockDomain()
  38                 self.comb += [
  39                         self.cd_eth_rx.clk.eq(ClockSignal()),
  40                         self.cd_eth_rx.rst.eq(ResetSignal()),
  41                         self.cd_eth_tx.clk.eq(ClockSignal()),
  42                         self.cd_eth_tx.rst.eq(ResetSignal()),
  43                 ]
  44
  45         def gen_simulation(self, selfp):
  46                 selfp.cd_eth_rx.rst = 1
  47                 selfp.cd_eth_tx.rst = 1
  48                 yield
  49                 selfp.cd_eth_rx.rst = 0
  50                 selfp.cd_eth_tx.rst = 0
  51
  52                 for i in range(100):
  53                         yield
  54
  55                 while True:
  56                         packet = Packet([i for i in range(64)])
  57                         yield from self.streamer.send(packet)
  58                         yield from self.logger.receive()
  59
  60                         # check results
  61                         s, l, e = check(packet, self.logger.packet)
  62                         print("shift "+ str(s) + " / length " + str(l) + " / errors " + str(e))
  63
  64
  65 if __name__ == "__main__":
  66         run_simulation(TB(), ncycles=2048, vcd_name="my.vcd", keep_files=True)