Merge pull request #312 from andlaus/db_via_diaglayer

DiagLayer: remember the database

odxtools/database.py

@@ -131,7 +131,7 @@ def refresh(self) -> None:
         # let the diaglayers sort out the inherited objects and the
         # short name references
         for dlc in self.diag_layer_containers:
-            dlc._finalize_init(self._odxlinks)
+            dlc._finalize_init(self, self._odxlinks)
 
     @property
     def odxlinks(self) -> OdxLinkDatabase:

odxtools/diaglayer.py

@@ -1,11 +1,12 @@
 # SPDX-License-Identifier: MIT
 import re
 import warnings
-from copy import copy
+from copy import copy, deepcopy
 from dataclasses import dataclass
 from functools import cached_property
 from itertools import chain
-from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar, Union, cast
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar,
+                    Union, cast)
 from xml.etree import ElementTree
 
 from deprecation import deprecated
@@ -40,6 +41,9 @@
 from .unitgroup import UnitGroup
 from .unitspec import UnitSpec
 
+if TYPE_CHECKING:
+    from .database import Database
+
 TNamed = TypeVar("TNamed", bound=OdxNamed)
 
 PrefixTree = Dict[int, Union[List[DiagService], "PrefixTree"]]
@@ -125,7 +129,22 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
 
         self.diag_layer_raw._resolve_odxlinks(odxlinks)
 
-    def _finalize_init(self, odxlinks: OdxLinkDatabase) -> None:
+    def __deepcopy__(self, memo: Dict[int, Any]) -> Any:
+        """Create a deep copy of the diagnostic layer
+
+        Note that the copied diagnostic layer is not fully
+        initialized, so `_finalize_init()` should to be called on it
+        before it can be used normally.
+        """
+        cls = self.__class__
+        result = cls.__new__(cls)
+        memo[id(self)] = result
+
+        result.diag_layer_raw = deepcopy(self.diag_layer_raw, memo)
+
+        return result
+
+    def _finalize_init(self, database: "Database", odxlinks: OdxLinkDatabase) -> None:
         """This method deals with everything inheritance related and
         -- after the final set of objects covered by the diagnostic
         layer is determined -- resolves any short name references in
@@ -141,6 +160,11 @@ def _finalize_init(self, odxlinks: OdxLinkDatabase) -> None:
         excessive memory consumption for large databases...
         """
 
+        # this attribute may be removed later. it is currently
+        # required to properly deal with auxiliary files within the
+        # diagnostic layer.
+        self._database = database
+
         #####
         # fill in all applicable objects that use value inheritance
         #####

odxtools/diaglayercontainer.py

@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
 from itertools import chain
-from typing import Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 from xml.etree import ElementTree
 
 from .admindata import AdminData
@@ -14,6 +14,9 @@
 from .specialdatagroup import SpecialDataGroup
 from .utils import dataclass_fields_asdict
 
+if TYPE_CHECKING:
+    from .database import Database
+
 
 @dataclass
 class DiagLayerContainer(IdentifiableElement):
@@ -127,17 +130,17 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         for ecu_variant in self.ecu_variants:
             ecu_variant._resolve_odxlinks(odxlinks)
 
-    def _finalize_init(self, odxlinks: OdxLinkDatabase) -> None:
+    def _finalize_init(self, database: "Database", odxlinks: OdxLinkDatabase) -> None:
         for ecu_shared_data in self.ecu_shared_datas:
-            ecu_shared_data._finalize_init(odxlinks)
+            ecu_shared_data._finalize_init(database, odxlinks)
         for protocol in self.protocols:
-            protocol._finalize_init(odxlinks)
+            protocol._finalize_init(database, odxlinks)
         for functional_group in self.functional_groups:
-            functional_group._finalize_init(odxlinks)
+            functional_group._finalize_init(database, odxlinks)
         for base_variant in self.base_variants:
-            base_variant._finalize_init(odxlinks)
+            base_variant._finalize_init(database, odxlinks)
         for ecu_variant in self.ecu_variants:
