Hardware
/
liteeth
mirror of https://github.com/enjoy-digital/liteeth.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

frontend/stream/LiteEthStream2UDPTX: Add optional CSR to allow dynamic configuration (Enable, IP Address and UDP Port).

This commit is contained in:
Florent Kermarrec 2024-12-13 21:27:47 +01:00
parent 9f4d9d20bc
commit aad9de7e53
1 changed files with 32 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
liteeth/frontend/stream.py
View File

@ -8,24 +8,29 @@ from litex.gen import *
from liteeth.common import * from liteeth.common import *
# Stream to UDP TX ----------------------------------------------------------------------------------- # Stream to UDP TX ---------------------------------------------------------------------------------
class LiteEthStream2UDPTX(LiteXModule): class LiteEthStream2UDPTX(LiteXModule):
def __init__(self, ip_address, udp_port, data_width=8, fifo_depth=None): def __init__(self, ip_address, udp_port, data_width=8, fifo_depth=None, with_csr=False):
self.sink = sink = stream.Endpoint(eth_tty_tx_description(data_width)) self.sink = sink = stream.Endpoint(eth_tty_tx_description(data_width))
self.source = source = stream.Endpoint(eth_udp_user_description(data_width)) self.source = source = stream.Endpoint(eth_udp_user_description(data_width))
# # # # # #
ip_address = convert_ip(ip_address) self.ip_address = Signal(32, reset=convert_ip(ip_address))
self.udp_port = Signal(16, reset=udp_port)
self.enable = Signal(reset=1)
if with_csr:
self.add_csr()
if fifo_depth is None: if fifo_depth is None:
self.comb += [ self.comb += [
sink.connect(source, keep={"valid", "ready", "data"}), sink.connect(source, keep={"valid", "ready", "data"}),
source.last.eq(1), source.last.eq(1),
source.src_port.eq(udp_port), source.src_port.eq(self.udp_port),
source.dst_port.eq(udp_port), source.dst_port.eq(self.udp_port),
source.ip_address.eq(ip_address), source.ip_address.eq(self.ip_address),
source.length.eq(data_width // 8) source.length.eq(data_width // 8)
] ]
else: else:
@ -38,12 +43,14 @@ class LiteEthStream2UDPTX(LiteXModule):
self.fifo = fifo = stream.SyncFIFO([("data", data_width)], fifo_depth, buffered=True) self.fifo = fifo = stream.SyncFIFO([("data", data_width)], fifo_depth, buffered=True)
self.comb += sink.connect(fifo.sink) self.comb += sink.connect(fifo.sink)
self.fsm = fsm = FSM(reset_state="IDLE") self.fsm = fsm = ResetInserter()(FSM(reset_state="IDLE"))
self.comb += fsm.reset.eq(~self.enable)
fsm.act("IDLE", fsm.act("IDLE",
NextValue(counter, 0), NextValue(counter, 0),
NextValue(_ip_address, ip_address), NextValue(_ip_address, self.ip_address),
NextValue(_udp_port, udp_port), NextValue(_udp_port, self.udp_port),
# Send FIFO contenst when: # Send FIFO contents when:
# - We have a full packet: # - We have a full packet:
If(fifo.sink.valid & fifo.sink.ready & fifo.sink.last, If(fifo.sink.valid & fifo.sink.ready & fifo.sink.last,
NextValue(level, fifo.level + 1), # +1 for level latency. NextValue(level, fifo.level + 1), # +1 for level latency.
@ -79,6 +86,20 @@ class LiteEthStream2UDPTX(LiteXModule):
) )
) )
def add_csr(self):
self._enable = CSRStorage(1, description="Enable Module", reset=1)
self._ip_address = CSRStorage(32, description="IP Address", reset=self.ip_address.reset.value)
self._udp_port = CSRStorage(16, description="UDP Port", reset=self.udp_port.reset.value)
# # #
self.comb += [
self.enable.eq(self._enable.storage),
self.ip_address.eq(self._ip_address.storage),
self.udp_port.eq(self._udp_port.storage),
]
# UDP to Stream RX --------------------------------------------------------------------------------- # UDP to Stream RX ---------------------------------------------------------------------------------
class LiteEthUDP2StreamRX(LiteXModule): class LiteEthUDP2StreamRX(LiteXModule):