Merge pull request #310 from andlaus/improve_aux_files

Improve auxiliary file handling

examples/pdxcopy.py

@@ -23,29 +23,16 @@
     metavar="OUTPUT_PDX_FILE",
     help="Path to the where the resulting .pdx file is written",
 )
-argparser.add_argument(
-    "aux_files",
-    metavar="AUX_FILES",
-    nargs="*",
-    default=[],
-    help="The names of the auxiliary files to be included in the resulting .pdx file",
-)
 
 args = argparser.parse_args()
 
 in_file_name = args.input_pdx_file
 out_file_name = args.output_pdx_file
-aux_file_names = args.aux_files
-
-# a content specifier is a tuple of (name_in_zipfile,
-# content_data). Here we simply read all all content from files on
-# the filesystem...
-auxiliary_content = [(x, open(x, "rb").read()) for x in aux_file_names]
 
 print(f"Loading input file '{in_file_name}'...", end="", flush=True)
 db = odxtools.load_pdx_file(in_file_name)
 print(" done")
 
 print(f"Writing output file '{out_file_name}'...", end="", flush=True)
-odxtools.write_pdx_file(out_file_name, db, auxiliary_content)
+odxtools.write_pdx_file(out_file_name, db)
 print(" done")

examples/somersaultecu.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: MIT
 import pathlib
 from enum import IntEnum
+from io import BytesIO
 from itertools import chain
 from typing import Any, Dict
 from xml.etree import ElementTree
@@ -1962,10 +1963,10 @@ class SomersaultSID(IntEnum):
             diagnostic_class=None,
             prog_codes=[
                 ProgCode(
-                    code_file="jobs.jar",
+                    code_file="jobs.py",
                     encryption=None,
-                    syntax="JAR",
-                    entrypoint="com.supervisor.jobs.CompulsoryProgram",
+                    syntax="PYTHON3",
+                    entrypoint="compulsory_program",
                     revision="1.23.4",
                     library_refs=[],
                 ),
@@ -2467,6 +2468,11 @@ class SomersaultSID(IntEnum):
 database = Database()
 database._diag_layer_containers = NamedItemList([somersault_dlc])
 database._comparam_subsets = NamedItemList(comparam_subsets)
+database.add_auxiliary_file("jobs.py",
+                            BytesIO(b"""
+def compulsory_program():
+  print("Hello, World")
+"""))
 
 # Create ID mapping and resolve references
 database.refresh()

odxtools/database.py

@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: MIT
 from itertools import chain
 from pathlib import Path
-from typing import List, Optional, OrderedDict
+from typing import IO, List, Optional, OrderedDict
 from xml.etree import ElementTree
 from zipfile import ZipFile
 
@@ -27,7 +27,7 @@ def __init__(self,
                  pdx_zip: Optional[ZipFile] = None,
                  odx_d_file_name: Optional[str] = None) -> None:
         self.model_version: Optional[Version] = None
-        self.auxiliary_files: OrderedDict[str, bytes] = OrderedDict()
+        self.auxiliary_files: OrderedDict[str, IO[bytes]] = OrderedDict()
 
         # create an empty database object
         self._diag_layer_containers = NamedItemList[DiagLayerContainer]()
@@ -37,28 +37,28 @@ def __init__(self,
     def add_pdx_file(self, pdx_file_name: str) -> None:
         pdx_zip = ZipFile(pdx_file_name)
 
-        names = list(pdx_zip.namelist())
-        names.sort()
-        for zip_member in names:
+        for zip_member in pdx_zip.namelist():
             # The name of ODX files can end with .odx, .odx-d,
             # .odx-c, .odx-cs, .odx-e, .odx-f, .odx-fd, .odx-m,
             # .odx-v .  We could test for all that, or just make
             # sure that the file's suffix starts with .odx
             p = Path(zip_member)
             if p.suffix.lower().startswith(".odx"):
-                root = ElementTree.fromstring(pdx_zip.read(zip_member))
+                root = ElementTree.parse(pdx_zip.open(zip_member)).getroot()
                 self._process_xml_tree(root)
             elif p.name.lower() != "index.xml":
-                self.add_auxiliary_file(zip_member, pdx_zip.read(zip_member))
+                self.add_auxiliary_file(zip_member, pdx_zip.open(zip_member))
 
     def add_odx_file(self, odx_file_name: str) -> None:
         self._process_xml_tree(ElementTree.parse(odx_file_name).getroot())
 
-    def add_auxiliary_file(self, aux_file_name: str, aux_file_data: Optional[bytes] = None) -> None:
-        if aux_file_data is None:
-            aux_file_data = open(aux_file_name, "rb").read()
+    def add_auxiliary_file(self,
+                           aux_file_name: str,
+                           aux_file_obj: Optional[IO[bytes]] = None) -> None:
+        if aux_file_obj is None:
+            aux_file_obj = open(aux_file_name, "rb")
 
-        self.auxiliary_files[aux_file_name] = aux_file_data
+        self.auxiliary_files[aux_file_name] = aux_file_obj
 
     def _process_xml_tree(self, root: ElementTree.Element) -> None:
         dlcs: List[DiagLayerContainer] = []
@@ -71,7 +71,7 @@ def _process_xml_tree(self, root: ElementTree.Element) -> None:
             odxraise(f"Different ODX versions used for the same database (ODX {model_version} "
                      f"and ODX {self.model_version}")
 
-            self.model_version = model_version
+        self.model_version = model_version
 
         dlc = root.find("DIAG-LAYER-CONTAINER")
         if dlc is not None:

odxtools/loadfile.py

@@ -58,7 +58,7 @@ def load_directory(dir_name: Union[str, Path]) -> Database:
         elif p.suffix.lower().startswith(".odx"):
             db.add_odx_file(str(p))
         elif p.name.lower() != "index.xml":
-            db.add_auxiliary_file(p.name, open(str(p), "rb").read())
+            db.add_auxiliary_file(p.name, open(str(p), "rb"))
 
     db.refresh()
     return db

odxtools/progcode.py

@@ -1,9 +1,9 @@
 # SPDX-License-Identifier: MIT
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
 from xml.etree import ElementTree
 
-from .exceptions import odxrequire
+from .exceptions import odxraise, odxrequire
 from .odxlink import OdxDocFragment, OdxLinkDatabase, OdxLinkId, OdxLinkRef
 
 if TYPE_CHECKING:
@@ -20,6 +20,10 @@ class ProgCode:
     entrypoint: Optional[str]
     library_refs: List[OdxLinkRef]
 
+    @property
+    def code(self) -> bytes:
+        return self._code
+
     @staticmethod
     def from_et(et_element: ElementTree.Element, doc_frags: List[OdxDocFragment]) -> "ProgCode":
         code_file = odxrequire(et_element.findtext("CODE-FILE"))
@@ -52,4 +56,15 @@ def _resolve_odxlinks(self, odxlinks: OdxLinkDatabase) -> None:
         pass
 
     def _resolve_snrefs(self, diag_layer: "DiagLayer") -> None:
-        pass
+        db = diag_layer._database
+
+        aux_file = db.auxiliary_files.get(self.code_file)
+
+        if aux_file is None:
+            odxraise(f"Reference to auxiliary file '{self.code_file}' "
+                     f"could not be resolved")
+            self._code: bytes = cast(bytes, None)
+            return
+
+        self._code = aux_file.read()
+        aux_file.seek(0)

odxtools/writepdxfile.py

@@ -1,10 +1,11 @@
 # SPDX-License-Identifier: MIT
 import datetime
 import inspect
+import mimetypes
 import os
 import time
 import zipfile
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, Optional
 
 import jinja2
 
@@ -60,17 +61,13 @@ def make_bool_xml_attrib(attrib_name: str, attrib_val: Optional[bool]) -> str:
 def write_pdx_file(
     output_file_name: str,
     database: Database,
-    auxiliary_content_specifiers: Optional[List[Tuple[str, bytes]]] = None,
     templates_dir: str = __templates_dir,
 ) -> bool:
     """
     Write an internalized database to a PDX file.
     """
     global odxdatabase
 
-    if auxiliary_content_specifiers is None:
-        auxiliary_content_specifiers = []
-
     odxdatabase = database
 
     file_index = []
@@ -96,12 +93,18 @@ def write_pdx_file(
                     # are written based on the database)
                     continue
 
-                template_file_mime_type = "text/plain"
+                template_file_mime_type = None
                 if template_file_name.endswith(".odx-cs"):
                     template_file_mime_type = "application/x-asam.odx.odx-cs"
                 elif template_file_name.endswith(".odx-d"):
                     template_file_mime_type = "application/x-asam.odx.odx-d"
 
+                guessed_mime_type, guessed_encoding = mimetypes.guess_type(template_file_name)
+                if template_file_mime_type is None and guessed_mime_type is not None:
+                    template_file_mime_type = guessed_mime_type
+                else:
+                    template_file_mime_type = "application/octet-stream"
+
                 in_path = [root]
                 in_path.append(template_file_name)
                 in_file_name = os.path.sep.join(in_path)
@@ -115,20 +118,26 @@ def write_pdx_file(
                     out_file.write(open(in_file_name, "rb").read())
 
         # write the auxiliary files
-        for output_file_name, data in auxiliary_content_specifiers:
+        for output_file_name, data_file in database.auxiliary_files.items():
             file_cdate = datetime.datetime.fromtimestamp(time.time())
             creation_date = file_cdate.strftime("%Y-%m-%dT%H:%M:%S")
 
-            mime_type = "text/plain"
+            mime_type = None
             if output_file_name.endswith(".odx-cs"):
                 mime_type = "application/x-asam.odx.odx-cs"
             elif output_file_name.endswith(".odx-d"):
                 mime_type = "application/x-asam.odx.odx-d"
 
+            guessed_mime_type, guessed_encoding = mimetypes.guess_type(output_file_name)
+            if mime_type is None and guessed_mime_type is not None:
+                mime_type = guessed_mime_type
+            else:
+                mime_type = "application/octet-stream"
+
             zf_name = os.path.basename(output_file_name)
             with zf.open(zf_name, "w") as out_file:
                 file_index.append((zf_name, creation_date, mime_type))
-                out_file.write(data)
+                out_file.write(data_file.read())
 
         jinja_env = jinja2.Environment(loader=jinja2.FileSystemLoader(templates_dir))
         jinja_env.globals["hasattr"] = hasattr
@@ -187,7 +196,7 @@ def write_pdx_file(
 
         # write the index.xml file
         vars["file_index"] = file_index
-        index_tpl = jinja_env.get_template("index.xml.xml.jinja2")
+        index_tpl = jinja_env.get_template("index.xml.jinja2")
         text = index_tpl.render(**vars)
         zf.writestr("index.xml", text)
 

tests/test_singleecujob.py

@@ -2,6 +2,7 @@
 import inspect
 import os
 import unittest
+from io import BytesIO
 from typing import NamedTuple, cast
 from xml.etree import ElementTree
 
@@ -468,7 +469,7 @@ def test_resolve_odxlinks(self) -> None:
 
         db = Database()
         db.add_auxiliary_file("abc.jar",
-                              b"this is supposed to be a JAR archive, but it isn't (HARR)")
+                              BytesIO(b"this is supposed to be a JAR archive, but it isn't (HARR)"))
 
         dl._resolve_odxlinks(odxlinks)
         dl._finalize_init(db, odxlinks)

tests/test_somersault.py

@@ -3,6 +3,8 @@
 from io import StringIO
 from unittest.mock import patch
 
+from packaging.version import Version
+
 from odxtools.description import Description
 from odxtools.exceptions import OdxError, odxrequire
 from odxtools.loadfile import load_pdx_file
@@ -14,6 +16,7 @@
 class TestDatabase(unittest.TestCase):
 
     def test_db_structure(self) -> None:
+        self.assertEqual(odxdb.model_version, Version("2.2.0"))
         self.assertEqual([x.short_name for x in odxdb.diag_layer_containers], ["somersault"])
 
         self.assertEqual(