-            ecu_variant._finalize_init(odxlinks)
+            ecu_variant._finalize_init(database, odxlinks)
 
     @property
     def diag_layers(self) -> NamedItemList[DiagLayer]:

odxtools/parentref.py

@@ -1,10 +1,12 @@
 # SPDX-License-Identifier: MIT
+from copy import deepcopy
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Dict, List
 from xml.etree import ElementTree
 
 from .exceptions import odxrequire
 from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
+from .utils import dataclass_fields_asdict
 
 if TYPE_CHECKING:
     from .diaglayer import DiagLayer
@@ -78,3 +80,14 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
 
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
         pass
+
+    def __deepcopy__(self, memo: Dict[int, Any]) -> Any:
+        cls = self.__class__
+        result = cls.__new__(cls)
+        memo[id(self)] = result
+
+        fields = dataclass_fields_asdict(self)
+        for name, value in fields.items():
+            setattr(result, name, deepcopy(value))
+
+        return result

tests/test_decoding.py

@@ -8,6 +8,7 @@
 from odxtools.compumethods.compuscale import CompuScale
 from odxtools.compumethods.identicalcompumethod import IdenticalCompuMethod
 from odxtools.compumethods.linearcompumethod import LinearCompuMethod
+from odxtools.database import Database
 from odxtools.dataobjectproperty import DataObjectProperty
 from odxtools.determinenumberofitems import DetermineNumberOfItems
 from odxtools.diagdatadictionaryspec import DiagDataDictionarySpec
