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