Commit aecc2b3c authored by Torben Miny's avatar Torben Miny
Browse files

Merge branch 'feature/aasx' into 'master'

Add AASX reading and writing functionality

Closes #38

See merge request acplt/pyaas!16
parents 81545c7e 65981a4b
Pipeline #259822 passed with stage
in 3 minutes and 4 seconds
This diff is collapsed.
......@@ -637,7 +637,18 @@ def create_example_asset_administration_shell(concept_dictionary: model.ConceptD
local=False,
value='https://acplt.org/Test_Submodel',
id_type=model.KeyType.IRI),),
model.Submodel)},
model.Submodel),
model.AASReference((model.Key(type_=model.KeyElements.SUBMODEL,
local=False,
value='http://acplt.org/Submodels/Assets/TestAsset/Identification',
id_type=model.KeyType.IRI),),
model.Submodel),
model.AASReference((model.Key(type_=model.KeyElements.SUBMODEL,
local=False,
value='http://acplt.org/Submodels/Assets/TestAsset/BillOfMaterial',
id_type=model.KeyType.IRI),),
model.Submodel),
},
concept_dictionary=[concept_dictionary],
view=[],
derived_from=model.AASReference((model.Key(type_=model.KeyElements.ASSET_ADMINISTRATION_SHELL,
......
......@@ -76,8 +76,10 @@ class DictObjectStore(AbstractObjectStore[_IT], Generic[_IT]):
"""
A local in-memory object store for Identifiable Objects, backed by a dict, mapping Identifier → Identifiable
"""
def __init__(self):
def __init__(self, objects: Iterable[_IT] = ()) -> None:
self._backend: Dict[Identifier, _IT] = {}
for x in objects:
self.add(x)
def get_identifiable(self, identifier: Identifier) -> _IT:
return self._backend[identifier]
......
# Copyright 2019 PyI40AAS Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
"""
A module with helper functions for traversing AAS object strcutures.
"""
from typing import Union, Iterator
from .. import model
def walk_submodel(collection: Union[model.Submodel, model.SubmodelElementCollection]) \
-> Iterator[model.SubmodelElement]:
"""
Traverse the SubmodelElements in a Submodel or a SubmodelElementCollection recursively in post-order tree-traversal.
This is a generator function, yielding all the SubmodelElements. No SubmodelElements should be added, removed or
moved while iterating, as this could result in undefined behaviour.
"""
elements = collection.submodel_element if isinstance(collection, model.Submodel) else collection.value
for element in elements:
if isinstance(element, model.SubmodelElementCollection):
yield from walk_submodel(element)
yield element
jsonschema>=3.2,<4.0
lxml>=4.2,<5
python-dateutil>=2.8,<3.0
\ No newline at end of file
python-dateutil>=2.8,<3.0
pyecma376-2>=0.2
......@@ -42,5 +42,6 @@ setuptools.setup(
install_requires=[
'python-dateutil>=2.8,<3',
'lxml>=4.2,<5',
'pyecma376-2>=0.2',
]
)
# Copyright 2019 PyI40AAS Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
import datetime
import hashlib
import io
import os
import tempfile
import unittest
import pyecma376_2
from aas import model
from aas.adapter import aasx
from aas.examples.data import example_aas, _helper
class TestAASXUtils(unittest.TestCase):
def test_name_friendlyfier(self) -> None:
friendlyfier = aasx.NameFriendlyfier()
name1 = friendlyfier.get_friendly_name(model.Identifier("http://example.com/AAS-a", model.IdentifierType.IRI))
self.assertEqual("http___example_com_AAS_a", name1)
name2 = friendlyfier.get_friendly_name(model.Identifier("http://example.com/AAS+a", model.IdentifierType.IRI))
self.assertEqual("http___example_com_AAS_a_1", name2)
def test_supplementary_file_container(self) -> None:
container = aasx.DictSupplementaryFileContainer()
with open(os.path.join(os.path.dirname(__file__), 'TestFile.pdf'), 'rb') as f:
new_name = container.add_file("/TestFile.pdf", f, "application/pdf")
# Name should not be modified, since there is no conflict
self.assertEqual("/TestFile.pdf", new_name)
f.seek(0)
container.add_file("/TestFile.pdf", f, "application/pdf")
# Name should not be modified, since there is still no conflict
self.assertEqual("/TestFile.pdf", new_name)
with open(__file__, 'rb') as f:
new_name = container.add_file("/TestFile.pdf", f, "application/pdf")
# Now, we have a conflict
self.assertNotEqual("/TestFile.pdf", new_name)
self.assertIn(new_name, container)
# Check metadata
self.assertEqual("application/pdf", container.get_content_type("/TestFile.pdf"))
self.assertEqual("b18229b24a4ee92c6c2b6bc6a8018563b17472f1150d35d5a5945afeb447ed44",
container.get_sha256("/TestFile.pdf").hex())
self.assertIn("/TestFile.pdf", container)
# Check contents
file_content = io.BytesIO()
container.write_file("/TestFile.pdf", file_content)
self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(), "78450a66f59d74c073bf6858db340090ea72a8b1")
class AASXWriterTest(unittest.TestCase):
def test_writing_reading_example_aas(self) -> None:
# Create example data and file_store
data = example_aas.create_full_example()
files = aasx.DictSupplementaryFileContainer()
with open(os.path.join(os.path.dirname(__file__), 'TestFile.pdf'), 'rb') as f:
files.add_file("/TestFile.pdf", f, "application/pdf")
f.seek(0)
# Create OPC/AASX core properties
cp = pyecma376_2.OPCCoreProperties()
cp.created = datetime.datetime.now()
cp.creator = "PyI40AAS Testing Framework"
# Write AASX file
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)
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
self.assertEqual(new_cp.created, cp.created)
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)
......@@ -15,7 +15,8 @@ import aas.compliance_tool.compliance_check_json as compliance_tool
from aas.compliance_tool.state_manager import ComplianceToolStateManager, Status
dirname = os.path.dirname
JSON_SCHEMA_FILE = os.path.join(dirname(dirname(dirname(__file__))), 'test\\adapter\\json\\aasJSONSchemaV2.0.json')
JSON_SCHEMA_FILE = os.path.join(dirname(dirname(dirname(__file__))), 'test', 'adapter', 'json',
'aasJSONSchemaV2.0.json')
class ComplianceToolJsonTest(unittest.TestCase):
......
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