Merge pull request #321 from andlaus/env_data_coding

implement de- and encoding of environment data descriptions

Author: andlaus

examples/somersault.pdx

@@ -34,8 +34,6 @@
 from odxtools.diaglayertype import DiagLayerType
 from odxtools.diagservice import DiagService
 from odxtools.docrevision import DocRevision
-from odxtools.environmentdata import EnvironmentData
-from odxtools.environmentdatadescription import EnvironmentDataDescription
 from odxtools.exceptions import odxrequire
 from odxtools.functionalclass import FunctionalClass
 from odxtools.modification import Modification
@@ -45,7 +43,6 @@
 from odxtools.parameters.codedconstparameter import CodedConstParameter
 from odxtools.parameters.matchingrequestparameter import MatchingRequestParameter
 from odxtools.parameters.nrcconstparameter import NrcConstParameter
-from odxtools.parameters.physicalconstantparameter import PhysicalConstantParameter
 from odxtools.parameters.tablekeyparameter import TableKeyParameter
 from odxtools.parameters.tablestructparameter import TableStructParameter
 from odxtools.parameters.valueparameter import ValueParameter
@@ -1270,65 +1267,6 @@ class SomersaultSID(IntEnum):
         )
 }
 
-# env-data
-somersault_env_datas = {
-    "flip_env_data":
-        EnvironmentData(
-            odx_id=OdxLinkId("somersault.env_data.flip_env_data", doc_frags),
-            short_name="flip_env_data",
-            long_name="Flip Env Data",
-            description=None,
-            admin_data=None,
-            sdgs=[],
-            byte_size=None,
-            dtc_values=[],
-            parameters=NamedItemList([
-                ValueParameter(
-                    short_name="flip_speed",
-                    long_name="Flip Speed",
-                    description=None,
-                    physical_default_value_raw=None,
-                    byte_position=0,
-                    semantic="DATA",
-                    dop_ref=OdxLinkRef.from_id(somersault_dops["num_flips"].odx_id),
-                    dop_snref=None,
-                    bit_position=None,
-                    sdgs=[],
-                ),
-                PhysicalConstantParameter(
-                    short_name="flip_direction",
-                    long_name="Flip Direction",
-                    description=None,
-                    byte_position=1,
-                    semantic="DATA",
-                    physical_constant_value_raw="1",
-                    dop_ref=OdxLinkRef.from_id(somersault_dops["num_flips"].odx_id),
-                    dop_snref=None,
-                    bit_position=None,
-                    sdgs=[],
-                ),
-            ]),
-            all_value=True,
-        )
-}
-
-# env-data-desc
-somersault_env_data_descs = {
-    "flip_env_data_desc":
-        EnvironmentDataDescription(
-            odx_id=OdxLinkId("somersault.env_data_desc.flip_env_data_desc", doc_frags),
-            short_name="flip_env_data_desc",
-            long_name="Flip Env Data Desc",
-            description=None,
-            admin_data=None,
-            param_snref="flip_speed",
-            param_snpathref=None,
-            env_datas=[],
-            env_data_refs=[OdxLinkRef("somersault.env_data.flip_env_data", doc_frags)],
-            sdgs=[],
-        )
-}
-
 # requests
 somersault_requests = {
     "start_session":
@@ -2063,8 +2001,8 @@ class SomersaultSID(IntEnum):
     ),
     tables=NamedItemList(somersault_tables.values()),
     muxs=NamedItemList(),
-    env_datas=NamedItemList(somersault_env_datas.values()),
-    env_data_descs=NamedItemList(somersault_env_data_descs.values()),
+    env_datas=NamedItemList(),
+    env_data_descs=NamedItemList(),
     dtc_dops=NamedItemList(),
     structures=NamedItemList(),
     static_fields=NamedItemList(),
@@ -2338,8 +2276,8 @@ class SomersaultSID(IntEnum):
         dynamic_length_fields=NamedItemList(),
         dynamic_endmarker_fields=NamedItemList(),
         tables=NamedItemList(),
-        env_data_descs=NamedItemList(),
         env_datas=NamedItemList(),
+        env_data_descs=NamedItemList(),
         muxs=NamedItemList(),
         unit_spec=None,
         sdgs=[],

examples/somersault_modified.pdx

@@ -10,7 +10,7 @@
 from .dataobjectproperty import DataObjectProperty
 from .decodestate import DecodeState
 from .encodestate import EncodeState
-from .exceptions import EncodeError, OdxWarning, odxassert, odxraise, strict_mode
+from .exceptions import EncodeError, OdxWarning, odxassert, odxraise
 from .nameditemlist import NamedItemList
 from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId
 from .odxtypes import ParameterDict, ParameterValue, ParameterValueDict
@@ -78,6 +78,7 @@ def coded_const_prefix(self, request_prefix: bytes = b'') -> bytes:
                 param.encode_into_pdu(physical_value=None, encode_state=encode_state)
             else:
                 break
+
         return encode_state.coded_message
 
     @property
@@ -159,8 +160,8 @@ def encode_into_pdu(self, physical_value: Optional[ParameterValue],
         orig_is_end_of_pdu = encode_state.is_end_of_pdu
         encode_state.is_end_of_pdu = False
 
-        # in strict mode, ensure that no values for unknown parameters are specified.
-        if strict_mode:
+        # ensure that no values for unknown parameters are specified.
+        if not encode_state.allow_unknown_parameters:
             param_names = {param.short_name for param in self.parameters}
             for param_value_name in physical_value:
                 if param_value_name not in param_names:
@@ -192,8 +193,10 @@ def encode_into_pdu(self, physical_value: Optional[ParameterValue],
                 odxraise(f"No value for required parameter {param.short_name} specified",
                          EncodeError)
 
-            param.encode_into_pdu(
-                physical_value=physical_value.get(param.short_name), encode_state=encode_state)
+            param_phys_value = physical_value.get(param.short_name)
+            param.encode_into_pdu(physical_value=param_phys_value, encode_state=encode_state)
+
+            encode_state.journal.append((param, param_phys_value))
 
         encode_state.is_end_of_pdu = False
         if self.byte_size is not None:
@@ -235,6 +238,7 @@ def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
         for param in self.parameters:
             value = param.decode_from_pdu(decode_state)
 
+            decode_state.journal.append((param, value))
             result[param.short_name] = value
 
         # decoding of the structure finished. go back the original origin.

examples/somersaultecu.py

@@ -1,18 +1,19 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Dict, cast
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, cast
 
 import odxtools.exceptions as exceptions
 
 from .exceptions import DecodeError
-from .odxtypes import AtomicOdxType, DataType
+from .odxtypes import AtomicOdxType, DataType, ParameterValue
 
 try:
     import bitstruct.c as bitstruct
 except ImportError:
     import bitstruct
 
 if TYPE_CHECKING:
+    from .parameters.parameter import Parameter
     from .tablerow import TableRow
 
 
@@ -46,6 +47,11 @@ class DecodeState:
     #: values of the table key parameters decoded so far
     table_keys: Dict[str, "TableRow"] = field(default_factory=dict)
 
+    #: List of parameters that have been decoded so far. The journal
+    #: is used by some types of parameters which depend on the values of
+    #: other parameters; i.e., environment data description parameters
+    journal: List[Tuple["Parameter", Optional[ParameterValue]]] = field(default_factory=list)
+
     def extract_atomic_value(
         self,
         bit_length: int,

odxtools/basicstructure.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional
 from xml.etree import ElementTree
 
 from .element import IdentifiableElement
@@ -17,7 +17,7 @@ class DiagnosticTroubleCode(IdentifiableElement):
     trouble_code: int
     text: Optional[str]
     display_trouble_code: Optional[str]
-    level: Union[int, None]
+    level: Optional[int]
     is_temporary_raw: Optional[bool]
     sdgs: List[SpecialDataGroup]
 

odxtools/decodestate.py

@@ -130,22 +130,35 @@ def decode_from_pdu(self, decode_state: DecodeState) -> ParameterValue:
             sdgs=[],
         )
 
-    @override
-    def encode_into_pdu(self, physical_value: Optional[ParameterValue],
-                        encode_state: EncodeState) -> None:
-        if isinstance(physical_value, DiagnosticTroubleCode):
-            trouble_code = physical_value.trouble_code
-        elif isinstance(physical_value, int):
+    def convert_to_numerical_trouble_code(self, dtc_value: ParameterValue) -> int:
+        if isinstance(dtc_value, DiagnosticTroubleCode):
+            return dtc_value.trouble_code
+        elif isinstance(dtc_value, int):
             # assume that physical value is the trouble_code
-            trouble_code = physical_value
-        elif isinstance(physical_value, str):
+            return dtc_value
+        elif isinstance(dtc_value, str):
             # assume that physical value is the short_name
-            dtcs = [dtc for dtc in self.dtcs if dtc.short_name == physical_value]
-            odxassert(len(dtcs) == 1)
-            trouble_code = dtcs[0].trouble_code
+            dtcs = [dtc for dtc in self.dtcs if dtc.short_name == dtc_value]
+            if len(dtcs) != 1:
+                odxraise(f"No DTC named {dtc_value} found for DTC-DOP "
+                         f"{self.short_name}.", EncodeError)
+                return cast(int, None)
+
+            return dtcs[0].trouble_code
         else:
-            raise EncodeError(f"The DTC-DOP {self.short_name} expected a"
-                              f" DiagnosticTroubleCode but got {physical_value!r}.")
+            odxraise(
+                f"The DTC-DOP {self.short_name} expected a"
+                f" diagnostic trouble code but got {type(dtc_value).__name__}", EncodeError)
+            return cast(int, None)
+
+    @override
+    def encode_into_pdu(self, physical_value: Optional[ParameterValue],
+                        encode_state: EncodeState) -> None:
+        if physical_value is None:
+            odxraise(f"No DTC specified", EncodeError)
+            return
+
+        trouble_code = self.convert_to_numerical_trouble_code(physical_value)
 
         internal_trouble_code = int(self.compu_method.convert_physical_to_internal(trouble_code))
 

odxtools/diagnostictroublecode.py

@@ -1,16 +1,19 @@
 # SPDX-License-Identifier: MIT
 import warnings
 from dataclasses import dataclass, field
-from typing import Dict, Optional, SupportsBytes
+from typing import TYPE_CHECKING, Dict, List, Optional, SupportsBytes, Tuple
 
 from .exceptions import EncodeError, OdxWarning, odxassert, odxraise
-from .odxtypes import AtomicOdxType, DataType
+from .odxtypes import AtomicOdxType, DataType, ParameterValue
 
 try:
     import bitstruct.c as bitstruct
 except ImportError:
     import bitstruct
 
+if TYPE_CHECKING:
+    from .parameters.parameter import Parameter
+
 
 @dataclass
 class EncodeState:
@@ -56,6 +59,15 @@ class EncodeState:
     #: (needed for MinMaxLengthType, EndOfPduField, etc.)
     is_end_of_pdu: bool = True
 
+    #: list of parameters that have been encoded so far. The journal
+    #: is used by some types of parameters which depend on the values of
+    #: other parameters; e.g., environment data description parameters
+    journal: List[Tuple["Parameter", Optional[ParameterValue]]] = field(default_factory=list)
+
+    #: If this is True, specifying unknown parameters for encoding
+    #: will raise an OdxError exception in strict mode.
+    allow_unknown_parameters = False
+
     def __post_init__(self) -> None:
         # if a coded message has been specified, but no used_mask, we
         # assume that all of the bits of the coded message are