diff --git a/filip/models/ngsi_ld/context.py b/filip/models/ngsi_ld/context.py index 85d6c34f..cbd77960 100644 --- a/filip/models/ngsi_ld/context.py +++ b/filip/models/ngsi_ld/context.py @@ -4,8 +4,11 @@ import logging from typing import Any, List, Dict, Union, Optional +from geojson_pydantic import Point, MultiPoint, LineString, MultiLineString, Polygon, \ + MultiPolygon, GeometryCollection +from typing_extensions import Self from aenum import Enum -from pydantic import field_validator, ConfigDict, BaseModel, Field +from pydantic import field_validator, ConfigDict, BaseModel, Field, model_validator from filip.models.ngsi_v2 import ContextEntity from filip.utils.validators import FiwareRegex, \ validate_fiware_datatype_string_protect, validate_fiware_standard_regex @@ -118,8 +121,6 @@ def check_property_type(cls, value): return value - - class NamedContextProperty(ContextProperty): """ Context properties are properties of context entities. For example, the current speed of a car could be modeled @@ -159,48 +160,31 @@ class ContextGeoPropertyValue(BaseModel): """ type: Optional[str] = Field( - default="Point", + default=None, title="type", frozen=True ) - coordinates: List[float] = Field( - default=None, - title="Geo property coordinates", - description="the actual coordinates" - ) - @field_validator("type") - @classmethod - def check_geoproperty_value_type(cls, value): - """ - Force property type to be "Point" - Args: - value: value field - Returns: - value - """ - if not value == "Point": - logging.warning(msg='NGSI_LD GeoProperty values must have type "Point"') - value = "Point" - return value + model_config = ConfigDict(extra='allow') - @field_validator("coordinates") - @classmethod - def check_geoproperty_value_coordinates(cls, value): + @model_validator(mode='after') + def check_geoproperty_value(self) -> Self: """ - Force property coordinates to be lis of two floats - Args: - value: value field - Returns: - value + Check if the value is a valid GeoProperty """ - if not isinstance(value, list) or len(value) != 2: - logging.error(msg='NGSI_LD GeoProperty values must have coordinates as list with length two') - raise ValueError - for element in value: - if not isinstance(element, float): - logging.error(msg='NGSI_LD GeoProperty values must have coordinates as list of floats') - raise TypeError - return value + if self.model_dump().get("type") == "Point": + return Point(**self.model_dump()) + elif self.model_dump().get("type") == "LineString": + return LineString(**self.model_dump()) + elif self.model_dump().get("type") == "Polygon": + return Polygon(**self.model_dump()) + elif self.model_dump().get("type") == "MultiPoint": + return MultiPoint(**self.model_dump()) + elif self.model_dump().get("type") == "MultiLineString": + return MultiLineString(**self.model_dump()) + elif self.model_dump().get("type") == "MultiPolygon": + return MultiPolygon(**self.model_dump()) + elif self.model_dump().get("type") == "GeometryCollection": + return GeometryCollection(**self.model_dump()) class ContextGeoProperty(BaseModel): @@ -229,7 +213,10 @@ class ContextGeoProperty(BaseModel): title="type", frozen=True ) - value: Optional[ContextGeoPropertyValue] = Field( + value: Optional[Union[ContextGeoPropertyValue, + Point, LineString, Polygon, + MultiPoint, MultiPolygon, + MultiLineString, GeometryCollection]] = Field( default=None, title="GeoProperty value", description="the actual data" @@ -252,28 +239,6 @@ class ContextGeoProperty(BaseModel): ) field_validator("datasetId")(validate_fiware_datatype_string_protect) - @field_validator("type") - @classmethod - def check_geoproperty_type(cls, value): - """ - Force property type to be "GeoProperty" - Args: - value: value field - Returns: - value - """ - if not value == "GeoProperty": - if value == "Relationship": - value == "Relationship" - elif value == "TemporalProperty": - value == "TemporalProperty" - else: - logging.warning(msg='NGSI_LD GeoProperties must have type "GeoProperty" ' - '-> They are checked first, so if no GeoProperties are used ignore this warning!') - raise ValueError('NGSI_LD GeoProperties must have type "GeoProperty" ' - '-> They are checked first, so if no GeoProperties are used ignore this warning!') - return value - class NamedContextGeoProperty(ContextGeoProperty): """ @@ -538,7 +503,6 @@ def __init__(self, data.update(self._validate_attributes(data)) super().__init__(id=id, type=type, **data) - # TODO we should distinguish between context relationship @classmethod def _validate_attributes(cls, data: Dict): fields = set([field.validation_alias for (_, field) in cls.model_fields.items()] + @@ -551,10 +515,15 @@ def _validate_attributes(cls, data: Dict): for key, attr in data.items(): # Check if the keyword is not already present in the fields if key not in fields: - try: + if attr.get("type") == "Relationship": + attrs[key] = ContextRelationship.model_validate(attr) + elif attr.get("type") == "GeoProperty": attrs[key] = ContextGeoProperty.model_validate(attr) - except ValueError: + elif attr.get("type") == "Property": attrs[key] = ContextProperty.model_validate(attr) + else: + raise ValueError(f"Attribute {attr.get('type')} " + "is not a valid type") return attrs model_config = ConfigDict(extra='allow', validate_default=True, validate_assignment=True) @@ -691,6 +660,22 @@ def delete_properties(self, props: Union[Dict[str, ContextProperty], for name in names: delattr(self, name) + def add_geo_properties(self, attrs: Union[Dict[str, ContextGeoProperty], + List[NamedContextGeoProperty]]) -> None: + """ + Add property to entity + Args: + attrs: + Returns: + None + """ + if isinstance(attrs, list): + attrs = {attr.name: ContextGeoProperty(**attr.model_dump(exclude={'name'}, + exclude_unset=True)) + for attr in attrs} + for key, attr in attrs.items(): + self.__setattr__(name=key, value=attr) + def add_properties(self, attrs: Union[Dict[str, ContextProperty], List[NamedContextProperty]]) -> None: """ diff --git a/tests/models/test_ngsi_ld_context.py b/tests/models/test_ngsi_ld_context.py index 2dfd873a..535d8511 100644 --- a/tests/models/test_ngsi_ld_context.py +++ b/tests/models/test_ngsi_ld_context.py @@ -1,13 +1,14 @@ """ Test module for context broker models """ - import unittest +from geojson_pydantic import Point, MultiPoint, LineString, Polygon, GeometryCollection from pydantic import ValidationError from filip.models.ngsi_ld.context import \ - ContextLDEntity, ContextProperty, NamedContextProperty + ContextLDEntity, ContextProperty, NamedContextProperty, \ + ContextGeoPropertyValue, ContextGeoProperty, NamedContextGeoProperty class TestLDContextModels(unittest.TestCase): @@ -48,7 +49,7 @@ def setUp(self) -> None: "type": "GeoProperty", "value": { "type": "Point", - "coordinates": [-8.5, 41.2] + "coordinates": (-8.5, 41.2) # coordinates are normally a tuple } }, "@context": [ @@ -61,7 +62,7 @@ def setUp(self) -> None: "type": "GeoProperty", "value": { "type": "Point", - "coordinates": [-8.5, 41.2] + "coordinates": (-8.5, 41.2) } }, "totalSpotNumber": { @@ -163,6 +164,79 @@ def setUp(self) -> None: ], } } + self.testpoint_value = { + "type": "Point", + "coordinates": (-8.5, 41.2) + } + self.testmultipoint_value = { + "type": "MultiPoint", + "coordinates": ( + (-3.80356167695194, 43.46296641666926), + (-3.804056, 43.464638) + ) + } + self.testlinestring_value = { + "type": "LineString", + "coordinates": ( + (-3.80356167695194, 43.46296641666926), + (-3.804056, 43.464638) + ) + } + self.testpolygon_value = { + "type": "Polygon", + "coordinates": ( + ( + (-3.80356167695194, 43.46296641666926), + (-3.804056, 43.464638), + (-3.805056, 43.463638), + (-3.80356167695194, 43.46296641666926) + ) + ) + } + self.testgeometrycollection_value = { + "type": "GeometryCollection", + "geometries": [ + { + "type": "Point", + "coordinates": (-3.80356167695194, 43.46296641666926) + }, + { + "type": "LineString", + "coordinates": ( + (-3.804056, 43.464638), + (-3.805056, 43.463638) + ) + } + ] + } + self.entity_geo_dict = { + "id": "urn:ngsi-ld:Geometry:001", + "type": "MyGeometry", + "testpoint": { + "type": "GeoProperty", + "value": self.testpoint_value + }, + "testmultipoint": { + "type": "GeoProperty", + "value": self.testmultipoint_value, + "observedAt": "2023-09-12T12:35:00Z" + }, + "testlinestring": { + "type": "GeoProperty", + "value": self.testlinestring_value, + "observedAt": "2023-09-12T12:35:30Z" + }, + "testpolygon": { + "type": "GeoProperty", + "value": self.testpolygon_value, + "observedAt": "2023-09-12T12:36:00Z" + }, + "testgeometrycollection": { + "type": "GeoProperty", + "value": self.testgeometrycollection_value, + "observedAt": "2023-09-12T12:36:30Z" + } + } def test_cb_attribute(self) -> None: """ @@ -181,6 +255,42 @@ def test_entity_id(self) -> None: with self.assertRaises(ValidationError): ContextLDEntity(**{'id': 'MyId', 'type': 'MyType'}) + def test_geo_property(self) -> None: + """ + Test ContextGeoPropertyValue models + Returns: + None + """ + geo_entity = ContextLDEntity(**self.entity_geo_dict) + new_entity = ContextLDEntity(id="urn:ngsi-ld:Geometry:002", type="MyGeometry") + test_point = NamedContextGeoProperty( + name="testpoint", + type="GeoProperty", + value=Point(**self.testpoint_value) + ) + test_MultiPoint = NamedContextGeoProperty( + name="testmultipoint", + type="GeoProperty", + value=MultiPoint(**self.testmultipoint_value) + ) + test_LineString = NamedContextGeoProperty( + name="testlinestring", + type="GeoProperty", + value=LineString(**self.testlinestring_value) + ) + test_Polygon = NamedContextGeoProperty( + name="testpolygon", + type="GeoProperty", + value=Polygon(**self.testpolygon_value) + ) + test_GeometryCollection = NamedContextGeoProperty( + name="testgeometrycollection", + type="GeoProperty", + value=GeometryCollection(**self.testgeometrycollection_value) + ) + new_entity.add_geo_properties([test_point, test_MultiPoint, test_LineString, + test_Polygon, test_GeometryCollection]) + def test_cb_entity(self) -> None: """ Test context entity models