@@ -224,8 +225,9 @@ def test_prefix_tree_construction(self) -> None:
             prot_stack_snref=None,
         )
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         self.assertEqual(
             diag_layer._prefix_tree,
@@ -344,8 +346,9 @@ def test_decode_request_coded_const(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         coded_message = bytes([0x7D, 0xAB])
         expected_message = Message(
@@ -486,8 +489,9 @@ def test_decode_request_coded_const_undefined_byte_position(self) -> None:
             prot_stack_snref=None,
         )
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         self.assertDictEqual(diag_layer._prefix_tree,
                              {0x12: {
@@ -695,8 +699,9 @@ def test_decode_request_structure(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         coded_message = bytes([0x12, 0x34])
         expected_message = Message(
@@ -902,8 +907,9 @@ def test_static_field_coding(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         expected_message = Message(
             coded_message=bytes([0x12, 0x34, 0x56, 0x00, 0x78, 0x9a, 0x00]),
@@ -1235,8 +1241,9 @@ def test_dynamic_endmarker_field_coding(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         ######
         ## test with endmarker termination
@@ -1508,8 +1515,9 @@ def test_dynamic_length_field_coding(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         expected_message = Message(
             coded_message=bytes([0x12, 0x00, 0x18, 0x00, 0x34, 0x44, 0x54]),
@@ -1738,8 +1746,9 @@ def test_decode_request_end_of_pdu_field(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         coded_message = bytes([0x12, 0x34, 0x54])
         expected_message = Message(
@@ -1917,8 +1926,9 @@ def test_decode_request_linear_compu_method(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         coded_message = bytes([0x7D, 0x12])
         # The physical value of the second parameter is decode(0x12) = decode(18) = 5 * 18 + 1 = 91
@@ -2102,8 +2112,9 @@ def test_decode_response(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         for sid, message in [(0x34, pos_response), (0x56, neg_response)]:
             coded_message = bytes([sid, 0xAB])

tests/test_diag_coded_types.py

@@ -10,6 +10,7 @@
 from odxtools.compumethods.identicalcompumethod import IdenticalCompuMethod
 from odxtools.compumethods.linearcompumethod import LinearCompuMethod
 from odxtools.createanydiagcodedtype import create_any_diag_coded_type_from_et
+from odxtools.database import Database
 from odxtools.dataobjectproperty import DataObjectProperty
 from odxtools.decodestate import DecodeState
 from odxtools.description import Description
@@ -297,8 +298,9 @@ def test_end_to_end(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         # Test decoding.
         coded_request = bytes([
@@ -619,8 +621,9 @@ def test_end_to_end(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         # Test decoding.
         coded_request = bytes([
@@ -955,8 +958,9 @@ def test_end_to_end(self) -> None:
         diag_layer = DiagLayer(diag_layer_raw=diag_layer_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(diag_layer._build_odxlinks())
+        db = Database()
         diag_layer._resolve_odxlinks(odxlinks)
-        diag_layer._finalize_init(odxlinks)
+        diag_layer._finalize_init(db, odxlinks)
 
         # Test decoding.
         coded_request = bytes([

tests/test_ecu_variant_matching.py

@@ -4,6 +4,7 @@
 
 import pytest
 
+from odxtools.database import Database
 from odxtools.diaglayer import DiagLayer
 from odxtools.diaglayerraw import DiagLayerRaw
 from odxtools.diaglayertype import DiagLayerType
@@ -221,8 +222,9 @@ def ecu_variant_1(
     )
     result = DiagLayer(diag_layer_raw=raw_layer)
     odxlinks.update(result._build_odxlinks())
+    db = Database()
     result._resolve_odxlinks(odxlinks)
-    result._finalize_init(odxlinks)
+    result._finalize_init(db, odxlinks)
     return result
 
 
@@ -259,8 +261,9 @@ def ecu_variant_2(
     )
     result = DiagLayer(diag_layer_raw=raw_layer)
     odxlinks.update(result._build_odxlinks())
+    db = Database()
     result._resolve_odxlinks(odxlinks)
-    result._finalize_init(odxlinks)
+    result._finalize_init(db, odxlinks)
     return result
 
 
@@ -298,8 +301,9 @@ def ecu_variant_3(
     )
     result = DiagLayer(diag_layer_raw=raw_layer)
     odxlinks.update(result._build_odxlinks())
+    db = Database()
     result._resolve_odxlinks(odxlinks)
-    result._finalize_init(odxlinks)
+    result._finalize_init(db, odxlinks)
     return result
 
 

tests/test_singleecujob.py

@@ -18,6 +18,7 @@
 from odxtools.compumethods.limit import Limit
 from odxtools.compumethods.linearcompumethod import LinearCompuMethod
 from odxtools.compumethods.texttablecompumethod import TexttableCompuMethod
+from odxtools.database import Database
 from odxtools.dataobjectproperty import DataObjectProperty
 from odxtools.description import Description
 from odxtools.diaglayer import DiagLayer
@@ -465,8 +466,12 @@ def test_resolve_odxlinks(self) -> None:
             self.context.negOutputDOP.odx_id: self.context.negOutputDOP,
         })
 
+        db = Database()
+        db.add_auxiliary_file("abc.jar",
+                              b"this is supposed to be a JAR archive, but it isn't (HARR)")
+
         dl._resolve_odxlinks(odxlinks)
-        dl._finalize_init(odxlinks)
+        dl._finalize_init(db, odxlinks)
 
         self.assertEqual(self.context.extensiveTask,
                          self.singleecujob_object.functional_classes.extensiveTask)

tests/test_unit_spec.py

@@ -4,6 +4,7 @@
 
 from odxtools.compumethods.compumethod import CompuCategory
 from odxtools.compumethods.identicalcompumethod import IdenticalCompuMethod
+from odxtools.database import Database
 from odxtools.dataobjectproperty import DataObjectProperty
 from odxtools.diagdatadictionaryspec import DiagDataDictionarySpec
 from odxtools.diaglayer import DiagLayer
@@ -210,8 +211,9 @@ def test_resolve_odxlinks(self) -> None:
         dl = DiagLayer(diag_layer_raw=dl_raw)
         odxlinks = OdxLinkDatabase()
         odxlinks.update(dl._build_odxlinks())
+        db = Database()
         dl._resolve_odxlinks(odxlinks)
-        dl._finalize_init(odxlinks)
+        dl._finalize_init(db, odxlinks)
 
         param = dl.requests[0].parameters[1]
         assert isinstance(param, ValueParameter)