"""Base class for errors in the bitstring module."""

def __init__(self, *params):

self.msg = params[0] if params else ''

self.params = params[1:]

def __str__(self):

if self.params:

return self.msg.format(*self.params)

"""Reading or peeking past the end of a bitstring."""

def __init__(self, *params):

"""Inappropriate interpretation of binary data."""

def __init__(self, *params):

"""Whole-byte position or length needed."""

def __init__(self, *params):

"""Inappropriate argument during bitstring creation."""

def __init__(self, *params):

"""Stores raw bytes together with a bit offset and length.

__slots__ = ('offset', '_rawarray', 'bitlength')

def __init__(self, data, bitlength=None, offset=None):

"""data is either a bytearray or a MmapByteArray"""

self._rawarray = data

if offset is None:

offset = 0

if bitlength is None:

bitlength = 8 * len(data) - offset

self.offset = offset

self.bitlength = bitlength

def getbit(self, pos):

assert 0 <= pos < self.bitlength

byte, bit = divmod(self.offset + pos, 8)

return bool(self._rawarray[byte] & (128 >> bit))

def getbyte(self, pos):

"""Direct access to byte data."""

return self._rawarray[pos]

def getbyteslice(self, start, end):

"""Direct access to byte data."""

c = self._rawarray[start:end]

return c

@property

def bytelength(self):

if not self.bitlength:

return 0

sb = self.offset // 8

eb = (self.offset + self.bitlength - 1) // 8

return eb - sb + 1

def __copy__(self):

return ByteStore(self._rawarray[:], self.bitlength, self.offset)

def _appendstore(self, store):

"""Join another store on to the end of this one."""

if not store.bitlength:

return

# Set new array offset to the number of bits in the final byte of current array.

store = offsetcopy(store, (self.offset + self.bitlength) % 8)

if store.offset:

# first do the byte with the join.

joinval = (self._rawarray.pop() & (255 ^ (255 >> store.offset)) |

(store.getbyte(0) & (255 >> store.offset)))

self._rawarray.append(joinval)

self._rawarray.extend(store._rawarray[1:])

else:

self._rawarray.extend(store._rawarray)

self.bitlength += store.bitlength

def _prependstore(self, store):

"""Join another store on to the start of this one."""

if not store.bitlength:

return

# Set the offset of copy of store so that it's final byte

# ends in a position that matches the offset of self,

# then join self on to the end of it.

store = offsetcopy(store, (self.offset - store.bitlength) % 8)

assert (store.offset + store.bitlength) % 8 == self.offset % 8

bit_offset = self.offset % 8

if bit_offset:

# first do the byte with the join.

store.setbyte(-1, (store.getbyte(-1) & (255 ^ (255 >> bit_offset)) | \

(self._rawarray[self.byteoffset] & (255 >> bit_offset))))

store._rawarray.extend(self._rawarray[self.byteoffset + 1: self.byteoffset + self.bytelength])

else:

store._rawarray.extend(self._rawarray[self.byteoffset: self.byteoffset + self.bytelength])

self._rawarray = store._rawarray

self.offset = store.offset

self.bitlength += store.bitlength

@property

def byteoffset(self):

return self.offset // 8

@property

def rawbytes(self):

"""Adding mutating methods to ConstByteStore

__slots__ = ()

def setbit(self, pos):

assert 0 <= pos < self.bitlength

byte, bit = divmod(self.offset + pos, 8)

self._rawarray[byte] |= (128 >> bit)

def unsetbit(self, pos):

assert 0 <= pos < self.bitlength

byte, bit = divmod(self.offset + pos, 8)

self._rawarray[byte] &= ~(128 >> bit)

def invertbit(self, pos):

assert 0 <= pos < self.bitlength

byte, bit = divmod(self.offset + pos, 8)

self._rawarray[byte] ^= (128 >> bit)

def setbyte(self, pos, value):

self._rawarray[pos] = value

def setbyteslice(self, start, end, value):

"""Return a copy of a ByteStore with the newoffset.

assert 0 <= newoffset < 8

if not s.bitlength:

return copy.copy(s)

else:

if newoffset == s.offset % 8:

return ByteStore(s.getbyteslice(s.byteoffset, s.byteoffset + s.bytelength), s.bitlength, newoffset)

newdata = []

d = s._rawarray

assert newoffset != s.offset % 8

if newoffset < s.offset % 8:

# We need to shift everything left

shiftleft = s.offset % 8 - newoffset

# First deal with everything except for the final byte

for x in range(s.byteoffset, s.byteoffset + s.bytelength - 1):

newdata.append(((d[x] << shiftleft) & 0xff) +\

(d[x + 1] >> (8 - shiftleft)))

bits_in_last_byte = (s.offset + s.bitlength) % 8

if not bits_in_last_byte:

bits_in_last_byte = 8

if bits_in_last_byte > shiftleft:

newdata.append((d[s.byteoffset + s.bytelength - 1] << shiftleft) & 0xff)

else: # newoffset > s._offset % 8

shiftright = newoffset - s.offset % 8

newdata.append(s.getbyte(0) >> shiftright)

for x in range(s.byteoffset + 1, s.byteoffset + s.bytelength):

newdata.append(((d[x - 1] << (8 - shiftright)) & 0xff) +\

(d[x] >> shiftright))

bits_in_last_byte = (s.offset + s.bitlength) % 8

if not bits_in_last_byte:

bits_in_last_byte = 8

if bits_in_last_byte + shiftright > 8:

newdata.append((d[s.byteoffset + s.bytelength - 1] << (8 - shiftright)) & 0xff)

new_s = ByteStore(bytearray(newdata), s.bitlength, newoffset)

assert new_s.offset == newoffset

"""Return True if ByteStores a == b.

# We want to return False for inequality as soon as possible, which

# means we get lots of special cases.

# First the easy one - compare lengths:

a_bitlength = a.bitlength

b_bitlength = b.bitlength

if a_bitlength != b_bitlength:

return False

if not a_bitlength:

assert b_bitlength == 0

return True

# Make 'a' the one with the smaller offset

if (a.offset % 8) > (b.offset % 8):

a, b = b, a

# and create some aliases

a_bitoff = a.offset % 8

b_bitoff = b.offset % 8

a_byteoffset = a.byteoffset

b_byteoffset = b.byteoffset

a_bytelength = a.bytelength

b_bytelength = b.bytelength

da = a._rawarray

db = b._rawarray

# If they are pointing to the same data, they must be equal

if da is db and a.offset == b.offset:

return True

if a_bitoff == b_bitoff:

bits_spare_in_last_byte = 8 - (a_bitoff + a_bitlength) % 8

if bits_spare_in_last_byte == 8:

bits_spare_in_last_byte = 0

# Special case for a, b contained in a single byte

if a_bytelength == 1:

a_val = ((da[a_byteoffset] << a_bitoff) & 0xff) >> (8 - a_bitlength)

b_val = ((db[b_byteoffset] << b_bitoff) & 0xff) >> (8 - b_bitlength)

return a_val == b_val

# Otherwise check first byte

if da[a_byteoffset] & (0xff >> a_bitoff) != db[b_byteoffset] & (0xff >> b_bitoff):

return False

# then everything up to the last

b_a_offset = b_byteoffset - a_byteoffset

for x in range(1 + a_byteoffset, a_byteoffset + a_bytelength - 1):

if da[x] != db[b_a_offset + x]:

return False

# and finally the last byte

return (da[a_byteoffset + a_bytelength - 1] >> bits_spare_in_last_byte ==

db[b_byteoffset + b_bytelength - 1] >> bits_spare_in_last_byte)

assert a_bitoff != b_bitoff

# This is how much we need to shift a to the right to compare with b:

shift = b_bitoff - a_bitoff

# Special case for b only one byte long

if b_bytelength == 1:

assert a_bytelength == 1

a_val = ((da[a_byteoffset] << a_bitoff) & 0xff) >> (8 - a_bitlength)

b_val = ((db[b_byteoffset] << b_bitoff) & 0xff) >> (8 - b_bitlength)

return a_val == b_val

# Special case for a only one byte long

if a_bytelength == 1:

assert b_bytelength == 2

a_val = ((da[a_byteoffset] << a_bitoff) & 0xff) >> (8 - a_bitlength)

b_val = ((db[b_byteoffset] << 8) + db[b_byteoffset + 1]) << b_bitoff

b_val &= 0xffff

b_val >>= 16 - b_bitlength

return a_val == b_val

# Compare first byte of b with bits from first byte of a

if (da[a_byteoffset] & (0xff >> a_bitoff)) >> shift != db[b_byteoffset] & (0xff >> b_bitoff):

return False

# Now compare every full byte of b with bits from 2 bytes of a

