Merge pull request #299 from andlaus/improve_snref_resolution

Improve short name reference handling

odxtools/basicstructure.py

@@ -23,7 +23,6 @@
 from .parameters.parameterwithdop import ParameterWithDOP
 from .parameters.physicalconstantparameter import PhysicalConstantParameter
 from .parameters.tablekeyparameter import TableKeyParameter
-from .parameters.tablestructparameter import TableStructParameter
 from .utils import dataclass_fields_asdict
 
 if TYPE_CHECKING:
@@ -307,7 +306,4 @@ def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         super()._resolve_snrefs(diag_layer)
 
         for param in self.parameters:
-            if isinstance(param, TableStructParameter):
-                param._table_struct_resolve_snrefs(diag_layer, param_list=self.parameters)
-            else:
-                param._resolve_snrefs(diag_layer)
+            param._parameter_resolve_snrefs(diag_layer, param_list=self.parameters)

odxtools/diagcomm.py

@@ -11,7 +11,7 @@
 from .exceptions import odxraise, odxrequire
 from .functionalclass import FunctionalClass
 from .nameditemlist import NamedItemList
-from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from .odxtypes import odxstr_to_bool
 from .specialdatagroup import SpecialDataGroup
 from .state import State
@@ -211,5 +211,13 @@ def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         for sdg in self.sdgs:
             sdg._resolve_snrefs(diag_layer)
 
-        self._protocols = NamedItemList(
-            [diag_layer.protocols[prot_snref] for prot_snref in self.protocol_snrefs])
+        if TYPE_CHECKING:
+            self._protocols = NamedItemList([
+                resolve_snref(prot_snref, diag_layer.protocols, DiagLayer)
+                for prot_snref in self.protocol_snrefs
+            ])
+        else:
+            self._protocols = NamedItemList([
+                resolve_snref(prot_snref, diag_layer.protocols)
+                for prot_snref in self.protocol_snrefs
+            ])

odxtools/diaglayerraw.py

@@ -19,7 +19,7 @@
 from .exceptions import odxassert, odxraise, odxrequire
 from .functionalclass import FunctionalClass
 from .nameditemlist import NamedItemList
-from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from .parentref import ParentRef
 from .protstack import ProtStack
 from .request import Request
@@ -281,8 +281,8 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         self._prot_stack: Optional[ProtStack] = None
         if self.prot_stack_snref is not None:
-            self._prot_stack = odxrequire(
-                odxrequire(self.comparam_spec).prot_stacks.get(self.prot_stack_snref))
+            self._prot_stack = resolve_snref(self.prot_stack_snref,
+                                             odxrequire(self.comparam_spec).prot_stacks, ProtStack)
 
         # do short-name reference resolution
         if self.admin_data is not None:
@@ -292,8 +292,8 @@ def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
 
         for company_data in self.company_datas:
             company_data._resolve_snrefs(diag_layer)
-        for functional_classe in self.functional_classes:
-            functional_classe._resolve_snrefs(diag_layer)
+        for functional_class in self.functional_classes:
+            functional_class._resolve_snrefs(diag_layer)
         for diag_comm in self.diag_comms:
             if isinstance(diag_comm, OdxLinkRef):
                 continue

odxtools/field.py

@@ -6,7 +6,7 @@
 from .complexdop import ComplexDop
 from .environmentdatadescription import EnvironmentDataDescription
 from .exceptions import odxassert, odxrequire
-from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkRef
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkRef, resolve_snref
 from .odxtypes import odxstr_to_bool
 from .utils import dataclass_fields_asdict
 
@@ -84,8 +84,9 @@ def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         """Recursively resolve any short-name references"""
         if self.structure_snref is not None:
             structures = diag_layer.diag_data_dictionary_spec.structures
-            self._structure = odxrequire(structures.get(self.structure_snref))
+            self._structure = resolve_snref(self.structure_snref, structures, BasicStructure)
 
         if self.env_data_desc_snref is not None:
             env_data_descs = diag_layer.diag_data_dictionary_spec.env_data_descs
-            self._env_data_desc = odxrequire(env_data_descs.get(self.env_data_desc_snref))
+            self._env_data_desc = resolve_snref(self.env_data_desc_snref, env_data_descs,
+                                                EnvironmentDataDescription)

odxtools/matchingparameter.py

@@ -32,8 +32,8 @@ def from_et(et_element: ElementTree.Element,
                 doc_frags: List[OdxDocFragment]) -> "MatchingParameter":
 
         expected_value = odxrequire(et_element.findtext("EXPECTED-VALUE"))
-        diag_com_snref_el = odxrequire(et_element.find("DIAG-COMM-SNREF"))
-        diag_comm_snref = odxrequire(diag_com_snref_el.get("SHORT-NAME"))
+        diag_comm_snref_el = odxrequire(et_element.find("DIAG-COMM-SNREF"))
+        diag_comm_snref = odxrequire(diag_comm_snref_el.get("SHORT-NAME"))
         out_param_snref_el = et_element.find("OUT-PARAM-IF-SNREF")
         out_param_snpathref_el = et_element.find("OUT-PARAM-IF-SNPATHREF")
         out_param_if = None

