Merge pull request #438 from andlaus/function_dictionary

Implement support for the `FUNCTION-DICT` category

Author: andlaus

examples/somersaultecu.py

@@ -845,8 +845,6 @@ class SomersaultSID(IntEnum):
 }
 
 # tables
-flip_quality_table_id = OdxLinkId("somersault.table.flip_quality", doc_frags)
-flip_quality_table_ref = OdxLinkRef.from_id(flip_quality_table_id)
 somersault_tables = {
     "last_flip_details":
         Table(

odxtools/basefunctionnode.py

@@ -0,0 +1,145 @@
+# SPDX-License-Identifier: MIT
+from dataclasses import dataclass, field
+from typing import Any
+from xml.etree import ElementTree
+
+from .admindata import AdminData
+from .audience import Audience
+from .componentconnector import ComponentConnector
+from .element import IdentifiableElement
+from .functioninparam import FunctionInParam
+from .functionoutparam import FunctionOutParam
+from .multipleecujob import MultipleEcuJob
+from .nameditemlist import NamedItemList
+from .odxdoccontext import OdxDocContext
+from .odxlink import OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .snrefcontext import SnRefContext
+from .specialdatagroup import SpecialDataGroup
+from .utils import dataclass_fields_asdict
+
+
+@dataclass(kw_only=True)
+class BaseFunctionNode(IdentifiableElement):
+    audience: Audience | None = None
+    function_in_params: NamedItemList[FunctionInParam] = field(default_factory=NamedItemList)
+    function_out_params: NamedItemList[FunctionOutParam] = field(default_factory=NamedItemList)
+    component_connectors: list[ComponentConnector] = field(default_factory=list)
+    multiple_ecu_job_refs: list[OdxLinkRef] = field(default_factory=list)
+    admin_data: AdminData | None = None
+    sdg: SpecialDataGroup | None = None
+
+    @property
+    def multiple_ecu_jobs(self) -> NamedItemList[MultipleEcuJob]:
+        return self._multiple_ecu_jobs
+
+    @staticmethod
+    def from_et(et_element: ElementTree.Element, context: OdxDocContext) -> "BaseFunctionNode":
+        kwargs = dataclass_fields_asdict(IdentifiableElement.from_et(et_element, context))
+
+        audience = None
+        if (audience_elem := et_element.find("AUDIENCE")) is not None:
+            audience = Audience.from_et(audience_elem, context)
+
+        function_in_params = NamedItemList([
+            FunctionInParam.from_et(fip_el, context)
+            for fip_el in et_element.iterfind("FUNCTION-IN-PARAMS/FUNCTION-IN-PARAM")
+        ])
+
+        function_out_params = NamedItemList([
+            FunctionOutParam.from_et(fop_el, context)
+            for fop_el in et_element.iterfind("FUNCTION-OUT-PARAMS/FUNCTION-OUT-PARAM")
+        ])
+
+        component_connectors = [
+            ComponentConnector.from_et(cc_el, context)
+            for cc_el in et_element.iterfind("COMPONENT-CONNECTORS/COMPONENT-CONNECTOR")
+        ]
+
+        multiple_ecu_job_refs = [
+            OdxLinkRef.from_et(el, context)
+            for el in et_element.iterfind("MULTIPLE-ECU-JOB-REFS/MULTIPLE-ECU-JOB-REF")
+        ]
+
+        admin_data = None
+        if (admin_data_elem := et_element.find("ADMIN-DATA")) is not None:
+            admin_data = AdminData.from_et(admin_data_elem, context)
+
+        # be a aware that for whatever reason, function nodes only
+        # exhibit at most one SDG. (quirk of the ODX standard?)
+        sdg = None
+        if (sdge := et_element.find("SDG")) is not None:
+            sdg = SpecialDataGroup.from_et(sdge, context)
+
+        return BaseFunctionNode(
+            audience=audience,
+            function_in_params=function_in_params,
+            function_out_params=function_out_params,
+            component_connectors=component_connectors,
+            multiple_ecu_job_refs=multiple_ecu_job_refs,
+            admin_data=admin_data,
+            sdg=sdg,
+            **kwargs)
+
+    def _build_odxlinks(self) -> dict[OdxLinkId, Any]:
+        result = {self.odx_id: self}
+
+        if self.audience is not None:
+            result.update(self.audience._build_odxlinks())
+
+        for function_in_param in self.function_in_params:
+            result.update(function_in_param._build_odxlinks())
+
+        for function_out_param in self.function_out_params:
+            result.update(function_out_param._build_odxlinks())
+
+        for component_connector in self.component_connectors:
+            result.update(component_connector._build_odxlinks())
+
+        if self.admin_data is not None:
+            result.update(self.admin_data._build_odxlinks())
+
+        if self.sdg is not None:
+            result.update(self.sdg._build_odxlinks())
+
+        return result
+
+    def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
+        self._multiple_ecu_jobs = NamedItemList(
+            [odxlinks.resolve(ref, MultipleEcuJob) for ref in self.multiple_ecu_job_refs])
+
+        if self.audience:
+            self.audience._resolve_odxlinks(odxlinks)
+
+        for function_in_param in self.function_in_params:
+            function_in_param._resolve_odxlinks(odxlinks)
+
+        for function_out_param in self.function_out_params:
+            function_out_param._resolve_odxlinks(odxlinks)
+
+        for component_connector in self.component_connectors:
+            component_connector._resolve_odxlinks(odxlinks)
+
+        if self.admin_data:
+            self.admin_data._resolve_odxlinks(odxlinks)
+
+        if self.sdg is not None:
+            self.sdg._resolve_odxlinks(odxlinks)
+
+    def _resolve_snrefs(self, context: SnRefContext) -> None:
+        if self.audience:
+            self.audience._resolve_snrefs(context)
+
+        for function_in_param in self.function_in_params:
+            function_in_param._resolve_snrefs(context)
+
+        for function_out_param in self.function_out_params:
+            function_out_param._resolve_snrefs(context)
+
+        for component_connector in self.component_connectors:
+            component_connector._resolve_snrefs(context)
+
+        if self.admin_data:
+            self.admin_data._resolve_snrefs(context)
+
+        if self.sdg is not None:
+            self.sdg._resolve_snrefs(context)

odxtools/componentconnector.py

@@ -0,0 +1,84 @@
+# SPDX-License-Identifier: MIT
+from dataclasses import dataclass, field
+from typing import Any
+from xml.etree import ElementTree
+
+from .diaglayers.basevariant import BaseVariant
+from .diaglayers.ecuvariant import EcuVariant
+from .diagobjectconnector import DiagObjectConnector
+from .exceptions import odxraise
+from .nameditemlist import NamedItemList
+from .odxdoccontext import OdxDocContext
+from .odxlink import OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .snrefcontext import SnRefContext
+
+
+@dataclass(kw_only=True)
+class ComponentConnector:
+    ecu_variant_refs: list[OdxLinkRef] = field(default_factory=list)
+    base_variant_ref: OdxLinkRef | None = None
+
+    diag_object_connector: DiagObjectConnector | None = None
+    diag_object_connector_ref: OdxLinkRef | None = None
+
+    @property
+    def ecu_variants(self) -> NamedItemList[EcuVariant]:
+        return self._ecu_variants
+
+    @property
+    def base_variant(self) -> BaseVariant | None:
+        return self._base_variant
+
+    @staticmethod
+    def from_et(et_element: ElementTree.Element, context: OdxDocContext) -> "ComponentConnector":
+        ecu_variant_refs = [
+            OdxLinkRef.from_et(el, context)
+            for el in et_element.iterfind("ECU-VARIANT-REFS/ECU-VARIANT-REF")
+        ]
+        base_variant_ref = OdxLinkRef.from_et(et_element.find("BASE-VARIANT-REF"), context)
+        diag_object_connector = None
+        if (doc_elem := et_element.find("DIAG-OBJECT-CONNECTOR")) is not None:
+            diag_object_connector = DiagObjectConnector.from_et(doc_elem, context)
+        diag_object_connector_ref = OdxLinkRef.from_et(
+            et_element.find("DIAG-OBJECT-CONNECTOR-REF"), context)
+
+        return ComponentConnector(
+            ecu_variant_refs=ecu_variant_refs,
+            base_variant_ref=base_variant_ref,
+            diag_object_connector=diag_object_connector,
+            diag_object_connector_ref=diag_object_connector_ref)
+
+    def _build_odxlinks(self) -> dict[OdxLinkId, Any]:
+        result = {}
+
+        if self.diag_object_connector_ref is None:
+            if self.diag_object_connector is None:
+                odxraise()
+            else:
+                result.update(self.diag_object_connector._build_odxlinks())
+
+        return result
+
+    def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
+        self._ecu_variants = NamedItemList(
+            [odxlinks.resolve(ev_ref, EcuVariant) for ev_ref in self.ecu_variant_refs])
+
+        self._base_variant = None
+        if self.base_variant_ref is not None:
+            self._base_variant = odxlinks.resolve(self.base_variant_ref, BaseVariant)
+
+        if self.diag_object_connector_ref is None:
+            if self.diag_object_connector is None:
+                odxraise()
+            self.diag_object_connector._resolve_odxlinks(odxlinks)
+        else:
+            self.diag_object_connector = odxlinks.resolve(self.diag_object_connector_ref,
+                                                          DiagObjectConnector)
+
+    def _resolve_snrefs(self, context: SnRefContext) -> None:
+        if self.diag_object_connector_ref is None:
+            if self.diag_object_connector is None:
+                odxraise()
+                return
+
+            self.diag_object_connector._resolve_snrefs(context)

odxtools/database.py

@@ -21,6 +21,7 @@
 from .ecuconfig import EcuConfig
 from .exceptions import odxraise, odxrequire
 from .flash import Flash
+from .functiondictionary import FunctionDictionary
 from .multipleecujobspec import MultipleEcuJobSpec
 from .nameditemlist import NamedItemList
 from .odxdoccontext import OdxDocContext
@@ -47,6 +48,7 @@ def __init__(self) -> None:
         self._vehicle_info_specs = NamedItemList[VehicleInfoSpec]()
         self._flashs = NamedItemList[Flash]()
         self._multiple_ecu_job_specs = NamedItemList[MultipleEcuJobSpec]()
+        self._function_dictionaries = NamedItemList[FunctionDictionary]()
         self._short_name = "odx_database"
 
     def add_pdx_file(self, pdx_file: Union[str, "PathLike[Any]", IO[bytes], ZipFile]) -> None:
@@ -135,6 +137,12 @@ def add_xml_tree(self, root: ElementTree.Element) -> None:
             context = OdxDocContext(model_version,
                                     (OdxDocFragment(category_sn, DocType.MULTIPLE_ECU_JOB_SPEC),))
             self._multiple_ecu_job_specs.append(MultipleEcuJobSpec.from_et(category_et, context))
+        elif category_tag == "FUNCTION-DICTIONARY":
+            context = OdxDocContext(model_version, (OdxDocFragment(
+                category_sn, DocType.FUNCTION_DICTIONARY_SPEC),))
+            self._function_dictionaries.append(FunctionDictionary.from_et(category_et, context))
+        else:
+            odxraise(f"Encountered unknown ODX category '{category_tag}' (non-conforming dataset?)")
 
     def refresh(self) -> None:
         # Create wrapper objects
@@ -178,6 +186,9 @@ def refresh(self) -> None:
         for multiple_ecu_job_spec in self.multiple_ecu_job_specs:
             multiple_ecu_job_spec._resolve_odxlinks(self._odxlinks)
 
+        for function_dictionary in self.function_dictionaries:
+            function_dictionary._resolve_odxlinks(self._odxlinks)
+
         # resolve short name references for containers which do not do
         # inheritance (we can call directly call _resolve_snrefs())
         context = SnRefContext()
@@ -198,6 +209,8 @@ def refresh(self) -> None:
             flash._finalize_init(self, self._odxlinks)
         for multiple_ecu_job_spec in self.multiple_ecu_job_specs:
             multiple_ecu_job_spec._finalize_init(self, self._odxlinks)
+        for function_dictionary in self.function_dictionaries:
+            function_dictionary._finalize_init(self, self._odxlinks)
 
         for subset in self.comparam_subsets:
             subset._resolve_snrefs(context)
@@ -213,6 +226,8 @@ def refresh(self) -> None:
             flash._resolve_snrefs(context)
         for multiple_ecu_job_spec in self.multiple_ecu_job_specs:
             multiple_ecu_job_spec._resolve_snrefs(context)
+        for function_dictionary in self.function_dictionaries:
+            function_dictionary._resolve_snrefs(context)
 
     def _build_odxlinks(self) -> dict[OdxLinkId, Any]:
         result: dict[OdxLinkId, Any] = {}
@@ -234,9 +249,13 @@ def _build_odxlinks(self) -> dict[OdxLinkId, Any]:
 
         for flash in self.flashs:
             result.update(flash._build_odxlinks())
+
         for multiple_ecu_job_spec in self.multiple_ecu_job_specs:
             result.update(multiple_ecu_job_spec._build_odxlinks())
 
+        for function_dictionary in self.function_dictionaries:
+            result.update(function_dictionary._build_odxlinks())
+
         return result
 
     @property
@@ -326,6 +345,10 @@ def flashs(self) -> NamedItemList[Flash]:
     def multiple_ecu_job_specs(self) -> NamedItemList[MultipleEcuJobSpec]:
         return self._multiple_ecu_job_specs
 
+    @property
+    def function_dictionaries(self) -> NamedItemList[FunctionDictionary]:
+        return self._function_dictionaries
+
     def __repr__(self) -> str:
         return f"Database(model_version={self.model_version}, " \
             f"protocols={[x.short_name for x in self.protocols]}, " \
@@ -336,4 +359,5 @@ def __repr__(self) -> str:
             f"ecu_configs={repr(self.ecu_configs)}, " \
             f"vehicle_info_specs={repr(self.vehicle_info_specs)}, " \
             f"multiple_ecu_job_specs={repr(self.multiple_ecu_job_specs)}, " \
+            f"function_dictionaries={repr(self.function_dictionaries)}, " \
             f"flashs={repr(self.flashs)})"