rate34 and rate12 data pdus implemented with tests, updated CRC9 interface for easier use, test that de-serialize and serialize burst provides the same data

Author: smarek

okdmr/dmrlib/etsi/crc/crc9.py

@@ -1,3 +1,5 @@
+from typing import Union
+
 from bitarray import bitarray
 from bitarray.util import int2ba, ba2int
 
@@ -20,17 +22,29 @@ def check(
         data: bytes, serial_number: int, crc9: int, mask: CrcMasks, crc32: bytes = None
     ) -> bool:
         assert crc9 <= 511, "CRC-9 check value is invalid, max. for 9-bit number is 511"
+        return (
+            CRC9.calculate_from_parts(
+                data=data, mask=mask, crc32=crc32, serial_number=serial_number
+            )
+            == crc9
+        )
 
+    @staticmethod
+    def calculate_from_parts(
+        data: bytes, serial_number: int, mask: CrcMasks, crc32: Union[int, bytes] = None
+    ):
         source_data: bitarray = bytes_to_bits(data, endian="big")
 
         if crc32 is not None:
+            if isinstance(crc32, int):
+                crc32 = crc32.to_bytes(4, byteorder="big")
             assert len(crc32) == 4, "32-bit CRC must be exactly 4-bytes long"
             source_data += bytes_to_bits(crc32, endian="big")
 
         dbsnba = int2ba(serial_number, length=7, endian="big", signed=False)
         source_data += dbsnba
 
-        return CRC9.calculate(source_data, mask) == crc9
+        return CRC9.calculate(data=source_data, mask=mask)
 
     @staticmethod
     def calculate(data: bitarray, mask: CrcMasks) -> int:

okdmr/dmrlib/etsi/layer2/burst.py

@@ -13,6 +13,7 @@
 from okdmr.dmrlib.etsi.layer2.pdu.data_header import DataHeader
 from okdmr.dmrlib.etsi.layer2.pdu.embedded_signalling import EmbeddedSignalling
 from okdmr.dmrlib.etsi.layer2.pdu.full_link_control import FullLinkControl
+from okdmr.dmrlib.etsi.layer2.pdu.rate12_data import Rate12Data
 from okdmr.dmrlib.etsi.layer2.pdu.rate34_data import Rate34Data
 from okdmr.dmrlib.etsi.layer2.pdu.slot_type import SlotType
 from okdmr.dmrlib.hytera.hytera_constants import IPSC_KAITAI_VOICE_SLOTS
@@ -102,6 +103,8 @@ def extract_data(self) -> Optional[BitsInterface]:
             return DataHeader.from_bits(self.info_bits_deinterleaved)
         elif self.data_type == DataTypes.Rate34Data:
             return Rate34Data.from_bits(self.info_bits_deinterleaved)
+        elif self.data_type == DataTypes.Rate12Data:
+            return Rate12Data.from_bits(self.info_bits_deinterleaved)
 
         return None
 

okdmr/dmrlib/etsi/layer2/pdu/data_header.py

