Hardware/liteeth
1
0
Fork 0
mirror of https://github.com/enjoy-digital/liteeth.git synced 2025-01-03 03:43:37 -05:00
Code Issues Projects Releases Packages Wiki Activity

frontend/stream: Add with_csr parameter to LiteEthUDP2StreamRX, similar to LiteEthStream2UDPTX.

Browse source
This commit is contained in:
Florent Kermarrec 2025-01-02 18:25:14 +01:00
parent 3897ed022c
commit 2d5b333e34
1 changed files with 26 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
liteeth/frontend/stream.py
View file

@ -103,22 +103,30 @@ class LiteEthStream2UDPTX(LiteXModule):
# UDP to Stream RX --------------------------------------------------------------------------------- # UDP to Stream RX ---------------------------------------------------------------------------------
class LiteEthUDP2StreamRX(LiteXModule): class LiteEthUDP2StreamRX(LiteXModule):
def __init__(self, ip_address=None, udp_port=None, data_width=8, fifo_depth=None, with_broadcast=True): def __init__(self, ip_address=0, udp_port=0, data_width=8, fifo_depth=None, with_broadcast=True, with_csr=False):
self.sink = sink = stream.Endpoint(eth_udp_user_description(data_width)) self.sink = sink = stream.Endpoint(eth_udp_user_description(data_width))
self.source = source = stream.Endpoint(eth_tty_rx_description(data_width)) self.source = source = stream.Endpoint(eth_tty_rx_description(data_width))
# # # # # #
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()
valid = Signal(reset=1) valid = Signal(reset=1)
# Disable RX when enable=0.
self.comb += If(~self.enable, valid.eq(0))
# Check UDP Port. # Check UDP Port.
assert udp_port is not None self.comb += If(sink.dst_port != self.udp_port, valid.eq(0))
self.comb += If(sink.dst_port != udp_port, valid.eq(0))
# Check IP Address (Optional). # Check IP Address (Optional).
if (ip_address is not None) and (not with_broadcast): if not with_broadcast:
ip_address = convert_ip(ip_address) self.comb += If(sink.ip_address != self.ip_address, valid.eq(0))
self.comb += If(sink.ip_address != ip_address, valid.eq(0))
# Data-Path / Buffering (Optional). # Data-Path / Buffering (Optional).
if fifo_depth is None: if fifo_depth is None:
@ -140,6 +148,18 @@ class LiteEthUDP2StreamRX(LiteXModule):
fifo.source.connect(source) fifo.source.connect(source)
] ]
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 Streamer ------------------------------------------------------------------------------------- # UDP Streamer -------------------------------------------------------------------------------------
class LiteEthUDPStreamer(LiteXModule): class LiteEthUDPStreamer(LiteXModule):