whitequark · May 10, 2025 01:52
diff --git a/amaranth_cobs.py b/amaranth_cobs.py
 from amaranth import *
 from amaranth.lib import data, wiring, memory, stream
 from amaranth.lib.stream import In, Out


 __all__ = ["Encoder", "Decoder"]


 class Encoder(wiring.Component):
    """`Consistent Overhead Byte Stuffing <cobs>`_ encoder combined with a FIFO.

    The encoder accepts a stream of tokens, which can be either _data_ or _end_, and produces
    a stream of bytes. Input data tokens produce non-NUL output bytes with a maximum latency
    of 256 cycles; input end tokens produce NUL output bytes.

    Since COBS encoding requires up to 254 bytes of lookahead, and a COBS encoder will be usually
    combined with a FIFO (either at the input or at the output), this encoder is combined with
    a FIFO to use limited memory resources more efficiently. All but one byte of the internal FIFO
    will be filled with data in case of output back-pressure.

    The latency of the encoder depends on the type and value of input tokens. Input non-NUL data
    tokens do not immediately produce output bytes; rather, bytes corresponding to these tokens
    appear at the output only after: (a) an input end token, or (b) an input NUL data token, or
    (c) 255th consecutive non-NUL data token.

    .. _cobs: https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing
    """

    i: In(stream.Signature(data.StructLayout({
        "data": 8,
        "end":  1
    })))
    o: Out(stream.Signature(8))

    def __init__(self, fifo_depth=256):
        if not (fifo_depth >= 256 and fifo_depth.bit_count() == 1):
            raise ValueError("COBS encoder requires a power-of-2 sized FIFO that is "
                             "at least 256 bytes deep")
        self.fifo_depth = fifo_depth

        super().__init__()

    def elaborate(self, platform):
        m = Module()

        # This implementation improves resource use efficiency by merging the two memories that
        # would otherwise be necessary in a typical implementation: the FIFO for buffering packet
        # data, and the lookahead memory for COBS encoding. Specifically, it reuses the "empty"
        # space in the FIFO for storing bytes that follow a yet-unknown COBS overhead byte; this is
        # called "staging". Once the value of the overhead byte becomes known, the FIFO write
        # pointer is advanced simultaneously with the overhead byte being overwriten; this is
        # called "committing".

        m.submodules.data = data = memory.Memory(shape=8, depth=self.fifo_depth, init=[])
        w_port = data.write_port()
        r_port = data.read_port(transparent_for=(w_port,))

        w_addr = Signal.like(w_port.addr)
        r_addr = Signal.like(r_port.addr)
        empty  = (w_addr == r_addr)
        full   = (w_addr == r_addr - 1)

        def write(at, data):
            m.d.comb += w_port.addr.eq(at)
            m.d.comb += w_port.data.eq(data)
            m.d.comb += w_port.en.eq(1)

        staged = Signal(8, init=1)

        def stage(data):
            write(w_addr + staged, data)
            m.d.sync += staged.eq(staged + 1)

        def commit():
            write(w_addr, staged)
            m.d.sync += w_addr.eq(w_addr + staged)
            m.d.sync += staged.eq(1)

        with m.FSM():
            with m.State("Data"):
                with m.If(self.i.valid & ~full):
                    with m.If(self.i.p.end):
                        m.d.comb += self.i.ready.eq(1)
                        commit()
                        m.next = "End"
                    with m.Elif(staged == 0xff):
                        commit()
                    with m.Elif(self.i.p.data == 0x00):
                        m.d.comb += self.i.ready.eq(1)
                        commit()
                    with m.Else():
                        m.d.comb += self.i.ready.eq(1)
                        stage(self.i.p.data)

            with m.State("End"):
                write(w_addr, 0x00)
                m.d.sync += w_addr.eq(w_addr + 1)
                m.next = "Data"

        m.d.comb += self.o.valid.eq(~empty)
        m.d.comb += self.o.payload.eq(r_port.data)
        with m.If(self.o.valid & self.o.ready):
            m.d.comb += r_port.addr.eq(r_addr + 1)
            m.d.sync += r_addr.eq(r_addr + 1)
        with m.Else():
            m.d.comb += r_port.addr.eq(r_addr)

        return m


 class Decoder(wiring.Component):
    """`Consistent Overhead Byte Stuffing <cobs>`_ decoder.

    Performs an inversion of the transformation done by :class:`Encoder` with a fixed 0 cycle
    latency.

    If invalid COBS data is encountered (namely: if a group header byte or data byte is NUL),
    the decoder transitions to an error state, signaled by the ``error`` output. This state is
    final, cleared only by a reset.

    .. _cobs: https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing
    """

    i: In(stream.Signature(8))
    o: Out(stream.Signature(data.StructLayout({
        "data": 8,
        "end":  1
    })))
    error: Out(1)

    def elaborate(self, platform):
        m = Module()

        count  = Signal(8)
        offset = Signal(8)

        with m.FSM():
            with m.State("Start"):
                m.d.comb += self.i.ready.eq(1)
                with m.If(self.i.valid & self.i.ready):
                    m.d.sync += count.eq(1)
                    with m.If(self.i.payload != 0x00):
                        m.d.sync += offset.eq(self.i.payload)
                        m.next = "Data"
                    with m.Else():
                        m.next = "Error"

            with m.State("Data"):
                m.d.comb += self.i.ready.eq(self.o.ready)
                with m.If(self.i.valid & self.i.ready):
                    with m.If(offset == count):
                        m.d.sync += count.eq(1)
                        with m.If(self.i.payload == 0x00):
                            m.d.comb += self.o.payload.end.eq(1)
                            m.d.comb += self.o.valid.eq(1)
                            m.next = "Start"
                        with m.Else():
                            m.d.comb += self.o.payload.data.eq(0x00)
                            m.d.comb += self.o.valid.eq(offset != 0xff)
                            m.d.sync += offset.eq(self.i.payload)
                    with m.Else():
                        m.d.sync += count.eq(count + 1)
                        with m.If(self.i.payload != 0x00):
                            m.d.comb += self.o.payload.data.eq(self.i.payload)
                            m.d.comb += self.o.valid.eq(1)
                        with m.Else():
                            m.next = "Error"

            with m.State("Error"):
                m.d.comb += self.error.eq(1)

        return m
