luke
/
manta
mirror of https://github.com/fischermoseley/manta.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

uart: handle case of 255 byte-long groups in COBS encoder

This commit is contained in:
Fischer Moseley 2026-02-23 21:48:11 -07:00
parent 6631ddde6a
commit 0c0f6121b3
2 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/manta/uart/cobs_encode.py
View File

@ -27,10 +27,16 @@ class COBSEncode(wiring.Component):
with m.State("COUNT_BYTES"):
with m.If(self.sink.valid & self.sink.ready):
# End of packet or zero found, clock out length
with m.If((self.sink.last) | (self.sink.data == 0)):
m.d.sync += fsm_data.eq(
fifo.r_level + fifo_written_to_last_cycle + 1
)
with m.If(
(self.sink.last) | (self.sink.data == 0) | (fifo.r_level == 253)
):
with m.If(fifo.r_level == 253):
m.d.sync += fsm_data.eq(255)
with m.Else():
m.d.sync += fsm_data.eq(
fifo.r_level + fifo_written_to_last_cycle + 1
)
m.d.sync += was_last.eq(self.sink.last)
m.next = "WAIT_FOR_LENGTH"

10
test/test_cobs_encode.py
View File

@ -20,7 +20,6 @@ async def test_cobs_encode(ctx):
ctx.set(ce.source.ready, 1)
await encode_and_compare(ctx, [0x11, 0x22, 0x33, 0x44])
# await encode_and_compare(ctx, [0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99])
await ctx.tick()
await ctx.tick()
await ctx.tick()
@ -37,7 +36,7 @@ async def test_cobs_encode_random(ctx):
num_tests = random.randint(1, 5)
for _ in range(num_tests):
length = random.randint(1, 254)
length = random.randint(1, 2000)
input_data = [random.randint(0, 255) for _ in range(length)]
result = await encode_and_compare(ctx, input_data, quiet=True)
await ctx.tick()
@ -59,7 +58,9 @@ async def encode(ctx, data):
while not (tx_done and rx_done):
# Feed data to encoder
if not tx_done:
tx_stall = random.randint(0, 1)
if not tx_done and not tx_stall:
ctx.set(ce.sink.valid, 1)
ctx.set(ce.sink.data, data[tx_index])
ctx.set(ce.sink.last, tx_index == len(data) - 1)
@ -76,8 +77,7 @@ async def encode(ctx, data):
ctx.set(ce.sink.last, 0)
# Randomly set source.ready
# ctx.set(ce.source.ready, random.randint(0, 1))
ctx.set(ce.source.ready, 1)
ctx.set(ce.source.ready, random.randint(0, 1))
# Pull output data from buffer
if ctx.get(ce.source.valid) and ctx.get(ce.source.ready):