Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 78 additions & 32 deletions sdk/basyx/aas/adapter/aasx.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
"""
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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::

Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down
4 changes: 0 additions & 4 deletions sdk/basyx/aas/adapter/xml/xml_serialization.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
31 changes: 27 additions & 4 deletions sdk/basyx/aas/model/base.py
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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_:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
12 changes: 11 additions & 1 deletion sdk/basyx/aas/model/submodel.py
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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):
"""
Expand Down
Loading
Loading