class UDP(Module):
def __init__(self, ip, ip_address, debug=False, loopback=False):
self.ip = ip
+ self.ip_address = ip_address
self.debug = debug
self.loopback = loopback
self.tx_packets = []
ip_packet.flags = 0
ip_packet.fragment_offset = 0
ip_packet.time_to_live = 0x80
- ip_packet.source_ip_address = ip_address,
+ ip_packet.source_ip_address = self.ip_address
ip_packet.destination_ip_address = 0x12345678 # XXX
+ ip_packet.header_checksum = 0
+ ip_packet.protocol = udp_protocol
self.ip.send(ip_packet)
def callback(self, packet):
class TB(Module):
def __init__(self):
- self.submodules.phy_model = phy.PHY(8, debug=True)
- self.submodules.mac_model = mac.MAC(self.phy_model, debug=True, loopback=False)
- self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address, debug=True)
- self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address, debug=True, loopback=False)
- self.submodules.udp_model = udp.UDP(self.ip_model, ip_address, debug=True, loopback=False)
+ self.submodules.phy_model = phy.PHY(8, debug=False)
+ self.submodules.mac_model = mac.MAC(self.phy_model, debug=False, loopback=False)
+ self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address, debug=False)
+ self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address, debug=False, loopback=False)
+ self.submodules.udp_model = udp.UDP(self.ip_model, ip_address, debug=False, loopback=True)
self.submodules.udp_ip = LiteEthUDPIPCore(self.phy_model, mac_address, ip_address)
self.submodules.streamer = PacketStreamer(eth_udp_user_description(8))
+ self.submodules.logger = PacketLogger(eth_udp_user_description(8))
self.comb += [
Record.connect(self.streamer.source, self.udp_ip.sink),
self.udp_ip.sink.ip_address.eq(0x12345678),
self.udp_ip.sink.source_port.eq(0x1234),
self.udp_ip.sink.destination_port.eq(0x5678),
- self.udp_ip.sink.length.eq(64)
+ self.udp_ip.sink.length.eq(64),
+ Record.connect(self.udp_ip.source, self.logger.sink)
]
# use sys_clk for each clock_domain
while True:
packet = Packet([i for i in range(64)])
yield from self.streamer.send(packet)
+ yield from self.logger.receive()
+
+ # check results
+ s, l, e = check(packet, self.logger.packet)
+ print("shift "+ str(s) + " / length " + str(l) + " / errors " + str(e))
+
if __name__ == "__main__":
run_simulation(TB(), ncycles=2048, vcd_name="my.vcd", keep_files=True)