Coverage for filip/models/ngsi_v2/context.py: 95%
191 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-02-19 11:48 +0000
1"""
2NGSIv2 models for context broker interaction
3"""
5import json
6from typing import Any, List, Dict, Union, Optional, Set, Tuple
8from aenum import Enum
9from pydantic import (
10 field_validator,
11 ConfigDict,
12 BaseModel,
13 Field,
14 model_validator,
15 SerializeAsAny,
16)
17from pydantic_core.core_schema import ValidationInfo
18from pydantic.types import OnErrorOmit
19from filip.models.ngsi_v2.base import (
20 EntityPattern,
21 Expression,
22 BaseAttribute,
23 BaseValueAttribute,
24 BaseNameAttribute,
25)
26from filip.models.base import DataType
27from filip.utils.validators import (
28 validate_fiware_datatype_standard,
29 validate_fiware_datatype_string_protect,
30 validate_fiware_attribute_value_regex,
31 validate_fiware_attribute_name_regex,
32)
35class GetEntitiesOptions(str, Enum):
36 """Options for queries"""
38 _init_ = "value __doc__"
40 NORMALIZED = "normalized", "Normalized message representation"
41 KEY_VALUES = (
42 "keyValues",
43 "Key value message representation."
44 "This mode represents the entity "
45 "attributes by their values only, leaving out "
46 "the information about type and metadata. "
47 "See example "
48 "below."
49 "Example: "
50 "{"
51 " 'id': 'R12345',"
52 " 'type': 'Room',"
53 " 'temperature': 22"
54 "}",
55 )
56 VALUES = (
57 "values",
58 "Key value message representation. "
59 "This mode represents the entity as an array of "
60 "attribute values. Information about id and type is "
61 "left out. See example below. The order of the "
62 "attributes in the array is specified by the attrs "
63 "URI param (e.g. attrs=branch,colour,engine). "
64 "If attrs is not used, the order is arbitrary. "
65 "Example:"
66 "[ 'Ford', 'black', 78.3 ]",
67 )
68 UNIQUE = (
69 "unique",
70 "unique mode. This mode is just like values mode, "
71 "except that values are not repeated",
72 )
75class PropertyFormat(str, Enum):
76 """
77 Format to decide if properties of ContextEntity class are returned as
78 List of NamedContextAttributes or as Dict of ContextAttributes.
79 """
81 LIST = "list"
82 DICT = "dict"
85class ContextAttribute(BaseAttribute, BaseValueAttribute):
86 """
87 Model for an attribute is represented by a JSON object with the following
88 syntax:
90 The attribute value is specified by the value property, whose value may
91 be any JSON datatype.
93 The attribute NGSI type is specified by the type property, whose value
94 is a string containing the NGSI type.
96 The attribute metadata is specified by the metadata property. Its value
97 is another JSON object which contains a property per metadata element
98 defined (the name of the property is the name of the metadata element).
99 Each metadata element, in turn, is represented by a JSON object
100 containing the following properties:
102 Values of entity attributes. For adding it you need to nest it into a
103 dict in order to give it a name.
105 Example:
107 >>> data = {"value": <...>,
108 "type": <...>,
109 "metadata": <...>}
110 >>> attr = ContextAttribute(**data)
112 """
114 # although `type` is a required field in the NGSIv2 specification, it is
115 # set to optional here to allow for the possibility of setting
116 # default-types in child classes. Pydantic will raise the correct error
117 # and also exports the correct json-schema.
118 def __init__(self, type: str = None, **data):
119 if type is None and self.model_fields["type"].default:
120 type = self.model_fields["type"].default
121 super().__init__(type=type, **data)
124class NamedContextAttribute(ContextAttribute, BaseNameAttribute):
125 """
126 Context attributes are properties of context entities. For example, the
127 current speed of a car could be modeled as attribute current_speed of entity
128 car-104.
130 In the NGSI data model, attributes have an attribute name, an attribute type
131 an attribute value and metadata.
132 """
134 pass
137class ContextEntityKeyValues(BaseModel):
138 """
139 Base Model for an entity is represented by a JSON object with the following
140 syntax.
142 The entity id is specified by the object's id property, whose value
143 is a string containing the entity id.
145 The entity type is specified by the object's type property, whose value
146 is a string containing the entity's type name.
148 """
150 model_config = ConfigDict(
151 extra="allow", validate_default=True, validate_assignment=True
152 )
153 id: str = Field(
154 ...,
155 title="Entity Id",
156 description="Id of an entity in an NGSI context broker. Allowed "
157 "characters are the ones in the plain ASCII set, except "
158 "the following ones: control characters, "
159 "whitespace, &, ?, / and #.",
160 json_schema_extra={"example": "Bcn-Welt"},
161 max_length=256,
162 min_length=1,
163 frozen=True,
164 )
165 valid_id = field_validator("id")(validate_fiware_datatype_standard)
166 type: Union[str, Enum] = Field(
167 ...,
168 title="Entity Type",
169 description="Id of an entity in an NGSI context broker. "
170 "Allowed characters are the ones in the plain ASCII set, "
171 "except the following ones: control characters, "
172 "whitespace, &, ?, / and #.",
173 json_schema_extra={"example": "Room"},
174 max_length=256,
175 min_length=1,
176 frozen=True,
177 )
178 valid_type = field_validator("type")(validate_fiware_datatype_standard)
180 # although `type` is a required field in the NGSIv2 specification, it is
181 # set to optional here to allow for the possibility of setting
182 # default-types in child classes. Pydantic will raise the correct error
183 # and also exports the correct json-schema.
184 def __init__(self, id: str, type: Union[str, Enum] = None, **data):
185 # this allows to set the type of the entity in child classes
186 if type is None:
187 if isinstance(self.model_fields["type"].default, str):
188 type = self.model_fields["type"].default
189 else:
190 # if this statement is reached not proper default-value for
191 # `type` was found and pydantic will raise the correct error
192 super().__init__(id=id, **data)
193 # This will result in usual behavior
194 data.update(self._validate_attributes(data))
195 super().__init__(id=id, type=type, **data)
197 # Validation of attributes
198 @classmethod
199 def _validate_attributes(cls, data: dict):
200 """
201 Validate attribute name and value of the entity in keyvalues format
202 """
203 for attr_name, attr_value in data.items():
204 if isinstance(attr_value, str):
205 validate_fiware_attribute_value_regex(attr_value)
206 validate_fiware_attribute_name_regex(attr_name)
207 return data
209 def get_attributes(self) -> dict:
210 """
211 Get the attribute of the entity with the given name in
212 dict format
214 Returns:
215 dict
216 """
217 return self.model_dump(exclude={"id", "type"})
220class ContextEntity(ContextEntityKeyValues):
221 """
222 Context entities, or simply entities, are the center of gravity in the
223 FIWARE NGSI information model. An entity represents a thing, i.e., any
224 physical or logical object (e.g., a sensor, a person, a room, an issue in
225 a ticketing system, etc.). Each entity has an entity id.
226 Furthermore, the type system of FIWARE NGSI enables entities to have an
227 entity type. Entity types are semantic types; they are intended to describe
228 the type of thing represented by the entity. For example, a context
229 entity #with id sensor-365 could have the type temperatureSensor.
231 Each entity is uniquely identified by the combination of its id and type.
233 The entity id is specified by the object's id property, whose value
234 is a string containing the entity id.
236 The entity type is specified by the object's type property, whose value
237 is a string containing the entity's type name.
239 Entity attributes are specified by additional properties, whose names are
240 the name of the attribute and whose representation is described by the
241 "ContextAttribute"-model. Obviously, `id` and `type` are
242 not allowed as attribute names.
244 Example::
246 >>> data = {'id': 'MyId',
247 'type': 'MyType',
248 'my_attr': {'value': 20, 'type': 'Number'}}
250 >>> entity = ContextEntity(**data)
252 """
254 model_config = ConfigDict(
255 extra="allow", validate_default=True, validate_assignment=True
256 )
258 # although `type` is a required field in the NGSIv2 specification, it is
259 # set to optional here to allow for the possibility of setting
260 # default-types in child classes. Pydantic will raise the correct error
261 # and also exports the correct json-schema.
262 def __init__(self, id: str, type: str = None, **data):
263 # There is currently no validation for extra fields
264 data.update(self._validate_attributes(data))
265 # case where type is None to raise correct error message
266 if type is None:
267 super().__init__(id=id, **data)
268 else:
269 super().__init__(id=id, type=type, **data)
271 # Validation of attributes
272 @classmethod
273 def _validate_attributes(cls, data: dict):
274 """
275 Validate attributes of the entity if the attribute is not a model
276 field and the type is not already a subtype of ContextAttribute
277 """
278 attrs = {
279 key: ContextAttribute.model_validate(attr)
280 for key, attr in data.items()
281 if (
282 # validate_fiware_attribute_value_regex(key) not in cls.model_fields
283 validate_fiware_attribute_name_regex(key) not in cls.model_fields
284 and not isinstance(attr, ContextAttribute)
285 # key not in cls.model_fields
286 # and not isinstance(attr, ContextAttribute)
287 )
288 }
290 return attrs
292 @field_validator("*")
293 @classmethod
294 def check_attributes(cls, value, info: ValidationInfo):
295 """
296 Check whether all model fields are of subtype of ContextAttribute to
297 ensure full functionality.
298 """
299 if info.field_name in ["id", "type"]:
300 return value
302 if info.field_name in cls.model_fields:
303 if not (
304 isinstance(value, ContextAttribute)
305 or value == cls.model_fields[info.field_name].default
306 ):
307 raise ValueError(
308 f"Attribute {info.field_name} must be a of "
309 f"type or subtype ContextAttribute"
310 )
311 return value
313 @model_validator(mode="after")
314 @classmethod
315 def check_attributes_after(cls, values):
316 try:
317 for attr in values.model_extra:
318 if not isinstance(values.__getattr__(attr), ContextAttribute):
319 raise ValueError(
320 f"Attribute {attr} must be a of type or "
321 f"subtype ContextAttribute. You most "
322 f"likely tried to directly assign an "
323 f"attribute without converting it to a "
324 f"proper Attribute-Type!"
325 )
326 except TypeError:
327 pass
328 return values
330 # API for attributes and commands
331 def add_attributes(
332 self, attrs: Union[Dict[str, ContextAttribute], List[NamedContextAttribute]]
333 ) -> None:
334 """
335 Add attributes (properties, relationships) to entity
337 Args:
338 attrs: Dict[str, ContextAttribute]: {NAME for attr : Attribute} or
339 List[NamedContextAttribute]
341 Returns:
342 None
343 """
344 if isinstance(attrs, list):
345 attrs = {
346 attr.name: ContextAttribute(**attr.model_dump(exclude={"name"}))
347 for attr in attrs
348 }
349 for key, attr in attrs.items():
350 self.__setattr__(name=key, value=attr)
352 def get_attributes(
353 self,
354 whitelisted_attribute_types: Optional[List[DataType]] = None,
355 blacklisted_attribute_types: Optional[List[DataType]] = None,
356 response_format: Union[str, PropertyFormat] = PropertyFormat.LIST,
357 strict_data_type: bool = True,
358 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]:
359 """
360 Get attributes or a subset from the entity.
362 Args:
363 whitelisted_attribute_types: Optional list, if given only
364 attributes matching one of the types are returned
365 blacklisted_attribute_types: Optional list, if given all
366 attributes are returned that do not match a list entry
367 response_format: Wanted result format,
368 List -> list of NamedContextAttributes
369 Dict -> dict of {name: ContextAttribute}
370 strict_data_type: whether to restrict the data type to pre-defined
371 types, True by default.
372 True -> Only return the attributes with pre-defined types,
373 False -> Do not restrict the data type.
374 Raises:
375 AssertionError, if both a white and a black list is given
376 Returns:
377 List[NamedContextAttribute] or Dict[str, ContextAttribute]
378 """
380 response_format = PropertyFormat(response_format)
382 assert (
383 whitelisted_attribute_types is None or blacklisted_attribute_types is None
384 ), "Only whitelist or blacklist is allowed"
386 if whitelisted_attribute_types is not None:
387 attribute_types = whitelisted_attribute_types
388 elif blacklisted_attribute_types is not None:
389 attribute_types = [
390 att_type
391 for att_type in list(DataType)
392 if att_type not in blacklisted_attribute_types
393 ]
394 else:
395 attribute_types = [att_type for att_type in list(DataType)]
397 if response_format == PropertyFormat.DICT:
398 if strict_data_type:
399 return {
400 key: ContextAttribute(**value)
401 for key, value in self.model_dump().items()
402 if key not in ContextEntity.model_fields
403 and value.get("type") in [att.value for att in attribute_types]
404 }
405 else:
406 return {
407 key: ContextAttribute(**value)
408 for key, value in self.model_dump().items()
409 if key not in ContextEntity.model_fields
410 }
411 else:
412 if strict_data_type:
413 return [
414 NamedContextAttribute(name=key, **value)
415 for key, value in self.model_dump().items()
416 if key not in ContextEntity.model_fields
417 and value.get("type") in [att.value for att in attribute_types]
418 ]
419 else:
420 return [
421 NamedContextAttribute(name=key, **value)
422 for key, value in self.model_dump().items()
423 if key not in ContextEntity.model_fields
424 ]
426 def update_attribute(
427 self, attrs: Union[Dict[str, ContextAttribute], List[NamedContextAttribute]]
428 ) -> None:
429 """
430 Update attributes of an entity. Overwrite the current held value
431 for the attribute with the value contained in the corresponding given
432 attribute
434 Args:
435 attrs: List of NamedContextAttributes,
436 Dict of {attribute_name: ContextAttribute}
437 Raises:
438 NameError, if the attribute does not currently exist in the entity
439 Returns:
440 None
441 """
442 if isinstance(attrs, list):
443 attrs = {
444 attr.name: ContextAttribute(**attr.model_dump(exclude={"name"}))
445 for attr in attrs
446 }
448 existing_attribute_names = self.get_attribute_names()
449 for key, attr in attrs.items():
450 if key not in existing_attribute_names:
451 raise NameError
452 self.__setattr__(name=key, value=attr)
454 def get_attribute_names(self) -> Set[str]:
455 """
456 Returns a set with all attribute names of this entity
458 Returns:
459 Set[str]
460 """
462 return {
463 key for key in self.model_dump() if key not in ContextEntity.model_fields
464 }
466 def delete_attributes(
467 self,
468 attrs: Union[
469 Dict[str, ContextAttribute], List[NamedContextAttribute], List[str]
470 ],
471 ):
472 """
473 Delete the given attributes from the entity
475 Args:
476 attrs: - Dict {name: ContextAttribute}
477 - List[NamedContextAttribute]
478 - List[str] -> names of attributes
479 Raises:
480 Exception: if one of the given attrs does not represent an
481 existing argument
482 """
484 names: List[str] = []
485 if isinstance(attrs, list):
486 for entry in attrs:
487 if isinstance(entry, str):
488 names.append(entry)
489 elif isinstance(entry, NamedContextAttribute):
490 names.append(entry.name)
491 else:
492 names.extend(list(attrs.keys()))
493 for name in names:
494 delattr(self, name)
496 def get_attribute(self, attribute_name: str) -> NamedContextAttribute:
497 """
498 Get the attribute of the entity with the given name
500 Args:
501 attribute_name (str): Name of attribute
503 Raises:
504 KeyError, if no attribute with given name exists
506 Returns:
507 NamedContextAttribute
508 """
509 for attr in self.get_attributes():
510 if attr.name == attribute_name:
511 return attr
512 raise KeyError(f"Attribute '{attribute_name}' not in entity")
514 def get_properties(
515 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
516 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]:
517 """
518 Returns all attributes of the entity that are not of type Relationship,
519 and are not auto generated command attributes
521 Args:
522 response_format: Wanted result format,
523 List -> list of NamedContextAttributes
524 Dict -> dict of {name: ContextAttribute}
526 Returns:
527 [NamedContextAttribute] or {name: ContextAttribute}
528 """
529 pre_filtered_attrs = self.get_attributes(
530 blacklisted_attribute_types=[DataType.RELATIONSHIP],
531 response_format=PropertyFormat.LIST,
532 )
534 all_command_attributes_names = set()
535 for command in self.get_commands():
536 (c, c_status, c_info) = self.get_command_triple(command.name)
537 all_command_attributes_names.update([c.name, c_status.name, c_info.name])
539 property_attributes = []
540 for attr in pre_filtered_attrs:
541 if attr.name not in all_command_attributes_names:
542 property_attributes.append(attr)
544 if response_format == PropertyFormat.LIST:
545 return property_attributes
546 else:
547 return {
548 p.name: ContextAttribute(**p.model_dump(exclude={"name"}))
549 for p in property_attributes
550 }
552 def get_relationships(
553 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
554 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]:
555 """
556 Get all relationships of the context entity
558 Args:
559 response_format: Wanted result format,
560 List -> list of NamedContextAttributes
561 Dict -> dict of {name: ContextAttribute}
563 Returns:
564 [NamedContextAttribute] or {name: ContextAttribute}
566 """
567 return self.get_attributes(
568 whitelisted_attribute_types=[DataType.RELATIONSHIP],
569 response_format=response_format,
570 )
572 def get_commands(
573 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
574 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]:
575 """
576 Get all commands of the context entity. Only works if the commands
577 were autogenerated by Fiware from an Device.
579 Args:
580 response_format: Wanted result format,
581 List -> list of NamedContextAttributes
582 Dict -> dict of {name: ContextAttribute}
584 Returns:
585 [NamedContextAttribute] or {name: ContextAttribute}
586 """
588 # if an attribute with name n is a command, its type does not need to
589 # be COMMAND.
590 # But the attributes name_info (type: commandResult) and
591 # name_status(type: commandStatus) need to exist. (Autogenerated)
593 # Search all attributes of type commandStatus, check for each if a
594 # corresponding _info exists and if also a fitting attribute exists
595 # we know: that is a command.
597 commands = []
598 for status_attribute in self.get_attributes(
599 whitelisted_attribute_types=[DataType.COMMAND_STATUS]
600 ):
601 if not status_attribute.name.split("_")[-1] == "status":
602 continue
603 base_name = status_attribute.name[:-7]
605 try:
606 info_attribute = self.get_attribute(f"{base_name}_info")
607 if not info_attribute.type == DataType.COMMAND_RESULT:
608 continue
610 attribute = self.get_attribute(base_name)
611 commands.append(attribute)
612 except KeyError:
613 continue
615 if response_format == PropertyFormat.LIST:
616 return commands
617 else:
618 return {
619 cmd.name: ContextAttribute(**cmd.model_dump(exclude={"name"}))
620 for cmd in commands
621 }
623 def get_command_triple(
624 self, command_attribute_name: str
625 ) -> Tuple[NamedContextAttribute, NamedContextAttribute, NamedContextAttribute]:
626 """
627 Returns for a given command attribute name all three corresponding
628 attributes as triple
630 Args:
631 command_attribute_name: Name of the command attribute
633 Raises:
634 KeyError, if the given name does not belong to a command attribute
636 Returns:
637 (Command, Command_status, Command_info)
638 """
640 commands = self.get_commands(response_format=PropertyFormat.DICT)
642 if command_attribute_name not in commands:
643 raise KeyError(f"Command '{command_attribute_name}' not in commands")
645 command = self.get_attribute(command_attribute_name)
647 # as the given name was found as a valid command, we know that the
648 # status and info attributes exist correctly
649 command_status = self.get_attribute(f"{command_attribute_name}_status")
650 command_info = self.get_attribute(f"{command_attribute_name}_info")
652 return command, command_status, command_info
655class ContextEntityList(BaseModel):
656 """
657 Collection model for a list of context entities
658 """
660 entities: List[OnErrorOmit[ContextEntity]]
663class ContextEntityKeyValuesList(BaseModel):
664 """
665 Collection model for a list of context entities in key-values format
666 """
668 entities: List[OnErrorOmit[ContextEntityKeyValues]]
671class Query(BaseModel):
672 """
673 Model for queries
674 """
676 entities: List[EntityPattern] = Field(
677 description="a list of entities to search for. Each element is "
678 "represented by a JSON object"
679 )
680 attrs: Optional[List[str]] = Field(
681 default=None,
682 description="List of attributes to be provided "
683 "(if not specified, all attributes).",
684 )
685 expression: Optional[Expression] = Field(
686 default=None,
687 description="An expression composed of q, mq, georel, geometry and " "coords",
688 )
689 metadata: Optional[List[str]] = Field(
690 default=None,
691 description="a list of metadata names to include in the response. "
692 'See "Filtering out attributes and metadata" section for '
693 "more detail.",
694 )
697class ActionType(str, Enum):
698 """
699 Options for queries
700 """
702 _init_ = "value __doc__"
703 APPEND = (
704 "append",
705 "maps to POST /v2/entities (if the entity does not "
706 "already exist) or POST /v2/entities/<id>/attrs (if "
707 "the entity already exists). ",
708 )
709 APPEND_STRICT = (
710 "appendStrict",
711 "maps to POST /v2/entities (if the "
712 "entity does not already exist) or POST "
713 "/v2/entities/<id>/attrs?options=append "
714 "(if the entity already exists).",
715 )
716 UPDATE = "update", "maps to PATCH /v2/entities/<id>/attrs."
717 DELETE = (
718 "delete",
719 "maps to DELETE /v2/entities/<id>/attrs/<attrName> on "
720 "every attribute included in the entity or to DELETE "
721 "/v2/entities/<id> if no attribute were included in "
722 "the entity.",
723 )
724 REPLACE = "replace", "maps to PUT /v2/entities/<id>/attrs"
727class Update(BaseModel):
728 """
729 Model for update action
730 """
732 action_type: Union[ActionType, str] = Field(
733 alias="actionType",
734 description="actionType, to specify the kind of update action to do: "
735 "either append, appendStrict, update, delete, or replace. ",
736 )
737 entities: SerializeAsAny[List[Union[ContextEntity, ContextEntityKeyValues]]] = (
738 Field(
739 description="an array of entities, each entity specified using the "
740 "JSON entity representation format "
741 )
742 )
744 @field_validator("action_type")
745 @classmethod
746 def check_action_type(cls, action):
747 """
748 validates action_type
749 Args:
750 action: field action_type
751 Returns:
752 action_type
753 """
754 return ActionType(action)
757class Command(BaseModel):
758 """
759 Class for sending commands to IoT Devices.
760 Note that the command must be registered via an IoT-Agent. Internally
761 FIWARE uses its registration mechanism in order to connect the command
762 with an IoT-Device
763 """
765 type: DataType = Field(
766 default=DataType.COMMAND,
767 description="Command must have the type command",
768 # const=True
769 )
770 value: Any = Field(
771 description="Any json serializable command that will "
772 "be forwarded to the connected IoT device"
773 )
775 @field_validator("value")
776 @classmethod
777 def check_value(cls, value):
778 """
779 Check if value is json serializable
780 Args:
781 value: value field
782 Returns:
783 value
784 """
785 try:
786 json.dumps(value)
787 except:
788 raise ValueError(f"Command value {value} " f"is not serializable")
789 return value
792class NamedCommand(Command):
793 """
794 Class for sending command to IoT-Device.
795 Extend :class: Command with command Name
796 """
798 name: str = Field(
799 description="Name of the command",
800 max_length=256,
801 min_length=1,
802 )
803 valid_name = field_validator("name")(validate_fiware_attribute_name_regex)