odxtools/multiplexercase.py

@@ -7,7 +7,7 @@
 from .compumethods.limit import Limit
 from .element import NamedElement
 from .exceptions import odxrequire
-from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from .odxtypes import AtomicOdxType, DataType
 from .utils import dataclass_fields_asdict
 
@@ -73,7 +73,7 @@ def _mux_case_resolve_odxlinks(self, odxlinks: OdxLinkDatabase, *,
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         if self.structure_snref:
             ddds = diag_layer.diag_data_dictionary_spec
-            self._structure = odxrequire(ddds.structures.get(self.structure_snref))
+            self._structure = resolve_snref(self.structure_snref, ddds.structures, BasicStructure)
 
     def applies(self, value: AtomicOdxType) -> bool:
         return self.lower_limit.complies_to_lower(value) \

odxtools/multiplexerdefaultcase.py

@@ -6,7 +6,7 @@
 from .basicstructure import BasicStructure
 from .element import NamedElement
 from .exceptions import odxrequire
-from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from .utils import dataclass_fields_asdict
 
 if TYPE_CHECKING:
@@ -20,7 +20,7 @@ class MultiplexerDefaultCase(NamedElement):
     structure_snref: Optional[str]
 
     def __post_init__(self) -> None:
-        self._structure: Optional[BasicStructure] = None
+        self._structure: BasicStructure
 
     @staticmethod
     def from_et(et_element: ElementTree.Element,
@@ -46,4 +46,8 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         if self.structure_snref:
             ddds = diag_layer.diag_data_dictionary_spec
-            self._structure = odxrequire(ddds.structures.get(self.structure_snref))
+            self._structure = resolve_snref(self.structure_snref, ddds.structures, BasicStructure)
+
+    @property
+    def structure(self) -> BasicStructure:
+        return self._structure

odxtools/odxlink.py

@@ -1,10 +1,11 @@
 # SPDX-License-Identifier: MIT
 import warnings
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Type, TypeVar, overload
+from typing import Any, Dict, Iterable, List, Optional, Type, TypeVar, overload
 from xml.etree import ElementTree
 
-from .exceptions import OdxWarning, odxassert
+from .exceptions import OdxWarning, odxassert, odxraise
+from .nameditemlist import OdxNamed, TNamed
 
 
 @dataclass(frozen=True)
@@ -270,3 +271,34 @@ def update(self, new_entries: Dict[OdxLinkId, Any]) -> None:
                     self._db[doc_frag] = {}
 
                 self._db[doc_frag][odx_id] = obj
+
+
+@overload
+def resolve_snref(target_short_name: str,
+                  items: Iterable[OdxNamed],
+                  expected_type: None = None) -> Any:
+    """Resolve a short name reference given a sequence of candidate objects"""
+    ...
+
+
+@overload
+def resolve_snref(target_short_name: str, items: Iterable[OdxNamed],
+                  expected_type: Type[TNamed]) -> TNamed:
+    ...
+
+
+def resolve_snref(target_short_name: str,
+                  items: Iterable[OdxNamed],
+                  expected_type: Any = None) -> Any:
+    candidates = [x for x in items if x.short_name == target_short_name]
+
+    if not candidates:
+        odxraise(f"Cannot resolve short name reference to '{target_short_name}'")
+        return None
+    elif len(candidates) > 1:
+        odxraise(f"Cannot uniquely resolve short name reference to '{target_short_name}'")
+    elif expected_type is not None and not isinstance(candidates[0], expected_type):
+        odxraise(f"Reference '{target_short_name}' points to a {type(candidates[0]).__name__}"
+                 f"object while expecting {expected_type.__name__}")
+
+    return candidates[0]

odxtools/parameters/codedconstparameter.py

@@ -59,8 +59,9 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         super()._resolve_odxlinks(odxlinks)
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
     @override
     def get_static_bit_length(self) -> Optional[int]:

odxtools/parameters/lengthkeyparameter.py

@@ -11,7 +11,7 @@
 from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId
 from ..odxtypes import ParameterValue
 from ..utils import dataclass_fields_asdict
-from .parameter import ParameterType
+from .parameter import Parameter, ParameterType
 from .parameterwithdop import ParameterWithDOP
 
 if TYPE_CHECKING:
@@ -60,8 +60,9 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         super()._resolve_odxlinks(odxlinks)
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
     @property
     @override

odxtools/parameters/nrcconstparameter.py

@@ -75,8 +75,9 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         super()._resolve_odxlinks(odxlinks)
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
     @override
     def get_static_bit_length(self) -> Optional[int]:

odxtools/parameters/parameter.py

@@ -82,7 +82,13 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         for sdg in self.sdgs:
             sdg._resolve_odxlinks(odxlinks)
 
+    @final
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
+        raise RuntimeError("Calling _resolve_snrefs() is not allowed for parameters. "
+                           "Use _parameter_resolve_snrefs() instead.")
+
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List["Parameter"]) -> None:
         for sdg in self.sdgs:
             sdg._resolve_snrefs(diag_layer)
 

odxtools/parameters/parameterwithdop.py

@@ -11,7 +11,7 @@
 from ..dtcdop import DtcDop
 from ..encodestate import EncodeState
 from ..exceptions import odxassert, odxrequire
-from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from ..odxtypes import AtomicOdxType, ParameterValue
 from ..physicaltype import PhysicalType
 from ..utils import dataclass_fields_asdict
@@ -61,12 +61,13 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
             self._dop = odxlinks.resolve_lenient(self.dop_ref)
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
         if self.dop_snref:
-            ddds = diag_layer.diag_data_dictionary_spec
-            self._dop = odxrequire(ddds.all_data_object_properties.get(self.dop_snref))
+            all_dops = diag_layer.diag_data_dictionary_spec.all_data_object_properties
+            self._dop = resolve_snref(self.dop_snref, all_dops, DopBase)
 
     @property
     def dop(self) -> DopBase:

odxtools/parameters/physicalconstantparameter.py

@@ -12,7 +12,7 @@
 from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId
 from ..odxtypes import ParameterValue
 from ..utils import dataclass_fields_asdict
-from .parameter import ParameterType
+from .parameter import Parameter, ParameterType
 from .parameterwithdop import ParameterWithDOP
 
 if TYPE_CHECKING:
@@ -50,8 +50,9 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         super()._resolve_odxlinks(odxlinks)
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
         dop = odxrequire(self.dop)
         if not isinstance(dop, DataObjectProperty):

odxtools/parameters/tablekeyparameter.py

@@ -8,7 +8,7 @@
 from ..decodestate import DecodeState
 from ..encodestate import EncodeState
 from ..exceptions import DecodeError, EncodeError, odxraise, odxrequire
-from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from ..odxtypes import ParameterValue
 from ..utils import dataclass_fields_asdict
 from .parameter import Parameter, ParameterType
@@ -94,19 +94,19 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
             self._table = self._table_row.table
 
     @override
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
         if self.table_snref is not None:
-            ddd_spec = diag_layer.diag_data_dictionary_spec
-            self._table = ddd_spec.tables[self.table_snref]
+            tables = diag_layer.diag_data_dictionary_spec.tables
+            self._table = resolve_snref(self.table_snref, tables, Table)
         if self.table_row_snref is not None:
             # make sure that we know the table to which the table row
             # SNREF is relative to.
-            table = odxrequire(
-                self._table, "If a table-row short name reference is defined, a "
-                "table must also be specified.")
-            self._table_row = table.table_rows[self.table_row_snref]
+            table = odxrequire(self._table,
+                               "If a table-row is referenced, a table must also be referenced.")
+            self._table_row = resolve_snref(self.table_row_snref, table.table_rows, TableRow)
 
     @property
     def table(self) -> "Table":

odxtools/parameters/tablestructparameter.py

@@ -3,12 +3,12 @@
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
 from xml.etree import ElementTree
 
-from typing_extensions import final, override
+from typing_extensions import override
 
 from ..decodestate import DecodeState
 from ..encodestate import EncodeState
 from ..exceptions import DecodeError, EncodeError, odxraise, odxrequire
-from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from ..odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef, resolve_snref
 from ..odxtypes import ParameterValue
 from ..utils import dataclass_fields_asdict
 from .parameter import Parameter, ParameterType
@@ -60,31 +60,12 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
             self._table_key = odxlinks.resolve(self.table_key_ref, TableKeyParameter)
 
     @override
-    @final
-    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        raise RuntimeError(f"Calling TableStructParameter._resolve_snref() is not allowed. "
-                           f"Use ._table_struct_resolve_snrefs() instead.")
-
-    def _table_struct_resolve_snrefs(self, diag_layer: "DiagLayer", *,
-                                     param_list: List[Parameter]) -> None:
-        super()._resolve_snrefs(diag_layer)
+    def _parameter_resolve_snrefs(self, diag_layer: "DiagLayer", *,
+                                  param_list: List[Parameter]) -> None:
+        super()._parameter_resolve_snrefs(diag_layer, param_list=param_list)
 
         if self.table_key_snref is not None:
-            tk_candidates = [p for p in param_list if p.short_name == self.table_key_snref]
-            if len(tk_candidates) > 1:
-                odxraise(f"Short name reference '{self.table_key_snref}' could "
-                         f"not be uniquely resolved.")
-            elif len(tk_candidates) == 0:
-                odxraise(f"Short name reference '{self.table_key_snref}' could "
-                         f"not be resolved.")
-                return
-
-            tk = tk_candidates[0]
-            if not isinstance(tk, TableKeyParameter):
-                odxraise(f"Table struct '{self.short_name}' references non-TableKey parameter "
-                         f"`{self.table_key_snref}' as its table key.")
-
-            self._table_key = tk
+            self._table_key = resolve_snref(self.table_key_snref, param_list, TableKeyParameter)
 
     @property
     def table_key(self) -> TableKeyParameter: