Coverage for filip/models/ngsi_ld/context.py: 91%
235 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-02-19 11:48 +0000
1"""
2NGSI LD models for context broker interaction
3"""
5import logging
6from typing import Any, List, Dict, Union, Optional
7from geojson_pydantic import (
8 Point,
9 MultiPoint,
10 LineString,
11 MultiLineString,
12 Polygon,
13 MultiPolygon,
14)
15from typing_extensions import Self
16from aenum import Enum
17from pydantic import field_validator, ConfigDict, BaseModel, Field, model_validator
18from filip.models.ngsi_v2 import ContextEntity
19from filip.utils.validators import (
20 FiwareRegex,
21 validate_fiware_datatype_string_protect,
22 validate_fiware_standard_regex,
23)
24from pydantic_core import ValidationError
27class DataTypeLD(str, Enum):
28 """
29 In NGSI-LD the data types on context entities are only divided into properties and relationships.
30 """
32 _init_ = "value __doc__"
33 GEOPROPERTY = "GeoProperty", "A property that represents a geometry value"
34 PROPERTY = "Property", "All attributes that do not represent a relationship"
35 RELATIONSHIP = (
36 "Relationship",
37 "Reference to another context entity, which can be identified with a URN.",
38 )
41# NGSI-LD entity models
42class ContextProperty(BaseModel):
43 """
44 The model for a property is represented by a JSON object with the following syntax:
46 The attribute value is specified by the value, whose value can be any data type. This does not need to be
47 specified further.
49 The NGSI type of the attribute is fixed and does not need to be specified.
50 Example:
52 >>> data = {"value": <...>}
54 >>> attr = ContextProperty(**data)
56 """
58 model_config = ConfigDict(extra="allow") # In order to allow nested properties
59 type: Optional[str] = Field(default="Property", title="type", frozen=True)
60 value: Optional[
61 Union[
62 Union[float, int, bool, str, List, Dict[str, Any]],
63 List[Union[float, int, bool, str, List, Dict[str, Any]]],
64 ]
65 ] = Field(default=None, title="Property value", description="the actual data")
66 observedAt: Optional[str] = Field(
67 None,
68 title="Timestamp",
69 description="Representing a timestamp for the "
70 "incoming value of the property.",
71 max_length=256,
72 min_length=1,
73 )
74 field_validator("observedAt")(validate_fiware_datatype_string_protect)
76 createdAt: Optional[str] = Field(
77 None,
78 title="Timestamp",
79 description="Representing a timestamp for the "
80 "creation time of the property.",
81 max_length=256,
82 min_length=1,
83 )
84 field_validator("createdAt")(validate_fiware_datatype_string_protect)
86 modifiedAt: Optional[str] = Field(
87 None,
88 title="Timestamp",
89 description="Representing a timestamp for the "
90 "last modification of the property.",
91 max_length=256,
92 min_length=1,
93 )
94 field_validator("modifiedAt")(validate_fiware_datatype_string_protect)
96 UnitCode: Optional[str] = Field(
97 None,
98 title="Unit Code",
99 description="Representing the unit of the value. "
100 "Should be part of the defined units "
101 "by the UN/ECE Recommendation No. 21"
102 "https://unece.org/fileadmin/DAM/cefact/recommendations/rec20/rec20_rev3_Annex2e.pdf ",
103 max_length=256,
104 min_length=1,
105 )
106 field_validator("UnitCode")(validate_fiware_datatype_string_protect)
108 datasetId: Optional[str] = Field(
109 None,
110 title="dataset Id",
111 description="It allows identifying a set or group of property values",
112 max_length=256,
113 min_length=1,
114 )
115 field_validator("datasetId")(validate_fiware_datatype_string_protect)
117 @classmethod
118 def get_model_fields_set(cls):
119 """
120 Get all names and aliases of the model fields.
121 """
122 return set(
123 [field.validation_alias for (_, field) in cls.model_fields.items()]
124 + [field_name for field_name in cls.model_fields]
125 )
127 @field_validator("type")
128 @classmethod
129 def check_property_type(cls, value):
130 """
131 Force property type to be "Property"
132 Args:
133 value: value field
134 Returns:
135 value
136 """
137 valid_property_types = ["Property", "Relationship", "TemporalProperty"]
138 if value not in valid_property_types:
139 msg = (
140 f"NGSI_LD Properties must have type {valid_property_types}, "
141 f'not "{value}"'
142 )
143 logging.warning(msg=msg)
144 raise ValueError(msg)
145 return value
148class NamedContextProperty(ContextProperty):
149 """
150 Context properties are properties of context entities. For example, the current speed of a car could be modeled
151 as the current_speed property of the car-104 entity.
153 In the NGSI-LD data model, properties have a name, the type "property" and a value.
154 """
156 name: str = Field(
157 title="Property name",
158 description="The property name describes what kind of property the "
159 "attribute value represents of the entity, for example "
160 "current_speed. Allowed characters "
161 "are the ones in the plain ASCII set, except the following "
162 "ones: control characters, whitespace, &, ?, / and #.",
163 max_length=256,
164 min_length=1,
165 )
166 field_validator("name")(validate_fiware_datatype_string_protect)
169class ContextGeoPropertyValue(BaseModel):
170 """
171 The value for a Geo property is represented by a JSON object with the following syntax:
173 A type with value "Point" and the
174 coordinates with a list containing the coordinates as value
176 Example:
177 "value": {
178 "type": "Point",
179 "coordinates": [
180 -3.80356167695194,
181 43.46296641666926
182 ]
183 }
184 }
186 """
188 type: Optional[str] = Field(default=None, title="type", frozen=True)
189 model_config = ConfigDict(extra="allow")
191 @model_validator(mode="after")
192 def check_geoproperty_value(self) -> Self:
193 """
194 Check if the value is a valid GeoProperty
195 """
196 if self.model_dump().get("type") == "Point":
197 return Point(**self.model_dump())
198 elif self.model_dump().get("type") == "LineString":
199 return LineString(**self.model_dump())
200 elif self.model_dump().get("type") == "Polygon":
201 return Polygon(**self.model_dump())
202 elif self.model_dump().get("type") == "MultiPoint":
203 return MultiPoint(**self.model_dump())
204 elif self.model_dump().get("type") == "MultiLineString":
205 return MultiLineString(**self.model_dump())
206 elif self.model_dump().get("type") == "MultiPolygon":
207 return MultiPolygon(**self.model_dump())
208 elif self.model_dump().get("type") == "GeometryCollection":
209 raise ValueError("GeometryCollection is not supported")
212class ContextGeoProperty(BaseModel):
213 """
214 The model for a Geo property is represented by a JSON object with the following syntax:
216 The attribute value is a JSON object with two contents.
218 Example:
220 {
221 "type": "GeoProperty",
222 "value": {
223 "type": "Point",
224 "coordinates": [
225 -3.80356167695194,
226 43.46296641666926
227 ]
228 }
230 """
232 model_config = ConfigDict(extra="allow")
233 type: Optional[str] = Field(default="GeoProperty", title="type", frozen=True)
234 value: Optional[
235 Union[
236 ContextGeoPropertyValue,
237 Point,
238 LineString,
239 Polygon,
240 MultiPoint,
241 MultiPolygon,
242 MultiLineString,
243 ]
244 ] = Field(default=None, title="GeoProperty value", description="the actual data")
245 observedAt: Optional[str] = Field(
246 default=None,
247 title="Timestamp",
248 description="Representing a timestamp for the "
249 "incoming value of the property.",
250 max_length=256,
251 min_length=1,
252 )
253 field_validator("observedAt")(validate_fiware_datatype_string_protect)
255 datasetId: Optional[str] = Field(
256 None,
257 title="dataset Id",
258 description="It allows identifying a set or group of property values",
259 max_length=256,
260 min_length=1,
261 )
262 field_validator("datasetId")(validate_fiware_datatype_string_protect)
265class NamedContextGeoProperty(ContextGeoProperty):
266 """
267 Context GeoProperties are geo properties of context entities. For example, the coordinates of a building .
269 In the NGSI-LD data model, properties have a name, the type "Geoproperty" and a value.
270 """
272 name: str = Field(
273 title="Property name",
274 description="The property name describes what kind of property the "
275 "attribute value represents of the entity, for example "
276 "current_speed. Allowed characters "
277 "are the ones in the plain ASCII set, except the following "
278 "ones: control characters, whitespace, &, ?, / and #.",
279 max_length=256,
280 min_length=1,
281 )
282 field_validator("name")(validate_fiware_datatype_string_protect)
285class ContextRelationship(BaseModel):
286 """
287 The model for a relationship is represented by a JSON object with the following syntax:
289 The attribute value is specified by the object, whose value can be a reference to another context entity. This
290 should be specified as the URN. The existence of this entity is not assumed.
292 The NGSI type of the attribute is fixed and does not need to be specified.
294 Example:
296 >>> data = {"object": <...>}
298 >>> attr = ContextRelationship(**data)
300 """
302 model_config = ConfigDict(extra="allow") # In order to allow nested relationships
303 type: Optional[str] = Field(default="Relationship", title="type", frozen=True)
304 object: Optional[
305 Union[
306 Union[float, int, bool, str, List, Dict[str, Any]],
307 List[Union[float, int, bool, str, List, Dict[str, Any]]],
308 ]
309 ] = Field(
310 default=None, title="Realtionship object", description="the actual object id"
311 )
313 datasetId: Optional[str] = Field(
314 None,
315 title="dataset Id",
316 description="It allows identifying a set or group of property values",
317 max_length=256,
318 min_length=1,
319 )
320 field_validator("datasetId")(validate_fiware_datatype_string_protect)
322 observedAt: Optional[str] = Field(
323 None,
324 titel="Timestamp",
325 description="Representing a timestamp for the "
326 "incoming value of the property.",
327 max_length=256,
328 min_length=1,
329 )
330 field_validator("observedAt")(validate_fiware_datatype_string_protect)
332 @field_validator("type")
333 @classmethod
334 def check_relationship_type(cls, value):
335 """
336 Force property type to be "Relationship"
337 Args:
338 value: value field
339 Returns:
340 value
341 """
342 if not value == "Relationship":
343 logging.warning(msg='NGSI_LD relationships must have type "Relationship"')
344 value = "Relationship"
345 return value
348class NamedContextRelationship(ContextRelationship):
349 """
350 Context Relationship are relations of context entities to each other.
351 For example, the current_speed of the entity car-104 could be modeled.
352 The location could be modeled as located_in the entity Room-001.
354 In the NGSI-LD data model, relationships have a name, the type "relationship" and an object.
355 """
357 name: str = Field(
358 title="Attribute name",
359 description="The attribute name describes what kind of property the "
360 "attribute value represents of the entity, for example "
361 "current_speed. Allowed characters "
362 "are the ones in the plain ASCII set, except the following "
363 "ones: control characters, whitespace, &, ?, / and #.",
364 max_length=256,
365 min_length=1,
366 # pattern=FiwareRegex.string_protect.value,
367 # Make it FIWARE-Safe
368 )
369 field_validator("name")(validate_fiware_datatype_string_protect)
372class ContextLDEntityKeyValues(BaseModel):
373 """
374 Base Model for an entity is represented by a JSON object with the following
375 syntax.
377 The entity id is specified by the object's id property, whose value
378 is a string containing the entity id.
380 The entity type is specified by the object's type property, whose value
381 is a string containing the entity's type name.
383 """
385 model_config = ConfigDict(
386 extra="allow", validate_default=True, validate_assignment=True
387 )
388 id: str = Field(
389 ...,
390 title="Entity Id",
391 description="Id of an entity in an NGSI context broker. Allowed "
392 "characters are the ones in the plain ASCII set, except "
393 "the following ones: control characters, "
394 "whitespace, &, ?, / and #."
395 "the id should be structured according to the urn naming scheme.",
396 json_schema_extra={"example": "urn:ngsi-ld:Room:001"},
397 max_length=256,
398 min_length=1,
399 # pattern=FiwareRegex.standard.value, # Make it FIWARE-Safe
400 frozen=True,
401 )
402 field_validator("id")(validate_fiware_standard_regex)
403 type: str = Field(
404 ...,
405 title="Entity Type",
406 description="Id of an entity in an NGSI context broker. "
407 "Allowed characters are the ones in the plain ASCII set, "
408 "except the following ones: control characters, "
409 "whitespace, &, ?, / and #.",
410 json_schema_extra={"example": "Room"},
411 max_length=256,
412 min_length=1,
413 # pattern=FiwareRegex.standard.value, # Make it FIWARE-Safe
414 frozen=True,
415 )
416 field_validator("type")(validate_fiware_standard_regex)
419class PropertyFormat(str, Enum):
420 """
421 Format to decide if properties of ContextEntity class are returned as
422 List of NamedContextAttributes or as Dict of ContextAttributes.
423 """
425 LIST = "list"
426 DICT = "dict"
429class ContextLDEntity(ContextLDEntityKeyValues):
430 """
431 Context LD entities, or simply entities, are the center of gravity in the
432 FIWARE NGSI-LD information model. An entity represents a thing, i.e., any
433 physical or logical object (e.g., a sensor, a person, a room, an issue in
434 a ticketing system, etc.). Each entity has an entity id.
435 Furthermore, the type system of FIWARE NGSI enables entities to have an
436 entity type. Entity types are semantic types; they are intended to describe
437 the type of thing represented by the entity. For example, a context
438 entity #with id sensor-365 could have the type temperatureSensor.
440 Each entity is uniquely identified by its id.
442 The entity id is specified by the object's id property, whose value
443 is a string containing the entity id.
445 The entity type is specified by the object's type property, whose value
446 is a string containing the entity's type name.
448 Entity attributes are specified by additional properties and relationships, whose names are
449 the name of the attribute and whose representation is described in the
450 "ContextProperty"/"ContextRelationship"-model. Obviously, id and type are
451 not allowed to be used as attribute names.
453 Example:
455 >>> data = {'id': 'MyId',
456 'type': 'MyType',
457 'my_attr': {'value': 20}}
459 >>> entity = ContextLDEntity(**data)
461 """
463 model_config = ConfigDict(
464 extra="allow",
465 validate_default=True,
466 validate_assignment=True,
467 populate_by_name=True,
468 )
470 observationSpace: Optional[ContextGeoProperty] = Field(
471 default=None,
472 title="Observation Space",
473 description="The geospatial Property representing "
474 "the geographic location that is being "
475 "observed, e.g. by a sensor. "
476 "For example, in the case of a camera, "
477 "the location of the camera and the "
478 "observationspace are different and "
479 "can be disjoint. ",
480 )
481 context: Optional[Union[str, List[str], Dict]] = Field(
482 title="@context",
483 default=None,
484 description="The @context in JSON-LD is used to expand terms, provided as short "
485 "hand strings, to concepts, specified as URIs, and vice versa, "
486 "to compact URIs into terms "
487 "The main implication of NGSI-LD API is that if the @context is "
488 "a compound one, i.e. an @context which references multiple "
489 "individual @context, served by resources behind different URIs, "
490 "then a wrapper @context has to be created and hosted.",
491 examples=["https://n5geh.github.io/n5geh.test-context.io/context_saref.jsonld"],
492 alias="@context",
493 validation_alias="@context",
494 frozen=False,
495 )
497 @field_validator("context")
498 @classmethod
499 def return_context(cls, context):
500 return context
502 operationSpace: Optional[ContextGeoProperty] = Field(
503 default=None,
504 title="Operation Space",
505 description="The geospatial Property representing "
506 "the geographic location in which an "
507 "Entity,e.g. an actuator is active. "
508 "For example, a crane can have a "
509 "certain operation space.",
510 )
512 createdAt: Optional[str] = Field(
513 None,
514 title="Timestamp",
515 description="Representing a timestamp for the "
516 "creation time of the property.",
517 max_length=256,
518 min_length=1,
519 )
520 field_validator("createdAt")(validate_fiware_datatype_string_protect)
522 modifiedAt: Optional[str] = Field(
523 None,
524 title="Timestamp",
525 description="Representing a timestamp for the "
526 "last modification of the property.",
527 max_length=256,
528 min_length=1,
529 )
530 field_validator("modifiedAt")(validate_fiware_datatype_string_protect)
532 def __init__(self, **data):
533 # There is currently no validation for extra fields
534 data.update(self._validate_attributes(data))
535 super().__init__(**data)
537 @classmethod
538 def get_model_fields_set(cls):
539 """
540 Get all names and aliases of the model fields.
541 """
542 return set(
543 [field.validation_alias for (_, field) in cls.model_fields.items()]
544 + [field_name for field_name in cls.model_fields]
545 )
547 @classmethod
548 def _validate_single_property(cls, attr) -> ContextProperty:
549 property_fields = ContextProperty.get_model_fields_set()
550 property_fields.remove(None)
551 # subattrs = {}
552 if attr.get("type") == "Relationship":
553 attr_instance = ContextRelationship.model_validate(attr)
554 elif attr.get("type") == "GeoProperty":
555 try:
556 attr_instance = ContextGeoProperty.model_validate(attr)
557 except Exception as e:
558 pass
559 elif attr.get("type") == "Property" or attr.get("type") is None:
560 attr_instance = ContextProperty.model_validate(attr)
561 else:
562 raise ValueError(f"Attribute {attr.get('type')} " "is not a valid type")
563 for subkey, subattr in attr.items():
564 if isinstance(subattr, dict) and subkey not in property_fields:
565 attr_instance.model_extra.update(
566 {subkey: cls._validate_single_property(attr=subattr)}
567 )
568 return attr_instance
570 @classmethod
571 def _validate_attributes(cls, data: Dict):
572 entity_fields = cls.get_model_fields_set()
573 entity_fields.remove(None)
574 # Initialize the attribute dictionary
575 attrs = {}
576 # Iterate through the data
577 for key, attr in data.items():
578 # Check if the keyword is not already present in the fields
579 if key not in entity_fields:
580 attrs[key] = cls._validate_single_property(attr=attr)
581 return attrs
583 def model_dump(self, *args, by_alias: bool = True, **kwargs):
584 return super().model_dump(*args, by_alias=by_alias, **kwargs)
586 @field_validator("id")
587 @classmethod
588 def _validate_id(cls, id: str):
589 if not id.startswith("urn:ngsi-ld:"):
590 logging.warning(
591 msg="It is recommended that the entity id to be a URN,"
592 'starting with the namespace "urn:ngsi-ld:"'
593 )
594 return id
596 def get_properties(
597 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
598 ) -> Union[List[NamedContextProperty], Dict[str, ContextProperty]]:
599 """
600 Get all properties of the entity.
601 Args:
602 response_format:
604 Returns:
606 """
607 response_format = PropertyFormat(response_format)
608 # response format dict:
609 if response_format == PropertyFormat.DICT:
610 final_dict = {}
611 for key, value in self.model_dump(exclude_unset=True).items():
612 if key not in ContextLDEntity.get_model_fields_set():
613 if value.get("type") != DataTypeLD.RELATIONSHIP:
614 if value.get("type") == DataTypeLD.GEOPROPERTY:
615 final_dict[key] = ContextGeoProperty(**value)
616 elif value.get("type") == DataTypeLD.PROPERTY:
617 final_dict[key] = ContextProperty(**value)
618 else: # named context property by default
619 final_dict[key] = ContextProperty(**value)
620 return final_dict
621 # response format list:
622 final_list = []
623 for key, value in self.model_dump(exclude_unset=True).items():
624 if key not in ContextLDEntity.get_model_fields_set():
625 if value.get("type") != DataTypeLD.RELATIONSHIP:
626 if value.get("type") == DataTypeLD.GEOPROPERTY:
627 final_list.append(NamedContextGeoProperty(name=key, **value))
628 elif value.get("type") == DataTypeLD.PROPERTY:
629 final_list.append(NamedContextProperty(name=key, **value))
630 else: # named context property by default
631 final_list.append(NamedContextProperty(name=key, **value))
632 return final_list
634 def delete_relationships(self, relationships: List[str]):
635 """
636 Delete the given relationships from the entity
638 Args:
639 relationships: List of relationship names
641 Returns:
643 """
644 all_relationships = self.get_relationships(response_format="dict")
645 for relationship in relationships:
646 # check they are relationships
647 if relationship not in all_relationships:
648 raise ValueError(f"Relationship {relationship} does not exist")
649 delattr(self, relationship)
651 def delete_properties(
652 self,
653 props: Union[Dict[str, ContextProperty], List[NamedContextProperty], List[str]],
654 ):
655 """
656 Delete the given properties from the entity
658 Args:
659 props: can be given in multiple forms
660 1) Dict: {"<property_name>": ContextProperty, ...}
661 2) List: [NamedContextProperty, ...]
662 3) List: ["<property_name>", ...]
664 Returns:
666 """
667 names: List[str] = []
668 if isinstance(props, list):
669 for entry in props:
670 if isinstance(entry, str):
671 names.append(entry)
672 elif isinstance(entry, NamedContextProperty):
673 names.append(entry.name)
674 else:
675 names.extend(list(props.keys()))
677 # check there are no relationships
678 relationship_names = [rel.name for rel in self.get_relationships()]
679 for name in names:
680 if name in relationship_names:
681 raise TypeError(f"{name} is a relationship")
683 for name in names:
684 delattr(self, name)
686 def add_geo_properties(
687 self, attrs: Union[Dict[str, ContextGeoProperty], List[NamedContextGeoProperty]]
688 ) -> None:
689 """
690 Add property to entity
691 Args:
692 attrs:
693 Returns:
694 None
695 """
696 if isinstance(attrs, list):
697 attrs = {
698 attr.name: ContextGeoProperty(
699 **attr.model_dump(exclude={"name"}, exclude_unset=True)
700 )
701 for attr in attrs
702 }
703 for key, attr in attrs.items():
704 self.__setattr__(name=key, value=attr)
706 def add_properties(
707 self, attrs: Union[Dict[str, ContextProperty], List[NamedContextProperty]]
708 ) -> None:
709 """
710 Add property to entity
711 Args:
712 attrs:
713 Returns:
714 None
715 """
716 if isinstance(attrs, list):
717 attrs = {
718 attr.name: ContextProperty(
719 **attr.model_dump(exclude={"name"}, exclude_unset=True)
720 )
721 for attr in attrs
722 }
723 for key, attr in attrs.items():
724 self.__setattr__(name=key, value=attr)
726 def add_relationships(
727 self,
728 relationships: Union[
729 Dict[str, ContextRelationship], List[NamedContextRelationship]
730 ],
731 ) -> None:
732 """
733 Add relationship to entity
734 Args:
735 relationships:
736 Returns:
737 None
738 """
739 if isinstance(relationships, list):
740 relationships = {
741 attr.name: ContextRelationship(**attr.dict(exclude={"name"}))
742 for attr in relationships
743 }
744 for key, attr in relationships.items():
745 self.__setattr__(name=key, value=attr)
747 def get_relationships(
748 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
749 ) -> Union[List[NamedContextRelationship], Dict[str, ContextRelationship]]:
750 """
751 Get all relationships of the context entity
753 Args:
754 response_format:
756 Returns:
758 """
759 response_format = PropertyFormat(response_format)
760 # response format dict:
761 if response_format == PropertyFormat.DICT:
762 final_dict = {}
763 for key, value in self.model_dump(exclude_unset=True).items():
764 if key not in ContextLDEntity.get_model_fields_set():
765 try:
766 if value.get("type") == DataTypeLD.RELATIONSHIP:
767 final_dict[key] = ContextRelationship(**value)
768 except AttributeError: # if context attribute
769 if isinstance(value, list):
770 pass
771 return final_dict
772 # response format list:
773 final_list = []
774 for key, value in self.model_dump(exclude_unset=True).items():
775 if key not in ContextLDEntity.get_model_fields_set():
776 if value.get("type") == DataTypeLD.RELATIONSHIP:
777 final_list.append(NamedContextRelationship(name=key, **value))
778 return final_list
780 def get_context(self):
781 """
782 Args:
783 response_format:
785 Returns: context of the entity as list
787 """
788 _, context = self.model_dump(include={"context"}).popitem()
789 if not context:
790 logging.warning("No context in entity")
791 return None
792 else:
793 return context
796class ActionTypeLD(str, Enum):
797 """
798 Options for queries
799 """
801 CREATE = "create"
802 UPSERT = "upsert"
803 UPDATE = "update"
804 DELETE = "delete"
807class UpdateLD(BaseModel):
808 """
809 Model for update action
810 """
812 entities: List[Union[ContextLDEntity, ContextLDEntityKeyValues]] = Field(
813 description="an array of entities, each entity specified using the "
814 "JSON entity representation format "
815 )