diff --git a/sdk/basyx/aas/adapter/aasx.py b/sdk/basyx/aas/adapter/aasx.py index 8bb5958f6..19ff35c8e 100644 --- a/sdk/basyx/aas/adapter/aasx.py +++ b/sdk/basyx/aas/adapter/aasx.py @@ -231,6 +231,8 @@ def _read_aas_part_into(self, part_name: str, read_identifiables.add(obj.id) if isinstance(obj, model.Submodel): self._collect_supplementary_files(part_name, obj, file_store) + elif isinstance(obj, model.AssetAdministrationShell): + self._collect_supplementary_files(part_name, obj, file_store) def _parse_aas_part(self, part_name: str, **kwargs) -> model.DictObjectStore: """ @@ -261,33 +263,59 @@ def _parse_aas_part(self, part_name: str, **kwargs) -> model.DictObjectStore: raise ValueError(error_message) return model.DictObjectStore() - def _collect_supplementary_files(self, part_name: str, submodel: model.Submodel, + def _collect_supplementary_files(self, part_name: str, + root_element: Union[model.AssetAdministrationShell, 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. + Helper function to search File objects within a single parsed AssetAdministrationShell or Submodel. + Resolve their absolute paths, and update the corresponding File/Thumbnail objects 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 root_element has been parsed from. This is used to resolve relative file paths. - :param submodel: The Submodel to process + :param root_element: The AssetAdministrationShell or 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(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(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 + if isinstance(root_element, model.AssetAdministrationShell): + if (root_element.asset_information.default_thumbnail and + root_element.asset_information.default_thumbnail.path): + file_name = self._add_supplementary_file(part_name, + root_element.asset_information.default_thumbnail.path, + file_store) + if file_name: + root_element.asset_information.default_thumbnail.path = file_name + if isinstance(root_element, model.Submodel): + for element in traversal.walk_submodel(root_element): + if isinstance(element, model.File): + if element.value is None: + continue + final_name = self._add_supplementary_file(part_name, element.value, file_store) + if final_name: + element.value = final_name + + def _add_supplementary_file(self, part_name: str, file_path: str, + file_store: "AbstractSupplementaryFileContainer") -> Optional[str]: + """ + Helper function to extract a single referenced supplementary file + and return the absolute path within the AASX package. + + :param part_name: The OPC part name of the part the root_element has been parsed from. This is used to resolve + relative file paths. + :param file_path: The file path or URI reference of the supplementary file to be extracted + :param file_store: The SupplementaryFileContainer to add the extracted supplementary files to + :return: The stored file name as returned by *file_store*, or ``None`` if the reference was skipped. + """ + # 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 file_path.startswith('//') or ':' in file_path.split('/')[0]: + logger.info(f"Skipping supplementary file {file_path}, since it seems to be an absolute URI or " + f"network-path URI reference") + return None + absolute_name = pyecma376_2.package_model.part_realpath(file_path, part_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)) + return final_name class AASXWriter: @@ -541,7 +569,8 @@ def write_all_aas_objects(self, contained objects into an ``aas_env`` part in the AASX package. If the ObjectStore 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 fetched from the ``file_store`` - and added to the AASX package. + and added to the AASX package. If the ObjectStore contains a thumbnail referenced by + ``default_thumbnail`` in :class:`~basyx.aas.model.aas.AssetInformation`, it is also added to the AASX package. .. attention:: @@ -563,17 +592,24 @@ def write_all_aas_objects(self, logger.debug(f"Writing AASX part {part_name} with AAS objects ...") supplementary_files: List[str] = [] + def _collect_supplementary_file(file_name: str) -> None: + # 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]: + return + supplementary_files.append(file_name) + # Retrieve objects and scan for referenced supplementary files for the_object in objects: + if isinstance(the_object, model.AssetAdministrationShell): + if (the_object.asset_information.default_thumbnail and + the_object.asset_information.default_thumbnail.path): + _collect_supplementary_file(the_object.asset_information.default_thumbnail.path) 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) + if element.value: + _collect_supplementary_file(element.value) # Add aas-spec relationship if not split_part: @@ -824,15 +860,25 @@ def add_file(self, name: str, file: IO[bytes], content_type: str) -> str: if hash not in self._store: self._store[hash] = data self._store_refcount[hash] = 0 - name_map_data = (hash, content_type) + return self._assign_unique_name(name, hash, content_type) + + def rename_file(self, old_name: str, new_name: str) -> str: + if old_name not in self._name_map: + raise KeyError(f"File with name {old_name} not found in SupplementaryFileContainer.") + if new_name == old_name: + return new_name + file_hash, file_content_type = self._name_map[old_name] + del self._name_map[old_name] + return self._assign_unique_name(new_name, file_hash, file_content_type) + + def _assign_unique_name(self, name: str, sha: bytes, content_type: str) -> str: new_name = name i = 1 while True: if new_name not in self._name_map: - self._name_map[new_name] = name_map_data - self._store_refcount[hash] += 1 + self._name_map[new_name] = (sha, content_type) return new_name - elif self._name_map[new_name] == name_map_data: + elif self._name_map[new_name] == (sha, content_type): return new_name new_name = self._append_counter(name, i) i += 1 diff --git a/sdk/basyx/aas/adapter/xml/xml_serialization.py b/sdk/basyx/aas/adapter/xml/xml_serialization.py index 2dc578ca0..60c5bf3fa 100644 --- a/sdk/basyx/aas/adapter/xml/xml_serialization.py +++ b/sdk/basyx/aas/adapter/xml/xml_serialization.py @@ -899,10 +899,6 @@ def object_to_xml_element(obj: object) -> etree._Element: return value_reference_pair_to_xml(obj) elif isinstance(obj, model.ConceptDescription): return concept_description_to_xml(obj) - elif isinstance(obj, model.LangStringSet): - # FIXME: `lang_string_set_to_xml` expects `tag` parameter, `tag` doesn't have default value - # Issue: https://github.com/eclipse-basyx/basyx-python-sdk/issues/397 - return lang_string_set_to_xml(obj) # type: ignore[call-arg] elif isinstance(obj, model.EmbeddedDataSpecification): return embedded_data_specification_to_xml(obj) elif isinstance(obj, model.DataSpecificationIEC61360): diff --git a/sdk/basyx/aas/model/base.py b/sdk/basyx/aas/model/base.py index 35ccad5a1..5e2b339f6 100644 --- a/sdk/basyx/aas/model/base.py +++ b/sdk/basyx/aas/model/base.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025 the Eclipse BaSyx Authors +# Copyright (c) 2026 the Eclipse BaSyx Authors # # This program and the accompanying materials are made available under the terms of the MIT License, available in # the LICENSE file of this project. @@ -291,7 +291,8 @@ class LangStringSet(MutableMapping[str, str]): """ def __init__(self, dict_: Dict[str, str]): self._dict: Dict[str, str] = {} - + if not isinstance(dict_, dict): + raise TypeError(f"A {self.__class__.__name__} must be initialized with a dict!, got {type(dict_)}") if len(dict_) < 1: raise ValueError(f"A {self.__class__.__name__} must not be empty!") for ltag in dict_: @@ -614,9 +615,9 @@ class Referable(HasExtension, metaclass=abc.ABCMeta): def __init__(self): super().__init__() self._id_short: Optional[NameType] = None - self.display_name: Optional[MultiLanguageNameType] = dict() + self._display_name: Optional[MultiLanguageNameType] = None self._category: Optional[NameType] = None - self.description: Optional[MultiLanguageTextType] = dict() + self._description: Optional[MultiLanguageTextType] = None # We use a Python reference to the parent Namespace instead of a Reference Object, as specified. This allows # simpler and faster navigation/checks and it has no effect in the serialized data formats anyway. self.parent: Optional[UniqueIdShortNamespace] = None @@ -827,6 +828,28 @@ def _set_id_short(self, id_short: Optional[NameType]): # Redundant to the line above. However, this way, we make sure that we really update the _id_short self._id_short = id_short + @property + def display_name(self) -> Optional[MultiLanguageNameType]: + """Display name of the element (MultiLanguageNameType).""" + return self._display_name + + @display_name.setter + def display_name(self, value: Union[MultiLanguageNameType, dict, None]) -> None: + if value is not None and not isinstance(value, MultiLanguageNameType): + value = MultiLanguageNameType(value) + self._display_name = value + + @property + def description(self) -> Optional[MultiLanguageTextType]: + """Description of the element (MultiLanguageTextType).""" + return self._description + + @description.setter + def description(self, value: Union[MultiLanguageTextType, dict, None]) -> None: + if value is not None and not isinstance(value, MultiLanguageTextType): + value = MultiLanguageTextType(value) + self._description = value + def update_from(self, other: "Referable"): """ Internal function to update the object's attributes from a different version of the exact same object. diff --git a/sdk/basyx/aas/model/submodel.py b/sdk/basyx/aas/model/submodel.py index 9e7321c41..733eaf58e 100644 --- a/sdk/basyx/aas/model/submodel.py +++ b/sdk/basyx/aas/model/submodel.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025 the Eclipse BaSyx Authors +# Copyright (c) 2026 the Eclipse BaSyx Authors # # This program and the accompanying materials are made available under the terms of the MIT License, available in # the LICENSE file of this project. @@ -344,6 +344,16 @@ def __init__(self, self.value: Optional[base.MultiLanguageTextType] = value self.value_id: Optional[base.Reference] = value_id + @property + def value(self) -> Optional[base.MultiLanguageTextType]: + return self._value + + @value.setter + def value(self, value: Union[base.MultiLanguageTextType, dict, None]) -> None: + if value is not None and not isinstance(value, base.MultiLanguageTextType): + value = base.MultiLanguageTextType(value) + self._value = value + class Range(DataElement): """ diff --git a/sdk/test/adapter/aasx/test_aasx.py b/sdk/test/adapter/aasx/test_aasx.py index a83c60186..931db528f 100644 --- a/sdk/test/adapter/aasx/test_aasx.py +++ b/sdk/test/adapter/aasx/test_aasx.py @@ -22,19 +22,43 @@ class TestAASXUtils(unittest.TestCase): 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") + saved_file_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) + self.assertEqual("/TestFile.pdf", saved_file_name) f.seek(0) - container.add_file("/TestFile.pdf", f, "application/pdf") + # Add the same file again with the same name + same_file_with_same_name = 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) + self.assertEqual("/TestFile.pdf", same_file_with_same_name) + # Add other file with the same name to create a conflict with open(__file__, 'rb') as f: - new_name = container.add_file("/TestFile.pdf", f, "application/pdf") + saved_file_name_2 = container.add_file("/TestFile.pdf", f, "application/pdf") # Now, we have a conflict - self.assertNotEqual("/TestFile.pdf", new_name) - self.assertIn(new_name, container) + self.assertNotEqual(saved_file_name, saved_file_name_2) + self.assertIn(saved_file_name_2, container) + + # Rename file to a new unique name + renamed = container.rename_file(saved_file_name_2, "/RenamedTestFile.pdf") + self.assertIn(renamed, container) + # Old name should no longer exist + self.assertNotIn(saved_file_name_2, container) + self.assertEqual(renamed, "/RenamedTestFile.pdf") + + # Renaming to the same name should be no-op + renamed_same = container.rename_file(renamed, renamed) + self.assertEqual(renamed, renamed_same) + + # Renaming to an existing name should create a conflict + renamed_conflict = container.rename_file(renamed, "/TestFile.pdf") + self.assertNotEqual(renamed_conflict, "/TestFile.pdf") + self.assertIn(renamed_conflict, container) + + # Renaming a non-existing file should raise KeyError + with self.assertRaises(KeyError): + container.rename_file("/NonExistingFile.pdf", "/AnotherName.pdf") + + new_name = renamed_conflict # Check metadata self.assertEqual("application/pdf", container.get_content_type("/TestFile.pdf")) @@ -68,22 +92,22 @@ def test_supplementary_file_container(self) -> None: 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() + data = example_aas.create_full_example() # creates a complete, valid example AAS + files = aasx.DictSupplementaryFileContainer() # in-memory store for attached files with open(os.path.join(os.path.dirname(__file__), 'TestFile.pdf'), 'rb') as f: - files.add_file("/TestFile.pdf", f, "application/pdf") + files.add_file("/TestFile.pdf", f, "application/pdf") # add a real supplementary pdf file f.seek(0) - # Create OPC/AASX core properties + # create AASX metadata (core properties) cp = pyecma376_2.OPCCoreProperties() cp.created = datetime.datetime.now() cp.creator = "Eclipse BaSyx Python Testing Framework" # Write AASX file - for write_json in (False, True): + for write_json in (False, True): # Loop over both XML and JSON modes with self.subTest(write_json=write_json): - fd, filename = tempfile.mkstemp(suffix=".aasx") - os.close(fd) + fd, filename = tempfile.mkstemp(suffix=".aasx") # create temporary file + os.close(fd) # close file descriptor # Write AASX file # the zipfile library reports errors as UserWarnings via the warnings library. Let's check for @@ -126,3 +150,86 @@ def test_writing_reading_example_aas(self) -> None: "78450a66f59d74c073bf6858db340090ea72a8b1") os.unlink(filename) + + +class AASXWriterReferencedSubmodelsTest(unittest.TestCase): + + def test_only_referenced_submodels(self): + """ + Test that verifies that all Submodels (referenced and unreferenced) are written to the AASX package when using + the convenience function write_all_aas_objects(). + When calling the higher-level function write_aas(), however, only + referenced Submodels in the ObjectStore should be included. + """ + # Create referenced and unreferenced Submodels + referenced_submodel = model.Submodel(id_="ref_submodel") + unreferenced_submodel = model.Submodel(id_="unref_submodel") + + aas = model.AssetAdministrationShell( + id_="Test_AAS", + asset_information=model.AssetInformation( + asset_kind=model.AssetKind.INSTANCE, + global_asset_id="http://acplt.org/Test_Asset" + ), + submodel={model.ModelReference.from_referable(referenced_submodel)} + ) + + # ObjectStore containing all objects + object_store = model.DictObjectStore([aas, referenced_submodel, unreferenced_submodel]) + + # Empty SupplementaryFileContainer (no files needed) + file_store = aasx.DictSupplementaryFileContainer() + + # --- Step 1: Check write_aas() behavior --- + for write_json in (False, True): + with self.subTest(method="write_aas", write_json=write_json): + fd, filename = tempfile.mkstemp(suffix=".aasx") + os.close(fd) + + with warnings.catch_warnings(record=True) as w: + with aasx.AASXWriter(filename) as writer: + # write_aas only takes the AAS id and ObjectStore + writer.write_aas( + aas_ids=[aas.id], + object_store=object_store, + file_store=file_store, + write_json=write_json + ) + + # Read back + 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) + + # Assertions + self.assertIn(referenced_submodel.id, new_data) # referenced Submodel is included + self.assertNotIn(unreferenced_submodel.id, new_data) # unreferenced Submodel is excluded + + os.unlink(filename) + + # --- Step 2: Check write_all_aas_objects --- + for write_json in (False, True): + with self.subTest(method="write_all_aas_objects", write_json=write_json): + fd, filename = tempfile.mkstemp(suffix=".aasx") + os.close(fd) + + with warnings.catch_warnings(record=True) as w: + with aasx.AASXWriter(filename) as writer: + writer.write_all_aas_objects( + part_name="/aasx/my_aas_part.xml", + objects=object_store, + file_store=file_store, + write_json=write_json + ) + + # Read back + 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) + + # Assertions + self.assertIn(referenced_submodel.id, new_data) + self.assertIn(unreferenced_submodel.id, new_data) # all objects are written + os.unlink(filename)