Skip to content

Commit

Permalink
More import reorganisation.
Browse files Browse the repository at this point in the history
  • Loading branch information
@scott-griffiths
scott-griffiths committed Jan 2, 2024
1 parent 4995afa commit d693026
Show file tree
Hide file tree
Showing 8 changed files with 156 additions and 174 deletions.
121 changes: 59 additions & 62 deletions bitstring/bits.py
View file

Large diffs are not rendered by default.

44 changes: 20 additions & 24 deletions bitstring/bitstore_helpers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,6 @@
import functools
from typing import Union, Optional, Dict, Callable
import bitarray
import bitarray.util
from bitstring.utils import tokenparser
from bitstring.exceptions import CreationError
from bitstring.fp8 import e4m3float_fmt, e5m2float_fmt
from bitstring.bitstore import BitStore
import bitstring

Expand All @@ -22,18 +18,18 @@
def tidy_input_string(s: str) -> str:
"""Return string made lowercase and with all whitespace and underscores removed."""
try:
l = s.split()
t = s.split()
except (AttributeError, TypeError):
raise ValueError(f"Expected str object but received a {type(s)} with value {s}.")
return ''.join(l).lower().replace('_', '')
return ''.join(t).lower().replace('_', '')


@functools.lru_cache(CACHE_SIZE)
def str_to_bitstore(s: str) -> BitStore:
try:
_, tokens = tokenparser(s)
_, tokens = bitstring.utils.tokenparser(s)
except ValueError as e:
raise CreationError(*e.args)
raise bitstring.CreationError(*e.args)
bs = BitStore()
for token in tokens:
bs += bitstore_from_token(*token)
Expand All @@ -51,7 +47,7 @@ def bin2bitstore_unsafe(binstring: str) -> BitStore:
try:
return BitStore(binstring)
except ValueError:
raise CreationError(f"Invalid character in bin initialiser {binstring}.")
raise bitstring.CreationError(f"Invalid character in bin initialiser {binstring}.")


def hex2bitstore(hexstring: str) -> BitStore:
Expand All @@ -60,7 +56,7 @@ def hex2bitstore(hexstring: str) -> BitStore:
try:
ba = bitarray.util.hex2ba(hexstring)
except ValueError:
raise CreationError("Invalid symbol in hex initialiser.")
raise bitstring.CreationError("Invalid symbol in hex initialiser.")
return BitStore(ba)


Expand All @@ -70,14 +66,14 @@ def oct2bitstore(octstring: str) -> BitStore:
try:
ba = bitarray.util.base2ba(8, octstring)
except ValueError:
raise CreationError("Invalid symbol in oct initialiser.")
raise bitstring.CreationError("Invalid symbol in oct initialiser.")
return BitStore(ba)


def ue2bitstore(i: Union[str, int]) -> BitStore:
i = int(i)
if i < 0:
raise CreationError("Cannot use negative initialiser for unsigned exponential-Golomb.")
raise bitstring.CreationError("Cannot use negative initialiser for unsigned exponential-Golomb.")
if i == 0:
return BitStore('1')
tmp = i + 1
Expand All @@ -101,7 +97,7 @@ def se2bitstore(i: Union[str, int]) -> BitStore:
def uie2bitstore(i: Union[str, int]) -> BitStore:
i = int(i)
if i < 0:
raise CreationError("Cannot use negative initialiser for unsigned interleaved exponential-Golomb.")
raise bitstring.CreationError("Cannot use negative initialiser for unsigned interleaved exponential-Golomb.")
return BitStore('1' if i == 0 else '0' + '0'.join(bin(i + 1)[3:]) + '1')


Expand Down Expand Up @@ -135,13 +131,13 @@ def bfloatle2bitstore(f: Union[str, float]) -> BitStore:

def e4m3float2bitstore(f: Union[str, float]) -> BitStore:
f = float(f)
u = e4m3float_fmt.float_to_int8(f)
u = bitstring.fp8.e4m3float_fmt.float_to_int8(f)
return uint2bitstore(u, 8)


def e5m2float2bitstore(f: Union[str, float]) -> BitStore:
f = float(f)
u = e5m2float_fmt.float_to_int8(f)
u = bitstring.fp8.e5m2float_fmt.float_to_int8(f)
return uint2bitstore(u, 8)