for x in range(1, b_bytelength - 1):

# Construct byte from 2 bytes in a to compare to byte in b

b_val = db[b_byteoffset + x]

a_val = ((da[a_byteoffset + x - 1] << 8) + da[a_byteoffset + x]) >> shift

a_val &= 0xff

if a_val != b_val:

return False

# Now check bits in final byte of b

final_b_bits = (b.offset + b_bitlength) % 8

if not final_b_bits:

final_b_bits = 8

b_val = db[b_byteoffset + b_bytelength - 1] >> (8 - final_b_bits)

final_a_bits = (a.offset + a_bitlength) % 8

if not final_a_bits:

final_a_bits = 8

if b.bytelength > a_bytelength:

assert b_bytelength == a_bytelength + 1

a_val = da[a_byteoffset + a_bytelength - 1] >> (8 - final_a_bits)

a_val &= 0xff >> (8 - final_b_bits)

return a_val == b_val

assert a_bytelength == b_bytelength

a_val = da[a_byteoffset + a_bytelength - 2] << 8

a_val += da[a_byteoffset + a_bytelength - 1]

a_val >>= (8 - final_a_bits)

a_val &= 0xff >> (8 - final_b_bits)

"""Looks like a bytearray, but from an mmap.

__slots__ = ('filemap', 'filelength', 'source', 'byteoffset', 'bytelength')

def __init__(self, source, bytelength=None, byteoffset=None):

self.source = source

source.seek(0, os.SEEK_END)

self.filelength = source.tell()

if byteoffset is None:

byteoffset = 0

if bytelength is None:

bytelength = self.filelength - byteoffset

self.byteoffset = byteoffset

self.bytelength = bytelength

self.filemap = mmap.mmap(source.fileno(), 0, access=mmap.ACCESS_READ)

def __getitem__(self, key):

try:

start = key.start

stop = key.stop

except AttributeError:

try:

assert 0 <= key < self.bytelength

return ord(self.filemap[key + self.byteoffset])

except TypeError:

# for Python 3

return self.filemap[key + self.byteoffset]

else:

if start is None:

start = 0

if stop is None:

stop = self.bytelength

assert key.step is None

assert 0 <= start < self.bytelength

assert 0 <= stop <= self.bytelength

s = slice(start + self.byteoffset, stop + self.byteoffset)

return bytearray(self.filemap.__getitem__(s))

def __len__(self):

'h': 'intbe:16', 'H': 'uintbe:16',

'l': 'intbe:32', 'L': 'uintbe:32',

'q': 'intbe:64', 'Q': 'uintbe:64',

'h': 'intle:16', 'H': 'uintle:16',

'l': 'intle:32', 'L': 'uintle:32',

'q': 'intle:64', 'Q': 'uintle:64',

"""Parse struct-like format string token into sub-token list."""

m = STRUCT_PACK_RE.match(token)

if not m:

return [token]

else:

endian = m.group('endian')

if endian is None:

return [token]

# Split the format string into a list of 'q', '4h' etc.

formatlist = re.findall(STRUCT_SPLIT_RE, m.group('fmt'))

# Now deal with mulitiplicative factors, 4h -> hhhh etc.

fmt = ''.join([f[-1] * int(f[:-1]) if len(f) != 1 else

f for f in formatlist])

if endian == '@':

# Native endianness

if byteorder == 'little':

endian = '<'

else:

assert byteorder == 'big'

endian = '>'

if endian == '<':

tokens = [REPLACEMENTS_LE[c] for c in fmt]

else:

assert endian == '>'

tokens = [REPLACEMENTS_BE[c] for c in fmt]

"""Divide the format string into tokens and parse them.

try:

return token_cache[(fmt, keys)]

except KeyError:

token_key = (fmt, keys)

# Very inefficient expanding of brackets.

fmt = expand_brackets(fmt)

# Split tokens by ',' and remove whitespace

# The meta_tokens can either be ordinary single tokens or multiple

# struct-format token strings.

meta_tokens = (''.join(f.split()) for f in fmt.split(','))

return_values = []

stretchy_token = False

for meta_token in meta_tokens:

# See if it has a multiplicative factor

m = MULTIPLICATIVE_RE.match(meta_token)

if not m:

factor = 1

else:

factor = int(m.group('factor'))

meta_token = m.group('token')

# See if it's a struct-like format

tokens = structparser(meta_token)

ret_vals = []

for token in tokens:

if keys and token in keys:

# Don't bother parsing it, it's a keyword argument

ret_vals.append([token, None, None])

continue

value = length = None

if token == '':

continue

# Match literal tokens of the form 0x... 0o... and 0b...

m = LITERAL_RE.match(token)

if m:

name = m.group('name')

value = m.group('value')

ret_vals.append([name, length, value])

continue

# Match everything else:

m1 = TOKEN_RE.match(token)

if not m1:

# and if you don't specify a 'name' then the default is 'uint':

m2 = DEFAULT_UINT.match(token)

if not m2:

raise ValueError("Don't understand token '{0}'.".format(token))

if m1:

name = m1.group('name')

length = m1.group('len')

if m1.group('value'):

value = m1.group('value')

else:

assert m2

name = 'uint'

length = m2.group('len')

if m2.group('value'):

value = m2.group('value')

if name == 'bool':

if length is not None:

raise ValueError("You can't specify a length with bool tokens - they are always one bit.")

length = 1

if length is None and name not in ('se', 'ue', 'sie', 'uie'):

stretchy_token = True

if length is not None:

# Try converting length to int, otherwise check it's a key.

try:

length = int(length)

if length < 0:

raise Error

# For the 'bytes' token convert length to bits.

if name == 'bytes':

length *= 8

except Error:

raise ValueError("Can't read a token with a negative length.")

except ValueError:

if not keys or length not in keys:

raise ValueError("Don't understand length '{0}' of token.".format(length))

ret_vals.append([name, length, value])

# This multiplies by the multiplicative factor, but this means that

# we can't allow keyword values as multipliers (e.g. n*uint:8).

# The only way to do this would be to return the factor in some fashion

# (we can't use the key's value here as it would mean that we couldn't

# sensibly continue to cache the function's results. (TODO).

return_values.extend(ret_vals * factor)

return_values = [tuple(x) for x in return_values]

if len(token_cache) < CACHE_SIZE:

token_cache[token_key] = stretchy_token, return_values

"""Remove whitespace and expand all brackets."""

s = ''.join(s.split())

while True:

start = s.find('(')

if start == -1:

break

count = 1 # Number of hanging open brackets

p = start + 1

while p < len(s):

if s[p] == '(':

count += 1

if s[p] == ')':

count -= 1

if not count:

break

p += 1

if count:

raise ValueError("Unbalanced parenthesis in '{0}'.".format(s))

if start == 0 or s[start - 1] != '*':

s = s[0:start] + s[start + 1:p] + s[p + 1:]

else:

m = BRACKET_RE.search(s)

if m:

factor = int(m.group('factor'))

matchstart = m.start('factor')

s = s[0:matchstart] + (factor - 1) * (s[start + 1:p] + ',') + s[start + 1:p] + s[p + 1:]

else:

raise ValueError("Failed to parse '{0}'.".format(s))

"""A container holding an immutable sequence of bits.

__slots__ = ('_datastore')

def __init__(self, auto=None, length=None, offset=None, **kwargs):

"""Either specify an 'auto' initialiser:

pass

def __new__(cls, auto=None, length=None, offset=None, _cache={}, **kwargs):

# For instances auto-initialised with a string we intern the

# instance for re-use.

try:

if isinstance(auto, basestring):

try:

return _cache[auto]

except KeyError:

x = object.__new__(Bits)

try:

_, tokens = tokenparser(auto)

except ValueError as e:

raise CreationError(*e.args)

x._datastore = ConstByteStore(bytearray(0), 0, 0)

for token in tokens:

x._datastore._appendstore(Bits._init_with_token(*token)._datastore)

assert x._assertsanity()

if len(_cache) < CACHE_SIZE:

_cache[auto] = x

return x

if type(auto) == Bits:

return auto

except TypeError:

pass

x = super(Bits, cls).__new__(cls)

x._initialise(auto, length, offset, **kwargs)

return x

def _initialise(self, auto, length, offset, **kwargs):

if length is not None and length < 0:

raise CreationError("bitstring length cannot be negative.")

if offset is not None and offset < 0:

raise CreationError("offset must be >= 0.")

if auto is not None:

self._initialise_from_auto(auto, length, offset)

return

if not kwargs:

# No initialisers, so initialise with nothing or zero bits

if length is not None and length != 0:

data = bytearray((length + 7) // 8)

self._setbytes_unsafe(data, length, 0)

return

self._setbytes_unsafe(bytearray(0), 0, 0)

return

k, v = kwargs.popitem()

try:

init_without_length_or_offset[k](self, v)

if length is not None or offset is not None:

raise CreationError("Cannot use length or offset with this initialiser.")

except KeyError:

try:

init_with_length_only[k](self, v, length)

if offset is not None:

raise CreationError("Cannot use offset with this initialiser.")

except KeyError:

if offset is None:

offset = 0

try:

init_with_length_and_offset[k](self, v, length, offset)

except KeyError:

raise CreationError("Unrecognised keyword '{0}' used to initialise.", k)

def _initialise_from_auto(self, auto, length, offset):

if offset is None:

offset = 0

self._setauto(auto, length, offset)

return

def __copy__(self):

"""Return a new copy of the Bits for the copy module."""

# Note that if you want a new copy (different ID), use _copy instead.

# The copy can return self as it's immutable.

return self

def __lt__(self, other):

raise TypeError("unorderable type: {0}".format(type(self).__name__))

def __gt__(self, other):

raise TypeError("unorderable type: {0}".format(type(self).__name__))

def __le__(self, other):

raise TypeError("unorderable type: {0}".format(type(self).__name__))

def __ge__(self, other):

raise TypeError("unorderable type: {0}".format(type(self).__name__))

def __add__(self, bs):

"""Concatenate bitstrings and return new bitstring.

bs = Bits(bs)

if bs.len <= self.len:

s = self._copy()

s._append(bs)

else:

s = bs._copy()

s = self.__class__(s)

s._prepend(self)

return s

def __radd__(self, bs):

"""Append current bitstring to bs and return new bitstring.

bs = self._converttobitstring(bs)

return bs.__add__(self)

def __getitem__(self, key):

"""Return a new bitstring representing a slice of the current bitstring.

length = self.len

try:

step = key.step if key.step is not None else 1

except AttributeError:

# single element

if key < 0:

key += length

if not 0 <= key < length:

raise IndexError("Slice index out of range.")

# Single bit, return True or False

return self._datastore.getbit(key)

else:

if step != 1:

# convert to binary string and use string slicing

bs = self.__class__()

bs._setbin_unsafe(self._getbin().__getitem__(key))

return bs

start, stop = 0, length

if key.start is not None:

start = key.start

if key.start < 0:

start += stop

if key.stop is not None:

stop = key.stop

if key.stop < 0:

stop += length

start = max(start, 0)

stop = min(stop, length)

if start < stop:

return self._slice(start, stop)

else:

return self.__class__()

def __len__(self):

"""Return the length of the bitstring in bits."""

return self._getlength()

def __str__(self):

"""Return approximate string representation of bitstring for printing.

length = self.len

if not length:

return ''

if length > MAX_CHARS * 4:

# Too long for hex. Truncate...

return ''.join(('0x', self._readhex(MAX_CHARS * 4, 0), '...'))

# If it's quite short and we can't do hex then use bin

if length < 32 and length % 4 != 0:

return '0b' + self.bin

# If we can use hex then do so

if not length % 4:

return '0x' + self.hex

# Otherwise first we do as much as we can in hex

# then add on 1, 2 or 3 bits on at the end

bits_at_end = length % 4

return ''.join(('0x', self._readhex(length - bits_at_end, 0),

', ', '0b',

self._readbin(bits_at_end, length - bits_at_end)))

def __repr__(self):

"""Return representation that could be used to recreate the bitstring.

length = self.len

if isinstance(self._datastore._rawarray, MmapByteArray):

offsetstring = ''

if self._datastore.byteoffset or self._offset:

offsetstring = ", offset=%d" % (self._datastore._rawarray.byteoffset * 8 + self._offset)

lengthstring = ", length=%d" % length

return "{0}(filename='{1}'{2}{3})".format(self.__class__.__name__,

self._datastore._rawarray.source.name, lengthstring, offsetstring)

else:

s = self.__str__()

lengthstring = ''

if s.endswith('...'):

lengthstring = " # length={0}".format(length)

return "{0}('{1}'){2}".format(self.__class__.__name__, s, lengthstring)

def __eq__(self, bs):

"""Return True if two bitstrings have the same binary representation.

try:

bs = Bits(bs)

except TypeError:

return False

return equal(self._datastore, bs._datastore)

def __ne__(self, bs):

"""Return False if two bitstrings have the same binary representation.

return not self.__eq__(bs)