rm.domain의 소스 코드

import struct

from abc import (
    ABC,
    abstractmethod,
)
from dataclasses import (
    dataclass,
)

from .error import (
    RmInvalidDomainValueError,
    RmInvalidDomainStringLengthError,
)


__all__ = (
    'RmValue',
    'RmDomain',
    'RmInt',
    'RmFloat',
    'RmString',
)


RmValue = int | float | str
"""
value의 타입.
"""




class RmDomain(ABC):
    """
    domain의 기반 추상 클래스입니다.
    """



    @abstractmethod
    def get_format(self) -> str:  # pragma: no cover
        """
        domain의 format을 반환합니다.

        Returns:
            str:
                domain의 format (`struct` 모듈에서 사용하는 format).
        """
        pass




    @abstractmethod
    def calc_size(self) -> int:  # pragma: no cover
        """
        domain의 크기를 반환합니다.

        Returns:
            int:
                domain의 크기.
        """
        pass




    @abstractmethod
    def validate(self, value: RmValue) -> None:  # pragma: no cover
        """
        value의 유효성을 검사합니다.

        Args:
            value (RmValue):
                대상 value.

        Raises:
            RmInvalidDomainValueError:
                `value` 가 유효하지 않은 경우.
        """
        pass




    @abstractmethod
    def encode(self, value: RmValue) -> bytes:  # pragma: no cover
        """
        value를 bytes로 인코딩합니다.

        Args:
            value (RmValue):
                대상 value.

        Returns:
            bytes:
                value를 bytes로 인코딩한 결과.
        """
        pass




    @abstractmethod
    def decode(self, data: bytes) -> RmValue:  # pragma: no cover
        """
        bytes를 value로 디코딩합니다.

        Args:
            data (bytes):
                대상 bytes.

        Returns:
            RmValue:
                bytes를 value로 디코딩한 결과.
        """
        pass






@dataclass(frozen=True)
class RmInt(RmDomain):
    """
    INT domain을 나타내는 클래스입니다.
    """



    def get_format(self) -> str:
        return '<i'




    def calc_size(self) -> int:
        return struct.calcsize(self.get_format())




    def validate(self, value: RmValue) -> None:
        if not isinstance(value, int):
            raise RmInvalidDomainValueError(
                f"RmInt의 `value`는 `int` 타입만 가능합니다 "
                f"(현재 타입: {type(value)}).",
            )




    def encode(self, value: RmValue) -> bytes:
        value = int(value)
        data = struct.pack(self.get_format(), value)
        return data




    def decode(self, data: bytes) -> int:
        (value,) = struct.unpack(self.get_format(), data)
        return int(value)






@dataclass(frozen=True)
class RmFloat(RmDomain):
    """
    FLOAT domain을 나타내는 클래스입니다.
    """



    def get_format(self) -> str:
        return '<f'




    def calc_size(self) -> int:
        return struct.calcsize(self.get_format())




    def validate(self, value: RmValue) -> None:
        if not isinstance(value, float):
            raise RmInvalidDomainValueError(
                f"RmFloat의 `value`는 `float` 타입만 가능합니다 "
                f"(현재 타입: {type(value)}).",
            )




    def encode(self, value: RmValue) -> bytes:
        value = float(value)
        data = struct.pack(self.get_format(), value)
        return data




    def decode(self, data: bytes) -> float:
        (value,) = struct.unpack(self.get_format(), data)
        return float(value)






@dataclass(frozen=True)
class RmString(RmDomain):
    """
    STRING domain을 나타내는 클래스입니다.

    Attributes:
        length (int):
            domain의 길이.
    """

    MAX_LENGTH = 255
    """
    domain의 최대 길이.
    """

    length: int
    """
    domain의 길이.
    """



    def __post_init__(self) -> None:
        """
        domain 초기화 후, `length` 를 검사합니다.

        Raises:
            RmInvalidDomainStringLengthError:
                `length` 가 유효하지 않은 경우.
        """
        if self.length < 1 or self.length > RmString.MAX_LENGTH:
            raise RmInvalidDomainStringLengthError(
                f"RmString의 길이는 1 이상, {RmString.MAX_LENGTH} 이하만 가능합니다 "
                f"(현재 길이: {self.length}).",
            )




    def get_format(self) -> str:
        return f'<{self.length}s'




    def calc_size(self) -> int:
        return self.length




    def validate(self, value: RmValue) -> None:
        if not isinstance(value, str):
            raise RmInvalidDomainValueError(
                f"RmString의 `value`는 `str` 타입만 가능합니다 "
                f"(현재 타입: {type(value)}).",
            )




    def encode(self, value: RmValue) -> bytes:
        value = str(value)
        data = value.encode()[:self.length]
        data = data.ljust(self.length, b'\x00')
        return data




    def decode(self, data: bytes) -> str:
        data = data[:self.length]
        value = data.decode()
        value = value.rstrip('\x00')
        return str(value)