@@ -47,6 +47,7 @@ def __init__(
         appended_blocks: int = 0,
         defined_data_format: Optional[DefinedDataFormats] = None,
         sarq: Optional[SARQ] = None,
+        bit_padding: Optional[bitarray] = bitarray(),
         # Unified Data Transport Header (UDT_HEAD) PDU
         is_emergency: Union[int, bool] = False,
         udt_option_flag: Optional[UDTOptionFlag] = None,
@@ -76,6 +77,7 @@ def __init__(
         self.appended_blocks: int = appended_blocks
         self.defined_data_format: Optional[DefinedDataFormats] = defined_data_format
         self.sarq: Optional[SARQ] = sarq
+        self.bit_padding: bitarray = bit_padding
         # Unified Data Transport Header (UDT_HEAD) PDU
         self.is_emergency: bool = is_emergency in (True, 1)
         self.udt_option_flag: Optional[UDTOptionFlag] = udt_option_flag
@@ -209,7 +211,7 @@ def as_bits(self) -> bitarray:
                 + int2ba(self.llid_source, length=24)
                 + self.defined_data_format.as_bits()
                 + bitarray([self.sarq.value, self.full_message_flag.value])
-                + bitarray([0] * 8)
+                + self.bit_padding
                 + self.crc
             )
         elif self.data_packet_format == DataPacketFormats.UnifiedDataTransport:
@@ -282,6 +284,7 @@ def from_bits(bits: bitarray) -> "DataHeader":
                 defined_data_format=DefinedDataFormats.from_bits(bits[64:70]),
                 sarq=SARQ(bits[70]),
                 full_message_flag=FullMessageFlag(bits[71]),
+                bit_padding=bits[72:80],
             )
         elif dpf == DataPacketFormats.DataPacketUnconfirmed:
             return DataHeader(

okdmr/dmrlib/etsi/layer2/pdu/rate12_data.py

@@ -0,0 +1,93 @@
+from typing import Union
+
+from bitarray import bitarray
+from bitarray.util import int2ba
+
+from okdmr.dmrlib.etsi.crc.crc9 import CRC9
+from okdmr.dmrlib.etsi.layer2.elements.crc_masks import CrcMasks
+from okdmr.dmrlib.utils.bits_bytes import bits_to_bytes, bytes_to_bits
+from okdmr.dmrlib.utils.bits_interface import BitsInterface
+
+
+class Rate12Data(BitsInterface):
+    """
+    ETSI TS 102 361-1 V2.5.1 (2017-10) - 9.2.7 Rate 1/2 coded packet Data (R_1_2_DATA) PDU
+    ETSI TS 102 361-1 V2.5.1 (2017-10) - 9.2.8 Rate 1/2 coded Last Data block (R_1_2_LDATA) PDU
+    """
+
+    def __init__(
+        self,
+        data: bytes,
+        dbsn: int = 0,
+        crc9: int = 0,
+        crc32: Union[int, bytes] = 0,
+    ):
+        self.data: bytes = data
+        self.dbsn: int = dbsn
+        self.crc32: int = (
+            crc32 if isinstance(crc32, int) else int.from_bytes(crc32, byteorder="big")
+        )
+
+        self.crc9: int = self.calculate_crc9()
+        self.crc9_ok: bool = self.crc9 == crc9 if crc9 > 0 else True
+
+    def calculate_crc9(self) -> int:
+        return CRC9.calculate_from_parts(
+            data=self.data,
+            serial_number=self.dbsn,
+            crc32=self.crc32,
+            mask=CrcMasks.Rate12DataContinuation,
+        )
+
+    def __repr__(self) -> str:
+        if len(self.data) == 12:
+            return f"[RATE 1/2 DATA] [DATA(12) {self.data.hex()}]"
+        elif len(self.data) == 10:
+            return (
+                f"[RATE 1/2 DATA CONFIRMED] [DATA(10) {self.data.hex()}]"
+                + f" [CRC9: {self.crc9}]"
+                + (" [CRC9 INVALID]" if not self.crc9_ok else "")
+            )
+        elif len(self.data) == 8:
+            return (
+                f"[RATE 1/2 DATA - LAST BLOCK UNCONFIRMED] [DATA(8) {self.data.hex()}]"
+                + f" [CRC32 int({self.crc32}) hex({self.crc32.to_bytes(4, byteorder='big').hex()})]"
+            )
+        elif len(self.data) == 6:
+            return (
+                f"[RATE 1/2 DATA - LAST BLOCK CONFIRMED] [DATA(6) {self.data.hex()}]"
+                + f" [CRC9: {self.crc9}]"
+                + (" [CRC9 INVALID]" if not self.crc9_ok else "")
+                + f" [CRC32 int({self.crc32}) hex({self.crc32.to_bytes(4, byteorder='big').hex()})]"
+            )
+        raise ValueError(f"__repr__ not implemented for data len {len(self.data)}")
+
+    @staticmethod
+    def from_bits(bits: bitarray) -> "Rate12Data":
+        assert (
+            len(bits) == 96
+        ), f"Rate 1/2 Data packet must be 96 bits (12 bytes) long, got {len(bits)} bits"
+        return Rate12Data(data=bits_to_bytes(bits))
+
+    def as_bits(self):
+        if len(self.data) == 12:
+            # R_1_2_DATA PDU content for unconfirmed data
+            return bytes_to_bits(self.data)
+        elif len(self.data) == 10:
+            # R_1_2_DATA PDU content for confirmed data
+            return (
+                int2ba(self.dbsn, length=7)
+                + int2ba(self.crc9, length=9)
+                + bytes_to_bits(self.data)
+            )
+        elif len(self.data) == 8:
+            # R_1_2_LDATA PDU content for confirmed data
+            return bytes_to_bits(self.data) + int2ba(self.crc32, length=32)
+        elif len(self.data) == 6:
+            # R_3_4_LDATA PDU content for confirmed data
+            return (
+                int2ba(self.dbsn, length=7)
+                + int2ba(self.calculate_crc9(), length=9)
+                + bytes_to_bits(self.data)
+                + int2ba(self.crc32, length=32)
+            )

okdmr/dmrlib/etsi/layer2/pdu/rate34_data.py

@@ -1,29 +1,66 @@
+from typing import Union
+
 from bitarray import bitarray
 from bitarray.util import int2ba
 
+from okdmr.dmrlib.etsi.crc.crc9 import CRC9
+from okdmr.dmrlib.etsi.layer2.elements.crc_masks import CrcMasks
 from okdmr.dmrlib.utils.bits_bytes import bits_to_bytes, bytes_to_bits
 from okdmr.dmrlib.utils.bits_interface import BitsInterface
 
 
 class Rate34Data(BitsInterface):
+    """
+    ETSI TS 102 361-1 V2.5.1 (2017-10) - 9.2.2 Rate ¾ coded packet Data (R_3_4_DATA) PDU
+    ETSI TS 102 361-1 V2.5.1 (2017-10) - 9.2.3 Rate ¾ coded Last Data block (R_3_4_LDATA) PDU
+    """
+
     def __init__(
         self,
         data: bytes,
         dbsn: int = 0,
         crc9: int = 0,
-        crc32: int = 0,
+        crc32: Union[int, bytes] = 0,
     ):
         self.data: bytes = data
         self.dbsn: int = dbsn
-        self.crc9: int = crc9
-        self.crc32: int = crc32
-
-        if len(data) in (14, 12):
-            # verify crc32
-            pass
-        if len(data) in (16, 12):
-            # verify crc9
-            pass
+        self.crc32: int = (
+            crc32 if isinstance(crc32, int) else int.from_bytes(crc32, byteorder="big")
+        )
+
+        self.crc9: int = self.calculate_crc9()
+        self.crc9_ok: bool = self.crc9 == crc9 if crc9 > 0 else True
+
+    def calculate_crc9(self) -> int:
+        return CRC9.calculate_from_parts(
+            data=self.data,
+            serial_number=self.dbsn,
+            crc32=self.crc32,
+            mask=CrcMasks.Rate34DataContinuation,
+        )
+
+    def __repr__(self) -> str:
+        if len(self.data) == 18:
+            return f"[RATE 3/4 DATA] [DATA(18) {self.data.hex()}]"
+        elif len(self.data) == 16:
+            return (
+                f"[RATE 3/4 DATA CONFIRMED] [DATA(16) {self.data.hex()}]"
+                + f" [CRC9: {self.crc9}]"
+                + (" [CRC9 INVALID]" if not self.crc9_ok else "")
+            )
+        elif len(self.data) == 14:
+            return (
+                f"[RATE 3/4 DATA - LAST BLOCK UNCONFIRMED] [DATA(14) {self.data.hex()}]"
+                + f" [CRC32 int({self.crc32}) hex({self.crc32.to_bytes(4, byteorder='big').hex()})]"
+            )
+        elif len(self.data) == 12:
+            return (
+                f"[RATE 3/4 DATA - LAST BLOCK CONFIRMED] [DATA(12) {self.data.hex()}]"
+                + f" [CRC9: {self.crc9}]"
+                + (" [CRC9 INVALID]" if not self.crc9_ok else "")
+                + f" [CRC32 int({self.crc32}) hex({self.crc32.to_bytes(4, byteorder='big').hex()})]"
+            )
+        raise ValueError(f"__repr__ not implemented for data len {len(self.data)}")
 
     @staticmethod
     def from_bits(bits: bitarray) -> "Rate34Data":

okdmr/tests/dmrlib/etsi/layer2/pdu/test_rate12_data.py

@@ -0,0 +1,29 @@
+from okdmr.dmrlib.etsi.layer2.pdu.rate12_data import Rate12Data
+
+
+def test_rate12_data():
+    r12u: Rate12Data = Rate12Data(data=b"\xc0K;\xb7NN\x83\xb5\xc9\x01\x03\xcb")
+    r12u_bits = r12u.as_bits()
+    assert len(repr(r12u))
+    assert Rate12Data.from_bits(r12u_bits).as_bits() == r12u_bits
+
+    r12c: Rate12Data = Rate12Data(data=b"t~\xa3B5\xcc\xc4JwK", dbsn=1)
+    r12c_bits = r12c.as_bits()
+    assert r12c.crc9_ok
+    assert Rate12Data.from_bits(r12c_bits).as_bits() == r12c_bits
+    assert len(repr(r12c))
+
+    r12ul: Rate12Data = Rate12Data(
+        data=b"y\x02\xe8\xf3c\xc3\x82\x88", crc32=b"\n\xfc\xb3\xe1"
+    )
+    r12ul_bits = r12ul.as_bits()
+    assert Rate12Data.from_bits(r12ul_bits).as_bits() == r12ul_bits
+    assert len(repr(r12ul))
+
+    r12cl: Rate12Data = Rate12Data(
+        data=b"\xaf\x12\xf4O\xdeK", crc32=b"\x82j\xfdX", dbsn=1
+    )
+    r12cl_bits = r12cl.as_bits()
+    assert r12cl.crc9_ok
+    assert len(repr(r12cl))
+    assert Rate12Data.from_bits(r12cl_bits).as_bits() == r12cl_bits

okdmr/tests/dmrlib/etsi/layer2/pdu/test_rate34_data.py

@@ -0,0 +1,36 @@
+from okdmr.dmrlib.etsi.layer2.pdu.rate34_data import Rate34Data
+
+
+def test_rate34_data():
+    # R_3_4_DATA PDU content for unconfirmed data
+    r34u: Rate34Data = Rate34Data(
+        data=b"\xdd\xda\xe4\xa1\xaeT'oe\xc9\xf6\xe4\x97\x0e#\xed\x17s"
+    )
+    r34u_bits = r34u.as_bits()
+    assert Rate34Data.from_bits(r34u_bits).as_bits() == r34u_bits
+    assert len(repr(r34u))
+
+    r34c: Rate34Data = Rate34Data(
+        data=b"\xc1\x1bb\x1a\xd4\x8b\x97\x91\x17A\x9e\xb1r\xeb\xe0\xb8",
+        dbsn=1,
+        crc9=306,
+    )
+    r34c_bits = r34c.as_bits()
+    assert r34c.crc9_ok
+    assert Rate34Data.from_bits(r34c_bits).as_bits() == r34c_bits
+    assert len(repr(r34c))
+
+    r34ul: Rate34Data = Rate34Data(
+        data=b"\x90\xbc\x85\x00\x89\xc1[K\xe4l\x83P\x88\xe0", crc32=b"\xdb\xb2p\xe5"
+    )
+    r34ul_bits = r34ul.as_bits()
+    assert Rate34Data.from_bits(r34ul_bits).as_bits() == r34ul_bits
+    assert len(repr(r34ul))
+
+    r34cl: Rate34Data = Rate34Data(
+        data=b"\xfe\x1ey\x9cR\xb5\xe4&T\xc2#\x83", dbsn=1, crc32=b'\x12"\x18N'
+    )
+    r34cl_bits = r34cl.as_bits()
+    assert Rate34Data.from_bits(r34cl_bits).as_bits() == r34cl_bits
+    assert r34cl.crc9_ok
+    assert len(repr(r34cl))

okdmr/tests/dmrlib/etsi/layer2/test_burst.py

@@ -106,12 +106,22 @@ def test_burst_as_bits():
             "2522222222a632b222222222560dff57d75df5dce2822222222791522222222c11",
             BurstTypes.DataAndControl,
         ),
+        (
+            "3a1f36af232d7afda01bd78255bdff57d75df5d55c045c2e3361260e501f863363",
+            BurstTypes.DataAndControl,
+        ),
+        (
+            "00e527076f2a4b0f03ed010115ddff57d75df5d6f145422817d6234b6e08802018",
+            BurstTypes.DataAndControl,
+        ),
     ]
     for (hexstr, burst_type) in bursts:
         _bytes: bytes = bytes.fromhex(hexstr)
         b: Burst = Burst.from_bytes(data=_bytes, burst_type=burst_type)
-        assert b.data.as_bits() == b.info_bits_deinterleaved
-        assert b.as_bits().tobytes() == _bytes
+        assert (
+            b.data.as_bits() == b.info_bits_deinterleaved
+        ), f"Mismatch in {repr(b)} {b.as_bits().tobytes().hex()}"
+        assert b.as_bits().tobytes() == _bytes, f"Mismatch in {repr(b)}"
 
 
 if __name__ == "__main__":