Abstract Data Record (#294)

Define common API for all Data Records.

README.rst

@@ -65,7 +65,7 @@ Link: https://github.com/pylessard/python-udsoncan
     https://udsoncan.readthedocs.io/en/latest/udsoncan/connection.html#available-connections
   - CAN bus fully supported with possibility to extension for other buses (requires custom code)
   - possibility to configure all transmission parameters for CAN using can-isotp package -
-    https://can-isotp.readthedocs.io/en/latest/isotp/implementation.html#
+    https://can-isotp.readthedocs.io/en/latest/isotp/implementation.html
   - handlers for multiple diagnostic services are implemented -
     https://udsoncan.readthedocs.io/en/latest/udsoncan/services.html
   - positive and negatives scenarios are handled - https://udsoncan.readthedocs.io/en/latest/udsoncan/exceptions.html
@@ -78,7 +78,7 @@ Link: https://github.com/pylessard/python-udsoncan
 - cons:
 
   - no support for full-duplex communication (sending and receiving at the same time)
-  - only Client side communication is implemented - https://udsoncan.readthedocs.io/en/latest/udsoncan/client.html#
+  - only Client side communication is implemented - https://udsoncan.readthedocs.io/en/latest/udsoncan/client.html
 
 
 python-uds

tests/software_tests/database/test_abstract_data_record.py

@@ -0,0 +1,37 @@
+import pytest
+from mock import Mock, patch
+
+from uds.database.abstract_data_record import AbstractDataRecord
+
+SCRIPT_LOCATION = "uds.database.abstract_data_record"
+
+
+class TestAbstractDataRecord:
+
+    def setup_method(self):
+        self.mock_data_record = Mock(spec=AbstractDataRecord)
+
+    # __init__
+
+    @patch(f"{SCRIPT_LOCATION}.isinstance")
+    def test_init__type_error(self, mock_isinstance):
+        mock_isinstance.return_value = False
+        mock_name = Mock()
+        with pytest.raises(TypeError):
+            AbstractDataRecord.__init__(self.mock_data_record, mock_name)
+        mock_isinstance.assert_called_once_with(mock_name, str)
+
+    @patch(f"{SCRIPT_LOCATION}.isinstance")
+    def test_init__valid(self, mock_isinstance):
+        mock_isinstance.return_value = True
+        mock_name = Mock()
+        assert AbstractDataRecord.__init__(self.mock_data_record, mock_name) is None
+        assert self.mock_data_record._AbstractDataRecord__name == mock_name.strip.return_value
+        mock_isinstance.assert_called_once_with(mock_name, str)
+        mock_name.strip.assert_called_once_with()
+
+    # name
+
+    def test_name(self):
+        self.mock_data_record._AbstractDataRecord__name = Mock()
+        assert AbstractDataRecord.name.fget(self.mock_data_record) == self.mock_data_record._AbstractDataRecord__name

uds/__init__.py

@@ -28,6 +28,7 @@
 
 
 import uds.can
+import uds.database
 import uds.message
 import uds.packet
 import uds.segmentation

uds/can/flow_control.py

@@ -57,9 +57,9 @@ class CanFlowStatus(ValidatedEnum, NibbleEnum):
 
     ContinueToSend: "CanFlowStatus" = 0x0  # type: ignore
     """Asks to resume Consecutive Frames transmission."""
-    Wait: "CanFlowStatus" = 0x1  # type: ignore  # noqa: F841
+    Wait: "CanFlowStatus" = 0x1  # type: ignore
     """Asks to pause Consecutive Frames transmission."""
-    Overflow: "CanFlowStatus" = 0x2  # type: ignore  # noqa: F841
+    Overflow: "CanFlowStatus" = 0x2  # type: ignore
     """Asks to abort transmission of a diagnostic message."""
 
 

uds/database/__init__.py

@@ -0,0 +1,7 @@
+"""
+Implementation for diagnostic messages databases.
+
+Tools for decoding and encoding information from/to diagnostic messages.
+"""
+
+from .abstract_data_record import AbstractDataRecord, DataRecordType, DecodedDataRecord

uds/database/abstract_data_record.py

@@ -0,0 +1,123 @@
+"""
+Definition of all Data Records types.
+
+Each Data Record contains mapping (translation) of raw data (sequence of bits in diagnostic message payload) to some
+meaningful information (e.g. physical value, text).
+"""
+
+__all__ = ["DataRecordType", "AbstractDataRecord", "DecodedDataRecord"]
+
+from abc import ABC, abstractmethod
+from typing import Tuple, TypedDict, Union, Optional
+
+from uds.utilities import ValidatedEnum
+
+DataRecordPhysicalValueAlias = Union[int, float, str, Tuple["DecodedDataRecord", ...]]
+"""Alias of Data Records' physical value."""
+
+
+class DecodedDataRecord(TypedDict):
+    """Structure of decoded Data Record."""
+
+    name: str
+    raw_value: int
+    physical_value: DataRecordPhysicalValueAlias  # noqa: F841
+
+
+class DataRecordType(ValidatedEnum):
+    """All Data Record types."""
+
+    # TODO: fill with following tasks:
+    #  - https://github.com/mdabrowski1990/uds/issues/2
+    #  - https://github.com/mdabrowski1990/uds/issues/6
+    #  - https://github.com/mdabrowski1990/uds/issues/8
+    #  - https://github.com/mdabrowski1990/uds/issues/9
+    #  - https://github.com/mdabrowski1990/uds/issues/10
+
+
+class AbstractDataRecord(ABC):
+    """Common implementation and interface for all Data Records."""
+
+    def __init__(self, name: str) -> None:
+        """
+        Initialize common part for all Data Records.
+
+        :param name: Name to assign to this Data Record.
+
+        :raise TypeError: Provided value of name is not str type.
+        """
+        if not isinstance(name, str):
+            raise TypeError("Provided name is not str type.")
+        self.__name = name.strip()
+
+    @property
+    def name(self) -> str:
+        """Name of this Data Record."""
+        return self.__name
+
+    @property  # noqa: F841
+    @abstractmethod
+    def data_record_type(self) -> DataRecordType:
+        """Type of this Data Record."""
+
+    @property  # noqa: F841
+    @abstractmethod
+    def length(self) -> int:
+        """Get number of bits that this Data Record is stored over."""
+
+    @property  # noqa: F841
+    @abstractmethod
+    def is_reoccurring(self) -> bool:
+        """
+        Whether this Data Record might occur multiple times.
+
+        Values meaning:
+        - False - exactly one occurrence in every diagnostic message
+        - True - number of occurrences might vary
+        """
+
+    @property  # noqa: F841
+    @abstractmethod
+    def min_occurrences(self) -> int:
+        """
+        Minimal number of this Data Record occurrences.
+
+        .. note:: Relevant only if :attr:`~uds.database.abstract_data_record.AbstractDataRecord.is_reoccurring`
+            equals True.
+        """
+
+    @property  # noqa: F841
+    @abstractmethod
+    def max_occurrences(self) -> Optional[int]:
+        """
+        Maximal number of this Data Record occurrences.
+
+        .. note:: Relevant only if :attr:`~uds.database.abstract_data_record.AbstractDataRecord.is_reoccurring`
+            equals True.
+        .. warning:: No maximal number (infinite number of occurrences) is represented by None value.
+        """
+
+    @property  # noqa: F841
+    @abstractmethod
+    def contains(self) -> Tuple["AbstractDataRecord", ...]:
+        """Get Data Records contained by this Data Record."""
+
+    @abstractmethod
+    def decode(self, raw_value: int) -> DecodedDataRecord:  # noqa: F841
+        """
+        Decode physical value for provided raw value.
+
+        :param raw_value: Raw (bit) value of Data Record.
+
+        :return: Dictionary with physical value for this Data Record.
+        """
+
+    @abstractmethod
+    def encode(self, physical_value: DataRecordPhysicalValueAlias) -> int:  # noqa: F841
+        """
+        Encode raw value for provided physical value.
+
+        :param physical_value: Physical (meaningful e.g. float, str type) value of this Data Record.
+
+        :return: Raw Value of this Data Record.
+        """

uds/transmission_attributes/transmission_direction.py

@@ -11,7 +11,7 @@
 class TransmissionDirection(ValidatedEnum, StrEnum):
     """Direction of a communication."""
 
-    RECEIVED: "TransmissionDirection" = "Rx"  # type: ignore  # noqa: F841
+    RECEIVED: "TransmissionDirection" = "Rx"  # type: ignore
     """Incoming transmission from the perspective of the code."""
-    TRANSMITTED: "TransmissionDirection" = "Tx"  # type: ignore  # noqa: F841
+    TRANSMITTED: "TransmissionDirection" = "Tx"  # type: ignore
     """Outgoing transmission from the perspective of the code."""