Expand All @@ -155,9 +151,9 @@ def uint2bitstore(uint: Union[str, int], length: int) -> BitStore:
if uint >= (1 << length):
msg = f"{uint} is too large an unsigned integer for a bitstring of length {length}. " \
f"The allowed range is [0, {(1 << length) - 1}]."
raise CreationError(msg)
raise bitstring.CreationError(msg)
if uint < 0:
raise CreationError("uint cannot be initialised with a negative number.")
raise bitstring.CreationError("uint cannot be initialised with a negative number.")
raise e
return x

Expand All @@ -170,7 +166,7 @@ def int2bitstore(i: Union[str, int], length: int) -> BitStore:
x = BitStore(bitarray.util.int2ba(i, length=length, endian='big', signed=True))
except OverflowError as e:
if i >= (1 << (length - 1)) or i < -(1 << (length - 1)):
raise CreationError(f"{i} is too large a signed integer for a bitstring of length {length}. "
raise bitstring.CreationError(f"{i} is too large a signed integer for a bitstring of length {length}. "
f"The allowed range is [{-(1 << (length - 1))}, {(1 << (length - 1)) - 1}].")
else:
raise e
Expand All @@ -179,26 +175,26 @@ def int2bitstore(i: Union[str, int], length: int) -> BitStore:

def uintbe2bitstore(i: Union[str, int], length: int) -> BitStore:
if length % 8 != 0:
raise CreationError(f"Big-endian integers must be whole-byte. Length = {length} bits.")
raise bitstring.CreationError(f"Big-endian integers must be whole-byte. Length = {length} bits.")
return uint2bitstore(i, length)


def intbe2bitstore(i: int, length: int) -> BitStore:
if length % 8 != 0:
raise CreationError(f"Big-endian integers must be whole-byte. Length = {length} bits.")
raise bitstring.CreationError(f"Big-endian integers must be whole-byte. Length = {length} bits.")
return int2bitstore(i, length)


def uintle2bitstore(i: int, length: int) -> BitStore:
if length % 8 != 0:
raise CreationError(f"Little-endian integers must be whole-byte. Length = {length} bits.")
raise bitstring.CreationError(f"Little-endian integers must be whole-byte. Length = {length} bits.")
x = uint2bitstore(i, length).tobytes()
return BitStore(frombytes=x[::-1])


def intle2bitstore(i: int, length: int) -> BitStore:
if length % 8 != 0:
raise CreationError(f"Little-endian integers must be whole-byte. Length = {length} bits.")
raise bitstring.CreationError(f"Little-endian integers must be whole-byte. Length = {length} bits.")
x = int2bitstore(i, length).tobytes()
return BitStore(frombytes=x[::-1])

Expand Down Expand Up @@ -253,7 +249,7 @@ def bitstore_from_token(name: str, token_length: Optional[int], value: Optional[
if name in literal_bit_funcs:
bs = literal_bit_funcs[name](value)
else:
raise CreationError(f"Can't parse token: {e}")
raise bitstring.CreationError(f"Can't parse token: {e}")
else:
b = bitstring.bits.Bits()
if value is None and name != 'pad':
Expand All @@ -262,6 +258,6 @@ def bitstore_from_token(name: str, token_length: Optional[int], value: Optional[
d.set_fn(b, value)
bs = b._bitstore
if token_length is not None and len(bs) != d.bitlength:
raise CreationError(f"Token with length {token_length} packed with value of length {len(bs)} "
raise bitstring.CreationError(f"Token with length {token_length} packed with value of length {len(bs)} "
f"({name}:{token_length}={value}).")
return bs
50 changes: 24 additions & 26 deletions bitstring/bitstream.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
from __future__ import annotations

import bitstring
from bitstring.bits import Bits, BitsType
from bitstring.bitarray_ import BitArray
from bitstring.exceptions import ReadError, ByteAlignError, CreationError
from bitstring.dtypes import Dtype
from typing import Union, List, Any, Optional, overload, TypeVar, Tuple
import copy
import numbers
Expand Down Expand Up @@ -125,7 +123,7 @@ def __init__(self, auto: Optional[Union[BitsType, int]] = None, /, length: Optio
if pos < 0:
pos += len(self._bitstore)
if pos < 0 or pos > len(self._bitstore):
raise CreationError(f"Cannot set pos to {pos} when length is {len(self._bitstore)}.")
raise bitstring.CreationError(f"Cannot set pos to {pos} when length is {len(self._bitstore)}.")
self._pos = pos
self._bitstore.immutable = True

Expand All @@ -136,7 +134,7 @@ def _setbytepos(self, bytepos: int) -> None:
def _getbytepos(self) -> int:
"""Return the current position in the stream in bytes. Must be byte aligned."""
if self._pos % 8:
raise ByteAlignError("Not byte aligned when using bytepos property.")
raise bitstring.ByteAlignError("Not byte aligned when using bytepos property.")
return self._pos // 8

def _setbitpos(self, pos: int) -> None:
Expand Down Expand Up @@ -201,7 +199,7 @@ def __xor__(self: TConstBitStream, bs: BitsType, /) -> TConstBitStream:
s._pos = 0
return s

def __add__(self: TConstBitStream, bs: BitsType) -> TConstBitStream:
def __add__(self: TConstBitStream, bs: BitsType, /) -> TConstBitStream:
"""Concatenate bitstrings and return new bitstring.
bs -- the bitstring to append.
Expand All @@ -211,7 +209,7 @@ def __add__(self: TConstBitStream, bs: BitsType) -> TConstBitStream:
s._pos = 0
return s

def append(self, bs: BitsType) -> None:
def append(self, bs: BitsType, /) -> None:
"""Append a bitstring to the current bitstring.
bs -- The bitstring to append.
Expand All @@ -230,8 +228,8 @@ def __repr__(self) -> str:
"""
return self._repr(self.__class__.__name__, len(self), self._bitstore.filename, self._pos)

def overwrite(self, bs: BitsType, pos: Optional[int] = None) -> None:
"""Overwrite with bs at bit position pos.
def overwrite(self, bs: BitsType, /, pos: Optional[int] = None) -> None:
"""Overwrite with bitstring at bit position pos.
bs -- The bitstring to overwrite with.
pos -- The bit position to begin overwriting from.
Expand All @@ -252,7 +250,7 @@ def overwrite(self, bs: BitsType, pos: Optional[int] = None) -> None:
self._overwrite(bs, pos)
self._pos = pos + len(bs)

def find(self, bs: BitsType, start: Optional[int] = None, end: Optional[int] = None,
def find(self, bs: BitsType, /, start: Optional[int] = None, end: Optional[int] = None,
bytealigned: Optional[bool] = None) -> Union[Tuple[int], Tuple[()]]:
"""Find first occurrence of substring bs.
Expand Down Expand Up @@ -280,7 +278,7 @@ def find(self, bs: BitsType, start: Optional[int] = None, end: Optional[int] = N
self._pos = p[0]
return p

def rfind(self, bs: BitsType, start: Optional[int] = None, end: Optional[int] = None,
def rfind(self, bs: BitsType, /, start: Optional[int] = None, end: Optional[int] = None,
bytealigned: Optional[bool] = None) -> Union[Tuple[int], Tuple[()]]:
"""Find final occurrence of substring bs.
Expand Down Expand Up @@ -353,11 +351,11 @@ def read(self, fmt: Union[int, str, Dtype]) -> Union[int, float, str, Bits, bool
if fmt < 0:
raise ValueError("Cannot read negative amount.")
if fmt > len(self) - self._pos:
raise ReadError(f"Cannot read {fmt} bits, only {len(self) - self._pos} available.")
raise bitstring.ReadError(f"Cannot read {fmt} bits, only {len(self) - self._pos} available.")
bs = self._slice(self._pos, self._pos + fmt)
self._pos += fmt
return bs
dtype = Dtype(fmt)
dtype = bitstring.dtypes.Dtype(fmt)
if dtype.bitlength is None and dtype.is_unknown_length is False:
# No length specified? Try again, but read to end.
bitlength = len(self) - self._pos
Expand All @@ -366,7 +364,7 @@ def read(self, fmt: Union[int, str, Dtype]) -> Union[int, float, str, Bits, bool
raise ValueError(
f"The '{dtype.name}' type must have a bit length that is a multiple of {dtype.bits_per_item}"
f" so cannot be read from the {bitlength} bits that are available.")
dtype = Dtype(fmt, items)
dtype = bitstring.dtypes.Dtype(fmt, items)
if dtype.bitlength is not None:
val = dtype.read_fn(self, self._pos)
self._pos += dtype.bitlength
Expand All @@ -375,7 +373,7 @@ def read(self, fmt: Union[int, str, Dtype]) -> Union[int, float, str, Bits, bool

if self._pos > len(self):
self._pos = p
raise ReadError(f"Reading off end of bitstring with fmt '{fmt}'. Only {len(self) - p} bits available.")
raise bitstring.ReadError(f"Reading off end of bitstring with fmt '{fmt}'. Only {len(self) - p} bits available.")
return val

def readlist(self, fmt: Union[str, List[Union[int, str]]], **kwargs) \
Expand Down Expand Up @@ -403,10 +401,10 @@ def readlist(self, fmt: Union[str, List[Union[int, str]]], **kwargs) \
value, self._pos = self._readlist(fmt, self._pos, **kwargs)
return value

def readto(self: TConstBitStream, bs: BitsType, bytealigned: Optional[bool] = None) -> TConstBitStream:
def readto(self: TConstBitStream, bs: BitsType, /, bytealigned: Optional[bool] = None) -> TConstBitStream:
"""Read up to and including next occurrence of bs and return result.
bs -- The bitstring to find. An integer is not permitted.
bs -- The bitstring to find.
bytealigned -- If True the bitstring will only be
found on byte boundaries.
Expand All @@ -420,7 +418,7 @@ def readto(self: TConstBitStream, bs: BitsType, bytealigned: Optional[bool] = No
oldpos = self._pos
p = self.find(bs, self._pos, bytealigned=bytealigned)
if not p:
raise ReadError("Substring not found")
raise bitstring.ReadError("Substring not found")
self._pos += len(bs)
return self._slice(oldpos, self._pos)

Expand Down Expand Up @@ -496,7 +494,7 @@ def bytealign(self) -> int:
""")


class BitStream(ConstBitStream, BitArray):
class BitStream(ConstBitStream, bitstring.BitArray):
"""A container or stream holding a mutable sequence of bits
Subclass of the ConstBitStream and BitArray classes. Inherits all of
Expand Down Expand Up @@ -630,8 +628,8 @@ def __copy__(self) -> BitStream:
s_copy._bitstore = self._bitstore.copy()
return s_copy

def __iadd__(self, bs: BitsType) -> BitStream:
"""Append bs to current bitstring. Return self.
def __iadd__(self, bs: BitsType, /) -> BitStream:
"""Append to current bitstring. Return self.
bs -- the bitstring to append.
Expand All @@ -641,7 +639,7 @@ def __iadd__(self, bs: BitsType) -> BitStream:
self._pos = len(self)
return self

def prepend(self, bs: BitsType) -> None:
def prepend(self, bs: BitsType, /) -> None:
"""Prepend a bitstring to the current bitstring.
bs -- The bitstring to prepend.
Expand All @@ -651,14 +649,14 @@ def prepend(self, bs: BitsType) -> None:
super().prepend(bs)
self._pos = 0

def __setitem__(self, key: Union[slice, int], value: BitsType) -> None:
def __setitem__(self, /, key: Union[slice, int], value: BitsType) -> None:
length_before = len(self)
super().__setitem__(key, value)
if len(self) != length_before:
self._pos = 0
return

def __delitem__(self, key: Union[slice, int]) -> None:
def __delitem__(self, /, key: Union[slice, int]) -> None:
"""Delete item or range.
>>> a = BitStream('0x001122')
Expand All @@ -672,8 +670,8 @@ def __delitem__(self, key: Union[slice, int]) -> None:
if len(self) != length_before:
self._pos = 0

def insert(self, bs: BitsType, pos: Optional[int] = None) -> None:
"""Insert bs at bit position pos.
def insert(self, bs: BitsType, /, pos: Optional[int] = None) -> None:
"""Insert bitstring at bit position pos.
bs -- The bitstring to insert.
pos -- The bit position to insert at.
Expand Down
Loading

0 comments on commit d693026

Please sign in to comment.