diff --git a/dmrtools/dmrproto/__init__.py b/dmrtools/dmrproto/__init__.py index f59c46d..6f67b82 100755 --- a/dmrtools/dmrproto/__init__.py +++ b/dmrtools/dmrproto/__init__.py @@ -1,2 +1,3 @@ from .mmdvm_l1 import * +from .etsi_l2 import * from .exceptions import * diff --git a/dmrtools/dmrproto/etsi_l2.py b/dmrtools/dmrproto/etsi_l2.py new file mode 100755 index 0000000..5d0bdef --- /dev/null +++ b/dmrtools/dmrproto/etsi_l2.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import enum + +from abc import ABC, abstractmethod +from bitarray import bitarray +from dmr_utils3.bptc import decode_full_lc + +from .exceptions import DMRPL2BadDataException + + +class DMRPL2Base(ABC): + DataTypes = enum.StrEnum('DataTypes', [ + 'UNKNOWN', 'FULL_LC', 'VOICE_BURST' + ]) + + def __init__(self, data: bytes) -> None: + self.set_data(data) + + def set_data(self, data: bytes) -> None: + if len(data) != 33: + DMRPL2BadDataException("Data must be 33 bytes long") + + self.bitdata: bitarray = bitarray(data, endian='big') + + def get_data(self) -> bytes: + return self.bitdata.tobytes() + + @abstractmethod + def get_data_type(self) -> DMRPL2Base.DataTypes: + pass + + +class DMRPL2FullLC(DMRPL2Base): + def get_data_type(self) -> DMRPL2Base.DataTypes: + return DMRPL2Base.DataTypes.FULL_LC + + def get_full_lc(self) -> bytes: + return decode_full_lc(self.bitdata[:98] + self.bitdata[-98:]).tobytes() + + +class DMRPL2VoiceBurst(DMRPL2Base): + def get_data_type(self) -> DMRPL2Base.DataTypes: + return DMRPL2Base.DataTypes.VOICE_BURST + + def get_voice_bits(self) -> tuple[bitarray, bitarray]: + return (self.bitdata[:108], self.bitdata[-108:]) + + def get_emb_lc(self) -> bytes: + return self.bitdata[116:148].tobytes() diff --git a/dmrtools/dmrproto/exceptions.py b/dmrtools/dmrproto/exceptions.py index 0f961fb..26567fe 100755 --- a/dmrtools/dmrproto/exceptions.py +++ b/dmrtools/dmrproto/exceptions.py @@ -18,8 +18,15 @@ class DMRPUnknownPacketTypeException(Exception): pass -class DMRPBadPacket(Exception): +class DMRPBadPacketException(Exception): """ - Exception raised when a packet is structurally invalid or corrupted. + Exception raised when a packet is structurally invalid or corrupted + """ + pass + + +class DMRPL2BadDataException(Exception): + """ + Exception raised when L2 data is invalid or corrupted. """ pass diff --git a/dmrtools/dmrproto/mmdvm_l1.py b/dmrtools/dmrproto/mmdvm_l1.py index f3b3df2..b463c25 100755 --- a/dmrtools/dmrproto/mmdvm_l1.py +++ b/dmrtools/dmrproto/mmdvm_l1.py @@ -8,9 +8,10 @@ from hashlib import sha256 from typing import Type, Self from .base_fields import DMRPFieldInt, DMRPFieldBytes, DMRPFieldStr -from .exceptions import DMRPBadPacket +from .exceptions import DMRPBadPacketException from .exceptions import DMRPFieldOutOfRangeException from .exceptions import DMRPUnknownPacketTypeException +from .etsi_l2 import DMRPL2Base, DMRPL2FullLC, DMRPL2VoiceBurst ############################# @@ -66,14 +67,14 @@ class DMRPBasePacket(ABC): data (bytes): The raw data from which to create a packet. Raises: - DMRPBadPacket: if size or type doesn't match + DMRPBadPacketException: if size or type doesn't match """ self._data = None if not data.startswith(self.PKT_TYPE): - raise DMRPBadPacket( + raise DMRPBadPacketException( f"Bad packet type: '{data[0:4]!r}', expected '{self.PKT_TYPE!r}'") if len(data) != self.PKT_SIZE: - raise DMRPBadPacket( + raise DMRPBadPacketException( f"Bad packet size: {len(data)}, expected {self.PKT_SIZE}") self._data = bytearray(data) @@ -296,6 +297,10 @@ class DMRPPacketData(DMRPBasePeerPacket): ber = DMRPFieldInt('ber', 53, 1) rssi = DMRPFieldInt('rssi', 54, 1) + def __init__(self, data: bytes|None = None) -> None: + self.__l2: DMRPL2Base|None = None + super().__init__(data) + def set_random_stream_id(self) -> None: self.stream_id = int.from_bytes(random.randbytes(4), byteorder="big") @@ -375,6 +380,42 @@ class DMRPPacketData(DMRPBasePeerPacket): def is_voice_term(self) -> bool: return self.voice_type == DMRPPacketData.VoiceType.TERM + def get_l2(self) -> DMRPL2Base|None: + if self.__l2 is not None: + return self.__l2 + + if self.voice_type in (DMRPPacketData.VoiceType.HEAD, + DMRPPacketData.VoiceType.TERM): + self.__l2 = DMRPL2FullLC(self.dmr_data) + + if self.voice_type in (DMRPPacketData.VoiceType.BURST_A, + DMRPPacketData.VoiceType.BURST_B, + DMRPPacketData.VoiceType.BURST_C, + DMRPPacketData.VoiceType.BURST_D, + DMRPPacketData.VoiceType.BURST_E, + DMRPPacketData.VoiceType.BURST_F): + self.__l2 = DMRPL2VoiceBurst(self.dmr_data) + + return self.__l2 + + def get_full_lc(self) -> bytes|None: + l2 = self.get_l2() + if type(l2) is DMRPL2FullLC: + return l2.get_full_lc() + return None + + def get_emb_lc(self) -> bytes|None: + l2 = self.get_l2() + if type(l2) is DMRPL2VoiceBurst: + return l2.get_emb_lc() + return None + + def get_lc(self) -> bytes: + if ((lcdata := self.get_full_lc()) is not None or + (lcdata := self.get_emb_lc()) is not None): + return lcdata + return b"" + def __str__(self) -> str: return self.format() @@ -391,7 +432,13 @@ class DMRPPacketData(DMRPBasePeerPacket): f"src:{self.src_id} dst:{self.dst_id} " f"bits:{self.bits:08b}") case 'ext': - return str(self) + f" data:{self.dmr_data.hex()}" + return (self.format('') + + f" data:{self.dmr_data.hex()}") + + case 'l2': + return (self.format('') + + f" lc:{self.get_lc().hex()}" + + f" data:{self.dmr_data.hex()}") # default return (f"{self.pkt_type} " diff --git a/requirements.txt b/requirements.txt index c855156..f6a3f5f 100755 --- a/requirements.txt +++ b/requirements.txt @@ -1,9 +1,12 @@ annotated-types==0.7.0 anyio==4.9.0 asyncio==3.4.3 +bitarray==3.4.1 +bitstring==4.3.1 certifi==2025.4.26 click==8.1.8 colorama==0.4.6 +dmr-utils3==0.1.29 dnspython==2.7.0 email_validator==2.2.0 fastapi==0.115.12