Experimental version with bit vector (#200)

* feat: adds BitVector

* docs: bit_vector

* chore: rename BitVector to bit_vector

* chore: renames data to bv (NMEASentence)

Author: M0r13n

pyais/bit_vector.py

@@ -0,0 +1,116 @@
+"""
+This bit vector uses pre-computed lookup tables and stores the entire payload as a single
+arbitrary-precision int so that every field extraction is a constant-time shift-and-mask operation.
+"""
+import typing as t
+
+# ASCII ordinal 6-bit AIS payload value
+# Valid input range: ordinals 48 ('0') through 119 ('w').
+# Everything outside that range maps to 0; upstream validation is expected.
+_PAYLOAD_ARMOR: t.Final[tuple[int, ...]] = tuple(
+    (c - 48 - 8 if c - 48 > 40 else c - 48) if 48 <= c <= 119 else 0 for c in range(256)
+)
+
+# 6-bit value decoded AIS character (for text fields)
+_SIXBIT_CHAR: t.Final[tuple[str, ...]] = tuple(
+    chr(v + 64) if v < 32 else chr(v) for v in range(64)
+)
+
+# Bitmasks: _MASK[n] = (1 << n) - 1
+# 257 entries covers every realistic AIS field width.
+_MASK: t.Final[tuple[int, ...]] = tuple((1 << n) - 1 for n in range(257))
+
+
+class bit_vector:
+
+    __slots__ = ("_value", "_length")
+
+    def __init__(self, data: bytes, pad: int = 0) -> None:
+        value = 0
+        for byte in data:
+            value = (value << 6) | _PAYLOAD_ARMOR[byte]
+
+        length = len(data) * 6
+        if pad:
+            value >>= pad
+            length -= pad
+
+        self._value: int = value
+        self._length: int = length
+
+    def get_num(self, start: int, width: int, signed: bool = False) -> int:
+        """Return an unsigned integer of *width* bits at bit position *start*.
+        Takes an additional argument *signed* if the number is signed.
+        Returns an unsigned integer by default.
+        """
+        if signed:
+            return self.get_signed(start, width)
+        return self.get(start, width)
+
+    def get(self, start: int, width: int) -> int:
+        """Return an unsigned integer of *width* bits at bit position *start*."""
+        # If the requested range extends beyond the end of the vector,
+        # only the available bits are returned
+        if width <= 0 or start >= self._length:
+            return 0
+        available = min(width, self._length - start)
+        shift = self._length - start - available
+        return (self._value >> shift) & _MASK[available]
+
+    def get_signed(self, start: int, width: int) -> int:
+        """Return a signed (two's-complement) integer."""
+        if width <= 0 or start >= self._length:
+            return 0
+        available = min(width, self._length - start)
+        shift = self._length - start - available
+        val = (self._value >> shift) & _MASK[available]
+        if val & (1 << (available - 1)):
+            val -= 1 << available
+        return val
+
+    def get_bool(self, start: int) -> bool:
+        """Return a single bit as a boolean."""
+        if start >= self._length:
+            return False
+        shift = self._length - start - 1
+        return bool((self._value >> shift) & 1)
+
+    def get_str(self, start: int, width: int) -> str:
+        """Return a 6-bit-encoded AIS text string."""
+        if width <= 0 or start >= self._length:
+            return ""
+        chars = _SIXBIT_CHAR
+        get = self.get
+        parts: list[str] = [chars[get(i, 6)] for i in range(start, start + width, 6)]
+        return "".join(parts).rstrip("@").strip()
+
+    def get_bytes(self, start: int, width: int) -> bytes:
+        """Return the value of *width* bits at bit position *start* as bytes."""
+        if width <= 0 or start >= self._length:
+            return b""
+
+        available = min(width, self._length - start)
+        shift = self._length - start - available
+        # Not using _MASK because the width might be larger than 257 bits
+        val = (self._value >> shift) & ((1 << available) - 1)
+
+        # Pad to byte boundary on the right
+        num_bytes = (available + 7) // 8
+        pad_bits = num_bytes * 8 - available
+        val <<= pad_bits
+
+        return val.to_bytes(num_bytes, "big")
+
+    def __len__(self) -> int:
+        return self._length
+
+    def __repr__(self) -> str:
+        return f"bit_vector(length={self._length})"
+
+    def __eq__(self, value: object) -> bool:
+        try:
+            if not isinstance(value, bit_vector):
+                return False
+            return self._length == value._length and self._value == value._value
+        except ValueError:
+            return False

pyais/messages.py

@@ -8,11 +8,12 @@
 
 import attr
 
+from pyais.bit_vector import bit_vector
 from pyais.constants import TalkerID, NavigationStatus, ManeuverIndicator, EpfdType, ShipType, NavAid, StationType, \
     TransmitMode, StationIntervals, TurnRate, InlandLoadedType
 from pyais.exceptions import InvalidNMEAMessageException, TagBlockNotInitializedException, UnknownMessageException, UnknownPartNoException, \
     InvalidDataTypeException, MissingPayloadException
-from pyais.util import SIX_BIT_ENCODING, SixBitNibleDecoder, SixBitNibleEncoder, checksum, compute_checksum, decode_bytes_as_ascii6, extract_bits, get_bytes, get_itdma_comm_state, get_num, get_sotdma_comm_state, chk_to_int, coerce_val, b64encode_str, is_auxiliary_craft
+from pyais.util import SIX_BIT_ENCODING, SixBitNibleEncoder, checksum, compute_checksum, get_itdma_comm_state, get_sotdma_comm_state, chk_to_int, coerce_val, b64encode_str, is_auxiliary_craft
 
 NMEA_VALUE = typing.Union[str, float, int, bool, bytes]
 
@@ -535,9 +536,8 @@ class AISSentence(NMEASentence):
         'frag_num',
         'seq_id',
         'payload',
-        'data',
+        'bv',
         'ais_id',
-        'total_bits',
         'channel',
     )
 
@@ -575,8 +575,8 @@ def __init__(self, raw: bytes) -> None:
             raise InvalidNMEAMessageException("Too many fragments")
 
         # Finally decode bytes into bits
-        self.data, self.total_bits = SixBitNibleDecoder().decode(self.payload, self.fill_bits)
-        self.ais_id = extract_bits(self.data, 0, 6, self.total_bits)
+        self.bv = bit_vector(self.payload, self.fill_bits)
+        self.ais_id = self.bv.get(0, 6)
 
     def asdict(self) -> Dict[str, Any]:
         """
@@ -641,9 +641,7 @@ def assemble_from_iterable(cls, messages: Sequence["AISSentence"]) -> "AISSenten
 
         messages[0].raw = raw
         messages[0].payload = payload
-        data, total_bits = SixBitNibleDecoder().decode(payload, fill_bits=messages[-1].fill_bits)
-        messages[0].data = data
-        messages[0].total_bits = total_bits
+        messages[0].bv = bit_vector(payload, messages[-1].fill_bits)
         messages[0].is_valid = is_valid
         return messages[0]
 
@@ -670,7 +668,7 @@ def decode(self) -> "ANY_MESSAGE":
         if not self.payload:
             raise MissingPayloadException(self.raw.decode())
         try:
-            return MSG_CLASS[self.ais_id].from_bytes(self.data, self.total_bits)
+            return MSG_CLASS[self.ais_id].from_bytes(self.bv)
         except KeyError as e:
             raise UnknownMessageException(f"The message {self} is not supported!") from e
 
@@ -818,16 +816,13 @@ def create(cls, **kwargs: NMEA_VALUE) -> "ANY_MESSAGE":
         return cls(**args)  # type:ignore
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
         cur: int = 0
         kwargs: typing.Dict[str, typing.Any] = {}
 
-        large_number = int.from_bytes(data, byteorder='big')
-        large_number = large_number >> (8 - (total_bits % 8)) % 8
-
         # Iterate over fields and data
         for field in cls.fields():
-            if cur >= total_bits:
+            if cur >= len(bv):
                 # All fields that did not fit into the bit array are None
                 kwargs[field.name] = None
                 continue
@@ -839,17 +834,17 @@ def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
             val: typing.Any
             # Get the correct data type and decoding function
             if d_type == int or d_type == bool or d_type == float:
-                val = get_num(large_number, cur, width, total_bits, signed=field.metadata['signed'])
+                val = bv.get_num(cur, width, field.metadata['signed'])
 
                 if d_type == float:
                     val = float(val)
                 elif d_type == bool:
                     val = bool(val)
 
             elif d_type == str:
-                val = decode_bytes_as_ascii6(data, cur, width)
+                val = bv.get_str(cur, width)
             elif d_type == bytes:
-                val = get_bytes(data, cur, width)
+                val = bv.get_bytes(cur, width)
             else:
                 raise InvalidDataTypeException(d_type)
 
@@ -1170,13 +1165,13 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
             return MessageType8Default.create(**kwargs)
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        dac: int = extract_bits(data, 40, 10)
-        fid: int = extract_bits(data, 50, 6)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        dac: int = bv.get(40, 10)
+        fid: int = bv.get(50, 6)
         if dac == 200 and fid == 10:
-            return MessageType8Dac200Fid10.from_bytes(data, total_bits)
+            return MessageType8Dac200Fid10.from_bytes(bv)
         else:
-            return MessageType8Default.from_bytes(data, total_bits)
+            return MessageType8Default.from_bytes(bv)
 
 
 @attr.s(slots=True)
@@ -1407,10 +1402,10 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
         return MessageType16DestinationA.create(**kwargs)
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        if total_bits > 96:
-            return MessageType16DestinationAB.from_bytes(data, total_bits)
-        return MessageType16DestinationA.from_bytes(data, total_bits)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        if len(bv) > 96:
+            return MessageType16DestinationAB.from_bytes(bv)
+        return MessageType16DestinationA.from_bytes(bv)
 
 
 @attr.s(slots=True)
@@ -1651,11 +1646,11 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
             return MessageType22Broadcast.create(**kwargs)
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        if extract_bits(data, 139, 1):
-            return MessageType22Addressed.from_bytes(data, total_bits)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        if bv.get(139, 1):
+            return MessageType22Addressed.from_bytes(bv)
         else:
-            return MessageType22Broadcast.from_bytes(data, total_bits)
+            return MessageType22Broadcast.from_bytes(bv)
 
 
 @attr.s(slots=True)
@@ -1767,15 +1762,15 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
             raise UnknownPartNoException(f"Partno {partno} is not allowed!")
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        mmsi: int = extract_bits(data, 8, 30)
-        partno: int = extract_bits(data, 38, 2)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        mmsi: int = bv.get(8, 30)
+        partno: int = bv.get(38, 2)
         if partno == 0:
-            return MessageType24PartA.from_bytes(data, total_bits)
+            return MessageType24PartA.from_bytes(bv)
         elif partno == 1:
             if is_auxiliary_craft(mmsi):
-                return MessageType24PartBAuxiliaryCraft.from_bytes(data, total_bits)
-            return MessageType24PartB.from_bytes(data, total_bits)
+                return MessageType24PartBAuxiliaryCraft.from_bytes(bv)
+            return MessageType24PartB.from_bytes(bv)
         else:
             raise UnknownPartNoException(f"Partno {partno} is not allowed!")
 
@@ -1859,20 +1854,20 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
                 return MessageType25BroadcastUnstructured.create(**kwargs)
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        addressed: int = extract_bits(data, 38, 1)
-        structured: int = extract_bits(data, 39, 1)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        addressed: int = bv.get(38, 1)
+        structured: int = bv.get(39, 1)
 
         if addressed:
             if structured:
-                return MessageType25AddressedStructured.from_bytes(data, total_bits)
+                return MessageType25AddressedStructured.from_bytes(bv)
             else:
-                return MessageType25AddressedUnstructured.from_bytes(data, total_bits)
+                return MessageType25AddressedUnstructured.from_bytes(bv)
         else:
             if structured:
-                return MessageType25BroadcastStructured.from_bytes(data, total_bits)
+                return MessageType25BroadcastStructured.from_bytes(bv)
             else:
-                return MessageType25BroadcastUnstructured.from_bytes(data, total_bits)
+                return MessageType25BroadcastUnstructured.from_bytes(bv)
 
 
 @attr.s(slots=True)
@@ -1959,20 +1954,20 @@ def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_ME
                 return MessageType26BroadcastUnstructured.create(**kwargs)
 
     @classmethod
-    def from_bytes(cls, data: bytes, total_bits: int) -> "ANY_MESSAGE":
-        addressed: int = extract_bits(data, 38, 1)
-        structured: int = extract_bits(data, 39, 1)
+    def from_bytes(cls, bv: bit_vector) -> "ANY_MESSAGE":
+        addressed: int = bv.get(38, 1)
+        structured: int = bv.get(39, 1)
 
         if addressed:
             if structured:
-                return MessageType26AddressedStructured.from_bytes(data, total_bits)
+                return MessageType26AddressedStructured.from_bytes(bv)
             else:
-                return MessageType26AddressedUnstructured.from_bytes(data, total_bits)
+                return MessageType26AddressedUnstructured.from_bytes(bv)
         else:
             if structured:
-                return MessageType26BroadcastStructured.from_bytes(data, total_bits)
+                return MessageType26BroadcastStructured.from_bytes(bv)
             else:
-                return MessageType26BroadcastUnstructured.from_bytes(data, total_bits)
+                return MessageType26BroadcastUnstructured.from_bytes(bv)
 
 
 @attr.s(slots=True)