Merge pull request #300 from andlaus/demf_drivebys

Dynamic endmarker fields drivebys

odxtools/dynamicendmarkerfield.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List, Sequence
 from xml.etree import ElementTree
 
 from typing_extensions import override
@@ -65,7 +65,7 @@ def encode_into_pdu(self, physical_value: ParameterValue, encode_state: EncodeSt
 
         odxassert(encode_state.cursor_bit_position == 0,
                   "No bit position can be specified for dynamic endmarker fields!")
-        if not isinstance(physical_value, (tuple, list)):
+        if not isinstance(physical_value, Sequence):
             odxraise(
                 f"Expected a sequence of values for dynamic endmarker field {self.short_name}, "
                 f"got {type(physical_value).__name__}", EncodeError)

odxtools/dynamiclengthfield.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List, Sequence
 from xml.etree import ElementTree
 
 from typing_extensions import override
@@ -55,7 +55,7 @@ def encode_into_pdu(self, physical_value: ParameterValue, encode_state: EncodeSt
         odxassert(encode_state.cursor_bit_position == 0,
                   "No bit position can be specified for dynamic length fields!")
 
-        if not isinstance(physical_value, list):
+        if not isinstance(physical_value, Sequence):
             odxraise(
                 f"Expected a list of values for dynamic length field {self.short_name}, "
                 f"got {type(physical_value)}", EncodeError)
@@ -77,8 +77,14 @@ def encode_into_pdu(self, physical_value: ParameterValue, encode_state: EncodeSt
         encode_state.cursor_byte_position = encode_state.origin_byte_position + self.offset
         encode_state.cursor_bit_position = 0
 
-        for value in physical_value:
+        orig_is_end_of_pdu = encode_state.is_end_of_pdu
+        encode_state.is_end_of_pdu = False
+        for i, value in enumerate(physical_value):
+            if i == len(physical_value) - 1:
+                encode_state.is_end_of_pdu = orig_is_end_of_pdu
+
             self.structure.encode_into_pdu(value, encode_state)
+        encode_state.is_end_of_pdu = orig_is_end_of_pdu
 
         # ensure the correct message size if the field is empty
         if len(physical_value) == 0:

odxtools/endofpdufield.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import List, Optional, Sequence
 from xml.etree import ElementTree
 
 from typing_extensions import override
@@ -46,21 +46,35 @@ def encode_into_pdu(self, physical_value: Optional[ParameterValue],
                         encode_state: EncodeState) -> None:
         odxassert(not encode_state.cursor_bit_position,
                   "No bit position can be specified for end-of-pdu fields!")
+        odxassert(encode_state.is_end_of_pdu,
+                  "End-of-pdu fields can only be located at the end of PDUs!")
 
-        if not isinstance(physical_value, list):
+        if not isinstance(physical_value, Sequence):
             odxraise(
                 f"Invalid type {type(physical_value).__name__} of physical "
                 f"value for end-of-pdu field, expected a list", EncodeError)
             return
 
-        for value in physical_value:
+        orig_is_end_of_pdu = encode_state.is_end_of_pdu
+        encode_state.is_end_of_pdu = False
+
+        for i, value in enumerate(physical_value):
+            if i == len(physical_value) - 1:
+                encode_state.is_end_of_pdu = orig_is_end_of_pdu
+
             self.structure.encode_into_pdu(value, encode_state)
 
+        encode_state.is_end_of_pdu = orig_is_end_of_pdu
+
     @override
     def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
         odxassert(not decode_state.cursor_bit_position,
                   "No bit position can be specified for end-of-pdu fields!")
 
+        orig_origin = decode_state.origin_byte_position
+        orig_cursor = decode_state.cursor_byte_position
+        decode_state.origin_byte_position = decode_state.cursor_byte_position
+
         result: List[ParameterValue] = []
         while decode_state.cursor_byte_position < len(decode_state.coded_message):
             # ATTENTION: the ODX specification is very misleading
@@ -69,4 +83,7 @@ def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
             # repeated are identical, not their values
             result.append(self.structure.decode_from_pdu(decode_state))
 
+        decode_state.origin_byte_position = orig_origin
+        decode_state.cursor_byte_position = max(orig_cursor, decode_state.cursor_byte_position)
+
         return result

odxtools/environmentdata.py

@@ -11,7 +11,14 @@
 
 @dataclass
 class EnvironmentData(BasicStructure):
-    """This class represents Environment Data that describes the circumstances in which the error occurred."""
+    """This class represents Environment Data that describes the
+    circumstances in which the error occurred.
+
+    This is one of the many multiplexer mechanisms specified by the
+    ODX standard, because an environment data parameter must only be
+    used if a DTC parameter has a certain set of values. (In this
+    sense, it is quite similar to NRC-CONST parameters.)
+    """
 
     all_value: Optional[bool]
     dtc_values: List[int]

odxtools/environmentdatadescription.py

@@ -20,8 +20,14 @@
 
 @dataclass
 class EnvironmentDataDescription(ComplexDop):
-    """This class represents Environment Data Description, which is a complex DOP
-    that is used to define the interpretation of environment data."""
+    """This class represents environment data descriptions
+
+    An environment data description provides a list of all environment
+    data objects that are potentially applicable to decode a given
+    response. (If a given environment data object is applicable
+    depends on the value of the DtcDOP that is associated with it.)
+
+    """
 
     # in ODX 2.0.0, ENV-DATAS seems to be a mandatory
     # sub-element of ENV-DATA-DESC, in ODX 2.2 it is not

odxtools/odxlink.py

@@ -211,8 +211,10 @@ def resolve(self, ref: OdxLinkRef, expected_type: Optional[Any] = None) -> Any:
 
                 return obj
 
-        raise KeyError(f"ODXLINK reference {ref} could not be resolved for any "
-                       f"of the document fragments {ref.ref_docs}")
+        odxraise(
+            f"ODXLINK reference {ref} could not be resolved for any "
+            f"of the document fragments {ref.ref_docs}", KeyError)
+        return None
 
     @overload
     def resolve_lenient(self, ref: OdxLinkRef, expected_type: None = None) -> Any:

odxtools/odxtypes.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: MIT
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Type, Union, overload
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterable, Optional, Tuple, Type, Union,
+                    overload)
 from xml.etree import ElementTree
 
 from .exceptions import odxassert, odxraise, odxrequire
@@ -28,7 +29,7 @@ def bytefield_to_bytearray(bytefield: str) -> bytearray:
 # multiple items, so this can be a list of objects.
 TableStructParameterValue = Tuple[str, "ParameterValue"]
 ParameterValue = Union[AtomicOdxType, "ParameterValueDict", TableStructParameterValue,
-                       List["ParameterValue"], "DiagnosticTroubleCode"]
+                       Iterable["ParameterValue"], "DiagnosticTroubleCode"]
 ParameterValueDict = Dict[str, ParameterValue]
 
 

odxtools/parameters/parameterwithdop.py

@@ -43,7 +43,7 @@ def from_et(et_element: ElementTree.Element,
     def __post_init__(self) -> None:
         odxassert(self.dop_snref is not None or self.dop_ref is not None,
                   f"Param {self.short_name} without a DOP-(SN)REF should not exist!")
-        self._dop: Optional[DopBase] = None
+        self._dop: DopBase
 
     @override
     def _build_odxlinks(self) -> Dict[OdxLinkId, Any]:
@@ -55,10 +55,7 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
 
         if self.dop_ref is not None:
             odxassert(self.dop_snref is None)
-            # TODO: do not do lenient resolves here. The problem is
-            # that currently not all kinds of DOPs are internalized
-            # (e.g., static and dynamic fields)
-            self._dop = odxlinks.resolve_lenient(self.dop_ref)
+            self._dop = odxlinks.resolve(self.dop_ref)
 
     @override
     def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
@@ -71,11 +68,9 @@ def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
 
     @property
     def dop(self) -> DopBase:
-        """may be a DataObjectProperty, a Structure or None"""
+        """This is usually a DataObjectProperty or a Structure object"""
 
-        return odxrequire(
-            self._dop, "Specifying a data object property is mandatory but it "
-            "could not be resolved")
+        return self._dop
 
     @override
     def get_static_bit_length(self) -> Optional[int]:

odxtools/staticfield.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List, Sequence
 from xml.etree import ElementTree
 
 from typing_extensions import override
@@ -51,15 +51,20 @@ def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
     def encode_into_pdu(self, physical_value: ParameterValue, encode_state: EncodeState) -> None:
 
         if not isinstance(physical_value,
-                          (tuple, list)) or len(physical_value) != self.fixed_number_of_items:
+                          Sequence) or len(physical_value) != self.fixed_number_of_items:
             odxraise(f"Value for static field '{self.short_name}' "
                      f"must be a list of size {self.fixed_number_of_items}")
 
-        for val in physical_value:
+        orig_is_end_of_pdu = encode_state.is_end_of_pdu
+        encode_state.is_end_of_pdu = False
+        for i, val in enumerate(physical_value):
             if not isinstance(val, dict):
                 odxraise(f"The individual parameter values for static field '{self.short_name}' "
                          f"must be dictionaries for structure '{self.structure.short_name}'")
 
+            if i == len(physical_value) - 1:
+                encode_state.is_end_of_pdu = orig_is_end_of_pdu
+
             pos_before = encode_state.cursor_byte_position
             self.structure.encode_into_pdu(val, encode_state)
             pos_after = encode_state.cursor_byte_position
@@ -75,12 +80,18 @@ def encode_into_pdu(self, physical_value: ParameterValue, encode_state: EncodeSt
                 encode_state.emplace_bytes(b'\x00' * (self.item_byte_size -
                                                       (pos_after - pos_before)))
 
+        encode_state.is_end_of_pdu = orig_is_end_of_pdu
+
     @override
     def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
 
         odxassert(decode_state.cursor_bit_position == 0,
                   "No bit position can be specified for static length fields!")
 
+        orig_origin = decode_state.origin_byte_position
+        orig_cursor = decode_state.cursor_byte_position
+        decode_state.origin_byte_position = decode_state.cursor_byte_position
+
         result: List[ParameterValue] = []
         for _ in range(self.fixed_number_of_items):
             orig_cursor = decode_state.cursor_byte_position
@@ -94,4 +105,7 @@ def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
 
             decode_state.cursor_byte_position = orig_cursor + self.item_byte_size
 
+        decode_state.origin_byte_position = orig_origin
+        decode_state.cursor_byte_position = max(orig_cursor, decode_state.cursor_byte_position)
+
         return result