Merge pull request #238 from rivian/protstack-and-comparamspec-support

Add support for ProtStack and ComparamSpec

examples/somersaultecu.py

@@ -2054,6 +2054,8 @@ class SomersaultSID(IntEnum):
     parent_refs=[],
     comparams=somersault_comparams,
     ecu_variant_patterns=[],
+    comparam_spec_ref=None,
+    prot_stack_snref=None,
 )
 somersault_diaglayer = DiagLayer(diag_layer_raw=somersault_diaglayer_raw)
 
@@ -2096,6 +2098,8 @@ class SomersaultSID(IntEnum):
     ],
     comparams=somersault_comparams,
     ecu_variant_patterns=[],
+    comparam_spec_ref=None,
+    prot_stack_snref=None,
 )
 somersault_lazy_diaglayer = DiagLayer(diag_layer_raw=somersault_lazy_diaglayer_raw)
 
@@ -2314,6 +2318,8 @@ class SomersaultSID(IntEnum):
     ],
     comparams=somersault_comparams,
     ecu_variant_patterns=[],
+    comparam_spec_ref=None,
+    prot_stack_snref=None,
 )
 somersault_assiduous_diaglayer = DiagLayer(diag_layer_raw=somersault_assiduous_diaglayer_raw)
 

odxtools/comparamspec.py

@@ -0,0 +1,94 @@
+# SPDX-License-Identifier: MIT
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from xml.etree import ElementTree
+
+from .admindata import AdminData
+from .companydata import CompanyData
+from .createcompanydatas import create_company_datas_from_et
+from .createsdgs import create_sdgs_from_et
+from .element import IdentifiableElement
+from .exceptions import odxrequire
+from .nameditemlist import NamedItemList
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId
+from .protstack import ProtStack
+from .specialdatagroup import SpecialDataGroup
+from .utils import dataclass_fields_asdict
+
+if TYPE_CHECKING:
+    from .diaglayer import DiagLayer
+
+
+@dataclass
+class ComparamSpec(IdentifiableElement):
+    admin_data: Optional[AdminData]
+    company_datas: NamedItemList[CompanyData]
+    sdgs: List[SpecialDataGroup]
+    prot_stacks: NamedItemList[ProtStack]
+
+    @staticmethod
+    def from_et(et_element: ElementTree.Element, doc_frags: List[OdxDocFragment]) -> "ComparamSpec":
+
+        short_name = odxrequire(et_element.findtext("SHORT-NAME"))
+        doc_frags = [OdxDocFragment(short_name, str(et_element.tag))]
+        kwargs = dataclass_fields_asdict(IdentifiableElement.from_et(et_element, doc_frags))
+
+        admin_data = AdminData.from_et(et_element.find("ADMIN-DATA"), doc_frags)
+        company_datas = create_company_datas_from_et(et_element.find("COMPANY-DATAS"), doc_frags)
+        sdgs = create_sdgs_from_et(et_element.find("SDGS"), doc_frags)
+        prot_stacks = NamedItemList([
+            ProtStack.from_et(dl_element, doc_frags)
+            for dl_element in et_element.iterfind("PROT-STACKS/PROT-STACK")
+        ])
+
+        return ComparamSpec(
+            admin_data=admin_data,
+            company_datas=company_datas,
+            sdgs=sdgs,
+            prot_stacks=prot_stacks,
+            **kwargs)
+
+    def _build_odxlinks(self) -> Dict[OdxLinkId, Any]:
+        odxlinks: Dict[OdxLinkId, Any] = {}
+        if self.odx_id is not None:
+            odxlinks[self.odx_id] = self
+
+        if self.admin_data is not None:
+            odxlinks.update(self.admin_data._build_odxlinks())
+
+        for cd in self.company_datas:
+            odxlinks.update(cd._build_odxlinks())
+
+        for sdg in self.sdgs:
+            odxlinks.update(sdg._build_odxlinks())
+
+        for ps in self.prot_stacks:
+            odxlinks.update(ps._build_odxlinks())
+
+        return odxlinks
+
+    def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
+        if self.admin_data is not None:
+            self.admin_data._resolve_odxlinks(odxlinks)
+
+        for cd in self.company_datas:
+            cd._resolve_odxlinks(odxlinks)
+
+        for sdg in self.sdgs:
+            sdg._resolve_odxlinks(odxlinks)
+
+        for ps in self.prot_stacks:
+            ps._resolve_odxlinks(odxlinks)
+
+    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
+        if self.admin_data is not None:
+            self.admin_data._resolve_snrefs(diag_layer)
+
+        for cd in self.company_datas:
+            cd._resolve_snrefs(diag_layer)
+
+        for sdg in self.sdgs:
+            sdg._resolve_snrefs(diag_layer)
+
+        for ps in self.prot_stacks:
+            ps._resolve_snrefs(diag_layer)

odxtools/database.py

@@ -5,6 +5,7 @@
 from xml.etree import ElementTree
 from zipfile import ZipFile
 
+from .comparamspec import ComparamSpec
 from .comparamsubset import ComparamSubset
 from .diaglayer import DiagLayer
 from .diaglayercontainer import DiagLayerContainer
@@ -32,6 +33,7 @@ def __init__(self,
             # create an empty database object
             self._diag_layer_containers = NamedItemList[DiagLayerContainer]()
             self._comparam_subsets = NamedItemList[ComparamSubset]()
+            self._comparam_specs = NamedItemList[ComparamSpec]()
             return
 
         if pdx_zip is not None and odx_d_file_name is not None:
@@ -55,6 +57,7 @@ def __init__(self,
 
         dlcs: List[DiagLayerContainer] = []
         comparam_subsets: List[ComparamSubset] = []
+        comparam_specs: List[ComparamSpec] = []
         for root in documents:
             # ODX spec version
             model_version = version(root.attrib.get("MODEL-VERSION", "2.0"))
@@ -70,14 +73,16 @@ def __init__(self,
                 if subset is not None:
                     comparam_subsets.append(ComparamSubset.from_et(subset, []))
             else:
-                subset = root.find("COMPARAM-SPEC")
-                if subset is not None:
-                    comparam_subsets.append(ComparamSubset.from_et(subset, []))
+                spec = root.find("COMPARAM-SPEC")
+                if spec is not None:
+                    comparam_specs.append(ComparamSpec.from_et(spec, []))
 
         self._diag_layer_containers = NamedItemList(dlcs)
         self._diag_layer_containers.sort(key=short_name_as_key)
         self._comparam_subsets = NamedItemList(comparam_subsets)
         self._comparam_subsets.sort(key=short_name_as_key)
+        self._comparam_specs = NamedItemList(comparam_specs)
+        self._comparam_specs.sort(key=short_name_as_key)
 
         self.refresh()
 
@@ -145,3 +150,7 @@ def diag_layer_containers(self, value: NamedItemList[DiagLayerContainer]) -> Non
     @property
     def comparam_subsets(self) -> NamedItemList[ComparamSubset]:
         return self._comparam_subsets
+
+    @property
+    def comparam_specs(self) -> NamedItemList[ComparamSpec]:
+        return self._comparam_specs

odxtools/diaglayer.py

@@ -290,6 +290,14 @@ def parent_refs(self) -> List[ParentRef]:
     def ecu_variant_patterns(self) -> List[EcuVariantPattern]:
         return self.diag_layer_raw.ecu_variant_patterns
 
+    @property
+    def comparam_spec_ref(self) -> Optional[OdxLinkRef]:
+        return self.diag_layer_raw.comparam_spec_ref
+
+    @property
+    def prot_stack_snref(self) -> Optional[str]:
+        return self.diag_layer_raw.prot_stack_snref
+
     #####
     # </properties forwarded to the "raw" diag layer>
     #####

odxtools/diaglayerraw.py

@@ -62,8 +62,8 @@ class DiagLayerRaw(IdentifiableElement):
     parent_refs: List[ParentRef]
     comparams: List[ComparamInstance]
     ecu_variant_patterns: List[EcuVariantPattern]
-    # comparam_spec: OdxLinkRef # TODO
-    # prot_stack_snref: str # TODO
+    comparam_spec_ref: Optional[OdxLinkRef]
+    prot_stack_snref: Optional[str]
     # diag_variables: List[DiagVariable] # TODO
     # diag_variable_groups: List[DiagVariableGroup] # TODO
     # dyn_defined_spec: Optional[DynDefinedSpec] # TODO
@@ -177,6 +177,9 @@ def from_et(et_element: ElementTree.Element, doc_frags: List[OdxDocFragment]) ->
                 len(ecu_variant_patterns) == 0,
                 "DiagLayer of type other than 'ECU-VARIANT' must not define a ECU-VARIANT-PATTERN")
 
+        comparam_spec_ref = OdxLinkRef.from_et(et_element.find("COMPARAM-SPEC-REF"), doc_frags)
+        prot_stack_snref = et_element.findtext("PROT-STACK-SNREF")
+
         # Create DiagLayer
         return DiagLayerRaw(
             variant_type=variant_type,
@@ -196,6 +199,8 @@ def from_et(et_element: ElementTree.Element, doc_frags: List[OdxDocFragment]) ->
             parent_refs=parent_refs,
             comparams=comparams,
             ecu_variant_patterns=ecu_variant_patterns,
+            comparam_spec_ref=comparam_spec_ref,
+            prot_stack_snref=prot_stack_snref,
             **kwargs)
 
     def _build_odxlinks(self) -> Dict[OdxLinkId, Any]:

odxtools/protstack.py

@@ -0,0 +1,51 @@
+# SPDX-License-Identifier: MIT
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Dict, List
+from xml.etree import ElementTree
+
+from .comparamsubset import ComparamSubset
+from .element import IdentifiableElement
+from .exceptions import odxrequire
+from .nameditemlist import NamedItemList
+from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .utils import dataclass_fields_asdict
+
+if TYPE_CHECKING:
+    from .diaglayer import DiagLayer
+
+
+@dataclass
+class ProtStack(IdentifiableElement):
+    # mandatory in ODX 2.2, but non existent in ODX 2.0
+    pdu_protocol_type: str
+    physical_link_type: str
+    comparam_subset_refs: List[OdxLinkRef]
+
+    @staticmethod
+    def from_et(et_element: ElementTree.Element, doc_frags: List[OdxDocFragment]) -> "ProtStack":
+        kwargs = dataclass_fields_asdict(IdentifiableElement.from_et(et_element, doc_frags))
+
+        pdu_protocol_type = odxrequire(et_element.findtext("PDU-PROTOCOL-TYPE"))
+        physical_link_type = odxrequire(et_element.findtext("PHYSICAL-LINK-TYPE"))
+        comparam_subset_refs = [
+            OdxLinkRef.from_et(csr_element, doc_frags)
+            for csr_element in et_element.iterfind("COMPARAM-SUBSET-REFS/"
+                                                   "COMPARAM-SUBSET-REF")
+        ]
+
+        return ProtStack(
+            pdu_protocol_type=pdu_protocol_type,
+            physical_link_type=physical_link_type,
+            comparam_subset_refs=comparam_subset_refs,
+            **kwargs)
+
+    def _build_odxlinks(self) -> Dict[OdxLinkId, Any]:
+        result = {self.odx_id: self}
+        return result
+
+    def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
+        self._comparam_subsets = NamedItemList[ComparamSubset](
+            [odxlinks.resolve(x, ComparamSubset) for x in self.comparam_subset_refs])
+
+    def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
+        pass

odxtools/templates/comparam-spec.odx-cs.xml.jinja2

@@ -0,0 +1,45 @@
+{#- -*- mode: sgml; tab-width: 1; indent-tabs-mode: nil -*-
+ #
+ # SPDX-License-Identifier: MIT
+ #
+ # This template writes an .odx-cs file for a communication
+ # parameter subset.
+-#}
+{%- import('macros/printAdminData.xml.jinja2') as pad -%}
+{%- import('macros/printCompanyData.xml.jinja2') as pcd -%}
+{%- import('macros/printProtStack.xml.jinja2') as pps %}
+{#- -#}
+
+<?xml version="1.0" encoding="UTF-8" standalone="no" ?>
+<ODX MODEL-VERSION="2.2.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="odx.xsd">
+<!-- Written using odxtools {{odxtools_version}} -->
+ <COMPARAM-SPEC ID="{{comparam_spec.odx_id.local_id}}">
+   <SHORT-NAME>{{comparam_spec.short_name}}</SHORT-NAME>
+   {%- if comparam_spec.long_name is not none %}
+   <LONG-NAME>{{comparam_spec.long_name|e}}</LONG-NAME>
+   {%- endif %}
+   {%- if comparam_spec.description and comparam_spec.description.strip() %}
+   <DESC>
+     {{comparam_spec.description}}
+   </DESC>
+   {%- endif %}
+   {%- if comparam_spec.admin_data is not none %}
+   {{- pad.printAdminData(comparam_spec.admin_data) | indent(3) }}
+   {%- endif %}
+   {%- if comparam_spec.company_datas %}
+   <COMPANY-DATAS>
+    {%- for cd in comparam_spec.company_datas %}
+     {{- pcd.printCompanyData(cd) | indent(5) -}}
+    {%- endfor %}
+   </COMPANY-DATAS>
+   {%- endif %}
+
+   {%- if comparam_spec.prot_stacks %}
+   <PROT-STACKS>
+    {%- for ps in comparam_spec.prot_stacks %}
+     {{- pps.printProtStack(ps) | indent(5) -}}
+    {%- endfor %}
+   </PROT-STACKS>
+   {%- endif %}
+ </COMPARAM-SPEC>
+</ODX>

odxtools/templates/macros/printCompanyData.xml.jinja2

@@ -6,7 +6,7 @@
 {%- import('macros/printElementID.xml.jinja2') as peid %}
 {%- import('macros/printSpecialData.xml.jinja2') as psd %}
 
-{%- macro printCompanyData(company_data) -%}
+{%- macro printCompanyData(company_data) %}
 <COMPANY-DATA ID="{{company_data.odx_id.local_id}}">
  {{ peid.printElementID(company_data)|indent(1) }}
  {%- if company_data.roles is not none %}

odxtools/templates/macros/printProtStack.xml.jinja2

@@ -0,0 +1,22 @@
+{#- -*- mode: sgml; tab-width: 1; indent-tabs-mode: nil -*-
+ #
+ # SPDX-License-Identifier: MIT
+-#}
+
+{%- import('macros/printElementID.xml.jinja2') as peid %}
+{%- import('macros/printSpecialData.xml.jinja2') as psd %}
+
+{%- macro printProtStack(ps) %}
+<PROT-STACK ID="{{ps.odx_id.local_id}}">
+ {{ peid.printElementID(ps) | indent(1) }}
+ <PDU-PROTOCOL-TYPE>{{ ps.pdu_protocol_type }}</PDU-PROTOCOL-TYPE>
+ <PHYSICAL-LINK-TYPE>{{ ps.physical_link_type }}</PHYSICAL-LINK-TYPE>
+ {%- if ps.comparam_subset_refs %}
+ <COMPARAM-SUBSET-REFS>
+  {%- for csr in ps.comparam_subset_refs %}
+  <COMPARAM-SUBSET-REF ID-REF="{{csr.ref_id}}" DOCREF="{{csr.ref_id}}" DOCTYPE="COMPARAM-SUBSET"/>
+  {%- endfor %}
+ </COMPARAM-SUBSET-REFS>
+ {%- endif %}
+</PROT-STACK>
+{%- endmacro %}

odxtools/templates/macros/printVariant.xml.jinja2

@@ -182,7 +182,10 @@
  </COMPARAM-REFS>
 {%- endif %}
 {%- if dlr.variant_type.value == "PROTOCOL" %}
-	<COMPARAM-SPEC-REF ID-REF="ISO_15765_3_on_ISO_15765_2" DOCREF="UDSOnCAN_CPS" DOCTYPE="COMPARAM-SPEC" />
+	<COMPARAM-SPEC-REF ID-REF="{{dlr.comparam_spec_ref.ref_id}}" DOCREF="{{dlr.comparam_spec_ref.ref_docs[0].doc_name}}" DOCTYPE="COMPARAM-SPEC"/>
+  {%- if dlr.prot_stack_snref is not none %}
+  <PROT-STACK-SNREF SHORT-NAME="{{dlr.prot_stack_snref}}"/>
+  {%- endif %}
 {%- endif %}
 {%- if dlr.parent_refs %}
  <PARENT-REFS>

odxtools/write_pdx_file.py

@@ -136,6 +136,22 @@ def write_pdx_file(
             zf.writestr(zf_file_name, comparam_subset_tpl.render(**vars))
         del vars["comparam_subset"]
 
+        # write the communication parameter specs
+        comparam_spec_tpl = jinja_env.get_template("comparam-spec.odx-cs.xml.jinja2")
+        for comparam_spec in database.comparam_specs:
+            zf_file_name = f"{comparam_spec.short_name}.odx-cs"
+            zf_file_cdate = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
+            zf_mime_type = "application/x-asam.odx.odx-cs"
+
+            vars["comparam_spec"] = comparam_spec
+
+            file_index.append((zf_file_name, zf_file_cdate, zf_mime_type))
+
+            zf.writestr(zf_file_name, comparam_spec_tpl.render(**vars))
+
+        if "comparam_spec" in vars:
+            del vars["comparam_spec"]
+
         # write the actual diagnostic data.
         dlc_tpl = jinja_env.get_template("diag_layer_container.odx-d.xml.jinja2")
         for dlc in database.diag_layer_containers: