diff --git a/filip/models/ngsi_ld/context.py b/filip/models/ngsi_ld/context.py index 85d6c34f..b2ce89a4 100644 --- a/filip/models/ngsi_ld/context.py +++ b/filip/models/ngsi_ld/context.py @@ -3,9 +3,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 @@ -17,7 +19,7 @@ class DataTypeLD(str, Enum): In NGSI-LD the data types on context entities are only divided into properties and relationships. """ _init_ = 'value __doc__' - + GEOPROPERTY = "GeoProperty", "A property that represents a geometry value" PROPERTY = "Property", "All attributes that do not represent a relationship" RELATIONSHIP = "Relationship", "Reference to another context entity, which can be identified with a URN." @@ -97,6 +99,15 @@ class ContextProperty(BaseModel): ) field_validator("datasetId")(validate_fiware_datatype_string_protect) + @classmethod + def get_model_fields_set(cls): + """ + Get all names and aliases of the model fields. + """ + return set([field.validation_alias + for (_, field) in cls.model_fields.items()] + + [field_name for field_name in cls.model_fields]) + @field_validator("type") @classmethod def check_property_type(cls, value): @@ -107,19 +118,15 @@ def check_property_type(cls, value): Returns: value """ - if not value == "Property": - if value == "Relationship": - value == "Relationship" - elif value == "TemporalProperty": - value == "TemporalProperty" - else: - logging.warning(msg='NGSI_LD Properties must have type "Property"') - value = "Property" + valid_property_types = ["Property", "Relationship", "TemporalProperty"] + if value not in valid_property_types: + msg = f'NGSI_LD Properties must have type {valid_property_types}, ' \ + f'not "{value}"' + logging.warning(msg=msg) + raise ValueError(msg) 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 +166,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): @@ -211,7 +201,7 @@ class ContextGeoProperty(BaseModel): Example: - "location": { + { "type": "GeoProperty", "value": { "type": "Point", @@ -220,7 +210,6 @@ class ContextGeoProperty(BaseModel): 43.46296641666926 ] } - } """ model_config = ConfigDict(extra='allow') @@ -229,7 +218,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 +244,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,23 +508,52 @@ 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 get_model_fields_set(cls): + """ + Get all names and aliases of the model fields. + """ + return set([field.validation_alias + for (_, field) in cls.model_fields.items()] + + [field_name for field_name in cls.model_fields]) + + # TODO should geoproperty has subproperties? and can geoproperty be subproperties? + @classmethod + def _validate_single_property(cls, attr) -> ContextProperty: + property_fields = ContextProperty.get_model_fields_set() + property_fields.remove(None) + # subattrs = {} + if attr.get("type") == "Relationship": + attr_instance = ContextRelationship.model_validate(attr) + elif attr.get("type") == "GeoProperty": + try: + attr_instance = ContextGeoProperty.model_validate(attr) + except Exception as e: + pass + elif attr.get("type") == "Property": + attr_instance = ContextProperty.model_validate(attr) + else: + raise ValueError(f"Attribute {attr.get('type')} " + "is not a valid type") + for subkey, subattr in attr.items(): + # TODO can we ensure that the subattr can only be dict? + if isinstance(subattr, dict) and subkey not in property_fields: + attr_instance.model_extra.update( + {subkey: cls._validate_single_property(attr=subattr)} + ) + return attr_instance + @classmethod def _validate_attributes(cls, data: Dict): - fields = set([field.validation_alias for (_, field) in cls.model_fields.items()] + - [field_name for field_name in cls.model_fields]) - fields.remove(None) + entity_fields = cls.get_model_fields_set() + entity_fields.remove(None) # Initialize the attribute dictionary attrs = {} - # Iterate through the data for key, attr in data.items(): # Check if the keyword is not already present in the fields - if key not in fields: - try: - attrs[key] = ContextGeoProperty.model_validate(attr) - except ValueError: - attrs[key] = ContextProperty.model_validate(attr) + if key not in entity_fields: + attrs[key] = cls._validate_single_property(attr=attr) return attrs model_config = ConfigDict(extra='allow', validate_default=True, validate_assignment=True) @@ -575,17 +574,6 @@ def _validate_id(cls, id: str): 'starting with the namespace "urn:ngsi-ld:"') return id - @classmethod - def _validate_properties(cls, data: Dict): - attrs = {} - for key, attr in data.items(): - if key not in ContextEntity.model_fields: - if attr["type"] == DataTypeLD.RELATIONSHIP: - attrs[key] = ContextRelationship.model_validate(attr) - else: - attrs[key] = ContextProperty.model_validate(attr) - return attrs - def get_properties(self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST) -> \ @@ -603,30 +591,26 @@ def get_properties(self, if response_format == PropertyFormat.DICT: final_dict = {} for key, value in self.model_dump(exclude_unset=True).items(): - if key not in ContextLDEntity.model_fields: - try: - if value.get('type') != DataTypeLD.RELATIONSHIP: - try: - final_dict[key] = ContextGeoProperty(**value) - except ValueError: # if context attribute - final_dict[key] = ContextProperty(**value) - except AttributeError: - if isinstance(value, list): - pass + if key not in ContextLDEntity.get_model_fields_set(): + if value.get('type') != DataTypeLD.RELATIONSHIP: + if value.get('type') == DataTypeLD.GEOPROPERTY: + final_dict[key] = ContextGeoProperty(**value) + elif value.get('type') == DataTypeLD.PROPERTY: + final_dict[key] = ContextProperty(**value) + else: # named context property by default + final_dict[key] = ContextProperty(**value) return final_dict # response format list: final_list = [] for key, value in self.model_dump(exclude_unset=True).items(): - if key not in ContextLDEntity.model_fields: - try: - if value.get('type') != DataTypeLD.RELATIONSHIP: - try: - final_list.append(NamedContextGeoProperty(name=key, **value)) - except ValueError: # if context attribute - final_list.append(NamedContextProperty(name=key, **value)) - except AttributeError: - if isinstance(value, list): - pass + if key not in ContextLDEntity.get_model_fields_set(): + if value.get('type') != DataTypeLD.RELATIONSHIP: + if value.get('type') == DataTypeLD.GEOPROPERTY: + final_list.append(NamedContextGeoProperty(name=key, **value)) + elif value.get('type') == DataTypeLD.PROPERTY: + final_list.append(NamedContextProperty(name=key, **value)) + else: # named context property by default + final_list.append(NamedContextProperty(name=key, **value)) return final_list def add_attributes(self, **kwargs): @@ -691,6 +675,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: """ @@ -741,7 +741,7 @@ def get_relationships(self, if response_format == PropertyFormat.DICT: final_dict = {} for key, value in self.model_dump(exclude_unset=True).items(): - if key not in ContextLDEntity.model_fields: + if key not in ContextLDEntity.get_model_fields_set(): try: if value.get('type') == DataTypeLD.RELATIONSHIP: final_dict[key] = ContextRelationship(**value) @@ -752,13 +752,9 @@ def get_relationships(self, # response format list: final_list = [] for key, value in self.model_dump(exclude_unset=True).items(): - if key not in ContextLDEntity.model_fields: - try: - if value.get('type') == DataTypeLD.RELATIONSHIP: - final_list.append(NamedContextRelationship(name=key, **value)) - except AttributeError: # if context attribute - if isinstance(value, list): - pass + if key not in ContextLDEntity.get_model_fields_set(): + if value.get('type') == DataTypeLD.RELATIONSHIP: + final_list.append(NamedContextRelationship(name=key, **value)) return final_list def get_context(self): @@ -769,15 +765,12 @@ def get_context(self): Returns: context of the entity as list """ - found_list = False - for key, value in self.model_dump(exclude_unset=True).items(): - if key not in ContextLDEntity.model_fields: - if isinstance(value, list): - found_list = True - return value - if not found_list: + _, context = self.model_dump(include={"context"}).popitem() + if not context: logging.warning("No context in entity") - return None + return None + else: + return context class ActionTypeLD(str, Enum): diff --git a/tests/models/test_ngsi_ld_context.py b/tests/models/test_ngsi_ld_context.py index 2dfd873a..7a26dbb3 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": { @@ -132,54 +133,152 @@ def setUp(self) -> None: } } } - # The entity for testing the nested structure of properties - self.entity_sub_props_dict = { - "id": "urn:ngsi-ld:Vehicle:test1243", - "type": "Vehicle", - "prop1": { - "type": "Property", - "value": 1, - "sub_property": { - "type": "Property", - "value": 10, - "sub_sub_property": { - "type": "Property", - "value": 100 - } + # # The entity for testing the nested structure of properties + # self.entity_sub_props_dict_wrong = { + # "id": "urn:ngsi-ld:OffStreetParking:Downtown1", + # "type": "OffStreetParking", + # "name": { + # "type": "Property", + # "value": "Downtown One" + # }, + # "totalSpotNumber": { + # "type": "Property", + # "value": 200 + # }, + # "location": { + # "type": "GeoProperty", + # "value": { + # "type": "Point", + # "coordinates": [-8.5, 41.2] + # } + # }, + # "@context": [ + # "http://example.org/ngsi-ld/latest/parking.jsonld", + # "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.3.jsonld" + # ] + # } + 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) }, - "sub_properties_list": [ - { - "sub_prop_1": { - "value": 100, - "type": "Property" - } - }, - { - "sub_prop_2": { - "value": 200, - "type": "Property" - } - } - ], + { + "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: + def test_cb_property(self) -> None: """ - Test context attribute models + Test context property models Returns: None """ - attr = ContextProperty(**{'value': "20"}) - self.assertIsInstance(attr.value, str) - attr = ContextProperty(**{'value': 20.53}) - self.assertIsInstance(attr.value, float) - attr = ContextProperty(**{'value': 20}) - self.assertIsInstance(attr.value, int) + prop = ContextProperty(**{'value': "20"}) + self.assertIsInstance(prop.value, str) + prop = ContextProperty(**{'value': 20.53}) + self.assertIsInstance(prop.value, float) + prop = ContextProperty(**{'value': 20}) + self.assertIsInstance(prop.value, int) - 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: """ @@ -187,6 +286,7 @@ def test_cb_entity(self) -> None: Returns: None """ + test = ContextLDEntity.get_model_fields_set() entity1 = ContextLDEntity(**self.entity1_dict) entity2 = ContextLDEntity(**self.entity2_dict) @@ -227,41 +327,71 @@ def test_cb_entity(self) -> None: properties = entity2.get_properties(response_format='list') self.assertIn("new_prop", [prop.name for prop in properties]) + def test_validate_subproperties_dict(self) -> None: + """ + Test the validation of multi-level properties in entities + Returns: + None + """ + entity4 = ContextLDEntity(**self.entity1_dict) + + def test_validate_subproperties_dict_wrong(self) -> None: + """ + Test the validation of multi-level properties in entities + Returns: + None + """ + entity_sub_props_dict_wrong_1 = self.entity1_dict.copy() + entity_sub_props_dict_wrong_1[ + "availableSpotNumber"]["reliability"]["type"] = "NotProperty" + with self.assertRaises(ValueError): + entity5 = ContextLDEntity(**entity_sub_props_dict_wrong_1) + entity_sub_props_dict_wrong_2 = self.entity1_dict.copy() + entity_sub_props_dict_wrong_2[ + "availableSpotNumber"]["providedBy"]["type"] = "NotRelationship" + with self.assertRaises(ValueError): + entity5 = ContextLDEntity(**entity_sub_props_dict_wrong_2) + def test_get_properties(self): """ Test the get_properties method """ - pass - entity = ContextLDEntity(id="urn:ngsi-ld:test", type="Tester") + entity = ContextLDEntity(id="urn:ngsi-ld:test", + type="Tester", + hasLocation={ + "type": "Relationship", + "object": "urn:ngsi-ld:test2" + }) properties = [ - NamedContextProperty(name="attr1"), - NamedContextProperty(name="attr2"), + NamedContextProperty(name="prop1"), + NamedContextProperty(name="prop2"), ] entity.add_properties(properties) + entity.get_properties(response_format="list") self.assertEqual(entity.get_properties(response_format="list"), properties) - def test_entity_delete_attributes(self): + def test_entity_delete_properties(self): """ - Test the delete_attributes methode + Test the delete_properties method """ - attr = ContextProperty(**{'value': 20, 'type': 'Text'}) - named_attr = NamedContextProperty(**{'name': 'test2', + prop = ContextProperty(**{'value': 20, 'type': 'Property'}) + named_prop = NamedContextProperty(**{'name': 'test2', 'value': 20, - 'type': 'Text'}) - attr3 = ContextProperty(**{'value': 20, 'type': 'Text'}) + 'type': 'Property'}) + prop3 = ContextProperty(**{'value': 20, 'type': 'Property'}) entity = ContextLDEntity(id="urn:ngsi-ld:12", type="Test") - entity.add_properties({"test1": attr, "test3": attr3}) - entity.add_properties([named_attr]) + entity.add_properties({"test1": prop, "test3": prop3}) + entity.add_properties([named_prop]) - entity.delete_properties({"test1": attr}) + entity.delete_properties({"test1": prop}) self.assertEqual(set([_prop.name for _prop in entity.get_properties()]), {"test2", "test3"}) - entity.delete_properties([named_attr]) + entity.delete_properties([named_prop]) self.assertEqual(set([_prop.name for _prop in entity.get_properties()]), {"test3"}) @@ -280,9 +410,10 @@ def test_get_context(self): self.assertEqual(self.entity1_context, context_entity1) - # test here if entity without context can be validated and get_context works accordingly: + # test here if entity without context can be validated and get_context + # works accordingly: entity3 = ContextLDEntity(**self.entity3_dict) context_entity3 = entity3.get_context() self.assertEqual(None, - context_entity3) \ No newline at end of file + context_entity3) diff --git a/tests/models/test_ngsi_ld_subscriptions.py b/tests/models/test_ngsi_ld_subscriptions.py index e02f8ffc..aa4d76a0 100644 --- a/tests/models/test_ngsi_ld_subscriptions.py +++ b/tests/models/test_ngsi_ld_subscriptions.py @@ -231,11 +231,4 @@ def test_query_string_serialization(self): examples[9] = 'address[city]=="Berlin".' examples[10] = 'sensor.rawdata[airquality.particulate]==40' for example in examples.values(): - validate_ngsi_ld_query(example) - - def tearDown(self) -> None: - """ - Cleanup test server - """ - clear_all(fiware_header=self.fiware_header, - cb_url=settings.CB_URL) \ No newline at end of file + validate_ngsi_ld_query(example) \ No newline at end of file