model.base: improve NamespaceSet

basyx/aas/adapter/json/aasJSONSchema.json

@@ -444,8 +444,12 @@
  "maxLength": 2000,
  "pattern": "^([\\t\\n\\r -\ud7ff\ue000-\ufffd]|\\ud800[\\udc00-\\udfff]|[\\ud801-\\udbfe][\\udc00-\\udfff]|\\udbff[\\udc00-\\udfff])*$"
  },
- "specificAssetId": {
- "$ref": "#/definitions/SpecificAssetId"
+ "specificAssetIds": {
+ "type": "array",
+ "items": {
+ "$ref": "#/definitions/SpecificAssetId"
+ },
+ "minItems": 1
  }
  },
  "required": [

basyx/aas/adapter/json/json_deserialization.py

@@ -417,13 +417,19 @@ def _construct_value_reference_pair(cls, dct: Dict[str, object],
  @classmethod
  def _construct_asset_information(cls, dct: Dict[str, object], object_class=model.AssetInformation)\
  -> model.AssetInformation:
- ret = object_class(asset_kind=ASSET_KIND_INVERSE[_get_ts(dct, 'assetKind', str)])
- cls._amend_abstract_attributes(ret, dct)
+ global_asset_id = None
  if 'globalAssetId' in dct:
- ret.global_asset_id = _get_ts(dct, 'globalAssetId', str)
+ global_asset_id = _get_ts(dct, 'globalAssetId', str)
+ specific_asset_id = set()
  if 'specificAssetIds' in dct:
  for desc_data in _get_ts(dct, "specificAssetIds", list):
- ret.specific_asset_id.add(cls._construct_specific_asset_id(desc_data, model.SpecificAssetId))
+ specific_asset_id.add(cls._construct_specific_asset_id(desc_data, model.SpecificAssetId))
+
+ ret = object_class(asset_kind=ASSET_KIND_INVERSE[_get_ts(dct, 'assetKind', str)],
+ global_asset_id=global_asset_id,
+ specific_asset_id=specific_asset_id)
+ cls._amend_abstract_attributes(ret, dct)
+
  if 'assetType' in dct:
  ret.asset_type = _get_ts(dct, 'assetType', str)
  if 'defaultThumbnail' in dct:
@@ -497,9 +503,10 @@ def _construct_entity(cls, dct: Dict[str, object], object_class=model.Entity) ->
  global_asset_id = None
  if 'globalAssetId' in dct:
  global_asset_id = _get_ts(dct, 'globalAssetId', str)
- specific_asset_id = None
+ specific_asset_id = set()
  if 'specificAssetIds' in dct:
- specific_asset_id = cls._construct_specific_asset_id(_get_ts(dct, 'specificAssetIds', dict))
+ for desc_data in _get_ts(dct, "specificAssetIds", list):
+ specific_asset_id.add(cls._construct_specific_asset_id(desc_data, model.SpecificAssetId))
 
  ret = object_class(id_short=None,
  entity_type=ENTITY_TYPES_INVERSE[_get_ts(dct, "entityType", str)],

basyx/aas/adapter/json/json_serialization.py

@@ -631,7 +631,7 @@ def _entity_to_json(cls, obj: model.Entity) -> Dict[str, object]:
  if obj.global_asset_id:
  data['globalAssetId'] = obj.global_asset_id
  if obj.specific_asset_id:
- data['specificAssetIds'] = obj.specific_asset_id
+ data['specificAssetIds'] = list(obj.specific_asset_id)
  return data
 
  @classmethod

basyx/aas/adapter/xml/AAS.xsd

@@ -284,7 +284,13 @@
  </xs:restriction>
  </xs:simpleType>
  </xs:element>
- <xs:element name="specificAssetId" type="specificAssetId_t" minOccurs="0" maxOccurs="1"/>
+ <xs:element name="specificAssetIds" minOccurs="0" maxOccurs="1">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="specificAssetId" type="specificAssetId_t" minOccurs="1" maxOccurs="unbounded"/>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
  </xs:sequence>
  </xs:group>
  <xs:group name="environment">

basyx/aas/adapter/xml/xml_deserialization.py

@@ -797,13 +797,17 @@ def construct_capability(cls, element: etree.Element, object_class=model.Capabil
 
  @classmethod
  def construct_entity(cls, element: etree.Element, object_class=model.Entity, **_kwargs: Any) -> model.Entity:
- global_asset_id = _get_text_or_none(element.find(NS_AAS + "globalAssetId"))
- specific_asset_id = _failsafe_construct(element.find(NS_AAS + "specificAssetId"),
- cls.construct_specific_asset_id, cls.failsafe)
+ specific_asset_id = set()
+ specific_assset_ids = element.find(NS_AAS + "specificAssetIds")
+ if specific_assset_ids is not None:
+ for id in _child_construct_multiple(specific_assset_ids, NS_AAS + "specificAssetId",
+ cls.construct_specific_asset_id, cls.failsafe):
+ specific_asset_id.add(id)
+
  entity = object_class(
  id_short=None,
  entity_type=_child_text_mandatory_mapped(element, NS_AAS + "entityType", ENTITY_TYPES_INVERSE),
- global_asset_id=global_asset_id,
+ global_asset_id=_get_text_or_none(element.find(NS_AAS + "globalAssetId")),
  specific_asset_id=specific_asset_id)
 
  if not cls.stripped:
@@ -994,17 +998,19 @@ def construct_specific_asset_id(cls, element: etree.Element, object_class=model.
  @classmethod
  def construct_asset_information(cls, element: etree.Element, object_class=model.AssetInformation, **_kwargs: Any) \
  -> model.AssetInformation:
- asset_information = object_class(
- _child_text_mandatory_mapped(element, NS_AAS + "assetKind", ASSET_KIND_INVERSE),
- )
- global_asset_id = _get_text_or_none(element.find(NS_AAS + "globalAssetId"))
- if global_asset_id is not None:
- asset_information.global_asset_id = global_asset_id
+ specific_asset_id = set()
  specific_assset_ids = element.find(NS_AAS + "specificAssetIds")
  if specific_assset_ids is not None:
  for id in _child_construct_multiple(specific_assset_ids, NS_AAS + "specificAssetId",
  cls.construct_specific_asset_id, cls.failsafe):
- asset_information.specific_asset_id.add(id)
+ specific_asset_id.add(id)
+
+ asset_information = object_class(
+ _child_text_mandatory_mapped(element, NS_AAS + "assetKind", ASSET_KIND_INVERSE),
+ global_asset_id=_get_text_or_none(element.find(NS_AAS + "globalAssetId")),
+ specific_asset_id=specific_asset_id,
+ )
+
  asset_type = _get_text_or_none(element.find(NS_AAS + "assetType"))
  if asset_type is not None:
  asset_information.asset_type = asset_type

basyx/aas/adapter/xml/xml_serialization.py

@@ -801,7 +801,10 @@ def entity_to_xml(obj: model.Entity,
  if obj.global_asset_id:
  et_entity.append(_generate_element(NS_AAS + "globalAssetId", text=obj.global_asset_id))
  if obj.specific_asset_id:
- et_entity.append(specific_asset_id_to_xml(obj.specific_asset_id, NS_AAS + "specificAssetId"))
+ et_specific_asset_id = _generate_element(name=NS_AAS + "specificAssetIds")
+ for specific_asset_id in obj.specific_asset_id:
+ et_specific_asset_id.append(specific_asset_id_to_xml(specific_asset_id, NS_AAS + "specificAssetId"))
+ et_entity.append(et_specific_asset_id)
  return et_entity
 
 

basyx/aas/examples/data/_helper.py

@@ -454,7 +454,7 @@ def _find_reference(self, object_: model.Reference, search_list: Iterable) -> Un
  return element
  return None
 
- def _find_specific_asset_id(self, object_: model.SpecificAssetId, search_list: Union[Set, List]) \
+ def _find_specific_asset_id(self, object_: model.SpecificAssetId, search_list: Iterable) \
  -> Union[model.SpecificAssetId, None]:
  """
  Find a SpecificAssetId in an list
@@ -604,25 +604,26 @@ def check_entity_equal(self, object_: model.Entity, expected_value: model.Entity
  self._check_abstract_attributes_submodel_element_equal(object_, expected_value)
  self.check_attribute_equal(object_, 'entity_type', expected_value.entity_type)
  self.check_attribute_equal(object_, 'global_asset_id', expected_value.global_asset_id)
- if object_.specific_asset_id and expected_value.specific_asset_id:
- self.check_specific_asset_id(object_.specific_asset_id, expected_value.specific_asset_id)
- else:
- if expected_value.specific_asset_id:
- self.check(expected_value.specific_asset_id is not None,
- 'SpecificAssetId {} must exist'.format(repr(expected_value.specific_asset_id)),
- value=object_.specific_asset_id)
- else:
- if object_.specific_asset_id:
- self.check(expected_value.specific_asset_id is None, 'Enity {} must not have a '
- 'specificAssetId'.format(repr(object_)),
- value=expected_value.specific_asset_id)
+ self._check_specific_asset_ids_equal(object_.specific_asset_id, expected_value.specific_asset_id, object_)
  self.check_contained_element_length(object_, 'statement', model.SubmodelElement, len(expected_value.statement))
  for expected_element in expected_value.statement:
  element = object_.get_referable(expected_element.id_short)
- self.check(element is not None, 'Entity {} must exist'.format(repr(expected_element)))
+ self.check(element is not None, f'Entity {repr(expected_element)} must exist')
 
  found_elements = self._find_extra_elements_by_id_short(object_.statement, expected_value.statement)
- self.check(found_elements == set(), 'Enity {} must not have extra statements'.format(repr(object_)),
+ self.check(found_elements == set(), f'Enity {repr(object_)} must not have extra statements',
+ value=found_elements)
+
+ def _check_specific_asset_ids_equal(self, object_: Iterable[model.SpecificAssetId],
+ expected_value: Iterable[model.SpecificAssetId],
+ object_parent):
+ for expected_pair in expected_value:
+ pair = self._find_specific_asset_id(expected_pair, object_)
+ if self.check(pair is not None, f'SpecificAssetId {repr(expected_pair)} must exist'):
+ self.check_specific_asset_id(pair, expected_pair) # type: ignore
+
+ found_elements = self._find_extra_object(object_, expected_value, model.SpecificAssetId)
+ self.check(found_elements == set(), f'{repr(object_parent)} must not have extra specificAssetIds',
  value=found_elements)
 
  def _check_event_element_equal(self, object_: model.EventElement, expected_value: model.EventElement):
@@ -722,16 +723,7 @@ def check_asset_information_equal(self, object_: model.AssetInformation, expecte
  self.check_attribute_equal(object_, 'global_asset_id', expected_value.global_asset_id)
  self.check_contained_element_length(object_, 'specific_asset_id', model.SpecificAssetId,
  len(expected_value.specific_asset_id))
- for expected_pair in expected_value.specific_asset_id:
- pair = self._find_specific_asset_id(expected_pair, object_.specific_asset_id)
- if self.check(pair is not None, 'SpecificAssetId {} must exist'.format(repr(expected_pair))):
- self.check_specific_asset_id(pair, expected_pair) # type: ignore
-
- found_elements = self._find_extra_object(object_.specific_asset_id, expected_value.specific_asset_id,
- model.SpecificAssetId)
- self.check(found_elements == set(), '{} must not have extra '
- 'specificAssetIds'.format(repr(object_)),
- value=found_elements)
+ self._check_specific_asset_ids_equal(object_.specific_asset_id, expected_value.specific_asset_id, object_)
  self.check_attribute_equal(object_, 'asset_type', object_.asset_type)
  if object_.default_thumbnail and expected_value.default_thumbnail:
  self.check_resource_equal(object_.default_thumbnail, expected_value.default_thumbnail)

basyx/aas/examples/data/example_aas.py

@@ -244,11 +244,12 @@ def create_example_bill_of_material_submodel() -> model.Submodel:
  entity_type=model.EntityType.SELF_MANAGED_ENTITY,
  statement={submodel_element_property, submodel_element_property2},
  global_asset_id='http://acplt.org/TestAsset/',
- specific_asset_id=model.SpecificAssetId(name="TestKey",
- value="TestValue",
- external_subject_id=model.ExternalReference(
- (model.Key(type_=model.KeyTypes.GLOBAL_REFERENCE,
- value='http://acplt.org/SpecificAssetId/'),))),
+ specific_asset_id={
+ model.SpecificAssetId(name="TestKey", value="TestValue",
+ external_subject_id=model.ExternalReference(
+ (model.Key(type_=model.KeyTypes.GLOBAL_REFERENCE,
+ value='http://acplt.org/SpecificAssetId/'),))
+ )},
  category="PARAMETER",
  description=model.MultiLanguageTextType({
  'en-US': 'Legally valid designation of the natural or judicial person which '
@@ -276,7 +277,7 @@ def create_example_bill_of_material_submodel() -> model.Submodel:
  entity_type=model.EntityType.CO_MANAGED_ENTITY,
  statement=(),
  global_asset_id=None,
- specific_asset_id=None,
+ specific_asset_id=(),
  category="PARAMETER",
  description=model.MultiLanguageTextType({
  'en-US': 'Legally valid designation of the natural or judicial person which '

basyx/aas/examples/data/example_aas_mandatory_attributes.py

@@ -201,7 +201,8 @@ def create_example_empty_asset_administration_shell() -> model.AssetAdministrati
  :return: example asset administration shell
  """
  asset_administration_shell = model.AssetAdministrationShell(
- asset_information=model.AssetInformation(),
+ asset_information=model.AssetInformation(
+ global_asset_id='http://acplt.org/TestAsset2_Mandatory/'),
  id_='https://acplt.org/Test_AssetAdministrationShell2_Mandatory')
  return asset_administration_shell
 

basyx/aas/examples/tutorial_serialization_deserialization.py

@@ -47,7 +47,7 @@
 )
 aashell = model.AssetAdministrationShell(
  id_='https://acplt.org/Simple_AAS',
- asset_information=model.AssetInformation(),
+ asset_information=model.AssetInformation(global_asset_id="test"),
  submodel={model.ModelReference.from_referable(submodel)}
 )
 

basyx/aas/model/aas.py

@@ -29,6 +29,8 @@ class AssetInformation:
  identifiers. However, to support the corner case of very first phase of lifecycle where a stabilised/constant
  global asset identifier does not already exist, the corresponding attribute “globalAssetId” is optional.
 
+ *Constraint AASd-131*: The globalAssetId or at least one specificAssetId shall be defined for AssetInformation.
+
  :ivar asset_kind: Denotes whether the Asset is of :class:`~aas.model.base.AssetKind` "TYPE" or "INSTANCE".
  Default is "INSTANCE".
  :ivar global_asset_id: :class:`~aas.model.base.Identifier` modelling the identifier of the asset the AAS is
@@ -52,30 +54,65 @@ class AssetInformation:
  def __init__(self,
  asset_kind: base.AssetKind = base.AssetKind.INSTANCE,
  global_asset_id: Optional[base.Identifier] = None,
- specific_asset_id: Optional[Set[base.SpecificAssetId]] = None,
+ specific_asset_id: Iterable[base.SpecificAssetId] = (),
  asset_type: Optional[base.Identifier] = None,
  default_thumbnail: Optional[base.Resource] = None):
 
  super().__init__()
  self.asset_kind: base.AssetKind = asset_kind
- self._global_asset_id: Optional[base.Identifier] = global_asset_id
- self.specific_asset_id: Set[base.SpecificAssetId] = set() if specific_asset_id is None \
- else specific_asset_id
  self.asset_type: Optional[base.Identifier] = asset_type
  self.default_thumbnail: Optional[base.Resource] = default_thumbnail
+ # assign private attributes, bypassing setters, as constraints will be checked below
+ self._specific_asset_id: base.ConstrainedList[base.SpecificAssetId] = base.ConstrainedList(
+ specific_asset_id,
+ item_set_hook=self._check_constraint_set_spec_asset_id,
+ item_del_hook=self._check_constraint_del_spec_asset_id
+ )
+ self._global_asset_id: Optional[base.Identifier] = global_asset_id
+ self._validate_global_asset_id(global_asset_id)
+ self._validate_aasd_131(global_asset_id, bool(specific_asset_id))
 
- def _get_global_asset_id(self):
+ @property
+ def global_asset_id(self) -> Optional[base.Identifier]:
  return self._global_asset_id
 
- def _set_global_asset_id(self, global_asset_id: Optional[base.Identifier]):
- if global_asset_id is None:
- if self.specific_asset_id is None or not self.specific_asset_id:
- raise ValueError("either global or specific asset id must be set")
- else:
- _string_constraints.check_identifier(global_asset_id)
+ @global_asset_id.setter
+ def global_asset_id(self, global_asset_id: Optional[base.Identifier]) -> None:
+ self._validate_global_asset_id(global_asset_id)
+ self._validate_aasd_131(global_asset_id, bool(self.specific_asset_id))
  self._global_asset_id = global_asset_id
 
- global_asset_id = property(_get_global_asset_id, _set_global_asset_id)
+ @property
+ def specific_asset_id(self) -> base.ConstrainedList[base.SpecificAssetId]:
+ return self._specific_asset_id
+
+ @specific_asset_id.setter
+ def specific_asset_id(self, specific_asset_id: Iterable[base.SpecificAssetId]) -> None:
+ # constraints are checked via _check_constraint_set_spec_asset_id() in this case
+ self._specific_asset_id[:] = specific_asset_id
+
+ def _check_constraint_set_spec_asset_id(self, items_to_replace: List[base.SpecificAssetId],
+ new_items: List[base.SpecificAssetId],
+ old_list: List[base.SpecificAssetId]) -> None:
+ self._validate_aasd_131(self.global_asset_id,
+ len(old_list) - len(items_to_replace) + len(new_items) > 0)
+
+ def _check_constraint_del_spec_asset_id(self, _item_to_del: base.SpecificAssetId,
+ old_list: List[base.SpecificAssetId]) -> None:
+ self._validate_aasd_131(self.global_asset_id, len(old_list) > 1)
+
+ @staticmethod
+ def _validate_global_asset_id(global_asset_id: Optional[base.Identifier]) -> None:
+ if global_asset_id is not None:
+ _string_constraints.check_identifier(global_asset_id)
+
+ @staticmethod
+ def _validate_aasd_131(global_asset_id: Optional[base.Identifier], specific_asset_id_nonempty: bool) -> None:
+ if global_asset_id is None and not specific_asset_id_nonempty:
+ raise base.AASConstraintViolation(131,
+ "An AssetInformation has to have a globalAssetId or a specificAssetId")
+ if global_asset_id is not None:
+ _string_constraints.check_identifier(global_asset_id)
 
  def __repr__(self) -> str:
  return "AssetInformation(assetKind={}, globalAssetId={}, specificAssetId={}, assetType={}, " \