From 4db12d074c046d629bbd3c905f26537d51b987e1 Mon Sep 17 00:00:00 2001 From: Alexander UR6LKW Date: Wed, 28 May 2025 20:36:47 +0300 Subject: [PATCH] REFACT Switch to unified factory class --- dmrtools/dmrproto/exceptions.py | 20 +++++++----- dmrtools/dmrproto/factory.py | 50 ++++++++++++++++++++++++++++++ dmrtools/dmrproto/mmdvm_l1.py | 55 +++++---------------------------- 3 files changed, 70 insertions(+), 55 deletions(-) create mode 100755 dmrtools/dmrproto/factory.py diff --git a/dmrtools/dmrproto/exceptions.py b/dmrtools/dmrproto/exceptions.py index 26567fe..d1e02bb 100755 --- a/dmrtools/dmrproto/exceptions.py +++ b/dmrtools/dmrproto/exceptions.py @@ -10,14 +10,6 @@ class DMRPFieldOutOfRangeException(Exception): super().__init__(f"Field out of range: {field} must be {typename}") -class DMRPUnknownPacketTypeException(Exception): - """ - Exception raised when the packet factory cannot recognize the packet type - from the given input data. - """ - pass - - class DMRPBadPacketException(Exception): """ Exception raised when a packet is structurally invalid or corrupted @@ -30,3 +22,15 @@ class DMRPL2BadDataException(Exception): Exception raised when L2 data is invalid or corrupted. """ pass + + +class FactoryException(Exception): + pass + + +class DMRPUnknownPacketTypeException(FactoryException): + """ + Exception raised when the packet factory cannot recognize the packet type + from the given input data. + """ + pass diff --git a/dmrtools/dmrproto/factory.py b/dmrtools/dmrproto/factory.py new file mode 100755 index 0000000..462016e --- /dev/null +++ b/dmrtools/dmrproto/factory.py @@ -0,0 +1,50 @@ +from abc import ABC, abstractmethod +from typing import Self, Type, NoReturn + +from .exceptions import FactoryException + + +class IFactoryProduced(ABC): + @abstractmethod + def __init__(self, data: bytes) -> None: + pass + + @classmethod + @abstractmethod + def detect_by_data(cls, data: bytes) -> bool: + pass + + +class BaseFactory(ABC): + __instance: Self|None = None + + @classmethod + def fd(cls, data: bytes) -> IFactoryProduced: + """ + Short singleton version of from_data method + """ + if cls.__instance is None: + cls.__instance = cls() + return cls.__instance.from_data(data) + + def __init__(self, + classes: list[Type[IFactoryProduced]]|None = None) -> None: + self._classes: list[Type[IFactoryProduced]] = ( + classes if classes is not None else []) + + def register(self, cls: Type[IFactoryProduced]) -> None: + self._classes.append(cls) + + def from_data(self, data: bytes) -> IFactoryProduced: + """ + Attempts to create an instance of corresponding + IFactoryProduced-implementing class based on data + """ + for cls in self._classes: + if cls.detect_by_data(data): + return cls(data) + + self.not_found(data) + + def not_found(self, data: bytes) -> NoReturn: + raise FactoryException(f"No class found for {data.hex()}") diff --git a/dmrtools/dmrproto/mmdvm_l1.py b/dmrtools/dmrproto/mmdvm_l1.py index cdfb113..a857716 100755 --- a/dmrtools/dmrproto/mmdvm_l1.py +++ b/dmrtools/dmrproto/mmdvm_l1.py @@ -4,14 +4,15 @@ import random from abc import ABC from hashlib import sha256 -from typing import Type, Self, TypeAlias +from typing import Self, TypeAlias, NoReturn from .base_fields import DMRPFieldInt, DMRPFieldBytes, DMRPFieldStr from .enums import CallType, VoiceType +from .etsi_l2 import DMRPL2Base, DMRPL2FullLC, DMRPL2VoiceBurst from .exceptions import DMRPBadPacketException from .exceptions import DMRPFieldOutOfRangeException from .exceptions import DMRPUnknownPacketTypeException -from .etsi_l2 import DMRPL2Base, DMRPL2FullLC, DMRPL2VoiceBurst +from .factory import BaseFactory, IFactoryProduced ############################# @@ -24,7 +25,7 @@ def calc_password_hash(salt: bytes, password: str) -> bytes: ############################# # Packet classes hierarchy ############################# -class DMRPBasePacket(ABC): +class DMRPBasePacket(IFactoryProduced, ABC): """ Abstract base packet class. @@ -436,66 +437,26 @@ class DMRPPacketData(DMRPBasePeerPacket): f"vt:{'T' if self.is_voice_term else 'f'}") -class DMRPPacketFactory: +class DMRPPacketFactory(BaseFactory): """ A factory class responsible for creating instances of DMRP packet classes based on packet data. This class supports both predefined packet types and user-registered custom packet types. """ - __instance: DMRPPacketFactory|None = None - - @classmethod - def fd(cls, data: bytes) -> DMRPBasePacket: - """ - Short singleton version of from_data method - """ - if cls.__instance is None: - cls.__instance = cls() - return cls.__instance.from_data(data) - def __init__(self) -> None: """ Initializes the packet factory with a list of all packet classes. - These classes must implement a `detect_by_data` class method to - determine whether they can handle the given input data. """ - self.__pclasses = [ + super().__init__([ DMRPPacketMasterNoAck, DMRPPacketMasterClose, DMRPPacketRepeaterClose, DMRPPacketLogin, DMRPPacketAck, DMRPPacketAuth, DMRPPacketConfig, DMRPPacketPing, DMRPPacketPong, DMRPPacketSalt, DMRPPacketBeacon, DMRPPacketData, DMRPPacketTalkerAlias, - ] - - def register_custom_packet(self, cls: Type[DMRPBasePacket]) -> None: - """ - Registers a custom packet class to the factory. - - Args: - cls: A class that implements the static method `detect_by_data`. - If this method returns True, the class is used to create a packet instance. - """ - self.__pclasses.append(cls) - - def from_data(self, data: bytes) -> DMRPBasePacket: - """ - Attempts to create an DMRP packet instance of corresponding packet - class based on packet data. - - Args: - data (bytes): The raw data from which to create a packet. - - Returns: - DMRPBasePacket: An instance of a subclass of DMRPBasePacket that matches the data. - - Raises: - DMRPUnknownPacketTypeException: If no registered packet class can handle the data. - """ - for cls in self.__pclasses: - if cls.detect_by_data(data): - return cls(data) + ]) + def not_found(self, data: bytes) -> NoReturn: ptypestr = data[0:4].decode(encoding='ascii', errors='ignore') raise DMRPUnknownPacketTypeException(f"Unknown packet type {ptypestr}")