diff --git a/test_cobs.py b/test_cobs.py
 from amaranth import *
 from amaranth.sim import Simulator

 from amaranth_cobs import *

 from cobs import cobs


 def ref_encode(data_i: bytes) -> bytes:
    return b"\0".join(cobs.encode(chunk) if chunk else b"" for chunk in data_i.split(b"$"))


 def rtl_encode(data_i: bytes, dollar=True) -> bytes:
    dut = Encoder()

    async def testbench_i(ctx):
        ctx.set(dut.i.valid, 1)
        for byte in data_i:
            if byte == ord("$") and dollar:
                ctx.set(dut.i.payload, {"end": 1})
            else:
                ctx.set(dut.i.payload, {"data": byte})
            await ctx.tick().until(dut.i.ready)
        if not dollar:
            ctx.set(dut.i.payload, {"end": 1})
            await ctx.tick().until(dut.i.ready)
        ctx.set(dut.i.valid, 0)

    data_o = bytearray()
    async def testbench_o(ctx):
        ctx.set(dut.o.ready, 1)
        empty_for = 0
        while empty_for < len(data_i) + 3:
            _, _, valid, payload = await ctx.tick().sample(dut.o.valid).sample(dut.o.payload)
            if valid:
                data_o.append(payload)
                empty_for  = 0
            else:
                empty_for += 1
        ctx.set(dut.o.ready, 0)

    sim = Simulator(dut)
    sim.add_clock(1e-6)
    sim.add_testbench(testbench_i)
    sim.add_testbench(testbench_o)
    sim.run()

    return data_o


 def rtl_decode(data_i: bytes, dollar=True) -> bytes:
    dut = Decoder()

    async def testbench_i(ctx):
        ctx.set(dut.i.valid, 1)
        for byte in data_i:
            ctx.set(dut.i.payload, byte)
            await ctx.tick().until(dut.i.ready)
        ctx.set(dut.i.valid, 0)

    data_o = []
    async def testbench_o(ctx):
        pending = bytearray()
        ctx.set(dut.o.ready, 1)
        for _ in range(data_i.count(b"\0")):
            while True:
                _, _, valid, payload = await ctx.tick().sample(dut.o.valid).sample(dut.o.payload)
                if valid:
                    if payload.end:
                        data_o.append(bytes(pending))
                        pending.clear()
                        break
                    else:
                        pending.append(payload.data)
        ctx.set(dut.o.ready, 0)

    sim = Simulator(dut)
    sim.add_clock(1e-6)
    sim.add_testbench(testbench_i)
    sim.add_testbench(testbench_o)
    sim.run()

    return data_o


 def cases() -> list[bytes]:
    return [
        b"\0$",
        b"\0\0$",
        b"\0A\0$",
        b"AB\0C$",
        b"ABCD$",
        b"A\0\0\0$",
        b"A"*254+b"$",
        b"\0"+b"A"*254+b"$",
        b"A"*500+b"$",
        b"foo$bar$",
    ]


 def test_encoder_simple():
    for case in cases():
        assert rtl_encode(case) == ref_encode(case)


 def test_encoder_vmlinuz():
    assert rtl_encode(vmlinuz, dollar=False) == cobs.encode(vmlinuz) + b"\0"


 def test_decoder_simple():
    for case in cases():
        print(ref_encode(case))
        assert rtl_decode(ref_encode(case)) == case.split(b"$")[:-1]


 def test_decoder_vmlinuz():
    assert rtl_decode(cobs.encode(vmlinuz) + b"\0") == [vmlinuz]


 vmlinuz = bytes.fromhex("""
 4d5a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 000000000000cd238281400000005045000064860400000000000000000001000000a00006020b02021400d0b70000700600
 00000000fd5cb7000050000000000000000000000010000000020000000000000300000000000000000000000090be000010
 00001a42b8000a00000100000000000000000000000000000000000000000000000000000000000000000000000006000000
 00000000000000000000000000000000000000000000000000000000000000000032b800c005000000000000000000002e73
 65747570000000300000001000000030000000100000000000000000000000000000400000422e636f6d7061740000100000
 004000000010000000400000000000000000000000000000400000422e7465787400000000d0b7000050000000d0b7000050
 0000000000000000000000000000200000602e64617461000000007006000020b800001200000020b8000000000000000000
 00000000400000c0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff
 ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff270100
 207e0b000000ffff000055aaeb6a486472530f0200000000001000430001008000001000000000000000000000000000005c
 000000000000ffffff7f0000200001157f00ff070000000000000000000000000000cc0200001e6ab4000000000000000000
 000000010000000000d07e0360a9b6003cb1b7008cd88ec0fc8cd239c289e27416ba005af60611028074048b16240281c200
 04730231d283e2fc7503bafcff8ed0660fb7e2fb1e68a302cb66813e784655aa5a5a7517bf8046b9035a6631c029f9c1e902
 f366ab66e84512000066b8eb03000066e8fe000000f4ebfd3806ff027405a2ff02eb00669c0fa00fa8666083ec2c89d689e7
 b90b00f366a566610fa90fa1071f669dcd00669c1e060fa00fa86660fc660fb7e48cc88ed88ec0678b7c244421ff740889e6
 b90b00f366a583c42c66610fa90fa1669d66c3665666536683ec2c6689c36683f80a750c66b80d00000066e8e3ffffff6689
 e066e8d01c000067c7442410070067c7442418010067c644241d0e67885c241c6631c96689e266b81000000066e850ffffff
 66833ecc5700743966beffff000066a1cc576683c005660fb7c066ff16a047a8207416660fb716cc57660fb6c36683c42c66
 5b665eff26a447664e74e6f390ebcd6683c42c665b665e66c366536689c367660fbe0384c0740a664366e84effffffebed66
 5b66c3074e6f207365747570207369676e617475726520666f756e642e2e2e0a0066ba800000006631c0ff26a44766566653
 66bba086010066be2000000066e8ddffffff66b86400000066ff16a0473cff7506664e7506eb1fa801741366e8beffffff66
 b86000000066ff16a047eb04a802740a664b75c66683c8ffeb036631c0665b665e66c366556657665666536689c66631c08e
 e06683c8ff8ee866bf000200006467668b2f6689e86683ee01722567668d5801646766891f66e860ffffff66b81002000065
 67668b006639c374da6631d8eb036631c066ba00020000646766892a665b665e665f665d66c3665666536683ec2c66bbff00
 000066b82000000066e87fffffff6685c00f85f5006689e066e8291b000067c744241c01246631c96689e266b81500000066
 e8bbfdffff66b82000000066e84affffff6685c00f85c00066e8e4feffff6689c666b82000000066e82effffff6685c00f85
 e8f3caffff66c706ec5701000000678a5424486683e27f6631c038d37429660fb606b64738d075066683c8ffeb1967894424
 1c6631c96689e266b81000000066e8b3caffffebe16683c458665b665e66c367660fb600e966ff665566576656665366a1f8
 5766486631ff6683f8010f87f8008a1eb6476631c08ee066e89cf8ffff6689c566a1044666406683e0fe66a3044666a3e845
 66be1401000066a10046662b0604466683f8070f8eb3006689f066e8fbf3ffff6685c00f85960067668d8600ffffff66e8f4
 feffff6685c00f85810066ba1000000066b8c003000066e8b7feffffa801756b66ba0600000066b8ce03000066e8a1feffff
 a8017555660fb7c566ba0f00000066e88dfeffff84c0754166a1044666406683e0fe67668d5008668916044667893067c740
 06000066ba4a04000064678b126789500266ba8404000064678a12660fb6d2664267895004664766466681fe800100000f85
 3cff660fb6c366e855feffff6689f8665b665e665f665d66c38ed98ec18ee18ee98ed101dc0f00df31c931d231db31ed31ff
 0f00d1ffe0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 65722027666f7263657061652720746f20656e61626c6520617420796f7572206f776e207269736b210a006561726c797072
 696e746b0073657269616c003078007474795300636f6e736f6c650075617274383235302c696f2c00756172742c696f2c00
 65646400736b69706d627200736b6970006f6666006f6e0071756965740050726f62696e672045444420286564643d6f6666
 20746f2064697361626c65292e2e2e20006f6b0a006561726c7920636f6e736f6c6520696e20736574757020636f64650a00
 6465627567005741524e494e473a20416e6369656e7420626f6f746c6f616465722c20736f6d652066756e6374696f6e616c
 697479206d6179206265206c696d69746564210a00556e61626c6520746f20626f6f74202d20706c65617365207573652061
 206b65726e656c20617070726f70726961746520666f7220796f7572204350552e0a004132302067617465206e6f74207265
 73706f6e64696e672c20756e61626c6520746f20626f6f742e2e2e0a005072657373203c454e5445523e20746f2073656520
 766964656f206d6f64657320617661696c61626c652c203c53504143453e20746f20636f6e74696e75652c206f7220776169
 74203330207365630a004d6f64653a205265736f6c7574696f6e3a2020547970653a20002564782564002563202530335820
 25346478252d377320252d367300456e746572206120766964656f206d6f6465206f7220227363616e2220746f207363616e
 20666f72206164646974696f6e616c206d6f6465733a200008200800556e646566696e656420766964656f206d6f6465206e
 756d6265723a2025780a004347412f4d44412f484743004547410056474100564553410042494f5300000000000000000000
 00000000000000000000000000006670750000056d737200000670616500000863783800000f636d6f760000186678737200
 001973736500001a7373653200011d6c6d0003146e6f706c00151f0000000000000000000000000000000000000000000000
 0000000000000000618100070000002000000000000000000000000000000000000000000000000000000000000000000000
 00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000f8030000
 f802000000000000000000000000000000000000ffff0000009bcf00ffff00000093cf006700001000890000000000000000
 0000491d0000111f00001f1f00001f1f00001f1f00001f1f0000111f00001f1f00001f1f00001f1f00001f1f0000df1e0000
 3b1f0000961e00001f1f00001f1f0000d31d00001f1f0000171f00001f1f00001f1f0000031f000030313233343536373839
 4142434445460000000000000000362e31322e32352d616d643634202864656269616e2d6b65726e656c406c697374732e64
 656269616e2e6f72672920233120534d5020505245454d50545f44594e414d49432044656269616e20362e31322e32352d31
 2028323032352d30342d32352900eb320000aa320000f33200002f330000fb32000003330000173300000100000002000000
 070000006d430000794300007d4300002046000028460000404600007d430000143200003f2f000000000000000000000000
 000000000000814300003e3300006035000000000000000000000000000000020002864300002d3800003538000000000000
 0000000001000000000180000000000000000000005a0000005a000000000000000000000000000000000000000000000000
 0000000f500019000000000f500019000000010f50002b0000000000000000000000000f500019000000010f500032000000
 020f50002b000000030f50001c000000050f50001e000000060f500022000000070f50003c00000055aa5a5a000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
 c2ffa6d3a2ff779c162b389071c9768dbe2ab48afc7f2d34f7af7e988a83885b99380aa91f8450ba9476a5bdca621b81edfc
 5786b9b1d842b457c541a5a442acc21f757eaa554b68312ca7ab0adc5d9589d9387072f08683ff0f8205f5567997e358c551
 e4d68cffffc980d8f3f6bad5ffbfecc38a7bf0dda78581eb318c29058754ac6306088dedc91134f66a67800062c037a68154
 8a0641c90f0769d62cb4960675352359c5bdf4a0b0bb2c5af86f4f905f8948771ff6c1048f1b02b470a0e32276e524b4c895
 82ffbf2e8364ae2514c9f9f2058a623e03d9d60cc25dbf14d4cec31819d06ce15f4729612ee24b5b7900a78df9af875409b3
 fff5ff6b0ef6f27f3da319fca066d72798ffbf5d9eaa45ae042ffccfff7f085b874f6232adaeada6e011c57c2da9fcd74598
 dbf8ff754b39c8b46690c34eed366d056f04752423fcffaf54b18d52283462ae5481ffe9cdcf7f60f0b9544b52f6b62e89d3
 e579f93f709dd2db677757aeb8d5565e6a7eb1db460c42f2da38e85e2dbf41708d29fb7f4f854efc2809f37f75abb99f8e50
 ec4359feab3f987b4f9307b3603e49a195ebe50ff32a46e6146a1453724ff35f338181ffd52bcc3d6d560d8fbe2be3ffab63
 bfa4d65642f09adeb5397a44fac9b919e26cfd8691163f33946f5e912a2ca031439f365f7599011cfce4e85670e8103fa16a
 7ee10782ffda1622f30fa9d2bd26b4d510b51f5c96e793cf944a51d7aef336e2bf105b581b91873922714561afc62528eb99
 8731fa51c4bcce26a4b0b6354899e6ffff288f4f031125071760a97a2399e5a428e651460bafe8f651d3ea70dced59b1f5d1
 941d98e08693fffeee3d4577ddf3e43b74dea743adc3a2de06abe551245991cf00c64e617ef09b3d403ccf560f169450325a
 64ddd6b074fa51ee6a9ded49c9faf0447fba437cd2ce1d783d20a17c483bb99111b00d7367a9d3a4c461468aa125d49a2b72
 3d5d8b882fef462bb6cea496cfddcdba738479b0f536eb6e0ae62137c7dfb0a0f0b0dc36225d7f4ad33f2d040253b6ce7be1
 23cc4f68ee1fdedb6ab3364efb8606b40e67744039369b75c3b1c8d5b00260ee69de9ba5e7d879e8a37ef97006fcaffebbe1
 14b99a24fe54c36535dc277df2372e591277c1dc06c8694466ef9653746124308787e8c0478eaa93e46a88bf2ac1fc0103b3
 22910ec21511d84d40ee99bddcb8030665aa1a9061de82abacd8ea22b17ba74a3bdb1e0a80392cc7210a8f41cedda5213276
 f7109fc2c0d8a3afbd031cf9eb955c4bb2a2f8921204730f1deabf17c80849e51766c338ebedfb64a821df7da101bbc1641a
 d38b541d6eb610c83b4d2855eb04a80bb82d5bfa2880b90efa9d69a71a135c74370773a731d51dc857995ede054108e10e45
 2d9ca0c0d25b5667951fff7f856a870a75e69f9d442f23bf773031773dd72d3aba4271feb4601773bfae3315a3493922396d
 f2374db00d6e6618c066fe28b706e65ff2552d2e3a3429853e812757e65729eedf0b9c7ff67f8358d533ca7f0db343ee2280
 6c37e705dd6e24af15b3dbba88f96e9dea740bc03b3ae8528c60e03c25c3869841ec07213dca5d524caac045cd7db4d08a41
 30fb5906f45b305a882b48fe60b399e9ecc7da2fce453dea062b42383b77f5f2ba2bde43062450b6800ec7112d31e5744f2d
 58abee60972dde85521e3c68611ff86e5992a93d29343a3ded1f2ac4e63bd1d9e087bcd43e8946d63d93ad468e3dca71c80e
 b846fb79ef21d459218f79c0b33cdd0dd287127d299663461b6da98527bc602c90b579903e962c2dc57eba72b400c7bd5e6a
 6316b06b8076533cf5d457ce6cfd6e45b81aea14b4704b25b21b197ae7afa3daa1c2ecc8c7bbdd4d9581f539ad5ba4a031ba
 6b85c671c8badacbc274fe41bc6d0c90691ec57d7fd2df3d2d21c6e9595ef86002df58103e3d934b17a0932afde9650bfa34
 47c7ec48b40780c7844eb54622091d81fd97bd6f5ed5eab22a25706892ec1a12d9d17df4338f337ef80084308a38776a0f6c
 c25205bdb84b437d0521867428e8f3112f01a54b9bb0bb4ee8dd06bd863ac35dece1a0f50fc7e6798da5aa3fabd7be0006f6
 6fe8e18ed6f68e6c6529375d6ee0b37c62a4230970800651bcd2a6a1c1ab69c8026696670472f802699f30858f903ee4c044
 68b346df0b227c29f75105e94d58024d573a8ad7eb8b3066cb3c0315ee871370f965dc67c9ab615fcc69aace42bc57aa2f50
 f8ec46a82dca4c0198a37b403208f4e9921e0ffe83d1e5bc5640f8db02250f1841c0acb9436615f0daf5fedc50bdf6c374e6
 f3e9bb1f56c0b865ad329dc1742d18a1c41ae6d0983b021654913262015eda33024ea65ad224c68d941843a33f5404e891d3
 024c5e554e432688b034cfe0a06530e64c2f6fe254a725f2cc308a647de42bbc4e3305fc95d3bb8b57fbc45a044af2394bb9
 02b2b2b4410dc307ba58fcbdc36c07e46ff4fb414eadbcdaf7482d6f4bb9d7f74bb3d223d11c6940a702da98f2b84b9721b4
 04d5c5fd50487ae9c14d00cad2d55d54fa589f01f4642846a1dfa0da4537c5af9c255b31193233d89863ea5419b243fba439
 a3427fd4ad7409b439aac841744e057d409aff7236dcf0ba0c97d69de66928f01e5947c68c6907f28136a4cf2d0a721eea3f
 dc246692518cb27fe1e8c0b3082f981ba5fc88cf77e268668b1914561ac108c30e60309dbf3ee9f53cb38c98bca17c8eeea9
 6aee71bdd6284aef16970e5ba2bd93ada3b0347da37c58b467c63b3a242c3d2346af81f846377fc433622b3cfa8d8660f87e
 23b536f7aeb9b5a9f55ac3d9bac42ff2a4048bea808c476e9233edc243354145263b8bd70f657438bb1f2725ea2a22a459e6
 39a874a89a96862c83743b1fb811bcc445d2ec795ec7d1224638b214d779b45f6ea156c1961bf219e4169027c62b409fb672
 24f2206e5576c8bf9a5f7bd0e84939dea4d29b3f9c28b9c5d6bb2d55213bda0dbcbdc56768eefcedc6fd18973b10ce651c0d
 9e04f6179c772986e6f67a403b0bda599ab7d440c1a0a67307a48dc9c034481d3cda3bbf23ad1b2020dff97c4ae829340bac
 849e66f988d263cad609945c46f0ab20a5473a1717e54aac20805c5a64f2af5512a898d9cab82522bc5e8fa475d764cbfc6f
 ef6e5ee355ba422b14f5e71857e65ec9c3aa92ad4149ee0545162255ca15a138d11a7009e86caeb4b649b32350795af7073f
 c1eb7abff4b1d4c4bb277c75c934e8482c38312394558f081926afcef9d869567de513439945798b71e2097d76cad3ebd26b
 4e85f02c8f88f6b5a63a90deaa49f272d94e6f7c04e1ad903145684003097e5212f874ed283b1acd9439e396c16739324670
 63da25508886f61aec9c44bc6b857b9834548da0406b9ccb4a40a10082bee7cb2e3c7731efcb4e208d39df8a2c73c25acf1d
 9f0f9672f8838c8db45e920d4aeb3f4cc9307e18ada37cb895e9ca9828819fe72341bb0d6137218416b229784b73523398f2
 0f5a526bd49a8d065351e004ed7cde529b74d0ad998dd2537e9d80612adb070b9e0929310bace82cf883e6bdb88a8e2a42fc
 98018529bbcd091b29428525482f45ad7208c98c570de1d7b3016485305a5138ab0a7f4bf23864114ce7c4255a46b4c43926
 be1e293f6e8047cf1198274c6ebdd36385109d87f746eba8466381af381054225ff349653c2b1e36bc442d595a37d940c5f4
 253487b7feb3a5c8098712bd4085b9c7385cfa60017d117e6d36772186b8825a1985652949ab2d8b00bbc2921c994c1d3a5a
 2747135eaa832185d2aaaf21434a377faec1c03ccfaf0dc8889270a43fe2b552d1f0728df5c64d37ea7a0fcb99e98c7606b4
 f76083555e9dea31a319f4c4d8ea33cee7ac1d85be4657c09667e9a1bbd8a0438c0eb679e86dccbc6f8342463ca2d0356ff1
 63073607ad201a5f5766693d089ca3bd9bf75db1097b2f8c239d5bf2ea5ec1ff0ed15d30eadbc47521fbb7d8979e69e16d87
 a296407e2bfe74a2bdd46c04f380d2dff244e60527453bd80b3e0abeaa162db0723e5981b768312247d97b2075038f63ae15
 96c61cda98ce6db8b855813179ecf2c15cf09646cc075bf61181ba5806e91b366aae3abc6d75b47320a943b8675ae57a8baf
 b81863085d4834c4149075211454bc7165d2dbc076833bdac2e54cfaf9235ec949e50730e4c8fad52842e67dd8d191c62adc
 718670b4b3a020603990e8e7a20711bc2aeae84f5a0c6d79860879dae8b49f787bbddb4082d643abc021d57f7c644f06028c
 3d3844e475c624287b45de5514b2a0b38bbd276e83e253c6afd45a0393151b7d7e50bbb0916e4d27760aa6229385c6363dd6
 3bb9c4b5a22fc6af4d6b9a3404595a0929f3107e071f5c4d292767af5aec9cd140a70c9df78a762e3758da49db0c84f972e5
 749faaf20272f4a72a31432d477c13e7d742c57b36f153c0f55c5e12d0dd72ae6d5e05e35a63440221afdc52157768f3ad4f
 be126f4bf55b747841c8f956a3a8226cd93a89eb8ad3f0cbb0f643c5ddf8a39fad03fb2ff9b7ac91dc86f64c978fd982e296
 7c0bbb8b714cb2581304270892806e7573060ede0ae87883f70665959ff5c052e433e831905e820a16a0093b05253b98d19e
 2c31e434e43af29216eccdd3c9a6c0e12fcd4e03036c1d3aa3096a092a5616996c2685e6ac8d1ec4d10f9199a337503141a5
 cd2025a8055861aae749ed885a3b0f15b4e513dedae9c33af5855337356e7e2ba09340a726edba58134cf361b526750853ee
 c6864eefb6f964619d1a8d08b69bdc71ef105d58376c0c93deb25ee037c48f70ac7269bb9b898edde47b6bdcc2a50fd51af3
 ec595aa7d3839d7d1a4766f1c6a130d448d42765d7d711a971f033b2eaad97b5821081be77fd68d808566492a9f07e8917e6
 795df5879c4de9b4b4c9a4eee900d87e791ee59dfc64a0472679f1e4516a05a0dd8b0a4075e3957ae52cf9758e367391f2bd
 d25331338e62484e4dbb3414f23bfe678e54839bc218e83aef7eb4b58920abd52eb4264aa830c2f54859103a04bd01686010
 bf602d7332231d504c81fa01d4da3eab884aafa86f6efb675b78e7b5ff4902bfea46fb3bb42ecde0cc9845cda95dec2a42f7
 85f168c1443aaf017258381c35023b382ff7ee99a16f68722d6db5729116e40473d7ff7de89d4f18c315050e871a8db74e6d
 f93efb46af589f1909d5507f9b09f9d9826f33a3210ee0c6916fb64ae072044db81c4dab8cfbca045e2d0c974e7a71dadf33
 b7f0e85543762f06d377a504e345944e6eaa11aade6cc7557d0215ab5605aaaae6eaacba24d87dfe237a60c36be320409ee7
 7073754d7559b28b7708d2d33e9a3d5af90c3fb65214b7cdd1ce8d5d5c1830ec72c8de8bad7cb07c369c6e7d955b9ff6f979
 7653cb539c0af5bdad7e4c3c42bfc2ad759a6adf3cdb52cb47759aaeb5146b813cd884f5c93aab3fd779ef9bb6fd875b1771
 ebf999766361a8a387ecc910518b38c6f0b82ba807285d6852b5073af314b4b6556b220f361ec3ef454469f52e9d4d236480
 7347c1928ba601d5b23467f0b3062eebba6222e42584299770153db295cba5f900b3ebca0608da2eadcdc0d59cff5ed5b972
 acbbc2359a7ce9ea445aef3ec1ce423a730315add23cbb9d655ae1cd3c047fc40f80c2f82fdf5e1935415ff307c2daa6bee2
 ec514dab0802393e47abf539edd458a9685496346ac092bc60e62c13c0eb0966d0474378f4609e570ebd2ccd7ed4447ad987
 5e524604b265dcd8c1966e33ad6572a77ba2ed7d152556dd0e56402b80499bf9038a0f7ce606e1240699681205b9ef89b1a3
 f88820467c0094d5e6278430cdee0b7e6629b73269970fc85fbc768b64f06b1f3fa7c29be6935173bed75c7f6a01a655a189
 2dfcea90d77b4b492735d4a63291243fdbd0a10c6e29f535d0ac7424c458e506b4d21cc0
 """.replace("\n", ""))
	from amaranth import *
	from amaranth.lib import data, wiring, memory, stream
	from amaranth.lib.stream import In, Out


	__all__ = ["Encoder", "Decoder"]


	class Encoder(wiring.Component):
	"""`Consistent Overhead Byte Stuffing <cobs>`_ encoder combined with a FIFO.

	The encoder accepts a stream of tokens, which can be either _data_ or _end_, and produces
	a stream of bytes. Input data tokens produce non-NUL output bytes with a maximum latency
	of 256 cycles; input end tokens produce NUL output bytes.

	Since COBS encoding requires up to 254 bytes of lookahead, and a COBS encoder will be usually
	combined with a FIFO (either at the input or at the output), this encoder is combined with
	a FIFO to use limited memory resources more efficiently. All but one byte of the internal FIFO
	will be filled with data in case of output back-pressure.

	The latency of the encoder depends on the type and value of input tokens. Input non-NUL data
	tokens do not immediately produce output bytes; rather, bytes corresponding to these tokens
	appear at the output only after: (a) an input end token, or (b) an input NUL data token, or
	(c) 255th consecutive non-NUL data token.

	.. _cobs: https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing
	"""

	i: In(stream.Signature(data.StructLayout({
	"data": 8,
	"end": 1
	})))
	o: Out(stream.Signature(8))

	def __init__(self, fifo_depth=256):
	if not (fifo_depth >= 256 and fifo_depth.bit_count() == 1):
	raise ValueError("COBS encoder requires a power-of-2 sized FIFO that is "
	"at least 256 bytes deep")
	self.fifo_depth = fifo_depth

	super().__init__()

	def elaborate(self, platform):
	m = Module()

	# This implementation improves resource use efficiency by merging the two memories that
	# would otherwise be necessary in a typical implementation: the FIFO for buffering packet
	# data, and the lookahead memory for COBS encoding. Specifically, it reuses the "empty"
	# space in the FIFO for storing bytes that follow a yet-unknown COBS overhead byte; this is
	# called "staging". Once the value of the overhead byte becomes known, the FIFO write
	# pointer is advanced simultaneously with the overhead byte being overwriten; this is
	# called "committing".

	m.submodules.data = data = memory.Memory(shape=8, depth=self.fifo_depth, init=[])
	w_port = data.write_port()
	r_port = data.read_port(transparent_for=(w_port,))

	w_addr = Signal.like(w_port.addr)
	r_addr = Signal.like(r_port.addr)
	empty = (w_addr == r_addr)
	full = (w_addr == r_addr - 1)

	def write(at, data):
	m.d.comb += w_port.addr.eq(at)
	m.d.comb += w_port.data.eq(data)
	m.d.comb += w_port.en.eq(1)

	staged = Signal(8, init=1)

	def stage(data):
	write(w_addr + staged, data)
	m.d.sync += staged.eq(staged + 1)

	def commit():
	write(w_addr, staged)
	m.d.sync += w_addr.eq(w_addr + staged)
	m.d.sync += staged.eq(1)

	with m.FSM():
	with m.State("Data"):
	with m.If(self.i.valid & ~full):
	with m.If(self.i.p.end):
	m.d.comb += self.i.ready.eq(1)
	commit()
	m.next = "End"
	with m.Elif(staged == 0xff):
	commit()
	with m.Elif(self.i.p.data == 0x00):
	m.d.comb += self.i.ready.eq(1)
	commit()
	with m.Else():
	m.d.comb += self.i.ready.eq(1)
	stage(self.i.p.data)

	with m.State("End"):
	write(w_addr, 0x00)
	m.d.sync += w_addr.eq(w_addr + 1)
	m.next = "Data"

	m.d.comb += self.o.valid.eq(~empty)
	m.d.comb += self.o.payload.eq(r_port.data)
	with m.If(self.o.valid & self.o.ready):
	m.d.comb += r_port.addr.eq(r_addr + 1)
	m.d.sync += r_addr.eq(r_addr + 1)
	with m.Else():
	m.d.comb += r_port.addr.eq(r_addr)

	return m


	class Decoder(wiring.Component):
	"""`Consistent Overhead Byte Stuffing <cobs>`_ decoder.

	Performs an inversion of the transformation done by :class:`Encoder` with a fixed 0 cycle
	latency.

	If invalid COBS data is encountered (namely: if a group header byte or data byte is NUL),
	the decoder transitions to an error state, signaled by the ``error`` output. This state is
	final, cleared only by a reset.

	.. _cobs: https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing
	"""

	i: In(stream.Signature(8))
	o: Out(stream.Signature(data.StructLayout({
	"data": 8,
	"end": 1
	})))
	error: Out(1)

	def elaborate(self, platform):
	m = Module()

	count = Signal(8)
	offset = Signal(8)

	with m.FSM():
	with m.State("Start"):
	m.d.comb += self.i.ready.eq(1)
	with m.If(self.i.valid & self.i.ready):
	m.d.sync += count.eq(1)
	with m.If(self.i.payload != 0x00):
	m.d.sync += offset.eq(self.i.payload)
	m.next = "Data"
	with m.Else():
	m.next = "Error"

	with m.State("Data"):
	m.d.comb += self.i.ready.eq(self.o.ready)
	with m.If(self.i.valid & self.i.ready):
	with m.If(offset == count):
	m.d.sync += count.eq(1)
	with m.If(self.i.payload == 0x00):
	m.d.comb += self.o.payload.end.eq(1)
	m.d.comb += self.o.valid.eq(1)
	m.next = "Start"
	with m.Else():
	m.d.comb += self.o.payload.data.eq(0x00)
	m.d.comb += self.o.valid.eq(offset != 0xff)
	m.d.sync += offset.eq(self.i.payload)
	with m.Else():
	m.d.sync += count.eq(count + 1)
	with m.If(self.i.payload != 0x00):
	m.d.comb += self.o.payload.data.eq(self.i.payload)
	m.d.comb += self.o.valid.eq(1)
	with m.Else():
	m.next = "Error"

	with m.State("Error"):
	m.d.comb += self.error.eq(1)

	return m