build: Initialize Python virtual environment and install project dependencies.
Some checks failed
WLED CI / wled_build (push) Has been cancelled

This commit is contained in:
2026-02-19 22:31:26 +08:00
parent ca1319462e
commit bb17574d0c
1668 changed files with 469578 additions and 0 deletions

View File

@@ -0,0 +1,110 @@
"""
#### ####
## #### ## ## #### ###### ##### ## ## #### ###### ## ##
## ## ## ### ## ## ## ## ## ## ## ## ## #### ##
## ## ## ###### ### ## ##### ## ## ## ## ##
## ## ## ## ### ## ## ## ## ## ## ## ## ##
#### #### ## ## #### ## ## ## ##### #### ## ######
Parsing made even more fun (and faster too)
Homepage:
http://construct.wikispaces.com (including online tutorial)
Typical usage:
>>> from construct import *
Hands-on example:
>>> from construct import *
>>> s = Struct("foo",
... UBInt8("a"),
... UBInt16("b"),
... )
>>> s.parse("\\x01\\x02\\x03")
Container(a = 1, b = 515)
>>> print s.parse("\\x01\\x02\\x03")
Container:
a = 1
b = 515
>>> s.build(Container(a = 1, b = 0x0203))
"\\x01\\x02\\x03"
"""
from .core import *
from .adapters import *
from .macros import *
from .debug import Probe, Debugger
#===============================================================================
# Metadata
#===============================================================================
__author__ = "tomer filiba (tomerfiliba [at] gmail.com)"
__maintainer__ = "Corbin Simpson <MostAwesomeDude@gmail.com>"
__version__ = "2.06"
#===============================================================================
# Shorthand expressions
#===============================================================================
Bits = BitField
Byte = UBInt8
Bytes = Field
Const = ConstAdapter
Tunnel = TunnelAdapter
Embed = Embedded
#===============================================================================
# Deprecated names
# Next scheduled name cleanout: 2.1
#===============================================================================
import functools, warnings
def deprecated(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
warnings.warn(
"This name is deprecated, use %s instead" % f.__name__,
DeprecationWarning, stacklevel=2)
return f(*args, **kwargs)
return wrapper
MetaBytes = deprecated(MetaField)
GreedyRepeater = deprecated(GreedyRange)
OptionalGreedyRepeater = deprecated(OptionalGreedyRange)
Repeater = deprecated(Range)
StrictRepeater = deprecated(Array)
MetaRepeater = deprecated(Array)
OneOfValidator = deprecated(OneOf)
NoneOfValidator = deprecated(NoneOf)
#===============================================================================
# exposed names
#===============================================================================
__all__ = [
'AdaptationError', 'Adapter', 'Alias', 'Aligned', 'AlignedStruct',
'Anchor', 'Array', 'ArrayError', 'BFloat32', 'BFloat64', 'Bit', 'BitField',
'BitIntegerAdapter', 'BitIntegerError', 'BitStruct', 'Bits', 'Bitwise',
'Buffered', 'Byte', 'Bytes', 'CString', 'CStringAdapter', 'Const',
'ConstAdapter', 'ConstError', 'Construct', 'ConstructError', 'Container',
'Debugger', 'Embed', 'Embedded', 'EmbeddedBitStruct', 'Enum', 'ExprAdapter',
'Field', 'FieldError', 'Flag', 'FlagsAdapter', 'FlagsContainer',
'FlagsEnum', 'FormatField', 'GreedyRange', 'GreedyRepeater',
'HexDumpAdapter', 'If', 'IfThenElse', 'IndexingAdapter', 'LFloat32',
'LFloat64', 'LazyBound', 'LengthValueAdapter', 'ListContainer',
'MappingAdapter', 'MappingError', 'MetaArray', 'MetaBytes', 'MetaField',
'MetaRepeater', 'NFloat32', 'NFloat64', 'Nibble', 'NoneOf',
'NoneOfValidator', 'Octet', 'OnDemand', 'OnDemandPointer', 'OneOf',
'OneOfValidator', 'OpenRange', 'Optional', 'OptionalGreedyRange',
'OptionalGreedyRepeater', 'PaddedStringAdapter', 'Padding',
'PaddingAdapter', 'PaddingError', 'PascalString', 'Pass', 'Peek',
'Pointer', 'PrefixedArray', 'Probe', 'Range', 'RangeError', 'Reconfig',
'Rename', 'RepeatUntil', 'Repeater', 'Restream', 'SBInt16', 'SBInt32',
'SBInt64', 'SBInt8', 'SLInt16', 'SLInt32', 'SLInt64', 'SLInt8', 'SNInt16',
'SNInt32', 'SNInt64', 'SNInt8', 'Select', 'SelectError', 'Sequence',
'SizeofError', 'SlicingAdapter', 'StaticField', 'StrictRepeater', 'String',
'StringAdapter', 'Struct', 'Subconstruct', 'Switch', 'SwitchError',
'SymmetricMapping', 'Terminator', 'TerminatorError', 'Tunnel',
'TunnelAdapter', 'UBInt16', 'UBInt32', 'UBInt64', 'UBInt8', 'ULInt16',
'ULInt32', 'ULInt64', 'ULInt8', 'UNInt16', 'UNInt32', 'UNInt64', 'UNInt8',
'Union', 'ValidationError', 'Validator', 'Value', "Magic",
]

View File

@@ -0,0 +1,470 @@
from .core import Adapter, AdaptationError, Pass
from .lib import int_to_bin, bin_to_int, swap_bytes
from .lib import FlagsContainer, HexString
from .lib.py3compat import BytesIO, decodebytes
#===============================================================================
# exceptions
#===============================================================================
class BitIntegerError(AdaptationError):
__slots__ = []
class MappingError(AdaptationError):
__slots__ = []
class ConstError(AdaptationError):
__slots__ = []
class ValidationError(AdaptationError):
__slots__ = []
class PaddingError(AdaptationError):
__slots__ = []
#===============================================================================
# adapters
#===============================================================================
class BitIntegerAdapter(Adapter):
"""
Adapter for bit-integers (converts bitstrings to integers, and vice versa).
See BitField.
Parameters:
* subcon - the subcon to adapt
* width - the size of the subcon, in bits
* swapped - whether to swap byte order (little endian/big endian).
default is False (big endian)
* signed - whether the value is signed (two's complement). the default
is False (unsigned)
* bytesize - number of bits per byte, used for byte-swapping (if swapped).
default is 8.
"""
__slots__ = ["width", "swapped", "signed", "bytesize"]
def __init__(self, subcon, width, swapped = False, signed = False,
bytesize = 8):
Adapter.__init__(self, subcon)
self.width = width
self.swapped = swapped
self.signed = signed
self.bytesize = bytesize
def _encode(self, obj, context):
if obj < 0 and not self.signed:
raise BitIntegerError("object is negative, but field is not signed",
obj)
obj2 = int_to_bin(obj, width = self.width)
if self.swapped:
obj2 = swap_bytes(obj2, bytesize = self.bytesize)
return obj2
def _decode(self, obj, context):
if self.swapped:
obj = swap_bytes(obj, bytesize = self.bytesize)
return bin_to_int(obj, signed = self.signed)
class MappingAdapter(Adapter):
"""
Adapter that maps objects to other objects.
See SymmetricMapping and Enum.
Parameters:
* subcon - the subcon to map
* decoding - the decoding (parsing) mapping (a dict)
* encoding - the encoding (building) mapping (a dict)
* decdefault - the default return value when the object is not found
in the decoding mapping. if no object is given, an exception is raised.
if `Pass` is used, the unmapped object will be passed as-is
* encdefault - the default return value when the object is not found
in the encoding mapping. if no object is given, an exception is raised.
if `Pass` is used, the unmapped object will be passed as-is
"""
__slots__ = ["encoding", "decoding", "encdefault", "decdefault"]
def __init__(self, subcon, decoding, encoding,
decdefault = NotImplemented, encdefault = NotImplemented):
Adapter.__init__(self, subcon)
self.decoding = decoding
self.encoding = encoding
self.decdefault = decdefault
self.encdefault = encdefault
def _encode(self, obj, context):
try:
return self.encoding[obj]
except (KeyError, TypeError):
if self.encdefault is NotImplemented:
raise MappingError("no encoding mapping for %r [%s]" % (
obj, self.subcon.name))
if self.encdefault is Pass:
return obj
return self.encdefault
def _decode(self, obj, context):
try:
return self.decoding[obj]
except (KeyError, TypeError):
if self.decdefault is NotImplemented:
raise MappingError("no decoding mapping for %r [%s]" % (
obj, self.subcon.name))
if self.decdefault is Pass:
return obj
return self.decdefault
class FlagsAdapter(Adapter):
"""
Adapter for flag fields. Each flag is extracted from the number, resulting
in a FlagsContainer object. Not intended for direct usage.
See FlagsEnum.
Parameters
* subcon - the subcon to extract
* flags - a dictionary mapping flag-names to their value
"""
__slots__ = ["flags"]
def __init__(self, subcon, flags):
Adapter.__init__(self, subcon)
self.flags = flags
def _encode(self, obj, context):
flags = 0
for name, value in self.flags.items():
if getattr(obj, name, False):
flags |= value
return flags
def _decode(self, obj, context):
obj2 = FlagsContainer()
for name, value in self.flags.items():
setattr(obj2, name, bool(obj & value))
return obj2
class StringAdapter(Adapter):
"""
Adapter for strings. Converts a sequence of characters into a python
string, and optionally handles character encoding.
See String.
Parameters:
* subcon - the subcon to convert
* encoding - the character encoding name (e.g., "utf8"), or None to
return raw bytes (usually 8-bit ASCII).
"""
__slots__ = ["encoding"]
def __init__(self, subcon, encoding = None):
Adapter.__init__(self, subcon)
self.encoding = encoding
def _encode(self, obj, context):
if self.encoding:
obj = obj.encode(self.encoding)
return obj
def _decode(self, obj, context):
if self.encoding:
obj = obj.decode(self.encoding)
return obj
class PaddedStringAdapter(Adapter):
r"""
Adapter for padded strings.
See String.
Parameters:
* subcon - the subcon to adapt
* padchar - the padding character. default is b"\x00".
* paddir - the direction where padding is placed ("right", "left", or
"center"). the default is "right".
* trimdir - the direction where trimming will take place ("right" or
"left"). the default is "right". trimming is only meaningful for
building, when the given string is too long.
"""
__slots__ = ["padchar", "paddir", "trimdir"]
def __init__(self, subcon, padchar = b"\x00", paddir = "right",
trimdir = "right"):
if paddir not in ("right", "left", "center"):
raise ValueError("paddir must be 'right', 'left' or 'center'",
paddir)
if trimdir not in ("right", "left"):
raise ValueError("trimdir must be 'right' or 'left'", trimdir)
Adapter.__init__(self, subcon)
self.padchar = padchar
self.paddir = paddir
self.trimdir = trimdir
def _decode(self, obj, context):
if self.paddir == "right":
obj = obj.rstrip(self.padchar)
elif self.paddir == "left":
obj = obj.lstrip(self.padchar)
else:
obj = obj.strip(self.padchar)
return obj
def _encode(self, obj, context):
size = self._sizeof(context)
if self.paddir == "right":
obj = obj.ljust(size, self.padchar)
elif self.paddir == "left":
obj = obj.rjust(size, self.padchar)
else:
obj = obj.center(size, self.padchar)
if len(obj) > size:
if self.trimdir == "right":
obj = obj[:size]
else:
obj = obj[-size:]
return obj
class LengthValueAdapter(Adapter):
"""
Adapter for length-value pairs. It extracts only the value from the
pair, and calculates the length based on the value.
See PrefixedArray and PascalString.
Parameters:
* subcon - the subcon returning a length-value pair
"""
__slots__ = []
def _encode(self, obj, context):
return (len(obj), obj)
def _decode(self, obj, context):
return obj[1]
class CStringAdapter(StringAdapter):
r"""
Adapter for C-style strings (strings terminated by a terminator char).
Parameters:
* subcon - the subcon to convert
* terminators - a sequence of terminator chars. default is b"\x00".
* encoding - the character encoding to use (e.g., "utf8"), or None to
return raw-bytes. the terminator characters are not affected by the
encoding.
"""
__slots__ = ["terminators"]
def __init__(self, subcon, terminators = b"\x00", encoding = None):
StringAdapter.__init__(self, subcon, encoding = encoding)
self.terminators = terminators
def _encode(self, obj, context):
return StringAdapter._encode(self, obj, context) + self.terminators[0:1]
def _decode(self, obj, context):
return StringAdapter._decode(self, b''.join(obj[:-1]), context)
class TunnelAdapter(Adapter):
"""
Adapter for tunneling (as in protocol tunneling). A tunnel is construct
nested upon another (layering). For parsing, the lower layer first parses
the data (note: it must return a string!), then the upper layer is called
to parse that data (bottom-up). For building it works in a top-down manner;
first the upper layer builds the data, then the lower layer takes it and
writes it to the stream.
Parameters:
* subcon - the lower layer subcon
* inner_subcon - the upper layer (tunneled/nested) subcon
Example:
# a pascal string containing compressed data (zlib encoding), so first
# the string is read, decompressed, and finally re-parsed as an array
# of UBInt16
TunnelAdapter(
PascalString("data", encoding = "zlib"),
GreedyRange(UBInt16("elements"))
)
"""
__slots__ = ["inner_subcon"]
def __init__(self, subcon, inner_subcon):
Adapter.__init__(self, subcon)
self.inner_subcon = inner_subcon
def _decode(self, obj, context):
return self.inner_subcon._parse(BytesIO(obj), context)
def _encode(self, obj, context):
stream = BytesIO()
self.inner_subcon._build(obj, stream, context)
return stream.getvalue()
class ExprAdapter(Adapter):
"""
A generic adapter that accepts 'encoder' and 'decoder' as parameters. You
can use ExprAdapter instead of writing a full-blown class when only a
simple expression is needed.
Parameters:
* subcon - the subcon to adapt
* encoder - a function that takes (obj, context) and returns an encoded
version of obj
* decoder - a function that takes (obj, context) and returns a decoded
version of obj
Example:
ExprAdapter(UBInt8("foo"),
encoder = lambda obj, ctx: obj / 4,
decoder = lambda obj, ctx: obj * 4,
)
"""
__slots__ = ["_encode", "_decode"]
def __init__(self, subcon, encoder, decoder):
Adapter.__init__(self, subcon)
self._encode = encoder
self._decode = decoder
class HexDumpAdapter(Adapter):
"""
Adapter for hex-dumping strings. It returns a HexString, which is a string
"""
__slots__ = ["linesize"]
def __init__(self, subcon, linesize = 16):
Adapter.__init__(self, subcon)
self.linesize = linesize
def _encode(self, obj, context):
return obj
def _decode(self, obj, context):
return HexString(obj, linesize = self.linesize)
class ConstAdapter(Adapter):
"""
Adapter for enforcing a constant value ("magic numbers"). When decoding,
the return value is checked; when building, the value is substituted in.
Parameters:
* subcon - the subcon to validate
* value - the expected value
Example:
Const(Field("signature", 2), "MZ")
"""
__slots__ = ["value"]
def __init__(self, subcon, value):
Adapter.__init__(self, subcon)
self.value = value
def _encode(self, obj, context):
if obj is None or obj == self.value:
return self.value
else:
raise ConstError("expected %r, found %r" % (self.value, obj))
def _decode(self, obj, context):
if obj != self.value:
raise ConstError("expected %r, found %r" % (self.value, obj))
return obj
class SlicingAdapter(Adapter):
"""
Adapter for slicing a list (getting a slice from that list)
Parameters:
* subcon - the subcon to slice
* start - start index
* stop - stop index (or None for up-to-end)
* step - step (or None for every element)
"""
__slots__ = ["start", "stop", "step"]
def __init__(self, subcon, start, stop = None):
Adapter.__init__(self, subcon)
self.start = start
self.stop = stop
def _encode(self, obj, context):
if self.start is None:
return obj
return [None] * self.start + obj
def _decode(self, obj, context):
return obj[self.start:self.stop]
class IndexingAdapter(Adapter):
"""
Adapter for indexing a list (getting a single item from that list)
Parameters:
* subcon - the subcon to index
* index - the index of the list to get
"""
__slots__ = ["index"]
def __init__(self, subcon, index):
Adapter.__init__(self, subcon)
if type(index) is not int:
raise TypeError("index must be an integer", type(index))
self.index = index
def _encode(self, obj, context):
return [None] * self.index + [obj]
def _decode(self, obj, context):
return obj[self.index]
class PaddingAdapter(Adapter):
r"""
Adapter for padding.
Parameters:
* subcon - the subcon to pad
* pattern - the padding pattern (character as byte). default is b"\x00"
* strict - whether or not to verify, during parsing, that the given
padding matches the padding pattern. default is False (unstrict)
"""
__slots__ = ["pattern", "strict"]
def __init__(self, subcon, pattern = b"\x00", strict = False):
Adapter.__init__(self, subcon)
self.pattern = pattern
self.strict = strict
def _encode(self, obj, context):
return self._sizeof(context) * self.pattern
def _decode(self, obj, context):
if self.strict:
expected = self._sizeof(context) * self.pattern
if obj != expected:
raise PaddingError("expected %r, found %r" % (expected, obj))
return obj
#===============================================================================
# validators
#===============================================================================
class Validator(Adapter):
"""
Abstract class: validates a condition on the encoded/decoded object.
Override _validate(obj, context) in deriving classes.
Parameters:
* subcon - the subcon to validate
"""
__slots__ = []
def _decode(self, obj, context):
if not self._validate(obj, context):
raise ValidationError("invalid object", obj)
return obj
def _encode(self, obj, context):
return self._decode(obj, context)
def _validate(self, obj, context):
raise NotImplementedError()
class OneOf(Validator):
"""
Validates that the object is one of the listed values.
:param ``Construct`` subcon: object to validate
:param iterable valids: a set of valid values
>>> OneOf(UBInt8("foo"), [4,5,6,7]).parse("\\x05")
5
>>> OneOf(UBInt8("foo"), [4,5,6,7]).parse("\\x08")
Traceback (most recent call last):
...
construct.core.ValidationError: ('invalid object', 8)
>>>
>>> OneOf(UBInt8("foo"), [4,5,6,7]).build(5)
'\\x05'
>>> OneOf(UBInt8("foo"), [4,5,6,7]).build(9)
Traceback (most recent call last):
...
construct.core.ValidationError: ('invalid object', 9)
"""
__slots__ = ["valids"]
def __init__(self, subcon, valids):
Validator.__init__(self, subcon)
self.valids = valids
def _validate(self, obj, context):
return obj in self.valids
class NoneOf(Validator):
"""
Validates that the object is none of the listed values.
:param ``Construct`` subcon: object to validate
:param iterable invalids: a set of invalid values
>>> NoneOf(UBInt8("foo"), [4,5,6,7]).parse("\\x08")
8
>>> NoneOf(UBInt8("foo"), [4,5,6,7]).parse("\\x06")
Traceback (most recent call last):
...
construct.core.ValidationError: ('invalid object', 6)
"""
__slots__ = ["invalids"]
def __init__(self, subcon, invalids):
Validator.__init__(self, subcon)
self.invalids = invalids
def _validate(self, obj, context):
return obj not in self.invalids

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,133 @@
"""
Debugging utilities for constructs
"""
from __future__ import print_function
import sys
import traceback
import pdb
import inspect
from .core import Construct, Subconstruct
from .lib import HexString, Container, ListContainer
class Probe(Construct):
"""
A probe: dumps the context, stack frames, and stream content to the screen
to aid the debugging process.
See also Debugger.
Parameters:
* name - the display name
* show_stream - whether or not to show stream contents. default is True.
the stream must be seekable.
* show_context - whether or not to show the context. default is True.
* show_stack - whether or not to show the upper stack frames. default
is True.
* stream_lookahead - the number of bytes to dump when show_stack is set.
default is 100.
Example:
Struct("foo",
UBInt8("a"),
Probe("between a and b"),
UBInt8("b"),
)
"""
__slots__ = [
"printname", "show_stream", "show_context", "show_stack",
"stream_lookahead"
]
counter = 0
def __init__(self, name = None, show_stream = True,
show_context = True, show_stack = True,
stream_lookahead = 100):
Construct.__init__(self, None)
if name is None:
Probe.counter += 1
name = "<unnamed %d>" % (Probe.counter,)
self.printname = name
self.show_stream = show_stream
self.show_context = show_context
self.show_stack = show_stack
self.stream_lookahead = stream_lookahead
def __repr__(self):
return "%s(%r)" % (self.__class__.__name__, self.printname)
def _parse(self, stream, context):
self.printout(stream, context)
def _build(self, obj, stream, context):
self.printout(stream, context)
def _sizeof(self, context):
return 0
def printout(self, stream, context):
obj = Container()
if self.show_stream:
obj.stream_position = stream.tell()
follows = stream.read(self.stream_lookahead)
if not follows:
obj.following_stream_data = "EOF reached"
else:
stream.seek(-len(follows), 1)
obj.following_stream_data = HexString(follows)
print
if self.show_context:
obj.context = context
if self.show_stack:
obj.stack = ListContainer()
frames = [s[0] for s in inspect.stack()][1:-1]
frames.reverse()
for f in frames:
a = Container()
a.__update__(f.f_locals)
obj.stack.append(a)
print("=" * 80)
print("Probe", self.printname)
print(obj)
print("=" * 80)
class Debugger(Subconstruct):
"""
A pdb-based debugger. When an exception occurs in the subcon, a debugger
will appear and allow you to debug the error (and even fix on-the-fly).
Parameters:
* subcon - the subcon to debug
Example:
Debugger(
Enum(UBInt8("foo"),
a = 1,
b = 2,
c = 3
)
)
"""
__slots__ = ["retval"]
def _parse(self, stream, context):
try:
return self.subcon._parse(stream, context)
except Exception:
self.retval = NotImplemented
self.handle_exc("(you can set the value of 'self.retval', "
"which will be returned)")
if self.retval is NotImplemented:
raise
else:
return self.retval
def _build(self, obj, stream, context):
try:
self.subcon._build(obj, stream, context)
except Exception:
self.handle_exc()
def handle_exc(self, msg = None):
print("=" * 80)
print("Debugging exception of %s:" % (self.subcon,))
print("".join(traceback.format_exception(*sys.exc_info())[1:]))
if msg:
print(msg)
pdb.post_mortem(sys.exc_info()[2])
print("=" * 80)

View File

@@ -0,0 +1,7 @@
from .binary import (
int_to_bin, bin_to_int, swap_bytes, encode_bin, decode_bin)
from .bitstream import BitStreamReader, BitStreamWriter
from .container import (Container, FlagsContainer, ListContainer,
LazyContainer)
from .hex import HexString, hexdump

View File

@@ -0,0 +1,117 @@
from .py3compat import int2byte
def int_to_bin(number, width=32):
r"""
Convert an integer into its binary representation in a bytes object.
Width is the amount of bits to generate. If width is larger than the actual
amount of bits required to represent number in binary, sign-extension is
used. If it's smaller, the representation is trimmed to width bits.
Each "bit" is either '\x00' or '\x01'. The MSBit is first.
Examples:
>>> int_to_bin(19, 5)
b'\x01\x00\x00\x01\x01'
>>> int_to_bin(19, 8)
b'\x00\x00\x00\x01\x00\x00\x01\x01'
"""
if number < 0:
number += 1 << width
i = width - 1
bits = bytearray(width)
while number and i >= 0:
bits[i] = number & 1
number >>= 1
i -= 1
return bytes(bits)
_bit_values = {
0: 0,
1: 1,
48: 0, # '0'
49: 1, # '1'
# The following are for Python 2, in which iteration over a bytes object
# yields single-character bytes and not integers.
'\x00': 0,
'\x01': 1,
'0': 0,
'1': 1,
}
def bin_to_int(bits, signed=False):
r"""
Logical opposite of int_to_bin. Both '0' and '\x00' are considered zero,
and both '1' and '\x01' are considered one. Set sign to True to interpret
the number as a 2-s complement signed integer.
"""
number = 0
bias = 0
ptr = 0
if signed and _bit_values[bits[0]] == 1:
bits = bits[1:]
bias = 1 << len(bits)
for b in bits:
number <<= 1
number |= _bit_values[b]
return number - bias
def swap_bytes(bits, bytesize=8):
r"""
Bits is a b'' object containing a binary representation. Assuming each
bytesize bits constitute a bytes, perform a endianness byte swap. Example:
>>> swap_bytes(b'00011011', 2)
b'11100100'
"""
i = 0
l = len(bits)
output = [b""] * ((l // bytesize) + 1)
j = len(output) - 1
while i < l:
output[j] = bits[i : i + bytesize]
i += bytesize
j -= 1
return b"".join(output)
_char_to_bin = {}
_bin_to_char = {}
for i in range(256):
ch = int2byte(i)
bin = int_to_bin(i, 8)
# Populate with for both keys i and ch, to support Python 2 & 3
_char_to_bin[ch] = bin
_char_to_bin[i] = bin
_bin_to_char[bin] = ch
def encode_bin(data):
"""
Create a binary representation of the given b'' object. Assume 8-bit
ASCII. Example:
>>> encode_bin('ab')
b"\x00\x01\x01\x00\x00\x00\x00\x01\x00\x01\x01\x00\x00\x00\x01\x00"
"""
return b"".join(_char_to_bin[ch] for ch in data)
def decode_bin(data):
"""
Locical opposite of decode_bin.
"""
if len(data) & 7:
raise ValueError("Data length must be a multiple of 8")
i = 0
j = 0
l = len(data) // 8
chars = [b""] * l
while j < l:
chars[j] = _bin_to_char[data[i:i+8]]
i += 8
j += 1
return b"".join(chars)

View File

@@ -0,0 +1,77 @@
from .binary import encode_bin, decode_bin
class BitStreamReader(object):
__slots__ = ["substream", "buffer", "total_size"]
def __init__(self, substream):
self.substream = substream
self.total_size = 0
self.buffer = ""
def close(self):
if self.total_size % 8 != 0:
raise ValueError("total size of read data must be a multiple of 8",
self.total_size)
def tell(self):
return self.substream.tell()
def seek(self, pos, whence = 0):
self.buffer = ""
self.total_size = 0
self.substream.seek(pos, whence)
def read(self, count):
if count < 0:
raise ValueError("count cannot be negative")
l = len(self.buffer)
if count == 0:
data = ""
elif count <= l:
data = self.buffer[:count]
self.buffer = self.buffer[count:]
else:
data = self.buffer
count -= l
bytes = count // 8
if count & 7:
bytes += 1
buf = encode_bin(self.substream.read(bytes))
data += buf[:count]
self.buffer = buf[count:]
self.total_size += len(data)
return data
class BitStreamWriter(object):
__slots__ = ["substream", "buffer", "pos"]
def __init__(self, substream):
self.substream = substream
self.buffer = []
self.pos = 0
def close(self):
self.flush()
def flush(self):
bytes = decode_bin("".join(self.buffer))
self.substream.write(bytes)
self.buffer = []
self.pos = 0
def tell(self):
return self.substream.tell() + self.pos // 8
def seek(self, pos, whence = 0):
self.flush()
self.substream.seek(pos, whence)
def write(self, data):
if not data:
return
if type(data) is not str:
raise TypeError("data must be a string, not %r" % (type(data),))
self.buffer.append(data)

View File

@@ -0,0 +1,161 @@
"""
Various containers.
"""
from pprint import pformat
from .py3compat import MutableMapping
def recursion_lock(retval, lock_name = "__recursion_lock__"):
def decorator(func):
def wrapper(self, *args, **kw):
if getattr(self, lock_name, False):
return retval
setattr(self, lock_name, True)
try:
return func(self, *args, **kw)
finally:
setattr(self, lock_name, False)
wrapper.__name__ = func.__name__
return wrapper
return decorator
class Container(MutableMapping):
"""
A generic container of attributes.
Containers are the common way to express parsed data.
"""
def __init__(self, **kw):
self.__dict__ = kw
# The core dictionary interface.
def __getitem__(self, name):
return self.__dict__[name]
def __delitem__(self, name):
del self.__dict__[name]
def __setitem__(self, name, value):
self.__dict__[name] = value
def keys(self):
return self.__dict__.keys()
def __len__(self):
return len(self.__dict__.keys())
# Extended dictionary interface.
def update(self, other):
self.__dict__.update(other)
__update__ = update
def __contains__(self, value):
return value in self.__dict__
# Rich comparisons.
def __eq__(self, other):
try:
return self.__dict__ == other.__dict__
except AttributeError:
return False
def __ne__(self, other):
return not self == other
# Copy interface.
def copy(self):
return self.__class__(**self.__dict__)
__copy__ = copy
# Iterator interface.
def __iter__(self):
return iter(self.__dict__)
def __repr__(self):
return "%s(%s)" % (self.__class__.__name__, repr(self.__dict__))
def __str__(self):
return "%s(%s)" % (self.__class__.__name__, str(self.__dict__))
class FlagsContainer(Container):
"""
A container providing pretty-printing for flags.
Only set flags are displayed.
"""
@recursion_lock("<...>")
def __str__(self):
d = dict((k, self[k]) for k in self
if self[k] and not k.startswith("_"))
return "%s(%s)" % (self.__class__.__name__, pformat(d))
class ListContainer(list):
"""
A container for lists.
"""
__slots__ = ["__recursion_lock__"]
@recursion_lock("[...]")
def __str__(self):
return pformat(self)
class LazyContainer(object):
__slots__ = ["subcon", "stream", "pos", "context", "_value"]
def __init__(self, subcon, stream, pos, context):
self.subcon = subcon
self.stream = stream
self.pos = pos
self.context = context
self._value = NotImplemented
def __eq__(self, other):
try:
return self._value == other._value
except AttributeError:
return False
def __ne__(self, other):
return not (self == other)
def __str__(self):
return self.__pretty_str__()
def __pretty_str__(self, nesting = 1, indentation = " "):
if self._value is NotImplemented:
text = "<unread>"
elif hasattr(self._value, "__pretty_str__"):
text = self._value.__pretty_str__(nesting, indentation)
else:
text = str(self._value)
return "%s: %s" % (self.__class__.__name__, text)
def read(self):
self.stream.seek(self.pos)
return self.subcon._parse(self.stream, self.context)
def dispose(self):
self.subcon = None
self.stream = None
self.context = None
self.pos = None
def _get_value(self):
if self._value is NotImplemented:
self._value = self.read()
return self._value
value = property(_get_value)
has_value = property(lambda self: self._value is not NotImplemented)

View File

@@ -0,0 +1,43 @@
from .py3compat import byte2int, int2byte, bytes2str
# Map an integer in the inclusive range 0-255 to its string byte representation
_printable = dict((i, ".") for i in range(256))
_printable.update((i, bytes2str(int2byte(i))) for i in range(32, 128))
def hexdump(data, linesize):
"""
data is a bytes object. The returned result is a string.
"""
prettylines = []
if len(data) < 65536:
fmt = "%%04X %%-%ds %%s"
else:
fmt = "%%08X %%-%ds %%s"
fmt = fmt % (3 * linesize - 1,)
for i in range(0, len(data), linesize):
line = data[i : i + linesize]
hextext = " ".join('%02x' % byte2int(b) for b in line)
rawtext = "".join(_printable[byte2int(b)] for b in line)
prettylines.append(fmt % (i, str(hextext), str(rawtext)))
return prettylines
class HexString(bytes):
"""
Represents bytes that will be hex-dumped to a string when its string
representation is requested.
"""
def __init__(self, data, linesize = 16):
self.linesize = linesize
def __new__(cls, data, *args, **kwargs):
return bytes.__new__(cls, data)
def __str__(self):
if not self:
return "''"
sep = "\n"
return sep + sep.join(
hexdump(self, self.linesize))

View File

@@ -0,0 +1,74 @@
#-------------------------------------------------------------------------------
# py3compat.py
#
# Some Python2&3 compatibility code
#-------------------------------------------------------------------------------
import sys
PY3 = sys.version_info[0] == 3
try:
from collections.abc import MutableMapping # python >= 3.3
except ImportError:
from collections import MutableMapping # python < 3.3
if PY3:
import io
StringIO = io.StringIO
BytesIO = io.BytesIO
def bchr(i):
""" When iterating over b'...' in Python 2 you get single b'_' chars
and in Python 3 you get integers. Call bchr to always turn this
to single b'_' chars.
"""
return bytes((i,))
def u(s):
return s
def int2byte(i):
return bytes((i,))
def byte2int(b):
return b
def str2bytes(s):
return s.encode("latin-1")
def str2unicode(s):
return s
def bytes2str(b):
return b.decode('latin-1')
def decodebytes(b, encoding):
return bytes(b, encoding)
advance_iterator = next
else:
import cStringIO
StringIO = BytesIO = cStringIO.StringIO
int2byte = chr
byte2int = ord
bchr = lambda i: i
def u(s):
return unicode(s, "unicode_escape")
def str2bytes(s):
return s
def str2unicode(s):
return unicode(s, "unicode_escape")
def bytes2str(b):
return b
def decodebytes(b, encoding):
return b.decode(encoding)
def advance_iterator(it):
return it.next()

View File

@@ -0,0 +1,634 @@
from .lib.py3compat import int2byte
from .lib import (BitStreamReader, BitStreamWriter, encode_bin,
decode_bin)
from .core import (Struct, MetaField, StaticField, FormatField,
OnDemand, Pointer, Switch, Value, RepeatUntil, MetaArray, Sequence, Range,
Select, Pass, SizeofError, Buffered, Restream, Reconfig)
from .adapters import (BitIntegerAdapter, PaddingAdapter,
ConstAdapter, CStringAdapter, LengthValueAdapter, IndexingAdapter,
PaddedStringAdapter, FlagsAdapter, StringAdapter, MappingAdapter)
#===============================================================================
# fields
#===============================================================================
def Field(name, length):
"""
A field consisting of a specified number of bytes.
:param str name: the name of the field
:param length: the length of the field. the length can be either an integer
(StaticField), or a function that takes the context as an argument and
returns the length (MetaField)
"""
if callable(length):
return MetaField(name, length)
else:
return StaticField(name, length)
def BitField(name, length, swapped = False, signed = False, bytesize = 8):
"""
BitFields, as the name suggests, are fields that operate on raw, unaligned
bits, and therefore must be enclosed in a BitStruct. Using them is very
similar to all normal fields: they take a name and a length (in bits).
:param str name: name of the field
:param int length: number of bits in the field, or a function that takes
the context as its argument and returns the length
:param bool swapped: whether the value is byte-swapped
:param bool signed: whether the value is signed
:param int bytesize: number of bits per byte, for byte-swapping
>>> foo = BitStruct("foo",
... BitField("a", 3),
... Flag("b"),
... Padding(3),
... Nibble("c"),
... BitField("d", 5),
... )
>>> foo.parse("\\xe1\\x1f")
Container(a = 7, b = False, c = 8, d = 31)
>>> foo = BitStruct("foo",
... BitField("a", 3),
... Flag("b"),
... Padding(3),
... Nibble("c"),
... Struct("bar",
... Nibble("d"),
... Bit("e"),
... )
... )
>>> foo.parse("\\xe1\\x1f")
Container(a = 7, b = False, bar = Container(d = 15, e = 1), c = 8)
"""
return BitIntegerAdapter(Field(name, length),
length,
swapped=swapped,
signed=signed,
bytesize=bytesize
)
def Padding(length, pattern = b"\x00", strict = False):
r"""a padding field (value is discarded)
* length - the length of the field. the length can be either an integer,
or a function that takes the context as an argument and returns the
length
* pattern - the padding pattern (character/byte) to use. default is b"\x00"
* strict - whether or not to raise an exception is the actual padding
pattern mismatches the desired pattern. default is False.
"""
return PaddingAdapter(Field(None, length),
pattern = pattern,
strict = strict,
)
def Flag(name, truth = 1, falsehood = 0, default = False):
"""
A flag.
Flags are usually used to signify a Boolean value, and this construct
maps values onto the ``bool`` type.
.. note:: This construct works with both bit and byte contexts.
.. warning:: Flags default to False, not True. This is different from the
C and Python way of thinking about truth, and may be subject to change
in the future.
:param str name: field name
:param int truth: value of truth (default 1)
:param int falsehood: value of falsehood (default 0)
:param bool default: default value (default False)
"""
return SymmetricMapping(Field(name, 1),
{True : int2byte(truth), False : int2byte(falsehood)},
default = default,
)
#===============================================================================
# field shortcuts
#===============================================================================
def Bit(name):
"""a 1-bit BitField; must be enclosed in a BitStruct"""
return BitField(name, 1)
def Nibble(name):
"""a 4-bit BitField; must be enclosed in a BitStruct"""
return BitField(name, 4)
def Octet(name):
"""an 8-bit BitField; must be enclosed in a BitStruct"""
return BitField(name, 8)
def UBInt8(name):
"""unsigned, big endian 8-bit integer"""
return FormatField(name, ">", "B")
def UBInt16(name):
"""unsigned, big endian 16-bit integer"""
return FormatField(name, ">", "H")
def UBInt32(name):
"""unsigned, big endian 32-bit integer"""
return FormatField(name, ">", "L")
def UBInt64(name):
"""unsigned, big endian 64-bit integer"""
return FormatField(name, ">", "Q")
def SBInt8(name):
"""signed, big endian 8-bit integer"""
return FormatField(name, ">", "b")
def SBInt16(name):
"""signed, big endian 16-bit integer"""
return FormatField(name, ">", "h")
def SBInt32(name):
"""signed, big endian 32-bit integer"""
return FormatField(name, ">", "l")
def SBInt64(name):
"""signed, big endian 64-bit integer"""
return FormatField(name, ">", "q")
def ULInt8(name):
"""unsigned, little endian 8-bit integer"""
return FormatField(name, "<", "B")
def ULInt16(name):
"""unsigned, little endian 16-bit integer"""
return FormatField(name, "<", "H")
def ULInt32(name):
"""unsigned, little endian 32-bit integer"""
return FormatField(name, "<", "L")
def ULInt64(name):
"""unsigned, little endian 64-bit integer"""
return FormatField(name, "<", "Q")
def SLInt8(name):
"""signed, little endian 8-bit integer"""
return FormatField(name, "<", "b")
def SLInt16(name):
"""signed, little endian 16-bit integer"""
return FormatField(name, "<", "h")
def SLInt32(name):
"""signed, little endian 32-bit integer"""
return FormatField(name, "<", "l")
def SLInt64(name):
"""signed, little endian 64-bit integer"""
return FormatField(name, "<", "q")
def UNInt8(name):
"""unsigned, native endianity 8-bit integer"""
return FormatField(name, "=", "B")
def UNInt16(name):
"""unsigned, native endianity 16-bit integer"""
return FormatField(name, "=", "H")
def UNInt32(name):
"""unsigned, native endianity 32-bit integer"""
return FormatField(name, "=", "L")
def UNInt64(name):
"""unsigned, native endianity 64-bit integer"""
return FormatField(name, "=", "Q")
def SNInt8(name):
"""signed, native endianity 8-bit integer"""
return FormatField(name, "=", "b")
def SNInt16(name):
"""signed, native endianity 16-bit integer"""
return FormatField(name, "=", "h")
def SNInt32(name):
"""signed, native endianity 32-bit integer"""
return FormatField(name, "=", "l")
def SNInt64(name):
"""signed, native endianity 64-bit integer"""
return FormatField(name, "=", "q")
def BFloat32(name):
"""big endian, 32-bit IEEE floating point number"""
return FormatField(name, ">", "f")
def LFloat32(name):
"""little endian, 32-bit IEEE floating point number"""
return FormatField(name, "<", "f")
def NFloat32(name):
"""native endianity, 32-bit IEEE floating point number"""
return FormatField(name, "=", "f")
def BFloat64(name):
"""big endian, 64-bit IEEE floating point number"""
return FormatField(name, ">", "d")
def LFloat64(name):
"""little endian, 64-bit IEEE floating point number"""
return FormatField(name, "<", "d")
def NFloat64(name):
"""native endianity, 64-bit IEEE floating point number"""
return FormatField(name, "=", "d")
#===============================================================================
# arrays
#===============================================================================
def Array(count, subcon):
"""
Repeats the given unit a fixed number of times.
:param int count: number of times to repeat
:param ``Construct`` subcon: construct to repeat
>>> c = Array(4, UBInt8("foo"))
>>> c.parse("\\x01\\x02\\x03\\x04")
[1, 2, 3, 4]
>>> c.parse("\\x01\\x02\\x03\\x04\\x05\\x06")
[1, 2, 3, 4]
>>> c.build([5,6,7,8])
'\\x05\\x06\\x07\\x08'
>>> c.build([5,6,7,8,9])
Traceback (most recent call last):
...
construct.core.RangeError: expected 4..4, found 5
"""
if callable(count):
con = MetaArray(count, subcon)
else:
con = MetaArray(lambda ctx: count, subcon)
con._clear_flag(con.FLAG_DYNAMIC)
return con
def PrefixedArray(subcon, length_field = UBInt8("length")):
"""an array prefixed by a length field.
* subcon - the subcon to be repeated
* length_field - a construct returning an integer
"""
return LengthValueAdapter(
Sequence(subcon.name,
length_field,
Array(lambda ctx: ctx[length_field.name], subcon),
nested = False
)
)
def OpenRange(mincount, subcon):
from sys import maxsize
return Range(mincount, maxsize, subcon)
def GreedyRange(subcon):
"""
Repeats the given unit one or more times.
:param ``Construct`` subcon: construct to repeat
>>> from construct import GreedyRange, UBInt8
>>> c = GreedyRange(UBInt8("foo"))
>>> c.parse("\\x01")
[1]
>>> c.parse("\\x01\\x02\\x03")
[1, 2, 3]
>>> c.parse("\\x01\\x02\\x03\\x04\\x05\\x06")
[1, 2, 3, 4, 5, 6]
>>> c.parse("")
Traceback (most recent call last):
...
construct.core.RangeError: expected 1..2147483647, found 0
>>> c.build([1,2])
'\\x01\\x02'
>>> c.build([])
Traceback (most recent call last):
...
construct.core.RangeError: expected 1..2147483647, found 0
"""
return OpenRange(1, subcon)
def OptionalGreedyRange(subcon):
"""
Repeats the given unit zero or more times. This repeater can't
fail, as it accepts lists of any length.
:param ``Construct`` subcon: construct to repeat
>>> from construct import OptionalGreedyRange, UBInt8
>>> c = OptionalGreedyRange(UBInt8("foo"))
>>> c.parse("")
[]
>>> c.parse("\\x01\\x02")
[1, 2]
>>> c.build([])
''
>>> c.build([1,2])
'\\x01\\x02'
"""
return OpenRange(0, subcon)
#===============================================================================
# subconstructs
#===============================================================================
def Optional(subcon):
"""an optional construct. if parsing fails, returns None.
* subcon - the subcon to optionally parse or build
"""
return Select(subcon.name, subcon, Pass)
def Bitwise(subcon):
"""converts the stream to bits, and passes the bitstream to subcon
* subcon - a bitwise construct (usually BitField)
"""
# subcons larger than MAX_BUFFER will be wrapped by Restream instead
# of Buffered. implementation details, don't stick your nose in :)
MAX_BUFFER = 1024 * 8
def resizer(length):
if length & 7:
raise SizeofError("size must be a multiple of 8", length)
return length >> 3
if not subcon._is_flag(subcon.FLAG_DYNAMIC) and subcon.sizeof() < MAX_BUFFER:
con = Buffered(subcon,
encoder = decode_bin,
decoder = encode_bin,
resizer = resizer
)
else:
con = Restream(subcon,
stream_reader = BitStreamReader,
stream_writer = BitStreamWriter,
resizer = resizer)
return con
def Aligned(subcon, modulus = 4, pattern = b"\x00"):
r"""aligns subcon to modulus boundary using padding pattern
* subcon - the subcon to align
* modulus - the modulus boundary (default is 4)
* pattern - the padding pattern (default is \x00)
"""
if modulus < 2:
raise ValueError("modulus must be >= 2", modulus)
def padlength(ctx):
return (modulus - (subcon._sizeof(ctx) % modulus)) % modulus
return SeqOfOne(subcon.name,
subcon,
# ??????
# ??????
# ??????
# ??????
Padding(padlength, pattern = pattern),
nested = False,
)
def SeqOfOne(name, *args, **kw):
"""a sequence of one element. only the first element is meaningful, the
rest are discarded
* name - the name of the sequence
* args - subconstructs
* kw - any keyword arguments to Sequence
"""
return IndexingAdapter(Sequence(name, *args, **kw), index = 0)
def Embedded(subcon):
"""embeds a struct into the enclosing struct.
* subcon - the struct to embed
"""
return Reconfig(subcon.name, subcon, subcon.FLAG_EMBED)
def Rename(newname, subcon):
"""renames an existing construct
* newname - the new name
* subcon - the subcon to rename
"""
return Reconfig(newname, subcon)
def Alias(newname, oldname):
"""creates an alias for an existing element in a struct
* newname - the new name
* oldname - the name of an existing element
"""
return Value(newname, lambda ctx: ctx[oldname])
#===============================================================================
# mapping
#===============================================================================
def SymmetricMapping(subcon, mapping, default = NotImplemented):
"""defines a symmetrical mapping: a->b, b->a.
* subcon - the subcon to map
* mapping - the encoding mapping (a dict); the decoding mapping is
achieved by reversing this mapping
* default - the default value to use when no mapping is found. if no
default value is given, and exception is raised. setting to Pass would
return the value "as is" (unmapped)
"""
reversed_mapping = dict((v, k) for k, v in mapping.items())
return MappingAdapter(subcon,
encoding = mapping,
decoding = reversed_mapping,
encdefault = default,
decdefault = default,
)
def Enum(subcon, **kw):
"""a set of named values mapping.
* subcon - the subcon to map
* kw - keyword arguments which serve as the encoding mapping
* _default_ - an optional, keyword-only argument that specifies the
default value to use when the mapping is undefined. if not given,
and exception is raised when the mapping is undefined. use `Pass` to
pass the unmapped value as-is
"""
return SymmetricMapping(subcon, kw, kw.pop("_default_", NotImplemented))
def FlagsEnum(subcon, **kw):
"""a set of flag values mapping.
* subcon - the subcon to map
* kw - keyword arguments which serve as the encoding mapping
"""
return FlagsAdapter(subcon, kw)
#===============================================================================
# structs
#===============================================================================
def AlignedStruct(name, *subcons, **kw):
"""a struct of aligned fields
* name - the name of the struct
* subcons - the subcons that make up this structure
* kw - keyword arguments to pass to Aligned: 'modulus' and 'pattern'
"""
return Struct(name, *(Aligned(sc, **kw) for sc in subcons))
def BitStruct(name, *subcons):
"""a struct of bitwise fields
* name - the name of the struct
* subcons - the subcons that make up this structure
"""
return Bitwise(Struct(name, *subcons))
def EmbeddedBitStruct(*subcons):
"""an embedded BitStruct. no name is necessary.
* subcons - the subcons that make up this structure
"""
return Bitwise(Embedded(Struct(None, *subcons)))
#===============================================================================
# strings
#===============================================================================
def String(name, length, encoding=None, padchar=None, paddir="right",
trimdir="right"):
"""
A configurable, fixed-length string field.
The padding character must be specified for padding and trimming to work.
:param str name: name
:param int length: length, in bytes
:param str encoding: encoding (e.g. "utf8") or None for no encoding
:param str padchar: optional character to pad out strings
:param str paddir: direction to pad out strings; one of "right", "left",
or "both"
:param str trim: direction to trim strings; one of "right", "left"
>>> from construct import String
>>> String("foo", 5).parse("hello")
'hello'
>>>
>>> String("foo", 12, encoding = "utf8").parse("hello joh\\xd4\\x83n")
u'hello joh\\u0503n'
>>>
>>> foo = String("foo", 10, padchar = "X", paddir = "right")
>>> foo.parse("helloXXXXX")
'hello'
>>> foo.build("hello")
'helloXXXXX'
"""
con = StringAdapter(Field(name, length), encoding=encoding)
if padchar is not None:
con = PaddedStringAdapter(con, padchar=padchar, paddir=paddir,
trimdir=trimdir)
return con
def PascalString(name, length_field=UBInt8("length"), encoding=None):
"""
A length-prefixed string.
``PascalString`` is named after the string types of Pascal, which are
length-prefixed. Lisp strings also follow this convention.
The length field will appear in the same ``Container`` as the
``PascalString``, with the given name.
:param str name: name
:param ``Construct`` length_field: a field which will store the length of
the string
:param str encoding: encoding (e.g. "utf8") or None for no encoding
>>> foo = PascalString("foo")
>>> foo.parse("\\x05hello")
'hello'
>>> foo.build("hello world")
'\\x0bhello world'
>>>
>>> foo = PascalString("foo", length_field = UBInt16("length"))
>>> foo.parse("\\x00\\x05hello")
'hello'
>>> foo.build("hello")
'\\x00\\x05hello'
"""
return StringAdapter(
LengthValueAdapter(
Sequence(name,
length_field,
Field("data", lambda ctx: ctx[length_field.name]),
)
),
encoding=encoding,
)
def CString(name, terminators=b"\x00", encoding=None,
char_field=Field(None, 1)):
"""
A string ending in a terminator.
``CString`` is similar to the strings of C, C++, and other related
programming languages.
By default, the terminator is the NULL byte (b``0x00``).
:param str name: name
:param iterable terminators: sequence of valid terminators, in order of
preference
:param str encoding: encoding (e.g. "utf8") or None for no encoding
:param ``Construct`` char_field: construct representing a single character
>>> foo = CString("foo")
>>> foo.parse(b"hello\\x00")
b'hello'
>>> foo.build(b"hello")
b'hello\\x00'
>>> foo = CString("foo", terminators = b"XYZ")
>>> foo.parse(b"helloX")
b'hello'
>>> foo.parse(b"helloY")
b'hello'
>>> foo.parse(b"helloZ")
b'hello'
>>> foo.build(b"hello")
b'helloX'
"""
return Rename(name,
CStringAdapter(
RepeatUntil(lambda obj, ctx: obj in terminators, char_field),
terminators=terminators,
encoding=encoding,
)
)
#===============================================================================
# conditional
#===============================================================================
def IfThenElse(name, predicate, then_subcon, else_subcon):
"""an if-then-else conditional construct: if the predicate indicates True,
`then_subcon` will be used; otherwise `else_subcon`
* name - the name of the construct
* predicate - a function taking the context as an argument and returning
True or False
* then_subcon - the subcon that will be used if the predicate returns True
* else_subcon - the subcon that will be used if the predicate returns False
"""
return Switch(name, lambda ctx: bool(predicate(ctx)),
{
True : then_subcon,
False : else_subcon,
}
)
def If(predicate, subcon, elsevalue = None):
"""an if-then conditional construct: if the predicate indicates True,
subcon will be used; otherwise, `elsevalue` will be returned instead.
* predicate - a function taking the context as an argument and returning
True or False
* subcon - the subcon that will be used if the predicate returns True
* elsevalue - the value that will be used should the predicate return False.
by default this value is None.
"""
return IfThenElse(subcon.name,
predicate,
subcon,
Value("elsevalue", lambda ctx: elsevalue)
)
#===============================================================================
# misc
#===============================================================================
def OnDemandPointer(offsetfunc, subcon, force_build = True):
"""an on-demand pointer.
* offsetfunc - a function taking the context as an argument and returning
the absolute stream position
* subcon - the subcon that will be parsed from the `offsetfunc()` stream
position on demand
* force_build - see OnDemand. by default True.
"""
return OnDemand(Pointer(offsetfunc, subcon),
advance_stream = False,
force_build = force_build
)
def Magic(data):
return ConstAdapter(Field(None, len(data)), data)