Select Git revision
large_residential_area.skp
-
Jonas Stienen authoredJonas Stienen authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
aasx.py 40.84 KiB
# Copyright (c) 2020 PyI40AAS Contributors
#
# This program and the accompanying materials are made available under the terms of the Eclipse Public License v. 2.0
# which is available at https://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 which is available
# at https://www.apache.org/licenses/LICENSE-2.0.
#
# SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
"""
.. _adapter.aasx:
Functionality for reading and writing AASX files according to "Details of the Asset Administration Shell Part 1 V2.0",
section 7.
The AASX file format is built upon the Open Packaging Conventions (OPC; ECMA 376-2). We use the `pyecma376_2` library
for low level OPC reading and writing. It currently supports all required features except for embedded digital
signatures.
Writing and reading of AASX packages is performed through the :class:`~.AASXReader` and :class:`~.AASXWriter` classes.
Each instance of these classes wraps an existing AASX file resp. a file to be created and allows to read/write the
included AAS objects into/form :class:`ObjectStores <aas.model.provider.AbstractObjectStore>`.
For handling of embedded supplementary files, this module provides the
:class:`~.AbstractSupplementaryFileContainer` class
interface and the :class:`~.DictSupplementaryFileContainer` implementation.
"""
import abc
import hashlib
import io
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
from .. import model
from .json import read_aas_json_file, write_aas_json_file
import pyecma376_2
from ..util import traversal
logger = logging.getLogger(__name__)
RELATIONSHIP_TYPE_AASX_ORIGIN = "http://www.admin-shell.io/aasx/relationships/aasx-origin"
RELATIONSHIP_TYPE_AAS_SPEC = "http://www.admin-shell.io/aasx/relationships/aas-spec"
RELATIONSHIP_TYPE_AAS_SPEC_SPLIT = "http://www.admin-shell.io/aasx/relationships/aas-spec-split"
RELATIONSHIP_TYPE_AAS_SUPL = "http://www.admin-shell.io/aasx/relationships/aas-suppl"
class AASXReader:
"""
An AASXReader wraps an existing AASX package file to allow reading its contents and metadata.
Basic usage:
.. code-block:: python
objects = DictObjectStore()
files = DictSupplementaryFileContainer()
with AASXReader("filename.aasx") as reader:
meta_data = reader.get_core_properties()
reader.read_into(objects, files)
"""
def __init__(self, file: Union[os.PathLike, str, IO]):
"""
Open an AASX reader for the given filename or file handle
The given file is opened as OPC ZIP package. Make sure to call `AASXReader.close()` after reading the file
contents to close the underlying ZIP file reader. You may also use the AASXReader as a context manager to ensure
closing under any circumstances.
:param file: A filename, file path or an open file-like object in binary mode
:raises ValueError: If the file is not a valid OPC zip package
"""
try:
logger.debug("Opening {} as AASX pacakge for reading ...".format(file))
self.reader = pyecma376_2.ZipPackageReader(file)
except Exception as e:
raise ValueError("{} is not a valid ECMA376-2 (OPC) file".format(file)) from e
def get_core_properties(self) -> pyecma376_2.OPCCoreProperties:
"""
Retrieve the OPC Core Properties (meta data) of the AASX package file.
If no meta data is provided in the package file, an emtpy OPCCoreProperties object is returned.
:return: The AASX package's meta data
"""
return self.reader.get_core_properties()
def get_thumbnail(self) -> Optional[bytes]:
"""
Retrieve the packages thumbnail image
The thumbnail image file is read into memory and returned as bytes object. You may use some python image library
for further processing or conversion, e.g. `pillow`:
.. code-block:: python
import io
from PIL import Image
thumbnail = Image.open(io.BytesIO(reader.get_thumbnail()))
:return: The AASX package thumbnail's file contents or None if no thumbnail is provided
"""
try:
thumbnail_part = self.reader.get_related_parts_by_type()[pyecma376_2.RELATIONSHIP_TYPE_THUMBNAIL][0]
except IndexError:
return None
with self.reader.open_part(thumbnail_part) as p:
return p.read()
def read_into(self, object_store: model.AbstractObjectStore,
file_store: "AbstractSupplementaryFileContainer",
override_existing: bool = False) -> Set[model.Identifier]:
"""
Read the contents of the AASX package and add them into a given
:class:`ObjectStore <aas.model.provider.AbstractObjectStore>`
This function does the main job of reading the AASX file's contents. It traverses the relationships within the
package to find AAS JSON or XML parts, parses them and adds the contained AAS objects into the provided
`object_store`. While doing so, it searches all parsed :class:`Submodels <aas.model.submodel.Submodel>` for
:class:`~aas.model.submodel.File` objects to extract the supplementary
files. The referenced supplementary files are added to the given `file_store` and the
:class:`~aas.model.submodel.File` objects' values are updated with the absolute name of the supplementary file
to allow for robust resolution the file within the
`file_store` later.
:param object_store: An :class:`ObjectStore <aas.model.provider.AbstractObjectStore>` to add the AAS 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
AASX that have the same :class:`~aas.model.base.Identifier`. Default behavior is to skip those objects from
the AASX.
:return: A set of the :class:`Identifiers <aas.model.base.Identifier>` of all
:class:`~aas.model.base.Identifiable` objects parsed from the AASX file
"""
# Find AASX-Origin part
core_rels = self.reader.get_related_parts_by_type()
try:
aasx_origin_part = core_rels[RELATIONSHIP_TYPE_AASX_ORIGIN][0]
except IndexError as e:
raise ValueError("Not a valid AASX file: aasx-origin Relationship is missing.") from e
read_identifiables: Set[model.Identifier] = set()
# Iterate AAS files
for aas_part in self.reader.get_related_parts_by_type(aasx_origin_part)[
RELATIONSHIP_TYPE_AAS_SPEC]:
self._read_aas_part_into(aas_part, object_store, file_store, read_identifiables, override_existing)
# Iterate split parts of AAS file
for split_part in self.reader.get_related_parts_by_type(aas_part)[
RELATIONSHIP_TYPE_AAS_SPEC_SPLIT]:
self._read_aas_part_into(split_part, object_store, file_store, read_identifiables, override_existing)
return read_identifiables
def close(self) -> None:
"""
Close the AASXReader and the underlying OPC / ZIP file readers. Must be called after reading the file.
"""
self.reader.close()
def __enter__(self) -> "AASXReader":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.close()
def _read_aas_part_into(self, part_name: str,
object_store: model.AbstractObjectStore,
file_store: "AbstractSupplementaryFileContainer",
read_identifiables: Set[model.Identifier],
override_existing: bool) -> None:
"""
Helper function for `read_into()` to read and process the contents of an AAS-spec part of the AASX file.
This method primarily checks for duplicate objects. It uses `_parse_aas_parse()` to do the actual parsing and
`_collect_supplementary_files()` for supplementary file processing of non-duplicate objects.
:param part_name: The OPC part name to read
:param object_store: An ObjectStore to add the AAS objects from the AASX file to
:param file_store: A SupplementaryFileContainer to add the embedded supplementary files to, which are reference
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
AASX that have the same Identifer. Default behavior is to skip those objects from the AASX.
"""
for obj in self._parse_aas_part(part_name):
if obj.identification in read_identifiables:
continue
if obj.identification in object_store:
if override_existing:
logger.info("Overriding existing object in ObjectStore with {} ...".format(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
object_store.add(obj)
read_identifiables.add(obj.identification)
if isinstance(obj, model.Submodel):
self._collect_supplementary_files(part_name, obj, file_store)
def _parse_aas_part(self, part_name: str) -> model.DictObjectStore:
"""
Helper function to parse the AAS objects from a single JSON or XML part of the AASX package.
This method chooses and calls the correct parser.
:param part_name: The OPC part name of the part to be parsed
:return: A DictObjectStore containing the parsed AAS objects
"""
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))
with self.reader.open_part(part_name) as p:
return read_aas_xml_file(p)
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))
with self.reader.open_part(part_name) as p:
return read_aas_json_file(io.TextIOWrapper(p, encoding='utf-8-sig'))
else:
logger.error("Could not determine part format of AASX part {} (Content Type: {}, extension: {}"
.format(part_name, content_type, extension))
return model.DictObjectStore()
def _collect_supplementary_files(self, part_name: str, submodel: model.Submodel,
file_store: "AbstractSupplementaryFileContainer") -> None:
"""
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
relative file paths.
:param submodel: The Submodel to process
:param file_store: The SupplementaryFileContainer to add the extracted supplementary files to
"""
for element in traversal.walk_submodel(submodel):
if isinstance(element, model.File):
if element.value is None:
continue
# Only absolute-path references and relative-path URI references (see RFC 3986, sec. 4.2) are considered
# 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)
continue
absolute_name = pyecma376_2.package_model.part_realpath(element.value, part_name)
logger.debug("Reading supplementary file {} from AASX package ...".format(absolute_name))
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
class AASXWriter:
"""
An AASXWriter wraps a new AASX package file to write its contents to it piece by piece.
Basic usage:
.. code-block:: python
# object_store and file_store are expected to be given (e.g. some storage backend or previously created data)
cp = OPCCoreProperties()
cp.creator = "ACPLT"
cp.created = datetime.datetime.now()
with AASXWriter("filename.aasx") as writer:
writer.write_aas(Identifier("https://acplt.org/AssetAdministrationShell", IdentifierType.IRI),
object_store,
file_store)
writer.write_aas(Identifier("https://acplt.org/AssetAdministrationShell2", IdentifierType.IRI),
object_store,
file_store)
writer.write_core_properties(cp)
**Attention:** The AASXWriter must always be closed using the :meth:`~.AASXWriter.close` method or its context
manager functionality (as shown above). Otherwise the resulting AASX file will lack important data structures
and will not be readable.
"""
AASX_ORIGIN_PART_NAME = "/aasx/aasx-origin"
def __init__(self, file: Union[os.PathLike, str, IO]):
"""
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 file: filename, path, or binary file handle opened for writing
"""
# names of aas-spec parts, used by `_write_aasx_origin_relationships()`
self._aas_part_names: List[str] = []
# name of the thumbnail part (if any)
self._thumbnail_part: Optional[str] = None
# name of the core properties part (if any)
self._properties_part: Optional[str] = None
# names and hashes of all supplementary file parts that have already been written
self._supplementary_part_names: Dict[str, Optional[bytes]] = {}
# Open OPC package writer
self.writer = pyecma376_2.ZipPackageWriter(file)
# Create AASX origin part
logger.debug("Creating AASX origin part in AASX package ...")
p = self.writer.open_part(self.AASX_ORIGIN_PART_NAME, "text/plain")
p.close()
def write_aas(self,
aas_ids: Union[model.Identifier, Iterable[model.Identifier]],
object_store: model.AbstractObjectStore,
file_store: "AbstractSupplementaryFileContainer",
write_json: bool = False) -> None:
"""
Convenience method to write one or more
:class:`AssetAdministrationShells <aas.model.aas.AssetAdministrationShell>` with all included and referenced
objects to the AASX package according to the part name conventions from DotAAS.
This method takes the AASs' :class:`Identifiers <aas.model.base.Identifier>` (as `aas_ids`) to retrieve the
AASs from the given object_store.
:class:`References <aas.model.base.Reference>` to :class:`Submodels <aas.model.submodel.Submodel>` and
:class:`ConceptDescriptions <aas.model.concept.ConceptDescription>` (via semanticId attributes) are also
resolved using the
`object_store`. All of these objects are written to an aas-spec part `/aasx/data.xml` or `/aasx/data.json` in
the AASX package, compliant to the convention presented in "Details of the Asset Administration Shell".
Supplementary files which are referenced by a :class:`~aas.model.submodel.File` object in any of the
:class:`Submodels <aas.model.submodel.Submodel>` are also added to the AASX
package.
This method uses `write_all_aas_objects()` to write the AASX part.
.. attention::
This method **must only be used once** on a single AASX package. Otherwise, the `/aasx/data.json`
(or `...xml`) part would be written twice to the package, hiding the first part and possibly causing
problems when reading the package.
To write multiple Asset Administration Shells to a single AASX package file, call this method once, passing
a list of AAS Identifiers to the `aas_ids` parameter.
:param aas_ids: :class:`~aas.model.base.Identifier` or Iterable of
:class:`Identifiers <aas.model.base.Identifier>` of the AAS(s) to be written to the AASX file
:param object_store: :class:`ObjectStore <aas.model.provider.AbstractObjectStore>` to retrieve the
:class:`~aas.model.base.Identifiable` AAS objects (:class:`~aas.model.aas.AssetAdministrationShell`,
:class:`~aas.model.aas.Asset`, :class:`~aas.model.concept.ConceptDescription` and
:class:`~aas.model.submodel.Submodel`) from
:param file_store: :class:`SupplementaryFileContainer <~.AbstractSupplementaryFileContainer>` to retrieve
supplementary files from, which are referenced by :class:`~aas.model.submodel.File` objects
:param write_json: If `True`, JSON parts are created for the AAS and each :class:`~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
:class:`Submodels <aas.model.submodel.Submodel>` and
:class:`ConceptDescriptions <aas.model.concept.ConceptDescription>` are skipped, logging a warning/info
message)
:raises TypeError: If one of the given AAS ids does not resolve to an AAS (but another
:class:`~aas.model.base.Identifiable` object)
"""
if isinstance(aas_ids, model.Identifier):
aas_ids = (aas_ids,)
objects_to_be_written: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
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 AssetAdminstrationShell object but to "
f"{aas!r}")
# Add the Asset object to the data part
objects_to_be_written.add(aas)
# Add referenced Submodels to the data part
for submodel_ref in aas.submodel:
try:
submodel = submodel_ref.resolve(object_store)
except KeyError:
logger.warning("Could not find submodel %s. Skipping it.", str(submodel_ref))
continue
objects_to_be_written.add(submodel)
# Traverse object tree and check if semanticIds are referencing to existing ConceptDescriptions in the
# ObjectStore
concept_descriptions: List[model.ConceptDescription] = []
for identifiable in objects_to_be_written:
for semantic_id in traversal.walk_semantic_ids_recursive(identifiable):
if not isinstance(semantic_id, model.AASReference) or semantic_id.type is not model.ConceptDescription:
logger.info("semanticId %s does not reference a ConceptDescription.", str(semantic_id))
continue
try:
cd = semantic_id.resolve(object_store)
except KeyError:
logger.info("ConceptDescription for semantidId %s not found in object store.", str(semantic_id))
continue
except model.UnexpectedTypeError as e:
logger.error("semantidId %s resolves to %s, which is not a ConceptDescription",
str(semantic_id), e.value)
continue
concept_descriptions.append(cd)
objects_to_be_written.update(concept_descriptions)
# Write AAS data part
self.write_all_aas_objects("/aasx/data.{}".format("json" if write_json else "xml"),
objects_to_be_written, file_store, write_json)
# TODO remove `method` parameter in future version.
# Not actually required since you can always create a local dict
def write_aas_objects(self,
part_name: str,
object_ids: Iterable[model.Identifier],
object_store: model.AbstractObjectStore,
file_store: "AbstractSupplementaryFileContainer",
write_json: bool = False,
split_part: bool = False,
additional_relationships: Iterable[pyecma376_2.OPCRelationship] = ()) -> None:
"""
A thin wrapper around :meth:`write_all_aas_objects` to ensure downwards compatibility
This method takes a list of :class:`~aas.model.base.Identifiers` (as `object_ids`) to retrieve the
identified objects from the given object_store. It then uses :meth:`write_all_aas_objects` to write the
objects and any referenced supplementary :class:`Files <aas.model.submodel.File>` from the `file_store` to the
AASX package.
.. attention::
You must make sure to call this method or `write_all_aas_objects` only once per unique `part_name` on a
single package instance.
:param part_name: Passed to :meth:`write_all_aas_objects`
:param object_ids: A list of identifiers of the objects to be written to the AASX package. Only these
Identifiable objects (and included Referable objects) are written to the package.
:param object_store: The objects store to retrieve the Identifiable objects from
:param file_store: Passed to :meth:`write_all_aas_objects`
:param write_json: Passed to :meth:`write_all_aas_objects`
:param split_part: Passed to :meth:`write_all_aas_objects`
:param additional_relationships: Passed to :meth:`write_all_aas_objects`
"""
logger.debug("Writing AASX part {} with AAS objects ...".format(part_name))
objects: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
# Retrieve objects and scan for referenced supplementary files
for identifier in object_ids:
try:
the_object = object_store.get_identifiable(identifier)
except KeyError:
logger.error("Could not find object {} in ObjectStore".format(identifier))
continue
objects.add(the_object)
self.write_all_aas_objects(part_name, objects, file_store, write_json, split_part, additional_relationships)
# TODO remove `split_part` parameter in future version.
# Not required anymore since changes from DotAAS version 2.0.1 to 3.0RC01
def write_all_aas_objects(self,
part_name: str,
objects: model.AbstractObjectStore[model.Identifiable],
file_store: "AbstractSupplementaryFileContainer",
write_json: bool = False,
split_part: bool = False,
additional_relationships: Iterable[pyecma376_2.OPCRelationship] = ()) -> None:
"""
Write all AAS objects in a given :class:`ObjectStore <aas.model.provider.AbstractObjectStore>` to an XML or
JSON part in the AASX package and add the referenced supplementary files to the package.
This method takes an :class:`ObjectStore <aas.model.provider.AbstractObjectStore>` and writes all contained
objects into an "aas_env" part in the AASX package. If
the ObjectStore includes :class:`~aas.model.submodel.Submodel` objects, supplementary files which are
referenced by :class:`~aas.model.submodel.File` objects
within those Submodels, are fetched from the `file_store` and added to the AASX package.
.. attention::
You must make sure to call this method only once per unique `part_name` on a single package instance.
:param part_name: Name of the Part within the AASX package to write the files to. Must be a valid ECMA376-2
part name and unique within the package. The extension of the part should match the data format (i.e.
'.json' if `write_json` else '.xml').
:param objects: The objects to be written to the AASX package. Only these Identifiable objects (and included
Referable objects) are written to the package.
:param file_store: The SupplementaryFileContainer to retrieve supplementary files from (if there are any `File`
objects within the written objects.
:param write_json: If True, the part is written as a JSON file instead of an XML file. Defaults to False.
:param split_part: If True, no aas-spec relationship is added from the aasx-origin to this part. You must make
sure to reference it via a aas-spec-split relationship from another aas-spec part
: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))
supplementary_files: List[str] = []
# Retrieve objects and scan for referenced supplementary files
for the_object in objects:
if isinstance(the_object, model.Submodel):
for element in traversal.walk_submodel(the_object):
if isinstance(element, model.File):
file_name = element.value
# Skip File objects with empty value URI references that are considered to be no local file
# (absolute URIs or network-path URI references)
if file_name is None or file_name.startswith('//') or ':' in file_name.split('/')[0]:
continue
supplementary_files.append(file_name)
# Add aas-spec relationship
if not split_part:
self._aas_part_names.append(part_name)
# Write part
# TODO allow writing xml *and* JSON part
with self.writer.open_part(part_name, "application/json" if write_json else "application/xml") as p:
if write_json:
write_aas_json_file(io.TextIOWrapper(p, encoding='utf-8'), objects)
else:
write_aas_xml_file(p, objects)
# 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
# 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))
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))
self.writer.write_relationships(
itertools.chain(
(pyecma376_2.OPCRelationship("r{}".format(i),
RELATIONSHIP_TYPE_AAS_SUPL,
submodel_file_name,
pyecma376_2.OPCTargetMode.INTERNAL)
for i, submodel_file_name in enumerate(supplementary_file_names)),
additional_relationships),
part_name)
def write_core_properties(self, core_properties: pyecma376_2.OPCCoreProperties):
"""
Write OPC Core Properties (meta data) to the AASX package file.
Attention: This method may only be called once for each AASXWriter!
:param core_properties: The OPCCoreProperties object with the meta data to be written to the package file
"""
if self._properties_part is not None:
raise RuntimeError("Core Properties have already been written.")
logger.debug("Writing core properties to AASX package ...")
with self.writer.open_part(pyecma376_2.DEFAULT_CORE_PROPERTIES_NAME, "application/xml") as p:
core_properties.write_xml(p)
self._properties_part = pyecma376_2.DEFAULT_CORE_PROPERTIES_NAME
def write_thumbnail(self, name: str, data: bytearray, content_type: str):
"""
Write an image file as thumbnail image to the AASX package.
Attention: This method may only be called once for each AASXWriter!
:param name: The OPC part name of the thumbnail part. Should not contain '/' or URI-encoded '/' or '\'.
:param data: The image file's binary contents to be written
: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))
with self.writer.open_part(name, content_type) as p:
p.write(data)
self._thumbnail_part = name
def close(self):
"""
Write relationships for all data files to package and close underlying OPC package and ZIP file.
"""
self._write_aasx_origin_relationships()
self._write_package_relationships()
self.writer.close()
def __enter__(self) -> "AASXWriter":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.close()
def _write_aasx_origin_relationships(self):
"""
Helper function to write aas-spec relationships of the aasx-origin part.
This method uses the list of aas-spec parts in `_aas_part_names`. It should be called just before closing the
file to make sure all aas-spec parts of the package have already been written.
"""
# Add relationships from AASX-origin part to AAS parts
logger.debug("Writing aas-spec relationships to AASX package ...")
self.writer.write_relationships(
(pyecma376_2.OPCRelationship("r{}".format(i), RELATIONSHIP_TYPE_AAS_SPEC,
aas_part_name,
pyecma376_2.OPCTargetMode.INTERNAL)
for i, aas_part_name in enumerate(self._aas_part_names)),
self.AASX_ORIGIN_PART_NAME)
def _write_package_relationships(self):
"""
Helper function to write package (root) relationships to the OPC package.
This method must be called just before closing the package file to make sure we write exactly the correct
relationships:
* aasx-origin (always)
* core-properties (if core properties have been added)
* thumbnail (if thumbnail part has been added)
"""
logger.debug("Writing package relationships to AASX package ...")
package_relationships: List[pyecma376_2.OPCRelationship] = [
pyecma376_2.OPCRelationship("r1", RELATIONSHIP_TYPE_AASX_ORIGIN,
self.AASX_ORIGIN_PART_NAME,
pyecma376_2.OPCTargetMode.INTERNAL),
]
if self._properties_part is not None:
package_relationships.append(pyecma376_2.OPCRelationship(
"r2", pyecma376_2.RELATIONSHIP_TYPE_CORE_PROPERTIES, self._properties_part,
pyecma376_2.OPCTargetMode.INTERNAL))
if self._thumbnail_part is not None:
package_relationships.append(pyecma376_2.OPCRelationship(
"r3", pyecma376_2.RELATIONSHIP_TYPE_THUMBNAIL, self._thumbnail_part,
pyecma376_2.OPCTargetMode.INTERNAL))
self.writer.write_relationships(package_relationships)
# TODO remove in future version.
# Not required anymore since changes from DotAAS version 2.0.1 to 3.0RC01
class NameFriendlyfier:
"""
A simple helper class to create unique "AAS friendly names" according to DotAAS, section 7.6.
Objects of this class store the already created friendly names to avoid name collisions within one set of names.
"""
RE_NON_ALPHANUMERICAL = re.compile(r"[^a-zA-Z0-9]")
def __init__(self) -> None:
self.issued_names: Set[str] = set()
def get_friendly_name(self, identifier: model.Identifier):
"""
Generate a friendly name from an AAS identifier.
TODO: This information is outdated. The whole class is no longer needed.
According to section 7.6 of "Details of the Asset Administration Shell", all non-alphanumerical characters are
replaced with underscores. We also replace all non-ASCII characters to generate valid URIs as the result.
If this replacement results in a collision with a previously generated friendly name of this NameFriendlifier,
a number is appended with underscore to the friendly name.
Example:
.. code-block:: python
friendlyfier = NameFriendlyfier()
friendlyfier.get_friendly_name(model.Identifier("http://example.com/AAS-a", model.IdentifierType.IRI))
> "http___example_com_AAS_a"
friendlyfier.get_friendly_name(model.Identifier("http://example.com/AAS+a", model.IdentifierType.IRI))
> "http___example_com_AAS_a_1"
"""
# friendlify name
raw_name = self.RE_NON_ALPHANUMERICAL.sub('_', identifier.id)
# Unify name (avoid collisions)
amended_name = raw_name
i = 1
while amended_name in self.issued_names:
amended_name = "{}_{}".format(raw_name, i)
i += 1
self.issued_names.add(amended_name)
return amended_name
class AbstractSupplementaryFileContainer(metaclass=abc.ABCMeta):
"""
Abstract interface for containers of supplementary files for AASs.
Supplementary files may be PDF files or other binary or textual files, referenced in a File object of an AAS by
their name. They are used to provide associated documents without embedding their contents (as
:class:`~aas.model.submodel.Blob` object) in the AAS.
A SupplementaryFileContainer keeps track of the name and content_type (MIME type) for each file. Additionally it
allows to resolve name conflicts by comparing the files' contents and providing an alternative name for a dissimilar
new file. It also provides each files sha256 hash sum to allow name conflict checking in other classes (e.g. when
writing AASX files).
"""
@abc.abstractmethod
def add_file(self, name: str, file: IO[bytes], content_type: str) -> str:
"""
Add a new file to the SupplementaryFileContainer and resolve name conflicts.
The file contents must be provided as a binary file-like object to be read by the SupplementaryFileContainer.
If the container already contains an equally named file, the content_type and file contents are compared (using
a hash sum). In case of dissimilar files, a new unique name for the new file is computed and returned. It should
be used to update in the File object of the AAS.
:param name: The file's proposed name. Should start with a '/'. Should not contain URI-encoded '/' or '\'
:param file: A binary file-like opened for reading the file contents
:param content_type: The file's content_type
:return: The file name as stored in the SupplementaryFileContainer. Typically `name` or a modified version of
`name` to resolve conflicts.
"""
pass # pragma: no cover
@abc.abstractmethod
def get_content_type(self, name: str) -> str:
"""
Get a stored file's content_type.
:param name: file name of questioned file
:return: The file's content_type
:raises KeyError: If no file with this name is stored
"""
pass # pragma: no cover
@abc.abstractmethod
def get_sha256(self, name: str) -> bytes:
"""
Get a stored file content's sha256 hash sum.
This may be used by other classes (e.g. the AASXWriter) to check for name conflicts.
:param name: file name of questioned file
:return: The file content's sha256 hash sum
:raises KeyError: If no file with this name is stored
"""
pass # pragma: no cover
@abc.abstractmethod
def write_file(self, name: str, file: IO[bytes]) -> None:
"""
Retrieve a stored file's contents by writing them into a binary writable file-like object.
:param name: file name of questioned file
:param file: A binary file-like object with write() method to write the file contents into
:raises KeyError: If no file with this name is stored
"""
pass # pragma: no cover
@abc.abstractmethod
def __contains__(self, item: str) -> bool:
"""
Check if a file with the given name is stored in this SupplementaryFileContainer.
"""
pass # pragma: no cover
@abc.abstractmethod
def __iter__(self) -> Iterator[str]:
"""
Return an iterator over all file names stored in this SupplementaryFileContainer.
"""
pass # pragma: no cover
class DictSupplementaryFileContainer(AbstractSupplementaryFileContainer):
"""
SupplementaryFileContainer implementation using a dict to store the file contents in-memory.
"""
def __init__(self):
# Stores the files' contents, identified by their sha256 hash
self._store: Dict[bytes, bytes] = {}
# Maps file names to (sha256, content_type)
self._name_map: Dict[str, Tuple[bytes, str]] = {}
def add_file(self, name: str, file: IO[bytes], content_type: str) -> str:
data = file.read()
hash = hashlib.sha256(data).digest()
if hash not in self._store:
self._store[hash] = data
name_map_data = (hash, content_type)
new_name = name
i = 1
while True:
if new_name not in self._name_map:
self._name_map[new_name] = name_map_data
return new_name
elif self._name_map[new_name] == name_map_data:
return new_name
new_name = self._append_counter(name, i)
i += 1
@staticmethod
def _append_counter(name: str, i: int) -> str:
split1 = name.split('/')
split2 = split1[-1].split('.')
index = -2 if len(split2) > 1 else -1
new_basename = "{}_{:04d}".format(split2[index], i)
split2[index] = new_basename
split1[-1] = ".".join(split2)
return "/".join(split1)
def get_content_type(self, name: str) -> str:
return self._name_map[name][1]
def get_sha256(self, name: str) -> bytes:
return self._name_map[name][0]
def write_file(self, name: str, file: IO[bytes]) -> None:
file.write(self._store[self._name_map[name][0]])
def __contains__(self, item: object) -> bool:
return item in self._name_map
def __iter__(self) -> Iterator[str]:
return iter(self._name_map)