API for Databases (#298)

- Abstract Database class defined
- tiny fix for Abstract Service (distinguishing between request and response message)
- adjust aliases of Data Records

tests/software_tests/database/service/test_abstract_service.py

@@ -0,0 +1,59 @@
+import pytest
+from mock import Mock
+
+from uds.database.service.abstract_service import AbstractService
+
+
+class TestAbstractService:
+    """Unit tests for Abstract Service."""
+
+    def setup_method(self):
+        self.mock_abstract_service = Mock(spec=AbstractService)
+
+    # encode
+
+    @pytest.mark.parametrize("sid, request_sid, response_sid", [
+        (1, 1, 2),
+        (0x11, 0x11, 0x51),
+    ])
+    @pytest.mark.parametrize("data_records_values", [
+        {},
+        {"a": 1, "b": 2, "c": 3, "xyz": None}
+    ])
+    def test_encode__request(self, sid, request_sid, response_sid, data_records_values):
+        self.mock_abstract_service.request_sid = request_sid
+        self.mock_abstract_service.response_sid = response_sid
+        assert (AbstractService.encode(self=self.mock_abstract_service, sid=sid, **data_records_values)
+                == self.mock_abstract_service.encode_request.return_value)
+        self.mock_abstract_service.encode_request.assert_called_once_with(**data_records_values)
+
+    @pytest.mark.parametrize("sid, request_sid, response_sid", [
+        (2, 1, 2),
+        (0x51, 0x11, 0x51),
+    ])
+    @pytest.mark.parametrize("data_records_values", [
+        {},
+        {"a": 1, "b": 2, "c": 3, "xyz": None}
+    ])
+    def test_encode__response(self, sid, request_sid, response_sid, data_records_values):
+        self.mock_abstract_service.request_sid = request_sid
+        self.mock_abstract_service.response_sid = response_sid
+        assert (AbstractService.encode(self=self.mock_abstract_service, sid=sid, **data_records_values)
+                == self.mock_abstract_service.encode_response.return_value)
+        self.mock_abstract_service.encode_response.assert_called_once_with(**data_records_values)
+
+    @pytest.mark.parametrize("sid, request_sid, response_sid", [
+        (0, 1, 2),
+        (0x11, 0x10, 0x50),
+    ])
+    @pytest.mark.parametrize("data_records_values", [
+        {},
+        {"a": 1, "b": 2, "c": 3, "xyz": None}
+    ])
+    def test_encode__value_error(self, sid, request_sid, response_sid, data_records_values):
+        self.mock_abstract_service.request_sid = request_sid
+        self.mock_abstract_service.response_sid = response_sid
+        with pytest.raises(ValueError):
+            AbstractService.encode(self=self.mock_abstract_service, sid=sid, **data_records_values)
+        self.mock_abstract_service.encode_request.asseert_not_called()
+        self.mock_abstract_service.encode_response.asseert_not_called()

tests/software_tests/database/test_abstract_database.py

@@ -0,0 +1,72 @@
+import pytest
+from mock import Mock, patch
+
+from uds.database.abstract_database import AbstractDatabase, RequestSID, ResponseSID, UdsMessage, UdsMessageRecord
+from uds.transmission_attributes import AddressingType
+
+SCRIPT_LOCATION = "uds.database.abstract_database"
+
+
+class TestAbstractDatabase:
+
+    def setup_method(self):
+        self.mock_database = Mock(spec=AbstractDatabase)
+
+    # encode
+
+    @pytest.mark.parametrize("sid", [Mock(), 1])
+    @pytest.mark.parametrize("data_records_values", [
+        {"data_record_1": 1, "data_record_2": "Some Value", "data_record_3": 543.234},
+        {"a": 2.3, "b": 543, "c": "xyz"},
+    ])
+    @patch(f"{SCRIPT_LOCATION}.isinstance")
+    def test_encode__type_error(self, mock_isinstance, sid, data_records_values):
+        mock_isinstance.return_value = False
+        with pytest.raises(TypeError):
+            AbstractDatabase.encode(self.mock_database, sid=sid, **data_records_values)
+        mock_isinstance.assert_called_once_with(sid, int)
+
+    @pytest.mark.parametrize("sid", [Mock(), 1, RequestSID.RequestDownload, ResponseSID.WriteDataByIdentifier])
+    @pytest.mark.parametrize("data_records_values", [
+        {"data_record_1": 1, "data_record_2": "Some Value", "data_record_3": 543.234},
+        {"a": 2.3, "b": 543, "c": "xyz"},
+    ])
+    @patch(f"{SCRIPT_LOCATION}.isinstance")
+    def test_encode__value_error(self, mock_isinstance, sid, data_records_values):
+        self.mock_database.services = {}
+        mock_isinstance.return_value = True
+        with pytest.raises(ValueError):
+            AbstractDatabase.encode(self.mock_database, sid=sid, **data_records_values)
+        mock_isinstance.assert_called_once_with(sid, int)
+
+    @pytest.mark.parametrize("sid", [1, RequestSID.RequestDownload, ResponseSID.WriteDataByIdentifier])
+    @pytest.mark.parametrize("data_records_values", [
+        {"data_record_1": 1, "data_record_2": "Some Value", "data_record_3": 543.234},
+        {"a": 2.3, "b": 543, "c": "xyz"},
+    ])
+    def test_encode(self, sid, data_records_values):
+        mock_service = Mock()
+        self.mock_database.services = {sid: mock_service}
+        assert (AbstractDatabase.encode(self.mock_database, sid=sid, **data_records_values)
+                == mock_service.encode.return_value)
+        mock_service.encode.assert_called_once_with(sid=sid, **data_records_values)
+
+    # decode
+
+    @pytest.mark.parametrize("message", [
+        UdsMessage(payload=[0x10, 0x03], addressing_type=AddressingType.PHYSICAL),
+        Mock(spec=UdsMessageRecord, payload=[0x62, *range(255)])
+    ])
+    def test_decode__value_error(self, message):
+        self.mock_database.services = {}
+        with pytest.raises(ValueError):
+            AbstractDatabase.decode(self.mock_database, message)
+
+    @pytest.mark.parametrize("message", [
+        UdsMessage(payload=[0x10, 0x03], addressing_type=AddressingType.PHYSICAL),
+        Mock(spec=UdsMessageRecord, payload=[0x62, *range(255)])
+    ])
+    def test_decode(self, message):
+        mock_service = Mock()
+        self.mock_database.services = {message.payload[0]: mock_service}
+        assert AbstractDatabase.decode(self.mock_database, message) == mock_service.decode.return_value

uds/database/__init__.py

@@ -4,5 +4,6 @@
 Tools for decoding and encoding information from/to diagnostic messages.
 """
 
+from .abstract_database import AbstractDatabase
 from .data_record import AbstractDataRecord, DecodedDataRecord, RawDataRecord
-from .services import AbstractService
+from .service import AbstractService

uds/database/abstract_database.py

@@ -0,0 +1,61 @@
+"""Definition of UDS messages database for data encoding and decoding."""
+
+__all__ = ["AbstractDatabase"]
+
+from abc import ABC, abstractmethod
+from typing import Dict, List, Union
+
+from uds.message import RequestSID, ResponseSID, UdsMessage, UdsMessageRecord
+from uds.utilities import RawBytesListAlias
+
+from .data_record import DataRecordValueAlias, DecodedDataRecord
+from .service import AbstractService
+
+
+class AbstractDatabase(ABC):
+    """Common interface and implementation for UDS message databases."""
+
+    @property
+    @abstractmethod
+    def services(self) -> Dict[int, AbstractService]:
+        """
+        Get mapping of diagnostic services.
+
+        Keys are SID (int) values.
+        Values are diagnostic services with dedicated decoding and encoding implementation.
+        """
+
+    def encode(self,
+               sid: Union[int, RequestSID, ResponseSID],
+               **data_records_values: DataRecordValueAlias) -> RawBytesListAlias:
+        """
+        Encode diagnostic message payload from data records values.
+
+        :param sid: Service Identifier of a diagnostic message.
+        :param data_records_values: Value for each Data Record that is part a service message.
+
+        :raise TypeError: Provided SID value is neither int, RequestSID nor ResponseSID type.
+        :raise ValueError: This database has no implementation for provided SID value.
+
+        :return: Payload of a diagnostic message.
+        """
+        if not isinstance(sid, int):
+            raise TypeError("Provided SID value is not int type.")
+        if sid not in self.services:
+            raise ValueError("Database has no encoding defined for provided SID value.")
+        return self.services[sid].encode(sid=sid, **data_records_values)
+
+    def decode(self, message: Union[UdsMessage, UdsMessageRecord]) -> List[DecodedDataRecord]:
+        """
+        Decode physical values carried in payload of a diagnostic message.
+
+        :param message: A diagnostic message that is carrying payload to decode.
+
+        :raise ValueError: This database has no service implementation for provided diagnostic message SID.
+
+        :return: Decoded Data Records values from provided diagnostic message.
+        """
+        sid = message.payload[0]
+        if sid not in self.services:
+            raise ValueError("Database has no decoding defined for SID value of provided message.")
+        return self.services[sid].decode(message.payload)

uds/database/data_record/__init__.py

@@ -5,5 +5,10 @@
 meaningful information (e.g. physical value, text).
 """
 
-from .abstract_data_record import AbstractDataRecord, DecodedDataRecord
+from .abstract_data_record import (
+    AbstractDataRecord,
+    DataRecordPhysicalValueAlias,
+    DataRecordValueAlias,
+    DecodedDataRecord,
+)
 from .raw_data_record import RawDataRecord

uds/database/data_record/abstract_data_record.py

@@ -1,12 +1,53 @@
 """Definition of AbstractDataRecord which is a base class for all Data Records."""
 
-__all__ = ["AbstractDataRecord", "DataRecordPhysicalValueAlias", "DecodedDataRecord"]
+__all__ = ["AbstractDataRecord", "DataRecordPhysicalValueAlias", "DecodedDataRecord", "DataRecordValueAlias"]
 
 from abc import ABC, abstractmethod
-from typing import Optional, Tuple, TypedDict, Union
-
-DataRecordPhysicalValueAlias = Union[int, float, str, Tuple["DecodedDataRecord", ...]]
-"""Alias of Data Records' physical value."""
+from typing import Dict, Optional, Sequence, Tuple, TypedDict, Union
+
+DataRecordValueAlias = Union[
+    int,  # raw value
+    float,  # physical value calculated through formula
+    str,  # text (physical) value from either Text Table or Text encoding
+    Dict[str, "DataRecordValueAlias"],  # value of container's children
+    Sequence[Union[int, Dict[str, "DataRecordValueAlias"]]],  # values for reoccurring container
+]
+"""
+Alias of Data Records' input value.
+
+Each type represent other data:
+
+- int type - raw value of a data record
+- float type - physical value of a formula data record
+
+    .. warning:: Providing physical value as float might sometime cause issues due
+        `floating-point precision <https://docs.python.org/3/tutorial/floatingpoint.html>`_.
+        The closest raw value would be evaluated and put into a payload.
+
+        To avoid rounding, provide raw value (int type).
+
+- str type - text (physical) value of either text table or text data record
+- dict type - values for children of a container data records
+- sequence type - values for following occurrences of a container data record
+"""
+
+DataRecordPhysicalValueAlias = Union[
+    int,  # physical value is the same as raw value
+    float,  # physical value calculated through formula
+    str,  # decoded text value
+    Tuple[Tuple["DecodedDataRecord", ...], ...]  # decoded container value, each element is another record
+]
+"""
+Alias of Data Records' physical value.
+
+Each type represent other data:
+
+- int type - physical value is the same as raw value
+- float type - value received through formula calculation
+- str type - text value received either through encoding (e.g. ASCII, UTF-8)
+  or mapping (each value has specific meaning)
+- tuple type - one element for each container occurrence; each element is a tuple with values for children data records
+"""
 
 
 class DecodedDataRecord(TypedDict):

uds/database/service/abstract_service.py

@@ -0,0 +1,75 @@
+"""Definition of UDS Service data encoding and decoding."""
+
+__all__ = ["AbstractService"]
+
+from abc import ABC, abstractmethod
+from typing import List, Union
+
+from uds.message import RequestSID, ResponseSID
+from uds.utilities import RawBytesAlias, RawBytesListAlias
+
+from ..data_record import DataRecordValueAlias, DecodedDataRecord
+
+
+class AbstractService(ABC):
+    """Common interface for all diagnostic services."""
+
+    @property  # noqa: F841
+    @abstractmethod
+    def request_sid(self) -> RequestSID:
+        """Service Identifier in request messages."""
+
+    @property  # noqa: F841
+    @abstractmethod
+    def response_sid(self) -> ResponseSID:
+        """Service Identifier in (positive) response messages."""
+
+    @abstractmethod
+    def decode(self, payload: RawBytesAlias) -> List[DecodedDataRecord]:
+        """
+        Decode physical values carried in payload of a diagnostic message.
+
+        :param payload: Payload of a diagnostic message.
+
+        :return: Decoded Data Records values from provided diagnostic message.
+        """
+
+    def encode(self,
+               sid: Union[int, RequestSID, ResponseSID],
+               **data_records_values: DataRecordValueAlias) -> RawBytesListAlias:
+        """
+        Encode diagnostic message payload for this service.
+
+        :param sid: Value of Service Identifier. It should be either equal to either `request_sid` or `response_sid`.
+        :param data_records_values: Value for each data record that is part of a service message.
+
+        :raise ValueError: Provided `sid` value is neither equal to request SID value nor response SID value for this
+            diagnostic service.
+
+        :return: Payload of a diagnostic message.
+        """
+        if sid == self.request_sid:
+            return self.encode_request(**data_records_values)
+        if sid == self.response_sid:
+            return self.encode_response(**data_records_values)
+        raise ValueError("Provided SID value is neither request or response SID value for this service.")
+
+    @abstractmethod
+    def encode_request(self, **data_records_values: DataRecordValueAlias) -> RawBytesListAlias:
+        """
+        Encode diagnostic message payload for this service's request message.
+
+        :param data_records_values: Value for each data record that is part of a service message.
+
+        :return: Payload of a request diagnostic message.
+        """
+
+    @abstractmethod
+    def encode_response(self, **data_records_values: DataRecordValueAlias) -> RawBytesListAlias:
+        """
+        Encode diagnostic message payload for this service's response message.
+
+        :param data_records_values: Value for each data record that is part of a service message.
+
+        :return: Payload of a response diagnostic message.
+        """