How to get construct.GreedyRange to give back a Byte?

Question

Okay, suppose I have this working exactly as expected:

from enum import IntEnum
from contstruct import *

class Char(IntEnum):
    START = 0xAB
    STOP = 0xBC
    ESC = 0xCD

MAPPING = Mapping(Byte, {x: x+1 for x in Char})

SLIP = GreedyRange(
    Select(
        FocusedSeq(
            'x',
            Const(Char.ESC, Byte), 
            Renamed(MAPPING, 'x')
        ),
        Byte
    )
)

Example:

>>> buffer = bytes([0x00, 0xAB, 0xBC, 0xCD, 0xFF])
>>> SLIP.build(buffer)
b'\x00\xcd\xac\xcd\xbd\xcd\xce\xff’

And:

>>> from operator import eq
>>> all(map(eq, SLIP.parse(SLIP.build(buffer)), buffer))
True

Now I need to wrap the encode/decode inside another struct:

PROTOCOL = FocusedSeq(
    'message',
    Const(Char.START, Byte),
    Renamed(SLIP, 'message'),
    Const(Char.STOP, Byte)
)

The build works exactly as expected:

>>> PROTOCOL.build(buffer)
b'\xab\x00\xcd\xac\xcd\xbd\xcd\xce\xff\xbc'

However, parsing, GreedyRange consumes 1 too many bytes:

>>> PROTOCOL.parse(b'\xab\x00\xcd\xac\xcd\xbd\xcd\xce\xff\xbc')
construct.core.StreamError: stream read less than specified amount, expected 1, found 0

How can I get GreedyRange to give back a byte?

Answer 1

The solution to this is NullTerminated(..., term=STOP), which internally buffers the the underlying stream and gives back when necessary.

PROTOCOL = FocusedSeq(
    'message'
    Const(Char.START, Byte),
    message=NullTerminated(
        # NOTE build consumes entire stream and appends STOP
        # NOTE parse consumes steam until STOP and passes buffer to GreedyRange
        GreedyRange(
            Select(
                FocusedSeq(
                    'x',
                    Const(Char.ESC, Byte),
                    x=MAPPING  # NOTE intentionally raises MappingError
                ),
                Byte  # NOTE fallback for MappingError
            )
        ),
        term=Byte.build(Char.STOP)
    )
)

Answer 2

In your case, you could simply rearrange fields of PROTOCOL and put SLIP at the end.

PROTOCOL = FocusedSeq(
    'message',
    Const(Char.START, Byte),
    Const(Char.STOP, Byte),
    Renamed(SLIP, 'message')
)

This way GreedyRange will not consume all bytes that caused stream parsing error: construct.core.StreamError: stream read less than specified amount, expected 1, found 0.

Here is a modified sample:

from construct import Byte, Const, FocusedSeq, GreedyRange, Mapping, Renamed, Select
from enum import IntEnum


class Char(IntEnum):
    START = 0xAB
    STOP = 0xBC
    ESC = 0xCD

MAPPING = Mapping(Byte, {x: x+1 for x in Char})

SLIP = GreedyRange(
    Select(
        FocusedSeq(
            'x',
            Const(Char.ESC, Byte),
            Renamed(MAPPING, 'x')
        ),
        Byte
    )
)
buffer = bytes([0x00, 0xAB, 0xBC, 0xCD, 0xFF])

slip_build = SLIP.build(buffer)
assert slip_build == b'\x00\xcd\xac\xcd\xbd\xcd\xce\xff'
slip_parsed = SLIP.parse(b'\x00\xcd\xac\xcd\xbd\xcd\xce\xff')

PROTOCOL = FocusedSeq(
    'message',
    Const(Char.START, Byte),
    Const(Char.STOP, Byte),
    Renamed(SLIP, 'message')
)

protocol_build = PROTOCOL.build(buffer)
assert protocol_build == b'\xab\xbc\x00\xcd\xac\xcd\xbd\xcd\xce\xff'
protocol_parsed = PROTOCOL.parse(protocol_build)
assert protocol_parsed == slip_parsed

Answer 3

One more way to do that is to use construct Adapter class to modify the byte sequence.

Here is another code sample:

from construct import Byte, Const, FocusedSeq, GreedyRange, \
    If, Mapping, Renamed, Select, this, Adapter
from enum import IntEnum


class Char(IntEnum):
    START = 0xAB
    STOP = 0xBC
    ESC = 0xCD

MAPPING = Mapping(Byte, {x: x+1 for x in Char})

SLIP = GreedyRange(
    Select(
        FocusedSeq(
            'x',
            Const(Char.ESC, Byte),
            Renamed(MAPPING, 'x')
        ),
        Byte
    )
)
buffer = bytes([0x00, 0xAB, 0xBC, 0xCD, 0xFF])

slip_build = SLIP.build(buffer)
assert slip_build == b'\x00\xcd\xac\xcd\xbd\xcd\xce\xff'
slip_parsed = SLIP.parse(b'\x00\xcd\xac\xcd\xbd\xcd\xce\xff')


class ProtocolAdapter(Adapter):
    def _decode(self, obj, context, path):

        # remove first and last bite
        obj.pop(0)
        obj.pop(-1)
        return obj

    def _encode(self, obj, context, path):
        return obj

PROTOCOL = FocusedSeq(
    "message",
    If(this._building == True, Const(Char.START, Byte)),
    "message" / SLIP,
    If(this._building == True, Const(Char.STOP, Byte))

)
ADAPTED_PROTOCOL = ProtocolAdapter(PROTOCOL)

protocol_build = ADAPTED_PROTOCOL.build(buffer)
assert protocol_build == b'\xab\x00\xcd\xac\xcd\xbd\xcd\xce\xff\xbc'
protocol_parsed = ADAPTED_PROTOCOL.parse(protocol_build)
assert protocol_parsed == slip_parsed