Source code for uds.addressing.abstract_addressing_information

"""Definition of UDS Addressing Information for storing Client/Server Addresses."""

__all__ = ["AbstractAddressingInformation"]

from abc import ABC, abstractmethod
from types import MappingProxyType
from typing import Any, Dict, Mapping, Optional

from uds.utilities import ReassignmentError

from .addressing_type import AddressingType




class AbstractAddressingInformation(ABC):
    """Storage for addressing related parameters for any UDS entity."""

    ADDRESSING_TYPE_NAME: str = "addressing_type"
    """Name of :ref:`Addressing Type <knowledge-base-addressing>` parameter in Addressing Information."""

    def __init__(self,
                 rx_physical_params: Mapping[str, Any],
                 tx_physical_params: Mapping[str, Any],
                 rx_functional_params: Mapping[str, Any],
                 tx_functional_params: Mapping[str, Any]) -> None:
        """
        Configure Addresses of UDS Entity (either a server or a client).

        :param rx_physical_params: Addressing parameters for incoming physically addressed communication.
        :param tx_physical_params: Addressing parameters for outgoing physically addressed communication.
        :param rx_functional_params: Addressing parameters for incoming functionally addressed communication.
        :param tx_functional_params: Addressing parameters for outgoing functionally addressed communication.
        """
        self.rx_physical_params = rx_physical_params
        self.tx_physical_params = tx_physical_params
        self.rx_functional_params = rx_functional_params
        self.tx_functional_params = tx_functional_params
        self._validate_addressing_information()



    def __eq__(self, other: Any) -> bool:
        """
        Compare with other object.

        :param other: Object to compare.

        :return: True if other object has the same type and carries the same Addressing Information, otherwise False.
        """
        if not isinstance(other, AbstractAddressingInformation):
            return False
        return (self.rx_physical_params == other.rx_physical_params
                and self.tx_physical_params == other.tx_physical_params
                and self.rx_functional_params == other.rx_functional_params
                and self.tx_functional_params == other.tx_functional_params)


    @property
    def rx_physical_params(self) -> Mapping[str, Any]:
        """Get addressing parameters for incoming physically addressed communication."""
        return self.__rx_physical_params

    @rx_physical_params.setter
    def rx_physical_params(self, addressing_params: Mapping[str, Any]) -> None:
        """
        Set addressing parameters for incoming physically addressed communication.

        :param addressing_params: Addressing parameters to set.

        :raise ReassignmentError: An attempt to change the value after object creation.
        """
        if hasattr(self, "_AbstractAddressingInformation__rx_physical_params"):
            raise ReassignmentError("Value of 'rx_physical_params' attribute cannot be changed once assigned.")
        params = dict(addressing_params)
        params[self.ADDRESSING_TYPE_NAME] = AddressingType.PHYSICAL
        self.__rx_physical_params: MappingProxyType[str, Any] \
            = MappingProxyType(self.validate_addressing_params(**params))

    @property
    def tx_physical_params(self) -> Mapping[str, Any]:
        """Get addressing parameters for outgoing physically addressed communication."""
        return self.__tx_physical_params

    @tx_physical_params.setter
    def tx_physical_params(self, addressing_params: Mapping[str, Any]) -> None:
        """
        Set addressing parameters for outgoing physically addressed communication.

        :param addressing_params: Addressing parameters to set.

        :raise ReassignmentError: An attempt to change the value after object creation.
        """
        if hasattr(self, "_AbstractAddressingInformation__tx_physical_params"):
            raise ReassignmentError("Value of 'tx_physical_params' attribute cannot be changed once assigned.")
        params = dict(addressing_params)
        params[self.ADDRESSING_TYPE_NAME] = AddressingType.PHYSICAL
        self.__tx_physical_params: MappingProxyType[str, Any] \
            = MappingProxyType(self.validate_addressing_params(**params))

    @property
    def rx_functional_params(self) -> Mapping[str, Any]:
        """Get addressing parameters for incoming functionally addressed communication."""
        return self.__rx_functional_params

    @rx_functional_params.setter
    def rx_functional_params(self, addressing_params: Mapping[str, Any]) -> None:
        """
        Set addressing parameters for incoming functionally addressed communication.

        :param addressing_params: Addressing parameters to set.

        :raise ReassignmentError: An attempt to change the value after object creation.
        """
        if hasattr(self, "_AbstractAddressingInformation__rx_functional_params"):
            raise ReassignmentError("Value of 'rx_functional_params' attribute cannot be changed once assigned.")
        params = dict(addressing_params)
        params[self.ADDRESSING_TYPE_NAME] = AddressingType.FUNCTIONAL
        self.__rx_functional_params: MappingProxyType[str, Any] \
            = MappingProxyType(self.validate_addressing_params(**params))

    @property
    def tx_functional_params(self) -> Mapping[str, Any]:
        """Get addressing parameters for outgoing functionally addressed communication."""
        return self.__tx_functional_params

    @tx_functional_params.setter
    def tx_functional_params(self, addressing_params: Mapping[str, Any]) -> None:
        """
        Set addressing parameters for outgoing functionally addressed communication.

        :param addressing_params: Addressing parameters to set.

        :raise ReassignmentError: An attempt to change the value after object creation.
        """
        if hasattr(self, "_AbstractAddressingInformation__tx_functional_params"):
            raise ReassignmentError("Value of 'tx_functional_params' attribute cannot be changed once assigned.")
        params = dict(addressing_params)
        params[self.ADDRESSING_TYPE_NAME] = AddressingType.FUNCTIONAL
        self.__tx_functional_params: MappingProxyType[str, Any] \
            = MappingProxyType(self.validate_addressing_params(**params))



    @abstractmethod
    def _validate_addressing_information(self) -> None:
        """Check whether the provided addressing information are valid."""




    @abstractmethod
    def validate_addressing_params(self, **addressing_params: Any) -> Dict[str, Any]:
        """Check whether the provided parameters are complete and correct."""




    @abstractmethod
    def is_input_packet(self, **frame_attributes: Any) -> Optional[AddressingType]:  # noqa: vulture
        """
        Check if a frame with provided attributes is an input packet for this UDS Entity.

        :param frame_attributes: Attributes of a frame to be checked.

        :return: Addressing Type used for transmission of this packet, None otherwise.
        """




    def get_other_end(self) -> "AbstractAddressingInformation":
        """
        Get Addressing Information of UDS entity on the other end of UDS communication.

        :return: UDS Addressing Information of a UDS entity that this one communicates with.
        """
        return self.__class__(rx_physical_params=self.tx_physical_params,
                              tx_physical_params=self.rx_physical_params,
                              rx_functional_params=self.tx_functional_params,
                              tx_functional_params=self.rx_functional_params)