Commit 3781cf04 authored by Michael Thies's avatar Michael Thies
Browse files

Merge branch 'feature/aasx_write_individual_objects' into 'master'

aasx: Allow writing individual objects

Closes #94

See merge request !59
parents 5de92d87 9fc8ed5a
Pipeline #349761 passed with stages
in 59 seconds
......@@ -28,7 +28,7 @@ import io
import logging
import os
import re
from typing import Dict, Tuple, IO, Union, List, Set, Optional
from typing import Dict, Tuple, IO, Union, List, Set, Optional, Iterable
from .xml import read_aas_xml_file, write_aas_xml_file
from .. import model
......@@ -299,14 +299,15 @@ class AASXWriter:
p = self.writer.open_part(self.AASX_ORIGIN_PART_NAME, "text/plain")
p.close()
# TODO allow to specify, which supplementary parts (submodels, conceptDescriptions) should be added to the package
def write_aas(self,
aas_id: model.Identifier,
object_store: model.AbstractObjectStore,
file_store: "AbstractSupplementaryFileContainer",
write_json: bool = False) -> None:
write_json: bool = False,
submodel_split_parts: bool = True) -> None:
"""
Add an Asset Administration Shell with all included and referenced objects to the AASX package.
Convenience method to add an Asset Administration Shell with all included and referenced objects to the AASX
package according to the part name conventions from DotAAS.
This method takes the AAS's Identifier (as `aas_id`) to retrieve it from the given object_store. References to
the Asset, ConceptDescriptions and Submodels are also resolved using the object_store. All of these objects are
......@@ -314,6 +315,9 @@ class AASXWriter:
Administration Shell". For each Submodel, a aas-spec-split part is used. Supplementary files which are
referenced by a File object in any of the Submodels, are also added to the AASX package.
Internally, this method uses `write_aas_objects()` to write the individual AASX parts for the AAS and each
submodel.
:param aas_id: Identifier of the AAS to be added to the AASX file
:param object_store: ObjectStore to retrieve the Identifiable AAS objects (AAS, Asset, ConceptDescriptions and
Submodels) from
......@@ -321,134 +325,151 @@ class AASXWriter:
objects
:param write_json: If True, JSON parts are created for the AAS and each submodel in the AASX package file
instead of XML parts. Defaults to False.
:param submodel_split_parts: If True (default), submodels are written to separate AASX parts instead of being
included in the AAS part with in the AASX package.
"""
aas_friendly_name = self._aas_name_friendlyfier.get_friendly_name(aas_id)
aas_part_name = "/aasx/{0}/{0}.aas.{1}".format(aas_friendly_name, "json" if write_json else "xml")
self._aas_part_names.append(aas_part_name)
aas_friendlyfier = NameFriendlyfier()
aas: model.AssetAdministrationShell = object_store.get_identifiable(aas_id) # type: ignore
objects_to_be_written: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
objects_to_be_written.add(aas)
aas = object_store.get_identifiable(aas_id)
if not isinstance(aas, model.AssetAdministrationShell):
raise ValueError(f"Identifier does not belong to an AssetAdminstrationShell object but to {aas!r}")
objects_to_be_written: Set[model.Identifier] = {aas.identification}
# Add the Asset object to the objects in the AAS part
try:
objects_to_be_written.add(aas.asset.resolve(object_store))
except KeyError as e:
logger.warning("Skipping Asset object, since {} could not be resolved: {}".format(aas.asset, e))
pass
objects_to_be_written.add(aas.asset.get_identifier())
# Add referenced ConceptDescriptions to the AAS part
for dictionary in aas.concept_dictionary:
for concept_rescription_ref in dictionary.concept_description:
try:
obj = concept_rescription_ref.resolve(object_store)
except KeyError as e:
logger.warning("Skipping ConceptDescription, since {} could not be resolved: {}"
.format(concept_rescription_ref, e))
continue
try:
objects_to_be_written.add(obj)
except KeyError:
# Ignore duplicate ConceptDescriptions (i.e. same Description referenced from multiple
# Dictionaries)
pass
objects_to_be_written.add(concept_rescription_ref.get_identifier())
if not submodel_split_parts:
for submodel_ref in aas.submodel:
objects_to_be_written.add(submodel_ref.get_identifier())
# Write AAS part
logger.debug("Writing AAS {} to part {} in AASX package ...".format(aas.identification, aas_part_name))
with self.writer.open_part(aas_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_to_be_written)
else:
write_aas_xml_file(p, objects_to_be_written)
# Create a AAS split part for each (available) submodel of the AAS
aas_split_part_names: List[str] = []
for submodel_ref in aas.submodel:
self.write_aas_objects(aas_part_name, objects_to_be_written, object_store, file_store, write_json)
if submodel_split_parts:
# Create a AAS split part for each (available) submodel of the AAS
aas_split_part_names: List[str] = []
aas_friendlyfier = NameFriendlyfier()
for submodel_ref in aas.submodel:
submodel_identification = submodel_ref.get_identifier()
submodel_friendly_name = aas_friendlyfier.get_friendly_name(submodel_identification)
submodel_part_name = "/aasx/{0}/{1}/{1}.submodel.{2}".format(aas_friendly_name, submodel_friendly_name,
"json" if write_json else "xml")
self.write_aas_objects(submodel_part_name, [submodel_identification], object_store, file_store,
write_json)
aas_split_part_names.append(submodel_part_name)
# Add relationships from AAS part to (submodel) split parts
logger.debug("Writing aas-spec-split relationships for AAS {} to AASX package ..."
.format(aas.identification))
self.writer.write_relationships(
(pyecma376_2.OPCRelationship("r{}".format(i),
RELATIONSHIP_TYPE_AAS_SPEC_SPLIT,
submodel_part_name,
pyecma376_2.OPCTargetMode.INTERNAL)
for i, submodel_part_name in enumerate(aas_split_part_names)),
aas_part_name)
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) -> None:
"""
Write a defined list of AAS objects to an XML or JSON part in the AASX package and append the referenced
supplementary files to the package.
This method takes the AAS's Identifier (as `aas_id`) to retrieve it from the given object_store. If the list
of written objects includes Submodel objects, Supplementary files which are referenced by File objects within
those submodels, are also added to the AASX package.
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 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 Identifable objects from
: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
"""
logger.debug("Writing AASX part {} with AAS objects ...".format(part_name))
objects: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
supplementary_files: List[str] = []
# Retrieve objects and scan for referenced supplementary files
for identifier in object_ids:
try:
submodel = submodel_ref.resolve(object_store)
except KeyError as e:
logger.warning("Skipping Submodel, since {} could not be resolved: {}".format(submodel_ref, e))
the_object = object_store.get_identifiable(identifier)
except KeyError:
logger.error("Could not find object {} in ObjectStore".format(identifier))
continue
submodel_friendly_name = aas_friendlyfier.get_friendly_name(submodel.identification)
submodel_part_name = "/aasx/{0}/{1}/{1}.submodel.{2}".format(aas_friendly_name, submodel_friendly_name,
"json" if write_json else "xml")
self._write_submodel_part(file_store, submodel, submodel_part_name, write_json)
aas_split_part_names.append(submodel_part_name)
# Add relationships from AAS part to (submodel) split parts
logger.debug("Writing aas-spec-split relationships for AAS {} to AASX package ..."
.format(aas.identification))
self.writer.write_relationships(
(pyecma376_2.OPCRelationship("r{}".format(i),
RELATIONSHIP_TYPE_AAS_SPEC_SPLIT,
submodel_part_name,
pyecma376_2.OPCTargetMode.INTERNAL)
for i, submodel_part_name in enumerate(aas_split_part_names)),
aas_part_name)
def _write_submodel_part(self, file_store: "AbstractSupplementaryFileContainer",
submodel: model.Submodel, submodel_part_name: str, write_json: bool = False) -> None:
"""
Helper function for `write_aas()` to write an aas-spec-split part for a Submodel object and add the relevant
supplementary files.
:param file_store: The SupplementaryFileContainer to retrieve supplementary files from
:param submodel: The submodel to be written into the AASX package
:param submodel_part_name: OPC part name of the aas-spec-split part for this Submodel
:param write_json: If True, the submodel is written as a JSON file instead of an XML file to the given OPC
part. Defaults to False.
"""
logger.debug("Writing Submodel {} to part {} in AASX package ..."
.format(submodel.identification, submodel_part_name))
submodel_file_objects: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
submodel_file_objects.add(submodel)
with self.writer.open_part(submodel_part_name, "application/json" if write_json else "application/xml") as p:
objects.add(the_object)
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
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'), submodel_file_objects)
write_aas_json_file(io.TextIOWrapper(p, encoding='utf-8'), objects)
else:
write_aas_xml_file(p, submodel_file_objects)
write_aas_xml_file(p, objects)
# Write submodel's supplementary files to AASX file
submodel_file_names = []
for element in traversal.walk_submodel(submodel):
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
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, referenced from {}."
.format(file_name, element))
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)
submodel_file_names.append(pyecma376_2.package_model.normalize_part_name(file_name))
self._supplementary_part_names[file_name] = hash
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
# TODO should the relationships be added from the AAS instead?
logger.debug("Writing aas-suppl relationships for Submodel {} to AASX package ..."
.format(submodel.identification))
logger.debug("Writing aas-suppl relationships for AAS object part {} to AASX package ...".format(part_name))
self.writer.write_relationships(
(pyecma376_2.OPCRelationship("r{}".format(i),
RELATIONSHIP_TYPE_AAS_SUPL,
submodel_file_name,
pyecma376_2.OPCTargetMode.INTERNAL)
for i, submodel_file_name in enumerate(submodel_file_names)),
submodel_part_name)
for i, submodel_file_name in enumerate(supplementary_file_names)),
part_name)
def write_core_properties(self, core_properties: pyecma376_2.OPCCoreProperties):
"""
......
......@@ -728,6 +728,22 @@ class AASReference(Reference, Generic[_RT]):
.format(item, self.type.__name__))
return item
def get_identifier(self) -> Identifier:
"""
Retrieve the Identifier of the Identifiable object, which is referenced or in which the referenced Referable is
contained.
:raises ValueError: If this Reference does not include a Key with global KeyType (IRDI, IRI, CUSTOM)
"""
try:
last_identifier = next(key.get_identifier()
for key in reversed(self.key)
if key.get_identifier())
return last_identifier # type: ignore # MyPy doesn't get the generator expression above
except StopIteration:
raise ValueError("Reference cannot be represented as an Identifier, since it does not contain a Key with "
"global KeyType (IRDI, IRI, CUSTOM)")
def __repr__(self) -> str:
return "AASReference(type={}, key={})".format(self.type.__name__, self.key)
......
......@@ -18,7 +18,7 @@ import unittest
import pyecma376_2
from aas import model
from aas.adapter import aasx
from aas.examples.data import example_aas, _helper
from aas.examples.data import example_aas, _helper, example_aas_mandatory_attributes
class TestAASXUtils(unittest.TestCase):
......@@ -72,41 +72,69 @@ class AASXWriterTest(unittest.TestCase):
cp.created = datetime.datetime.now()
cp.creator = "PyI40AAS Testing Framework"
# Write AASX file
for write_json in (False, True):
for submodel_split_parts in (False, True):
with self.subTest(write_json=write_json, submodel_split_parts=submodel_split_parts):
fd, filename = tempfile.mkstemp(suffix=".aasx")
os.close(fd)
with aasx.AASXWriter(filename) as writer:
writer.write_aas(model.Identifier(id_='https://acplt.org/Test_AssetAdministrationShell',
id_type=model.IdentifierType.IRI),
data, files, write_json=write_json, submodel_split_parts=submodel_split_parts)
writer.write_core_properties(cp)
# Read AASX file
new_data: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
new_files = aasx.DictSupplementaryFileContainer()
with aasx.AASXReader(filename) as reader:
reader.read_into(new_data, new_files)
new_cp = reader.get_core_properties()
# Check AAS objects
checker = _helper.AASDataChecker(raise_immediately=True)
example_aas.check_full_example(checker, new_data)
# Check core properties
assert(isinstance(cp.created, datetime.datetime)) # to make mypy happy
self.assertIsInstance(new_cp.created, datetime.datetime)
assert(isinstance(new_cp.created, datetime.datetime)) # to make mypy happy
self.assertAlmostEqual(new_cp.created, cp.created, delta=datetime.timedelta(milliseconds=20))
self.assertEqual(new_cp.creator, "PyI40AAS Testing Framework")
self.assertIsNone(new_cp.lastModifiedBy)
# Check files
self.assertEqual(new_files.get_content_type("/TestFile.pdf"), "application/pdf")
file_content = io.BytesIO()
new_files.write_file("/TestFile.pdf", file_content)
self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(),
"78450a66f59d74c073bf6858db340090ea72a8b1")
os.unlink(filename)
def test_writing_reading_objects_single_part(self) -> None:
# Create example data and file_store
data = example_aas_mandatory_attributes.create_full_example()
files = aasx.DictSupplementaryFileContainer()
# Write AASX file
for write_json in (False, True):
with self.subTest(write_json=write_json):
fd, filename = tempfile.mkstemp(suffix=".aasx")
os.close(fd)
with aasx.AASXWriter(filename) as writer:
writer.write_aas(model.Identifier(id_='https://acplt.org/Test_AssetAdministrationShell',
id_type=model.IdentifierType.IRI),
data, files, write_json=write_json)
writer.write_core_properties(cp)
writer.write_aas_objects('/aasx/aasx.{}'.format('json' if write_json else 'xml'),
[obj.identification for obj in data],
data, files, write_json)
# Read AASX file
new_data: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
new_files = aasx.DictSupplementaryFileContainer()
with aasx.AASXReader(filename) as reader:
reader.read_into(new_data, new_files)
new_cp = reader.get_core_properties()
# Check AAS objects
checker = _helper.AASDataChecker(raise_immediately=True)
example_aas.check_full_example(checker, new_data)
# Check core properties
assert(isinstance(cp.created, datetime.datetime)) # to make mypy happy
self.assertIsInstance(new_cp.created, datetime.datetime)
assert(isinstance(new_cp.created, datetime.datetime)) # to make mypy happy
self.assertAlmostEqual(new_cp.created, cp.created, delta=datetime.timedelta(milliseconds=20))
self.assertEqual(new_cp.creator, "PyI40AAS Testing Framework")
self.assertIsNone(new_cp.lastModifiedBy)
# Check files
self.assertEqual(new_files.get_content_type("/TestFile.pdf"), "application/pdf")
file_content = io.BytesIO()
new_files.write_file("/TestFile.pdf", file_content)
self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(),
"78450a66f59d74c073bf6858db340090ea72a8b1")
example_aas_mandatory_attributes.check_full_example(checker, new_data)
os.unlink(filename)
......@@ -517,6 +517,28 @@ class AASReferenceTest(unittest.TestCase):
ref5.resolve(DummyObjectProvider())
self.assertEqual('List of keys is empty', str(cm_5.exception))
def test_get_identifier(self) -> None:
ref = model.AASReference((model.Key(model.KeyElements.SUBMODEL, False, "urn:x-test:x", model.KeyType.IRI),),
model.Submodel)
self.assertEqual(model.Identifier("urn:x-test:x", model.IdentifierType.IRI), ref.get_identifier())
ref2 = model.AASReference((model.Key(model.KeyElements.SUBMODEL, False, "urn:x-test:x", model.KeyType.IRI),
model.Key(model.KeyElements.PROPERTY, False, "myProperty", model.KeyType.IDSHORT),),
model.Submodel)
self.assertEqual(model.Identifier("urn:x-test:x", model.IdentifierType.IRI), ref.get_identifier())
# People will do strange things ...
ref3 = model.AASReference((model.Key(model.KeyElements.ASSET_ADMINISTRATION_SHELL, False, "urn:x-test-aas:x",
model.KeyType.IRI),
model.Key(model.KeyElements.SUBMODEL, False, "urn:x-test:x", model.KeyType.IRI),),
model.Submodel)
self.assertEqual(model.Identifier("urn:x-test:x", model.IdentifierType.IRI), ref2.get_identifier())
ref4 = model.AASReference((model.Key(model.KeyElements.PROPERTY, False, "myProperty", model.KeyType.IDSHORT),),
model.Property)
with self.assertRaises(ValueError):
ref4.get_identifier()
def test_from_referable(self) -> None:
prop = model.Property("prop", model.datatypes.Int)
collection = model.SubmodelElementCollectionUnordered("collection", {prop})
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment