aasx.py: Add failsafe for writing and reading AASX files (#378)

Previously, the `failsafe` parameter, allowing to skip errors and only 
log them as warnings instead, only existed for JSON and XML 
(de-)serialization. 

This PR extends the feature to also be included when reading and 
writing AASX files. For this, we add the `failsafe` boolean attribute 
to the `AASXReader` and `AASXWriter` classes in `adapter.aasx`. 
If `failsafe` is `True`, the document is parsed in a failsafe way: 
Missing attributes and elements are logged instead of causing 
exceptions. Defect objects are skipped. The default value is `False`, 
keeping the existing behavior.

Furthermore, we do a little code-style clean up on string-formatting 
in the `adapter.aasx` module, updating all occurrences from 
(historically) different syntaxes to the more modern f-string syntax.

Fixes #228

Author: Frosty2500

sdk/basyx/aas/adapter/aasx.py

@@ -28,7 +28,6 @@
 import itertools
 import logging
 import os
-import re
 from typing import Dict, Tuple, IO, Union, List, Set, Optional, Iterable, Iterator
 
 from .xml import read_aas_xml_file, write_aas_xml_file
@@ -60,7 +59,7 @@ class AASXReader:
             reader.read_into(objects, files)
 
     """
-    def __init__(self, file: Union[os.PathLike, str, IO]):
+    def __init__(self, file: Union[os.PathLike, str, IO], failsafe: bool = True):
         """
         Open an AASX reader for the given filename or file handle
 
@@ -69,16 +68,19 @@ def __init__(self, file: Union[os.PathLike, str, IO]):
         closing under any circumstances.
 
         :param file: A filename, file path or an open file-like object in binary mode
+        :param failsafe: If ``True``, the document is parsed in a failsafe way: Missing attributes and elements are
+            logged instead of causing exceptions. Defect objects are skipped.
         :raises FileNotFoundError: If the file does not exist
         :raises ValueError: If the file is not a valid OPC zip package
         """
+        self.failsafe: bool = failsafe
         try:
-            logger.debug("Opening {} as AASX pacakge for reading ...".format(file))
+            logger.debug(f"Opening {file} as AASX package for reading ...")
             self.reader = pyecma376_2.ZipPackageReader(file)
         except FileNotFoundError:
             raise
         except Exception as e:
-            raise ValueError("{} is not a valid ECMA376-2 (OPC) file: {}".format(file, e)) from e
+            raise ValueError(f"{file} is not a valid ECMA376-2 (OPC) file: {e}") from e
 
     def get_core_properties(self) -> pyecma376_2.OPCCoreProperties:
         """
@@ -132,7 +134,7 @@ def read_into(self, object_store: model.AbstractObjectStore,
                              objects from the AASX file to
         :param file_store: A :class:`SupplementaryFileContainer <.AbstractSupplementaryFileContainer>` to add the
                            embedded supplementary files to
-        :param override_existing: If ``True``, existing objects in the object store are overridden with objects from the
+        :param override_existing: If ``True``, existing objects in the ObjectStore are overridden with objects from the
             AASX that have the same :class:`~basyx.aas.model.base.Identifier`. Default behavior is to skip those objects
             from the AASX.
         :return: A set of the :class:`Identifiers <basyx.aas.model.base.Identifier>` of all
@@ -172,7 +174,10 @@ def read_into(self, object_store: model.AbstractObjectStore,
                 self._read_aas_part_into(split_part, object_store, file_store,
                                          read_identifiables, override_existing, **kwargs)
         if no_aas_files_found:
-            logger.warning("No AAS files found in AASX package")
+            if self.failsafe:
+                logger.warning("No AAS files found in AASX package")
+            else:
+                raise ValueError("No AAS files found in AASX package")
 
         return read_identifiables
 
@@ -205,20 +210,23 @@ def _read_aas_part_into(self, part_name: str,
             from a File object of this part
         :param read_identifiables: A set of Identifiers of objects which have already been read. New objects'
             Identifiers are added to this set. Objects with already known Identifiers are skipped silently.
-        :param override_existing: If True, existing objects in the object store are overridden with objects from the
+        :param override_existing: If True, existing objects in the ObjectStore are overridden with objects from the
             AASX that have the same Identifier. Default behavior is to skip those objects from the AASX.
         """
         for obj in self._parse_aas_part(part_name, **kwargs):
             if obj.id in read_identifiables:
                 continue
             if obj.id in object_store:
                 if override_existing:
-                    logger.info("Overriding existing object in  ObjectStore with {} ...".format(obj))
+                    logger.info(f"Overriding existing object in  ObjectStore with {obj} ...")
                     object_store.discard(obj)
                 else:
-                    logger.warning("Skipping {}, since an object with the same id is already contained in the "
-                                   "ObjectStore".format(obj))
-                    continue
+                    if self.failsafe:
+                        logger.warning(f"Skipping {obj}, since an object with the same id is already contained in the "
+                                       "ObjectStore")
+                        continue
+                    else:
+                        raise ValueError(f"Object with id {obj} is already contained in the ObjectStore")
             object_store.add(obj)
             read_identifiables.add(obj.id)
             if isinstance(obj, model.Submodel):
@@ -236,17 +244,21 @@ def _parse_aas_part(self, part_name: str, **kwargs) -> model.DictObjectStore:
         content_type = self.reader.get_content_type(part_name)
         extension = part_name.split("/")[-1].split(".")[-1]
         if content_type.split(";")[0] in ("text/xml", "application/xml") or content_type == "" and extension == "xml":
-            logger.debug("Parsing AAS objects from XML stream in OPC part {} ...".format(part_name))
+            logger.debug(f"Parsing AAS objects from XML stream in OPC part {part_name} ...")
             with self.reader.open_part(part_name) as p:
-                return read_aas_xml_file(p, **kwargs)
+                return read_aas_xml_file(p, failsafe=self.failsafe, **kwargs)
         elif content_type.split(";")[0] in ("text/json", "application/json") \
                 or content_type == "" and extension == "json":
-            logger.debug("Parsing AAS objects from JSON stream in OPC part {} ...".format(part_name))
+            logger.debug(f"Parsing AAS objects from JSON stream in OPC part {part_name} ...")
             with self.reader.open_part(part_name) as p:
-                return read_aas_json_file(io.TextIOWrapper(p, encoding='utf-8-sig'), **kwargs)
+                return read_aas_json_file(io.TextIOWrapper(p, encoding='utf-8-sig'), failsafe=self.failsafe, **kwargs)
         else:
-            logger.error("Could not determine part format of AASX part {} (Content Type: {}, extension: {}"
-                         .format(part_name, content_type, extension))
+            error_message = (f"Could not determine part format of AASX part {part_name} (Content Type: {content_type},"
+                             f" extension: {extension}")
+            if self.failsafe:
+                logger.error(error_message)
+            else:
+                raise ValueError(error_message)
             return model.DictObjectStore()
 
     def _collect_supplementary_files(self, part_name: str, submodel: model.Submodel,
@@ -255,7 +267,7 @@ def _collect_supplementary_files(self, part_name: str, submodel: model.Submodel,
         Helper function to search File objects within a single parsed Submodel, extract the referenced supplementary
         files and update the File object's values with the absolute path.
 
-        :param part_name: The OPC part name of the part the submodel has been parsed from. This is used to resolve
+        :param part_name: The OPC part name of the part the Submodel has been parsed from. This is used to resolve
             relative file paths.
         :param submodel: The Submodel to process
         :param file_store: The SupplementaryFileContainer to add the extracted supplementary files to
@@ -268,11 +280,11 @@ def _collect_supplementary_files(self, part_name: str, submodel: model.Submodel,
                 # to refer to files within the AASX package. Thus, we must skip all other types of URIs (esp. absolute
                 # URIs and network-path references)
                 if element.value.startswith('//') or ':' in element.value.split('/')[0]:
-                    logger.info("Skipping supplementary file %s, since it seems to be an absolute URI or network-path "
-                                "URI reference", element.value)
+                    logger.info(f"Skipping supplementary file {element.value}, since it seems to be an absolute URI or "
+                                f"network-path URI reference")
                     continue
                 absolute_name = pyecma376_2.package_model.part_realpath(element.value, part_name)
-                logger.debug("Reading supplementary file {} from AASX package ...".format(absolute_name))
+                logger.debug(f"Reading supplementary file {absolute_name} from AASX package ...")
                 with self.reader.open_part(absolute_name) as p:
                     final_name = file_store.add_file(absolute_name, p, self.reader.get_content_type(absolute_name))
                 element.value = final_name
@@ -308,16 +320,19 @@ class AASXWriter:
     """
     AASX_ORIGIN_PART_NAME = "/aasx/aasx-origin"
 
-    def __init__(self, file: Union[os.PathLike, str, IO]):
+    def __init__(self, file: Union[os.PathLike, str, IO], failsafe: bool = True):
         """
         Create a new AASX package in the given file and open the AASXWriter to add contents to the package.
 
         Make sure to call ``AASXWriter.close()`` after writing all contents to write the aas-spec relationships for all
         AAS parts to the file and close the underlying ZIP file writer. You may also use the AASXWriter as a context
         manager to ensure closing under any circumstances.
 
+        :param failsafe: If ``True``, the document is written in a failsafe way: Missing attributes and elements are
+            logged instead of causing exceptions. Defect objects are skipped.
         :param file: filename, path, or binary file handle opened for writing
         """
+        self.failsafe: bool = failsafe
         # names of aas-spec parts, used by `_write_aasx_origin_relationships()`
         self._aas_part_names: List[str] = []
         # name of the thumbnail part (if any)
@@ -377,7 +392,7 @@ def write_aas(self,
         :param write_json:  If ``True``, JSON parts are created for the AAS and each
             :class:`~basyx.aas.model.submodel.Submodel` in the AASX package file instead of XML parts.
             Defaults to ``False``.
-        :raises KeyError: If one of the AAS could not be retrieved from the object store (unresolvable
+        :raises KeyError: If one of the AAS could not be retrieved from the ObjectStore (unresolvable
             :class:`Submodels <basyx.aas.model.submodel.Submodel>` and
             :class:`ConceptDescriptions <basyx.aas.model.concept.ConceptDescription>` are skipped, logging a
             warning/info message)
@@ -391,12 +406,15 @@ def write_aas(self,
         for aas_id in aas_ids:
             try:
                 aas = object_store.get_identifiable(aas_id)
-            # TODO add failsafe mode
-            except KeyError:
-                raise
-            if not isinstance(aas, model.AssetAdministrationShell):
-                raise TypeError(f"Identifier {aas_id} does not belong to an AssetAdministrationShell object but to "
-                                f"{aas!r}")
+                if not isinstance(aas, model.AssetAdministrationShell):
+                    raise TypeError(f"Identifier {aas_id} does not belong to an AssetAdministrationShell object but to "
+                                    f"{aas!r}")
+            except (KeyError, TypeError) as e:
+                if self.failsafe:
+                    logger.error(f"Skipping AAS {aas_id}: {e}")
+                    continue
+                else:
+                    raise
 
             # Add the AssetAdministrationShell object to the data part
             objects_to_be_written.add(aas)
@@ -406,8 +424,11 @@ def write_aas(self,
                 try:
                     submodel = submodel_ref.resolve(object_store)
                 except KeyError:
-                    logger.warning("Could not find submodel %s. Skipping it.", str(submodel_ref))
-                    continue
+                    if self.failsafe:
+                        logger.warning(f"Could not find Submodel {submodel_ref}. Skipping it.")
+                        continue
+                    else:
+                        raise KeyError(f"Could not find Submodel {submodel_ref!r}")
                 objects_to_be_written.add(submodel)
 
         # Traverse object tree and check if semanticIds are referencing to existing ConceptDescriptions in the
@@ -423,13 +444,20 @@ def write_aas(self,
                 try:
                     cd = semantic_id.resolve(object_store)
                 except KeyError:
-                    logger.warning("ConceptDescription for semanticId %s not found in object store. Skipping it.",
-                                   str(semantic_id))
-                    continue
+                    if self.failsafe:
+                        logger.warning(f"ConceptDescription for semanticId {semantic_id} not found in ObjectStore. "
+                                       f"Skipping it.")
+                        continue
+                    else:
+                        raise KeyError(f"ConceptDescription for semanticId {semantic_id!r} not found in ObjectStore.")
                 except model.UnexpectedTypeError as e:
-                    logger.error("semanticId %s resolves to %s, which is not a ConceptDescription. Skipping it.",
-                                 str(semantic_id), e.value)
-                    continue
+                    if self.failsafe:
+                        logger.error(f"semanticId {semantic_id} resolves to {e.value}, "
+                                     f"which is not a ConceptDescription. Skipping it.")
+                        continue
+                    else:
+                        raise TypeError(f"semanticId {semantic_id!r} resolves to {e.value!r}, which is not a"
+                                        f" ConceptDescription.") from e
                 concept_descriptions.append(cd)
         objects_to_be_written.update(concept_descriptions)
 
@@ -453,7 +481,7 @@ def write_aas_objects(self,
         This method takes the AAS's :class:`~basyx.aas.model.base.Identifier` (as ``aas_id``) to retrieve it
         from the given object_store. If the list of written objects includes :class:`~basyx.aas.model.submodel.Submodel`
         objects, Supplementary files which are referenced by :class:`~basyx.aas.model.submodel.File` objects within
-        those submodels, are also added to the AASX package.
+        those Submodels, are also added to the AASX package.
 
         .. attention::
 
@@ -478,7 +506,7 @@ def write_aas_objects(self,
         :param additional_relationships: Optional OPC/ECMA376 relationships which should originate at the AAS object
             part to be written, in addition to the aas-suppl relationships which are created automatically.
         """
-        logger.debug("Writing AASX part {} with AAS objects ...".format(part_name))
+        logger.debug(f"Writing AASX part {part_name} with AAS objects ...")
 
         objects: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
 
@@ -487,8 +515,11 @@ def write_aas_objects(self,
             try:
                 the_object = object_store.get_identifiable(identifier)
             except KeyError:
-                logger.error("Could not find object {} in ObjectStore".format(identifier))
-                continue
+                if self.failsafe:
+                    logger.error(f"Could not find object {identifier} in ObjectStore")
+                    continue
+                else:
+                    raise KeyError(f"Could not find object {identifier!r} in ObjectStore")
             objects.add(the_object)
 
         self.write_all_aas_objects(part_name, objects, file_store, write_json, split_part, additional_relationships)
@@ -529,7 +560,7 @@ def write_all_aas_objects(self,
         :param additional_relationships: Optional OPC/ECMA376 relationships which should originate at the AAS object
             part to be written, in addition to the aas-suppl relationships which are created automatically.
         """
-        logger.debug("Writing AASX part {} with AAS objects ...".format(part_name))
+        logger.debug(f"Writing AASX part {part_name} with AAS objects ...")
         supplementary_files: List[str] = []
 
         # Retrieve objects and scan for referenced supplementary files
@@ -556,29 +587,36 @@ def write_all_aas_objects(self,
             else:
                 write_aas_xml_file(p, objects)
 
-        # Write submodel's supplementary files to AASX file
+        # Write Submodel's supplementary files to AASX file
         supplementary_file_names = []
         for file_name in supplementary_files:
             try:
                 content_type = file_store.get_content_type(file_name)
                 hash = file_store.get_sha256(file_name)
             except KeyError:
-                logger.warning("Could not find file {} in file store.".format(file_name))
-                continue
+                if self.failsafe:
+                    logger.warning(f"Could not find file {file_name} in FileStore.")
+                    continue
+                else:
+                    raise KeyError(f"Could not find file {file_name} in FileStore.")
             # Check if this supplementary file has already been written to the AASX package or has a name conflict
             if self._supplementary_part_names.get(file_name) == hash:
                 continue
             elif file_name in self._supplementary_part_names:
-                logger.error("Trying to write supplementary file {} to AASX twice with different contents"
-                             .format(file_name))
-            logger.debug("Writing supplementary file {} to AASX package ...".format(file_name))
+                if self.failsafe:
+                    logger.error(f"Trying to write supplementary file {file_name} to AASX "
+                                 f"twice with different contents")
+                else:
+                    raise ValueError(f"Trying to write supplementary file {file_name} to AASX twice with"
+                                     f" different contents")
+            logger.debug(f"Writing supplementary file {file_name} to AASX package ...")
             with self.writer.open_part(file_name, content_type) as p:
                 file_store.write_file(file_name, p)
             supplementary_file_names.append(pyecma376_2.package_model.normalize_part_name(file_name))
             self._supplementary_part_names[file_name] = hash
 
-        # Add relationships from submodel to supplementary parts
-        logger.debug("Writing aas-suppl relationships for AAS object part {} to AASX package ...".format(part_name))
+        # Add relationships from Submodel to supplementary parts
+        logger.debug(f"Writing aas-suppl relationships for AAS object part {part_name} to AASX package ...")
         self.writer.write_relationships(
             itertools.chain(
                 (pyecma376_2.OPCRelationship("r{}".format(i),
@@ -617,7 +655,7 @@ def write_thumbnail(self, name: str, data: bytearray, content_type: str):
         :param content_type: OPC content type (MIME type) of the image file
         """
         if self._thumbnail_part is not None:
-            raise RuntimeError("package thumbnail has already been written to {}.".format(self._thumbnail_part))
+            raise RuntimeError(f"package thumbnail has already been written to {self._thumbnail_part}.")
         with self.writer.open_part(name, content_type) as p:
             p.write(data)
         self._thumbnail_part = name