diff --git a/sdk/basyx/aas/model/base.py b/sdk/basyx/aas/model/base.py index 84650fe17..638f03200 100644 --- a/sdk/basyx/aas/model/base.py +++ b/sdk/basyx/aas/model/base.py @@ -573,6 +573,75 @@ def _get_key_value_for_referable(referable: "Referable") -> str: ) +class NamespaceSetAttribute(Generic[_NSO]): + """ + Descriptor for the attributes of a :class:`~basyx.aas.model.base.Namespace` that hold a + :class:`~basyx.aas.model.base.NamespaceSet`, e.g. ``Submodel.submodel_element``. + + The corresponding ``__init__`` parameters accept any ``Iterable``, so assigning an iterable to the attribute + is the obvious thing to write. A plain assignment would rebind the attribute and thereby silently shadow the + NamespaceSet: the set would stay registered in ``Namespace.namespace_element_sets``, so its objects would + remain reachable via ``get_referable()`` and friends while being invisible in the attribute, uniqueness + constraints would no longer be checked and the ``parent`` of the assigned objects would never be set. + + This descriptor therefore redirects such an assignment to + :meth:`~basyx.aas.model.base.NamespaceSet.assign`, which replaces the *content* of the existing set:: + + submodel.submodel_element = [my_property] # replaces the content of the set + aas.extension = [] # clears the set + + The first assignment, done by the ``__init__`` of the owning class, binds the NamespaceSet itself. Classes + outside of this package can use this descriptor for their own NamespaceSet attributes in the same way. + """ + + _name: str + _storage_name: str + + def __set_name__(self, owner: type, name: str) -> None: + self._name = name + self._storage_name = f"_{name}_namespace_set" + self.__doc__ = ( + f"A :class:`~basyx.aas.model.base.NamespaceSet` holding the {name} of this element. Assigning an " + "iterable replaces its content, see " + ":class:`~basyx.aas.model.base.NamespaceSetAttribute`." + ) + + def __get__( + self, obj: Optional["Namespace"], objtype: Optional[type] = None + ) -> "NamespaceSet[_NSO]": + if obj is None: + # Class attribute access, e.g. by Sphinx autodoc + return self # type: ignore[return-value] + return getattr(obj, self._storage_name) + + def __set__(self, obj: "Namespace", value: Iterable[_NSO]) -> None: + current: Optional["NamespaceSet[_NSO]"] = getattr(obj, self._storage_name, None) + if current is None: + if not isinstance(value, NamespaceSet): + raise TypeError( + f"{type(obj).__name__}.{self._name} must be initialized with a NamespaceSet, " + f"got {type(value).__name__}" + ) + setattr(obj, self._storage_name, value) + elif current is not value: + current.assign(value) + + +class OrderedNamespaceSetAttribute(NamespaceSetAttribute[_NSO]): + """ + Descriptor for the attributes of a :class:`~basyx.aas.model.base.Namespace` that hold an + :class:`~basyx.aas.model.base.OrderedNamespaceSet`, see + :class:`~basyx.aas.model.base.NamespaceSetAttribute`. It only narrows the type of the attribute, the behavior + is identical. + """ + + if TYPE_CHECKING: + + def __get__( + self, obj: Optional["Namespace"], objtype: Optional[type] = None + ) -> "OrderedNamespaceSet[_NSO]": ... + + class Namespace(metaclass=abc.ABCMeta): """ Abstract baseclass for all objects which form a Namespace to hold objects and resolve them by their @@ -649,11 +718,12 @@ class HasExtension(Namespace, metaclass=abc.ABCMeta): :ivar extension: A :class:`~.NamespaceSet` of :class:`Extensions <.Extension>` of the element. """ + extension: "NamespaceSetAttribute[Extension]" = NamespaceSetAttribute() + @abc.abstractmethod def __init__(self) -> None: super().__init__() self.namespace_element_sets: List[NamespaceSet] = [] - self.extension: NamespaceSet[Extension] def get_extension_by_name(self, name: str) -> "Extension": """ @@ -1814,11 +1884,12 @@ class Qualifiable(Namespace, metaclass=abc.ABCMeta): qualifiable element. """ + qualifier: "NamespaceSetAttribute[Qualifier]" = NamespaceSetAttribute() + @abc.abstractmethod def __init__(self) -> None: super().__init__() self.namespace_element_sets: List[NamespaceSet] = [] - self.qualifier: NamespaceSet[Qualifier] def get_qualifier_by_type(self, qualifier_type: QualifierType) -> "Qualifier": """ @@ -2339,6 +2410,31 @@ def clear(self) -> None: for attr_name, (backend, case_sensitive) in self._backend.items(): backend.clear() + def assign(self, items: Iterable[_NSO]) -> None: + """ + Replace all objects of this set with the given ones, atomically. + + This is the implementation of assignments to NamespaceSet attributes, see ``Namespace.__setattr__()``. + If adding one of the new items fails, the previous content of the set is restored. + + :param items: The objects this set should contain afterwards + :raises AASConstraintViolation: If the given items violate a uniqueness constraint of this namespace + :raises ValueError: If one of the given items already belongs to another namespace + """ + # items may be a lazy iterable over this very set, e.g. filter(..., submodel.submodel_element), so it has + # to be consumed before the set is cleared + new_items: List[_NSO] = list(items) + previous: List[_NSO] = list(self) + self.clear() + try: + for item in new_items: + self.add(item) + except Exception: + self.clear() + for item in previous: + self.add(item) + raise + def get_object_by_attribute( self, attribute_name: str, attribute_value: ATTRIBUTE_TYPES ) -> _NSO: diff --git a/sdk/basyx/aas/model/submodel.py b/sdk/basyx/aas/model/submodel.py index c52551532..1aaf7c8e4 100644 --- a/sdk/basyx/aas/model/submodel.py +++ b/sdk/basyx/aas/model/submodel.py @@ -144,6 +144,10 @@ class Submodel( :ivar embedded_data_specifications: List of Embedded data specification. """ + submodel_element: "base.NamespaceSetAttribute[SubmodelElement]" = ( + base.NamespaceSetAttribute() + ) + def __init__( self, id_: base.Identifier, @@ -730,6 +734,8 @@ class SubmodelElementCollection(SubmodelElement, base.UniqueIdShortNamespace): :ivar embedded_data_specifications: List of Embedded data specification. """ + value: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute() + def __init__( self, id_short: Optional[base.NameType], @@ -757,9 +763,7 @@ def __init__( supplemental_semantic_id, embedded_data_specifications, ) - self.value: base.NamespaceSet[SubmodelElement] = base.NamespaceSet( - self, [("id_short", True)], value - ) + self.value = base.NamespaceSet(self, [("id_short", True)], value) _SE = TypeVar("_SE", bound=SubmodelElement) @@ -1100,6 +1104,8 @@ class AnnotatedRelationshipElement(RelationshipElement, base.UniqueIdShortNamesp :ivar embedded_data_specifications: List of Embedded data specification. """ + annotation: "base.NamespaceSetAttribute[DataElement]" = base.NamespaceSetAttribute() + def __init__( self, id_short: Optional[base.NameType], @@ -1177,6 +1183,10 @@ class Operation(SubmodelElement, base.UniqueIdShortNamespace): :ivar embedded_data_specifications: List of Embedded data specification. """ + input_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute() + output_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute() + in_output_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute() + def __init__( self, id_short: Optional[base.NameType], @@ -1316,6 +1326,8 @@ class Entity(SubmodelElement, base.UniqueIdShortNamespace): :ivar embedded_data_specifications: List of Embedded data specification. """ + statement: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute() + def __init__( self, id_short: Optional[base.NameType], diff --git a/sdk/test/model/test_base.py b/sdk/test/model/test_base.py index 2e2e75ca0..978d44d48 100644 --- a/sdk/test/model/test_base.py +++ b/sdk/test/model/test_base.py @@ -354,6 +354,11 @@ def test_update_commit_qualifier_extension_semantic_id(self): class ExampleNamespaceReferable( model.UniqueIdShortNamespace, model.UniqueSemanticIdNamespace, model.Identifiable ): + set1: "model.NamespaceSetAttribute[model.Referable]" = model.NamespaceSetAttribute() + set2: "model.NamespaceSetAttribute[model.Referable]" = model.NamespaceSetAttribute() + set3: "model.NamespaceSetAttribute[model.Extension]" = model.NamespaceSetAttribute() + set4: "model.NamespaceSetAttribute[model.Qualifier]" = model.NamespaceSetAttribute() + def __init__(self, values=()): super().__init__() # The 'id' is required by Referable.__repr__() in error messages. @@ -367,6 +372,8 @@ def __init__(self, values=()): class ExampleNamespaceQualifier(model.Qualifiable): + set1: "model.NamespaceSetAttribute[model.Qualifier]" = model.NamespaceSetAttribute() + def __init__(self, values=()): super().__init__() self.set1 = model.NamespaceSet(self, [("type", False)], values) @@ -464,6 +471,54 @@ def test_namespaceset_pop_removes_from_all_backends(self) -> None: self.namespace.set1.add(new_prop) self.assertEqual(1, len(self.namespace.set1)) + def test_namespaceset_attribute_assignment(self) -> None: + # Assigning an iterable to an attribute holding a NamespaceSet must replace the content of that + # NamespaceSet instead of replacing the NamespaceSet itself, see issue #65 + self.namespace.set2.add(self.prop1) + namespace_set = self.namespace.set2 + number_of_sets = len(self.namespace.namespace_element_sets) + + self.namespace.set2 = [self.prop2, self.prop4] + self.assertIs(namespace_set, self.namespace.set2) + self.assertEqual(number_of_sets, len(self.namespace.namespace_element_sets)) + self.assertEqual((self.prop2, self.prop4), tuple(self.namespace.set2)) + self.assertIsNone(self.prop1.parent) + self.assertIs(self.namespace, self.prop2.parent) + self.assertIs(self.namespace, self.prop4.parent) + self.assertIs(self.prop2, self.namespace.get_referable("Prop2")) + + self.namespace.set2 = [] + self.assertIs(namespace_set, self.namespace.set2) + self.assertEqual(0, len(self.namespace.set2)) + self.assertIsNone(self.prop2.parent) + self.assertIsNone(self.prop4.parent) + + def test_namespaceset_attribute_assignment_atomicity(self) -> None: + # If an item cannot be added, the previous content of the NamespaceSet must be restored + self.namespace.set2 = [self.prop1, self.prop2] + + with self.assertRaises(model.AASConstraintViolation) as cm: + self.namespace.set2 = [self.prop4, self.prop3, self.prop7] + self.assertEqual( + "Object with attribute (name='id_short', value='Prop2') is already present in this set " + "of objects (Constraint AASd-022)", + str(cm.exception), + ) + self.assertEqual((self.prop1, self.prop2), tuple(self.namespace.set2)) + self.assertIs(self.namespace, self.prop1.parent) + self.assertIs(self.namespace, self.prop2.parent) + self.assertIsNone(self.prop3.parent) + self.assertIsNone(self.prop4.parent) + self.assertIsNone(self.prop7.parent) + + def test_namespaceset_attribute_assignment_same_set(self) -> None: + self.namespace.set2.add(self.prop1) + namespace_set = self.namespace.set2 + self.namespace.set2 = namespace_set + self.assertIs(namespace_set, self.namespace.set2) + self.assertEqual((self.prop1,), tuple(self.namespace.set2)) + self.assertEqual(1, self.namespace.namespace_element_sets.count(namespace_set)) + def test_NamespaceSet(self) -> None: self.namespace.set1.add(self.prop1) self.assertEqual(1, len(self.namespace.set1)) @@ -881,6 +936,15 @@ def test_aasd_117(self) -> None: class ExampleOrderedNamespace( model.UniqueIdShortNamespace, model.UniqueSemanticIdNamespace, model.Identifiable ): + set1: "model.OrderedNamespaceSetAttribute[model.Referable]" = ( + model.OrderedNamespaceSetAttribute() + ) + set2: "model.OrderedNamespaceSetAttribute[model.Referable]" = ( + model.OrderedNamespaceSetAttribute() + ) + set3: "model.NamespaceSetAttribute[model.Extension]" = model.NamespaceSetAttribute() + set4: "model.NamespaceSetAttribute[model.Qualifier]" = model.NamespaceSetAttribute() + def __init__(self, values=()): super().__init__() # The 'id' is required by Referable.__repr__() in error messages. @@ -949,6 +1013,13 @@ def test_OrderedNamespace(self) -> None: str(cm2.exception), ) + def test_ordered_namespaceset_attribute_assignment_preserves_order(self) -> None: + self.namespace.set2 = [self.prop2, self.prop1, self.prop4] + self.assertEqual( + [self.prop2, self.prop1, self.prop4], list(self.namespace.set2) + ) + self.assertEqual(self.prop1, self.namespace.set2[1]) + def test_ordered_namespaceset_int_setitem_preserves_index(self) -> None: # __setitem__ int must place the new item at the exact index of the replaced item. # Items before and after the replaced index must not shift. @@ -1026,6 +1097,118 @@ def test_ordered_namespaceset_slice_setitem_preserves_order(self) -> None: self.assertIs(ns, new2.parent) +class NamespaceSetAttributeAssignmentTest(unittest.TestCase): + """ + Tests for assignments to attributes of the metamodel classes that hold a NamespaceSet, see issue #65 + """ + + def test_assignment(self) -> None: + submodel = model.Submodel("urn:x-test:submodel") + prop = model.Property("Prop", model.datatypes.Int) + submodel.submodel_element = [prop] + self.assertIsInstance(submodel.submodel_element, model.NamespaceSet) + self.assertIs(prop, submodel.get_referable("Prop")) + self.assertIs(submodel, prop.parent) + + qualifier = model.Qualifier("type1", model.datatypes.Int, 1) + submodel.qualifier = [qualifier] + self.assertIs(qualifier, submodel.get_qualifier_by_type("type1")) + + aas = model.AssetAdministrationShell( + model.AssetInformation(global_asset_id="urn:x-test:asset"), + "urn:x-test:aas", + ) + extension = model.Extension("Ext1", model.datatypes.Int, 1) + aas.extension = [extension] + self.assertIs(extension, aas.get_extension_by_name("Ext1")) + + def test_update_from(self) -> None: + # update_from() and therefore the ObjectStore backends rely on these attributes being NamespaceSets + submodel = model.Submodel("urn:x-test:submodel") + submodel.submodel_element = [model.Property("Prop", model.datatypes.Int, 1)] + other_submodel = model.Submodel( + "urn:x-test:submodel", + submodel_element=[model.Property("Prop", model.datatypes.Int, 2)], + ) + submodel.update_from(other_submodel) + updated_prop = submodel.get_referable("Prop") + self.assertIsInstance(updated_prop, model.Property) + assert isinstance(updated_prop, model.Property) + self.assertEqual(2, updated_prop.value) + + def test_assignment_of_all_protected_attributes(self) -> None: + # Every attribute that holds a NamespaceSet must be protected, so that none is forgotten + collection = model.SubmodelElementCollection("Collection") + collection.value = [model.Property("Prop", model.datatypes.Int)] + self.assertIs(collection, collection.get_referable("Prop").parent) + + annotated = model.AnnotatedRelationshipElement( + "Annotated", + model.ExternalReference( + (model.Key(model.KeyTypes.GLOBAL_REFERENCE, "urn:x-test:first"),) + ), + model.ExternalReference( + (model.Key(model.KeyTypes.GLOBAL_REFERENCE, "urn:x-test:second"),) + ), + ) + annotated.annotation = [model.Property("Annotation", model.datatypes.Int)] + self.assertIs(annotated, annotated.get_referable("Annotation").parent) + + operation = model.Operation("Operation") + operation.input_variable = [model.Property("In", model.datatypes.Int)] + operation.output_variable = [model.Property("Out", model.datatypes.Int)] + operation.in_output_variable = [model.Property("InOut", model.datatypes.Int)] + for id_short in ("In", "Out", "InOut"): + self.assertIs(operation, operation.get_referable(id_short).parent) + + entity = model.Entity( + "Entity", + model.EntityType.SELF_MANAGED_ENTITY, + global_asset_id="urn:x-test:asset", + ) + entity.statement = [model.Property("Statement", model.datatypes.Int)] + self.assertIs(entity, entity.get_referable("Statement").parent) + + def test_assignment_of_lazy_iterable_over_the_same_set(self) -> None: + # Filtering a NamespaceSet into itself is a common idiom, so the iterable must be consumed before the set + # is cleared + submodel = model.Submodel( + "urn:x-test:submodel", + submodel_element=[ + model.Property("Prop1", model.datatypes.Int), + model.Property("Prop2", model.datatypes.Int), + ], + ) + prop1 = submodel.get_referable("Prop1") + prop2 = submodel.get_referable("Prop2") + + submodel.submodel_element = filter( + lambda element: element.id_short == "Prop1", submodel.submodel_element + ) + self.assertEqual([prop1], list(submodel.submodel_element)) + self.assertIs(submodel, prop1.parent) + self.assertIsNone(prop2.parent) + + submodel.submodel_element = ( + element for element in submodel.submodel_element if False + ) + self.assertEqual(0, len(submodel.submodel_element)) + self.assertIsNone(prop1.parent) + + def test_initial_assignment_must_be_a_namespaceset(self) -> None: + # A class that never binds a NamespaceSet to the attribute must not end up with a plain iterable in it + class UninitializedQualifiable(model.Qualifiable): + def __init__(self) -> None: + super().__init__() + + with self.assertRaises(TypeError) as cm: + UninitializedQualifiable().qualifier = [] + self.assertEqual( + "UninitializedQualifiable.qualifier must be initialized with a NamespaceSet, got list", + str(cm.exception), + ) + + class ExternalReferenceTest(unittest.TestCase): def test_constraints(self): with self.assertRaises(ValueError) as cm: