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