Coverage for filip/models/ngsi_v2/iot.py: 88%
187 statements
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-26 14:36 +0000
1"""
2Module contains models for accessing and interaction with FIWARE's IoT-Agents.
3"""
5from __future__ import annotations
6import logging
7import itertools
8import warnings
9from enum import Enum
10from typing import Any, Dict, Optional, List, Union
11import pytz
12from pydantic import (
13 field_validator,
14 model_validator,
15 ConfigDict,
16 BaseModel,
17 Field,
18 AnyHttpUrl,
19 OnErrorOmit,
20)
21from filip.models.base import NgsiVersion, DataType
22from filip.models.ngsi_v2.base import (
23 BaseAttribute,
24 BaseValueAttribute,
25 BaseNameAttribute,
26)
27from filip.utils.validators import (
28 validate_fiware_datatype_string_protect,
29 validate_fiware_datatype_standard,
30 validate_jexl_expression,
31 validate_expression_language,
32 validate_fiware_attribute_name_regex,
33)
35logger = logging.getLogger()
38class ExpressionLanguage(str, Enum):
39 """
40 Options for expression language
41 """
43 LEGACY = "legacy"
44 JEXL = "jexl"
47class PayloadProtocol(str, Enum):
48 """
49 Options for payload protocols
50 """
52 IOTA_JSON = "IoTA-JSON"
53 IOTA_UL = "PDI-IoTA-UltraLight"
54 LORAWAN = "LoRaWAN"
57class TransportProtocol(str, Enum):
58 """
59 Options for transport protocols
60 """
62 MQTT = "MQTT"
63 AMQP = "AMQP"
64 HTTP = "HTTP"
67class IoTABaseAttribute(BaseAttribute, BaseNameAttribute):
68 """
69 Base model for device attributes
70 """
72 expression: Optional[str] = Field(
73 default=None,
74 description="indicates that the value of the target attribute will "
75 "not be the plain value or the measurement, but an "
76 "expression based on a combination of the reported values. "
77 "See the Expression Language definition for details "
78 "(https://iotagent-node-lib.readthedocs.io/en/latest/"
79 "api.html#expression-language-support)",
80 )
81 entity_name: Optional[str] = Field(
82 default=None,
83 description="entity_name: the presence of this attribute indicates "
84 "that the value will not be stored in the original device "
85 "entity but in a new entity with an ID given by this "
86 "attribute. The type of this additional entity can be "
87 "configured with the entity_type attribute. If no type is "
88 "configured, the device entity type is used instead. "
89 "Entity names can be defined as expressions, using the "
90 "Expression Language definition "
91 "(https://iotagent-node-lib.readthedocs.io/en/latest/"
92 "api.html#expression-language-support). Allowed characters are"
93 " the ones in the plain ASCII set, except the following "
94 "ones: control characters, whitespace, &, ?, / and #.",
95 max_length=256,
96 min_length=1,
97 )
98 valid_entity_name = field_validator("entity_name")(
99 validate_fiware_datatype_standard
100 )
101 entity_type: Optional[str] = Field(
102 default=None,
103 description="configures the type of an alternative entity. "
104 "Allowed characters "
105 "are the ones in the plain ASCII set, except the following "
106 "ones: control characters, whitespace, &, ?, / and #.",
107 max_length=256,
108 min_length=1,
109 )
110 valid_entity_type = field_validator("entity_type")(
111 validate_fiware_datatype_standard
112 )
113 reverse: Optional[str] = Field(
114 default=None,
115 description="add bidirectionality expressions to the attribute. See "
116 "the bidirectionality transformation plugin in the "
117 "Data Mapping Plugins section for details. "
118 "(https://iotagent-node-lib.readthedocs.io/en/latest/api/"
119 "index.html#data-mapping-plugins)",
120 )
122 def __eq__(self, other):
123 if isinstance(other, BaseAttribute):
124 return self.name == other.name
125 else:
126 return self.model_dump() == other
129class DeviceAttribute(IoTABaseAttribute):
130 """
131 Model for active device attributes
132 """
134 object_id: Optional[str] = Field(
135 default=None, description="name of the attribute as coming from the device."
136 )
139class LazyDeviceAttribute(BaseNameAttribute):
140 """
141 Model for lazy device attributes
142 """
144 type: Union[DataType, str] = Field(
145 default=DataType.TEXT,
146 description="The attribute type represents the NGSI value type of the "
147 "attribute value. Note that FIWARE NGSI has its own type "
148 "system for attribute values, so NGSI value types are not "
149 "the same as JSON types. Allowed characters "
150 "are the ones in the plain ASCII set, except the following "
151 "ones: control characters, whitespace, &, ?, / and #.",
152 max_length=256,
153 min_length=1,
154 )
155 valid_type = field_validator("type")(validate_fiware_datatype_string_protect)
158class DeviceCommand(BaseModel):
159 """
160 Model for commands
161 """
163 name: str = Field(
164 description="ID of the attribute in the target entity in the "
165 "Context Broker. Allowed characters "
166 "are the ones in the plain ASCII set, except the following "
167 "ones: control characters, whitespace, &, ?, / and #.",
168 max_length=256,
169 min_length=1,
170 )
171 valid_name = field_validator("name")(validate_fiware_attribute_name_regex)
172 type: Union[DataType, str] = Field(
173 description="name of the type of the attribute in the target entity. ",
174 default=DataType.COMMAND,
175 )
178class StaticDeviceAttribute(IoTABaseAttribute, BaseValueAttribute):
179 """
180 Model for static device attributes
181 """
183 pass
186class ServiceGroup(BaseModel):
187 """
188 Model for device service group.
189 https://iotagent-node-lib.readthedocs.io/en/latest/api/index.html#service-group-api
190 """
192 service: Optional[str] = Field(
193 default=None, description="ServiceGroup of the devices of this type"
194 )
195 subservice: Optional[str] = Field(
196 default=None,
197 description="Subservice of the devices of this type.",
198 pattern="^/",
199 )
200 resource: str = Field(
201 description="string representing the Southbound resource that will be "
202 "used to assign a type to a device (e.g.: pathname in the "
203 "southbound port)."
204 )
205 apikey: str = Field(
206 description="API Key string. It is a key used for devices belonging "
207 "to this service_group. If "
208 ", service_group does not use "
209 "apikey, but it must be specified."
210 )
211 timestamp: Optional[bool] = Field(
212 default=None,
213 description="Optional flag about whether or not to add the TimeInstant "
214 "attribute to the device entity created, as well as a "
215 "TimeInstant metadata to each attribute, with the current "
216 "timestamp. With NGSI-LD, the Standard observedAt "
217 "property-of-a-property is created instead.",
218 )
219 entity_type: Optional[str] = Field(
220 default=None,
221 description="name of the Entity type to assign to the group. "
222 "Allowed characters "
223 "are the ones in the plain ASCII set, except the following "
224 "ones: control characters, whitespace, &, ?, / and #.",
225 max_length=256,
226 min_length=1,
227 )
228 valid_entity_type = field_validator("entity_type")(
229 validate_fiware_datatype_standard
230 )
231 trust: Optional[str] = Field(
232 default=None,
233 description="trust token to use for secured access to the "
234 "Context Broker for this type of devices (optional; only "
235 "needed for secured scenarios).",
236 )
237 cbHost: Optional[AnyHttpUrl] = Field(
238 default=None,
239 description="Context Broker connection information. This options can "
240 "be used to override the global ones for specific types of "
241 "devices.",
242 )
244 @field_validator("cbHost")
245 @classmethod
246 def validate_cbHost(cls, value):
247 """
248 convert cbHost to str
249 Returns:
250 timezone
251 """
252 return str(value) if value else value
254 lazy: Optional[List[LazyDeviceAttribute]] = Field(
255 default=[],
256 description="list of common lazy attributes of the device. For each "
257 "attribute, its name and type must be provided.",
258 )
259 commands: Optional[List[DeviceCommand]] = Field(
260 default=[],
261 description="list of common commands attributes of the device. For each "
262 "attribute, its name and type must be provided, additional "
263 "metadata is optional",
264 )
265 attributes: Optional[List[DeviceAttribute]] = Field(
266 default=[],
267 description="list of common commands attributes of the device. For "
268 "each attribute, its name and type must be provided, "
269 "additional metadata is optional.",
270 )
271 static_attributes: Optional[List[StaticDeviceAttribute]] = Field(
272 default=[],
273 description="this attributes will be added to all the entities of this "
274 "group 'as is', additional metadata is optional.",
275 )
276 internal_attributes: Optional[List[Dict[str, Any]]] = Field(
277 default=[],
278 description="optional section with free format, to allow specific "
279 "IoT Agents to store information along with the devices "
280 "in the Device Registry.",
281 )
282 expressionLanguage: Optional[ExpressionLanguage] = Field(
283 default=ExpressionLanguage.JEXL,
284 description="optional boolean value, to set expression language used "
285 "to compute expressions, possible values are: "
286 "legacy or jexl, but legacy is deprecated. If it is set None, jexl is used.",
287 )
288 valid_expressionLanguage = field_validator("expressionLanguage")(
289 validate_expression_language
290 )
291 explicitAttrs: Optional[bool] = Field(
292 default=False,
293 description="optional boolean value, to support selective ignore "
294 "of measures so that IOTA does not progress. If not "
295 "specified default is false.",
296 )
297 autoprovision: Optional[bool] = Field(
298 default=True,
299 description="optional boolean: If false, autoprovisioned devices "
300 "(i.e. devices that are not created with an explicit "
301 "provision operation but when the first measure arrives) "
302 "are not allowed in this group. "
303 "Default (in the case of omitting the field) is true.",
304 )
305 ngsiVersion: Optional[NgsiVersion] = Field(
306 default="v2",
307 description="optional string value used in mixed mode to switch between"
308 " NGSI-v2 and NGSI-LD payloads. Possible values are: "
309 "v2 or ld. The default is v2. When not running in mixed "
310 "mode, this field is ignored.",
311 )
312 defaultEntityNameConjunction: Optional[str] = Field(
313 default=None,
314 description="optional string value to set default conjunction string "
315 "used to compose a default entity_name when is not "
316 "provided at device provisioning time.",
317 )
320class DeviceSettings(BaseModel):
321 """
322 Model for iot device settings
323 """
325 model_config = ConfigDict(validate_assignment=True)
326 timezone: Optional[str] = Field(
327 default="Europe/London", description="Time zone of the sensor if it has any"
328 )
329 timestamp: Optional[bool] = Field(
330 default=None,
331 description="Optional flag about whether or not to add the TimeInstant "
332 "attribute to the device entity created, as well as a "
333 "TimeInstant metadata to each attribute, with the current "
334 "timestamp. With NGSI-LD, the Standard observedAt "
335 "property-of-a-property is created instead.",
336 )
337 apikey: Optional[str] = Field(
338 default=None,
339 description="Optional Apikey key string to use instead of group apikey",
340 )
341 endpoint: Optional[AnyHttpUrl] = Field(
342 default=None,
343 description="Endpoint where the device is going to receive commands, "
344 "if any.",
345 )
346 protocol: Optional[Union[PayloadProtocol, str]] = Field(
347 default=None,
348 description="Name of the device protocol, for its use with an " "IoT Manager.",
349 )
350 transport: Optional[Union[TransportProtocol, str]] = Field(
351 default=None,
352 description="Name of the device transport protocol, for the IoT Agents "
353 "with multiple transport protocols.",
354 )
355 expressionLanguage: Optional[ExpressionLanguage] = Field(
356 default=ExpressionLanguage.JEXL,
357 description="optional boolean value, to set expression language used "
358 "to compute expressions, possible values are: "
359 "legacy or jexl, but legacy is deprecated. If it is set None, jexl is used.",
360 )
361 valid_expressionLanguage = field_validator("expressionLanguage")(
362 validate_expression_language
363 )
364 explicitAttrs: Optional[bool] = Field(
365 default=False,
366 description="optional boolean value, to support selective ignore "
367 "of measures so that IOTA does not progress. If not "
368 "specified default is false.",
369 )
372class Device(DeviceSettings):
373 """
374 Model for iot devices.
375 https://iotagent-node-lib.readthedocs.io/en/latest/api/index.html#device-api
376 """
378 model_config = ConfigDict(validate_default=True, validate_assignment=True)
379 device_id: str = Field(
380 description="Device ID that will be used to identify the device"
381 )
382 service: Optional[str] = Field(
383 default=None,
384 description="Name of the service the device belongs to "
385 "(will be used in the fiware-service header).",
386 max_length=50,
387 )
388 service_path: Optional[str] = Field(
389 default="/",
390 description="Name of the subservice the device belongs to "
391 "(used in the fiware-servicepath header).",
392 max_length=51,
393 pattern="^/",
394 )
395 entity_name: str = Field(
396 description="Name of the entity representing the device in "
397 "the Context Broker Allowed characters "
398 "are the ones in the plain ASCII set, except the following "
399 "ones: control characters, whitespace, &, ?, / and #.",
400 max_length=256,
401 min_length=1,
402 )
403 valid_entity_name = field_validator("entity_name")(
404 validate_fiware_datatype_standard
405 )
406 entity_type: str = Field(
407 description="Type of the entity in the Context Broker. "
408 "Allowed characters "
409 "are the ones in the plain ASCII set, except the following "
410 "ones: control characters, whitespace, &, ?, / and #.",
411 max_length=256,
412 min_length=1,
413 )
414 valid_entity_type = field_validator("entity_type")(
415 validate_fiware_datatype_standard
416 )
417 lazy: List[LazyDeviceAttribute] = Field(
418 default=[], description="List of lazy attributes of the device"
419 )
420 commands: List[DeviceCommand] = Field(
421 default=[], description="List of commands of the device"
422 )
423 attributes: List[DeviceAttribute] = Field(
424 default=[], description="List of active attributes of the device"
425 )
426 static_attributes: Optional[List[StaticDeviceAttribute]] = Field(
427 default=[],
428 description="List of static attributes to append to the entity. All the"
429 " updateContext requests to the CB will have this set of "
430 "attributes appended.",
431 )
432 internal_attributes: Optional[List[Dict[str, Any]]] = Field(
433 default=[],
434 description="List of internal attributes with free format for specific "
435 "IoT Agent configuration",
436 )
437 ngsiVersion: NgsiVersion = Field(
438 default=NgsiVersion.v2,
439 description="optional string value used in mixed mode to switch between"
440 " NGSI-v2 and NGSI-LD payloads. Possible values are: "
441 "v2 or ld. The default is v2. When not running in "
442 "mixed mode, this field is ignored.",
443 )
445 @field_validator("timezone")
446 @classmethod
447 def validate_timezone(cls, value):
448 """
449 validate timezone
450 Returns:
451 timezone
452 """
453 assert value in pytz.all_timezones
454 return value
456 @model_validator(mode="after")
457 def validate_device_attributes_expression(self):
458 """
459 Validates device attributes expressions based on the expression language (JEXL or Legacy, where Legacy is
460 deprecated).
462 Args:
463 self: The Device instance.
465 Returns:
466 The Device instance after validation.
467 """
468 if self.expressionLanguage == ExpressionLanguage.JEXL:
469 for attribute in self.attributes:
470 if attribute.expression:
471 validate_jexl_expression(
472 attribute.expression, attribute.name, self.device_id
473 )
474 elif self.expressionLanguage == ExpressionLanguage.LEGACY:
475 warnings.warn(
476 f"No validation for legacy expression language of Device {self.device_id}."
477 )
479 return self
481 @model_validator(mode="after")
482 def validate_duplicated_device_attributes(self):
483 """
484 Check whether device has identical attributes
485 Args:
486 self: dict of Device instance.
488 Returns:
489 The dict of Device instance after validation.
490 """
491 for i, attr in enumerate(self.attributes):
492 for other_attr in self.attributes[:i] + self.attributes[i + 1 :]:
493 if attr.model_dump() == other_attr.model_dump():
494 raise ValueError(f"Duplicated attributes found: {attr.name}")
495 return self
497 @model_validator(mode="after")
498 def validate_device_attributes_name_object_id(self):
499 """
500 Validate the device regarding the behavior with devices attributes.
501 According to https://iotagent-node-lib.readthedocs.io/en/latest/api.html and
502 based on our best practice, following rules are checked
503 - name is required, but not necessarily unique
504 - object_id is not required, if given must be unique, i.e. not equal to any
505 existing object_id and name
506 Args:
507 self: dict of Device instance.
509 Returns:
510 The dict of Device instance after validation.
511 """
512 for i, attr in enumerate(self.attributes):
513 for other_attr in self.attributes[:i] + self.attributes[i + 1 :]:
514 if (
515 attr.object_id
516 and other_attr.object_id
517 and attr.object_id == other_attr.object_id
518 ):
519 raise ValueError(f"object_id {attr.object_id} is not unique")
520 if attr.object_id and attr.object_id == other_attr.name:
521 raise ValueError(f"object_id {attr.object_id} is not unique")
522 return self
524 def get_attribute(
525 self, attribute_name: str
526 ) -> Union[
527 DeviceAttribute, LazyDeviceAttribute, StaticDeviceAttribute, DeviceCommand
528 ]:
529 """
531 Args:
532 attribute_name:
534 Returns:
536 """
537 for attribute in itertools.chain(
538 self.attributes,
539 self.lazy,
540 self.static_attributes,
541 self.internal_attributes,
542 self.commands,
543 ):
544 if attribute.name == attribute_name:
545 return attribute
546 msg = (
547 f"Device: {self.device_id}: Could not "
548 f"find attribute with name {attribute_name}"
549 )
550 logger.error(msg)
551 raise KeyError(msg)
553 def add_attribute(
554 self,
555 attribute: Union[
556 DeviceAttribute, LazyDeviceAttribute, StaticDeviceAttribute, DeviceCommand
557 ],
558 update: bool = False,
559 ) -> None:
560 """
562 Args:
563 attribute:
564 update (bool): If 'True' and attribute does already exists tries
565 to update the attribute if not
566 Returns:
567 None
568 """
569 try:
570 if type(attribute) == DeviceAttribute:
571 if attribute.model_dump(exclude_none=True) in [
572 attr.model_dump(exclude_none=True) for attr in self.attributes
573 ]:
574 raise ValueError
576 self.attributes.append(attribute)
577 self.__setattr__(name="attributes", value=self.attributes)
578 elif type(attribute) == LazyDeviceAttribute:
579 if attribute in self.lazy:
580 raise ValueError
582 self.lazy.append(attribute)
583 self.__setattr__(name="lazy", value=self.lazy)
584 elif type(attribute) == StaticDeviceAttribute:
585 if attribute in self.static_attributes:
586 raise ValueError
588 self.static_attributes.append(attribute)
589 self.__setattr__(name="static_attributes", value=self.static_attributes)
590 elif type(attribute) == DeviceCommand:
591 if attribute in self.commands:
592 raise ValueError
594 self.commands.append(attribute)
595 self.__setattr__(name="commands", value=self.commands)
596 else:
597 raise ValueError
598 except ValueError:
599 if update:
600 self.update_attribute(attribute, append=False)
601 logger.warning(
602 "Device: %s: Attribute already " "exists. Will update: \n %s",
603 self.device_id,
604 attribute.model_dump_json(indent=2),
605 )
606 else:
607 logger.error(
608 "Device: %s: Attribute already " "exists: \n %s",
609 self.device_id,
610 attribute.model_dump_json(indent=2),
611 )
612 raise
614 def update_attribute(
615 self,
616 attribute: Union[
617 DeviceAttribute, LazyDeviceAttribute, StaticDeviceAttribute, DeviceCommand
618 ],
619 append: bool = False,
620 ) -> None:
621 """
622 Updates existing device attribute
624 Args:
625 attribute: Attribute to add to device configuration
626 append (bool): Adds attribute if not exist
628 Returns:
629 None
630 """
631 try:
632 if type(attribute) == DeviceAttribute:
633 idx = self.attributes.index(attribute)
634 self.attributes[idx].model_dump().update(attribute.model_dump())
635 elif type(attribute) == LazyDeviceAttribute:
636 idx = self.lazy.index(attribute)
637 self.lazy[idx].model_dump().update(attribute.model_dump())
638 elif type(attribute) == StaticDeviceAttribute:
639 idx = self.static_attributes.index(attribute)
640 self.static_attributes[idx].model_dump().update(attribute.model_dump())
641 elif type(attribute) == DeviceCommand:
642 idx = self.commands.index(attribute)
643 self.commands[idx].model_dump().update(attribute.model_dump())
644 except ValueError:
645 if append:
646 logger.warning(
647 "Device: %s: Could not find " "attribute: \n %s",
648 self.device_id,
649 attribute.model_dump_json(indent=2),
650 )
651 self.add_attribute(attribute=attribute)
652 else:
653 msg = (
654 f"Device: {self.device_id}: Could not find "
655 f"attribute: \n {attribute.model_dump_json(indent=2)}"
656 )
657 raise KeyError(msg)
659 def delete_attribute(
660 self,
661 attribute: Union[
662 DeviceAttribute, LazyDeviceAttribute, StaticDeviceAttribute, DeviceCommand
663 ],
664 ):
665 """
666 Deletes attribute from device
667 Args:
668 attribute: ()
670 Returns:
672 """
673 try:
674 if type(attribute) == DeviceAttribute:
675 self.attributes.remove(attribute)
676 elif type(attribute) == LazyDeviceAttribute:
677 self.lazy.remove(attribute)
678 elif type(attribute) == StaticDeviceAttribute:
679 self.static_attributes.remove(attribute)
680 elif type(attribute) == DeviceCommand:
681 self.commands.remove(attribute)
682 else:
683 raise ValueError
684 except ValueError:
685 logger.warning(
686 "Device: %s: Could not delete " "attribute: \n %s",
687 self.device_id,
688 attribute.model_dump_json(indent=2),
689 )
690 raise
692 logger.info(
693 "Device: %s: Attribute deleted! \n %s",
694 self.device_id,
695 attribute.model_dump_json(indent=2),
696 )
698 def get_command(self, command_name: str):
699 """
700 Short for self.get_attributes
701 Args:
702 command_name (str):
703 Returns:
705 """
706 return self.get_attribute(attribute_name=command_name)
708 def add_command(self, command: DeviceCommand, update: bool = False):
709 """
710 Short for self.add_attribute
711 Args:
712 command (DeviceCommand):
713 update (bool): Update command if it already exists
714 Returns:
715 """
716 self.add_attribute(attribute=command, update=update)
718 def update_command(self, command: DeviceCommand, append: bool = False):
719 """
720 Short for self.update_attribute
721 Args:
722 command:
723 append:
724 Returns:
725 """
726 self.update_attribute(attribute=command, append=append)
728 def delete_command(self, command: DeviceCommand):
729 """
730 Short for self.delete_attribute
731 Args:
732 command:
734 Returns:
735 None
736 """
737 self.delete_attribute(attribute=command)
740class DeviceList(DeviceSettings):
741 """
742 Collection model for a list of devices
743 """
745 devices: List[OnErrorOmit[Device]]
748class DeviceValidationList(DeviceList):
749 """
750 Collection model for a list of valid and invalid devices
751 """
753 invalid_devices: List[str]