Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

300 ngsi ld revise geoproperty model #328

Closed
wants to merge 4 commits into from
Closed

300 ngsi ld revise geoproperty model #328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d705acd
feat: implement the rest geoproperty models
djs0109 Sep 18, 2024
edae39c
feat: add function to add geo_properties
djs0109 Sep 18, 2024
33bef2e
fix: validation of entity attributes
djs0109 Oct 2, 2024
0e72383
fix: change coordinates from list to tuple
djs0109 Oct 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 50 additions & 65 deletions filip/models/ngsi_ld/context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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"
Expand All @@ -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):
"""
Expand Down Expand Up @@ -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()] +
Expand All @@ -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)
Expand Down Expand Up @@ -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:
"""
Expand Down
118 changes: 114 additions & 4 deletions tests/models/test_ngsi_ld_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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": [
Expand All @@ -61,7 +62,7 @@ def setUp(self) -> None:
"type": "GeoProperty",
"value": {
"type": "Point",
"coordinates": [-8.5, 41.2]
"coordinates": (-8.5, 41.2)
}
},
"totalSpotNumber": {
Expand Down Expand Up @@ -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:
"""
Expand All @@ -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
Expand Down