From a3b486071d7a0b6f3bb95b5dad8c64255504a1f0 Mon Sep 17 00:00:00 2001 From: Alexander UR6LKW Date: Mon, 26 May 2025 20:30:52 +0300 Subject: [PATCH] ETSI Layer2 implementation; stage2: decode --- dmrtools/dmrproto/etsi_l2.py | 148 ++++++++++++++++++++++++++++++++ dmrtools/dmrproto/exceptions.py | 10 +++ dmrtools/dmrproto/util.py | 1 - 3 files changed, 158 insertions(+), 1 deletion(-) diff --git a/dmrtools/dmrproto/etsi_l2.py b/dmrtools/dmrproto/etsi_l2.py index 5d0bdef..d5c67d6 100755 --- a/dmrtools/dmrproto/etsi_l2.py +++ b/dmrtools/dmrproto/etsi_l2.py @@ -5,8 +5,14 @@ import enum from abc import ABC, abstractmethod from bitarray import bitarray from dmr_utils3.bptc import decode_full_lc +from typing import NoReturn +from .base_fields import DMRPFieldInt +from .enums import CallType +from .exceptions import DMRPFieldOutOfRangeException from .exceptions import DMRPL2BadDataException +from .exceptions import DMRPUnknownLCTypeException +from .factory import BaseFactory, IFactoryProduced class DMRPL2Base(ABC): @@ -48,3 +54,145 @@ class DMRPL2VoiceBurst(DMRPL2Base): def get_emb_lc(self) -> bytes: return self.bitdata[116:148].tobytes() + + +class LCBase(IFactoryProduced, ABC): + DATA_SIZE: int = 9 + FLCOS: list[int] = [] + + flco = DMRPFieldInt('flco', 0, 1) + + def __init__(self, data: bytes|None) -> None: + if data is None: + self.create() + else: + self.from_data(data) + + def create(self, flco: int|None = None): + self._data: bytearray = bytearray(b'\x00' * self.DATA_SIZE) + if flco == None and len(self.FLCOS) > 0: + self.flco = self.FLCOS[0] + + def from_data(self, data: bytes) -> None: + if len(data) != self.DATA_SIZE: + raise DMRPL2BadDataException( + f"LC data must be {self.DATA_SIZE} bytes long") + self._data = bytearray(data) + + @classmethod + def detect_by_data(cls, data: bytes) -> bool: + return len(data) == cls.DATA_SIZE and data[0] in cls.FLCOS + + +class LCCall(LCBase): + FLCOS = [0x00, 0x03] + CALL_TYPE_FLCO_MAP = { 0x00: CallType.GROUP, 0x03: CallType.UNIT } + + @property + def call_type(self) -> CallType: + flco = self._data[0] if len(self._data) > 0 else 0 + if flco not in self.CALL_TYPE_FLCO_MAP: + return CallType.GROUP + return self.CALL_TYPE_FLCO_MAP[flco] + + dst_id = DMRPFieldInt('dst_id', 3, 3) + src_id = DMRPFieldInt('src_id', 6, 3) + + def __str__(self) -> str: + return (f"LC Call {self.flco} {self.call_type.name} " + f"dst:{self.dst_id} src:{self.src_id}") + + +class LCLocation(LCBase): + FLCOS = [0x08] + + # lat + @property + def lat(self) -> float: + """Decode 24-bit signed latitude integer to decimal degrees.""" + lat_24bit = int.from_bytes(self._data[6:9], byteorder="big") + if lat_24bit & (1 << 23): # Check sign bit + lat_24bit -= (1 << 24) # Convert from two's complement + return lat_24bit * (180.0 / (1 << 24)) # Map to range [-90, +90] + + # lon + @property + def lon(self) -> float: + """Decode 25-bit signed longitude integer to decimal degrees.""" + lon_25bit = int.from_bytes(self._data[2:6], byteorder="big") + lon_25bit &= 0x1FFFFFF + if lon_25bit & (1 << 24): # Check sign bit + lon_25bit -= (1 << 25) # Convert from two's complement + return lon_25bit * (360.0 / (1 << 25)) # Map to range [-180, +180] + + def __str__(self) -> str: + return f"LC GPS {self.flco} {self.lat} {self.lon}" + + +class LCTalkerAlias(LCBase): + FLCOS = [0x04, 0x05, 0x06, 0x07] + + class Format(enum.Enum): + _7BIT = 0b00 + ISO8 = 0b01 + UTF8 = 0b10 + UTF16BF = 0b11 + + def __init__( + self, data: bytes|None, + format: Format = Format.ISO8) -> None: + super().__init__(data) + self.__format: LCTalkerAlias.Format = format + + # format + def get_format(self) -> Format: + if self.flco != 0x04: + return self.__format + return LCTalkerAlias.Format((self._data[2] & 0xC0) >> 6) + + def set_format(self, format: Format) -> None: + if self.flco != 0x04: + self.__format = format + self._data[2] = (self._data[2] & ~0xC0) | ((format.value << 6) & 0xC0) + + format = property(get_format, set_format) + + # len + def get_len(self) -> int: + if self.flco != 0x04: + return 0 + return (self._data[2] & 0x3E) >> 1 + + def set_len(self, len: int) -> None: + if self.flco != 0x04: + return + if not (0 <= len < 1<<5): + raise DMRPFieldOutOfRangeException("len", "uint5") + if self._data is not None: + self._data[2] = (self._data[2] & ~0x3E) | ((len << 1) & 0x3E) + + len = property(get_len, set_len) + + @property + def ta_str(self) -> str: + # TODO: decode different fmts + if self.flco == 0x04: + return self._data[3:9].decode(encoding='ascii') + return self._data[2:9].decode(encoding='ascii') + + def __str__(self) -> str: + return (f"LC TA {self.flco} fmt:{self.format.name} " + f"len:{self.len} str:{self.ta_str}") + + +class LCFactory(BaseFactory): + def __init__(self) -> None: + """ + Initializes the lc analyzer factory with a list of all lc + analyzer classes. + """ + super().__init__([LCCall, LCLocation, LCTalkerAlias]) + + def not_found(self, data: bytes) -> NoReturn: + lctype = data[0:1].hex() if len(data) > 0 else '' + raise DMRPUnknownLCTypeException(f"Unknown lc type '{lctype}'") diff --git a/dmrtools/dmrproto/exceptions.py b/dmrtools/dmrproto/exceptions.py index d1e02bb..3215567 100755 --- a/dmrtools/dmrproto/exceptions.py +++ b/dmrtools/dmrproto/exceptions.py @@ -34,3 +34,13 @@ class DMRPUnknownPacketTypeException(FactoryException): from the given input data. """ pass + + +class DMRPUnknownLCTypeException(FactoryException): + """ + Exception raised when the lc analyzer factory cannot recognize the + lc type from the given input data. + """ + pass + + diff --git a/dmrtools/dmrproto/util.py b/dmrtools/dmrproto/util.py index 1824b2b..789b346 100755 --- a/dmrtools/dmrproto/util.py +++ b/dmrtools/dmrproto/util.py @@ -58,7 +58,6 @@ class EmbLCAssembler: self.vseq = p.vseq # logging.debug(f"assembled emblc's: {self.lcs}") - return burst_n == 3 def decode(self) -> bytes|None: