diff --git a/basyx/aas/model/base.py b/basyx/aas/model/base.py index db2d35327..5f22e470e 100644 --- a/basyx/aas/model/base.py +++ b/basyx/aas/model/base.py @@ -1802,7 +1802,9 @@ class NamespaceSet(MutableSet[_NSO], Generic[_NSO]): """ def __init__(self, parent: Union[UniqueIdShortNamespace, UniqueSemanticIdNamespace, Qualifiable, HasExtension], attribute_names: List[Tuple[str, bool]], items: Iterable[_NSO] = (), - item_add_hook: Optional[Callable[[_NSO, Iterable[_NSO]], None]] = None) -> None: + item_add_hook: Optional[Callable[[_NSO, Iterable[_NSO]], None]] = None, + item_id_set_hook: Optional[Callable[[_NSO], None]] = None, + item_id_del_hook: Optional[Callable[[_NSO], None]] = None) -> None: """ Initialize a new NamespaceSet. @@ -1816,6 +1818,11 @@ def __init__(self, parent: Union[UniqueIdShortNamespace, UniqueSemanticIdNamespa :param item_add_hook: A function that is called for each item that is added to this NamespaceSet, even when it is initialized. The first parameter is the item that is added while the second is an iterator over all currently contained items. Useful for constraint checking. + :param item_id_set_hook: A function called to calculate the identifying attribute (e.g. id_short) of an object + on-the-fly when it is added. Used for the SubmodelElementList implementation. + :param item_id_del_hook: A function that is called for each item removed from this NamespaceSet. Used in + SubmodelElementList to unset id_shorts on removal. Should not be used for + constraint checking, as the hook is called after removal. :raises AASConstraintViolation: When `items` contains multiple objects with same unique attribute or when an item doesn't has an identifying attribute """ @@ -1823,6 +1830,8 @@ def __init__(self, parent: Union[UniqueIdShortNamespace, UniqueSemanticIdNamespa parent.namespace_element_sets.append(self) self._backend: Dict[str, Tuple[Dict[ATTRIBUTE_TYPES, _NSO], bool]] = {} self._item_add_hook: Optional[Callable[[_NSO, Iterable[_NSO]], None]] = item_add_hook + self._item_id_set_hook: Optional[Callable[[_NSO], None]] = item_id_set_hook + self._item_id_del_hook: Optional[Callable[[_NSO], None]] = item_id_del_hook for name, case_sensitive in attribute_names: self._backend[name] = ({}, case_sensitive) try: @@ -1869,6 +1878,8 @@ def add(self, value: _NSO): if value.parent is not None and value.parent is not self.parent: raise ValueError("Object has already a parent, but it must not be part of two namespaces.") # TODO remove from current parent instead (allow moving)? + if self._item_id_set_hook is not None: + self._item_id_set_hook(value) for set_ in self.parent.namespace_element_sets: for attr_name, (backend, case_sensitive) in set_._backend.items(): if hasattr(value, attr_name): @@ -1883,7 +1894,12 @@ def add(self, value: _NSO): "this set of objects" if set_ is self else "another set in the same namespace")) if self._item_add_hook is not None: - self._item_add_hook(value, self.__iter__()) + try: + self._item_add_hook(value, self.__iter__()) + except Exception: + if self._item_id_del_hook is not None: + self._item_id_del_hook(value) + raise value.parent = self.parent for attr_name, (backend, case_sensitive) in self._backend.items(): backend[self._get_attribute(value, attr_name, case_sensitive)] = value @@ -1901,9 +1917,15 @@ def remove(self, item: _NSO) -> None: break if not item_found: raise KeyError("Object not found in NamespaceDict") + # parent needs to be unset first, otherwise generated id_shorts cannot be unset + # see SubmodelElementList item.parent = None + # item has to be removed from backend before _item_del_hook() is called, as the hook may unset the id_short, + # as in SubmodelElementLists for attr_name, (backend, case_sensitive) in self._backend.items(): del backend[self._get_attribute(item, attr_name, case_sensitive)] + if self._item_id_del_hook is not None: + self._item_id_del_hook(item) def discard(self, x: _NSO) -> None: if x not in self: @@ -1912,13 +1934,19 @@ def discard(self, x: _NSO) -> None: def pop(self) -> _NSO: _, value = next(iter(self._backend.values()))[0].popitem() + if self._item_id_del_hook is not None: + self._item_id_del_hook(value) value.parent = None return value def clear(self) -> None: for attr_name, (backend, case_sensitive) in self._backend.items(): for value in backend.values(): + # parent needs to be unset first, otherwise generated id_shorts cannot be unset + # see SubmodelElementList value.parent = None + if self._item_id_del_hook is not None: + self._item_id_del_hook(value) for attr_name, (backend, case_sensitive) in self._backend.items(): backend.clear() @@ -1999,7 +2027,9 @@ class OrderedNamespaceSet(NamespaceSet[_NSO], MutableSequence[_NSO], Generic[_NS """ def __init__(self, parent: Union[UniqueIdShortNamespace, UniqueSemanticIdNamespace, Qualifiable, HasExtension], attribute_names: List[Tuple[str, bool]], items: Iterable[_NSO] = (), - item_add_hook: Optional[Callable[[_NSO, Iterable[_NSO]], None]] = None) -> None: + item_add_hook: Optional[Callable[[_NSO, Iterable[_NSO]], None]] = None, + item_id_set_hook: Optional[Callable[[_NSO], None]] = None, + item_id_del_hook: Optional[Callable[[_NSO], None]] = None) -> None: """ Initialize a new OrderedNamespaceSet. @@ -2013,11 +2043,16 @@ def __init__(self, parent: Union[UniqueIdShortNamespace, UniqueSemanticIdNamespa :param item_add_hook: A function that is called for each item that is added to this NamespaceSet, even when it is initialized. The first parameter is the item that is added while the second is an iterator over all currently contained items. Useful for constraint checking. + :param item_id_set_hook: A function called to calculate the identifying attribute (e.g. id_short) of an object + on-the-fly when it is added. Used for the SubmodelElementList implementation. + :param item_id_del_hook: A function that is called for each item removed from this NamespaceSet. Used in + SubmodelElementList to unset id_shorts on removal. Should not be used for + constraint checking, as the hook is called after removal. :raises AASConstraintViolation: When `items` contains multiple objects with same unique attribute or when an item doesn't has an identifying attribute """ self._order: List[_NSO] = [] - super().__init__(parent, attribute_names, items, item_add_hook) + super().__init__(parent, attribute_names, items, item_add_hook, item_id_set_hook, item_id_del_hook) def __iter__(self) -> Iterator[_NSO]: return iter(self._order) diff --git a/test/model/test_base.py b/test/model/test_base.py index 65b8e02f6..a8b1e347a 100644 --- a/test/model/test_base.py +++ b/test/model/test_base.py @@ -7,7 +7,7 @@ import unittest from unittest import mock -from typing import Dict, Optional, List +from typing import Callable, Dict, Iterable, List, Optional, Type, TypeVar from collections import OrderedDict from basyx.aas import model @@ -460,46 +460,106 @@ def test_NamespaceSet(self) -> None: "of objects (Constraint AASd-022)", str(cm.exception)) - def test_namespaceset_item_add_hook(self) -> None: - new_item = None - existing_items = [] - - class DummyNamespace(model.UniqueIdShortNamespace): - def __init__(self, items): - def dummy_hook(new, existing): - nonlocal new_item, existing_items - new_item = new - # Create a new list to prevent an error when checking the assertions: - # RuntimeError: dictionary changed size during iteration - existing_items = list(existing) - - super().__init__() - self.set1 = model.NamespaceSet(self, [('id_short', True)], items, dummy_hook) - - cap = model.Capability("test_cap") - dummy_ns = DummyNamespace({cap}) - self.assertIs(new_item, cap) - self.assertEqual(len(existing_items), 0) - - mlp = model.MultiLanguageProperty("test_mlp") - dummy_ns.add_referable(mlp) - self.assertIs(new_item, mlp) - self.assertEqual(len(existing_items), 1) - self.assertIn(cap, existing_items) - - prop = model.Property("test_prop", model.datatypes.Int) - dummy_ns.set1.add(prop) - self.assertIs(new_item, prop) - self.assertEqual(len(existing_items), 2) - self.assertIn(cap, existing_items) - self.assertIn(mlp, existing_items) - - dummy_ns.remove_referable("test_cap") - dummy_ns.add_referable(cap) - self.assertIs(new_item, cap) - self.assertEqual(len(existing_items), 2) - self.assertIn(mlp, existing_items) - self.assertIn(prop, existing_items) + def test_namespaceset_hooks(self) -> None: + T = TypeVar("T", bound=model.Referable) + nss_types: List[Type[model.NamespaceSet]] = [model.NamespaceSet, model.OrderedNamespaceSet] + for nss_type in nss_types: + new_item = None + old_item = None + existing_items = [] + + class DummyNamespace(model.UniqueIdShortNamespace): + def __init__(self, items: Iterable[T], item_add_hook: Optional[Callable[[T, Iterable[T]], None]] = None, + item_id_set_hook: Optional[Callable[[T], None]] = None, + item_id_del_hook: Optional[Callable[[T], None]] = None): + super().__init__() + self.set1 = nss_type(self, [('id_short', True)], items, item_add_hook=item_add_hook, + item_id_set_hook=item_id_set_hook, + item_id_del_hook=item_id_del_hook) + + def add_hook(new: T, existing: Iterable[T]) -> None: + nonlocal new_item, existing_items + new_item = new + # Create a new list to prevent an error when checking the assertions: + # RuntimeError: dictionary changed size during iteration + existing_items = list(existing) + + def id_set_hook(new: T) -> None: + if new.id_short is not None: + new.id_short += "new" + + def id_del_hook(old: T) -> None: + nonlocal old_item + old_item = old + if old.id_short is not None: + # remove "new" suffix + old.id_short = old.id_short[:-3] + + cap = model.Capability("test_cap") + dummy_ns = DummyNamespace({cap}, item_add_hook=add_hook, item_id_set_hook=id_set_hook, + item_id_del_hook=id_del_hook) + self.assertEqual(cap.id_short, "test_capnew") + self.assertIs(new_item, cap) + self.assertEqual(len(existing_items), 0) + + mlp = model.MultiLanguageProperty("test_mlp") + dummy_ns.add_referable(mlp) + self.assertEqual(mlp.id_short, "test_mlpnew") + self.assertIs(new_item, mlp) + self.assertEqual(len(existing_items), 1) + self.assertIn(cap, existing_items) + + prop = model.Property("test_prop", model.datatypes.Int) + dummy_ns.set1.add(prop) + self.assertEqual(prop.id_short, "test_propnew") + self.assertIs(new_item, prop) + self.assertEqual(len(existing_items), 2) + self.assertIn(cap, existing_items) + self.assertIn(mlp, existing_items) + + dummy_ns.remove_referable("test_capnew") + self.assertIs(old_item, cap) + self.assertEqual(cap.id_short, "test_cap") + + dummy_ns.set1.remove(prop) + self.assertIs(old_item, prop) + self.assertEqual(prop.id_short, "test_prop") + + dummy_ns.set1.remove_by_id("id_short", "test_mlpnew") + self.assertIs(old_item, mlp) + self.assertEqual(mlp.id_short, "test_mlp") + + self.assertEqual(len(list(dummy_ns)), 0) + + # test atomicity + add_hook_counter: int = 0 + + def add_hook_constraint(_new: T, _existing: Iterable[T]) -> None: + nonlocal add_hook_counter + add_hook_counter += 1 + if add_hook_counter > 1: + raise ValueError + + self.assertEqual(cap.id_short, "test_cap") + self.assertEqual(mlp.id_short, "test_mlp") + with self.assertRaises(ValueError): + DummyNamespace([cap, mlp], item_add_hook=add_hook_constraint, item_id_set_hook=id_set_hook, + item_id_del_hook=id_del_hook) + self.assertEqual(cap.id_short, "test_cap") + self.assertIsNone(cap.parent) + self.assertEqual(mlp.id_short, "test_mlp") + self.assertIsNone(mlp.parent) + + dummy_ns = DummyNamespace((), item_add_hook=add_hook_constraint, item_id_set_hook=id_set_hook, + item_id_del_hook=id_del_hook) + add_hook_counter = 0 + dummy_ns.add_referable(cap) + self.assertIs(cap.parent, dummy_ns) + + with self.assertRaises(ValueError): + dummy_ns.set1.add(prop) + self.assertEqual(prop.id_short, "test_prop") + self.assertIsNone(prop.parent) def test_Namespace(self) -> None: with self.assertRaises(model.AASConstraintViolation) as cm: