model.base: add item_id_{set,del}_hook to NamespaceSet

These hooks allow dynamically assigning a value to the identifying
attribute of an object whenever it is added to a `NamespaceSet`
(`item_id_set_hook`). Furthermore, the `item_id_del_hook` allows
unsetting the identifying attribute whenever an object is removed
from a `NamespaceSet`.
This functionality is necessary for the implementation of
`SubmodelElementLists`, because there we need to store items without
an identifying attribute in a `NamespaceSet`.

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,13 +1818,20 @@ 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.parent = parent
  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)

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: