feat(attribute-completeness): support custom filter

CHANGELOG.md

@@ -1,6 +1,12 @@
 # Changelog
 
-## Current Main 
+## Current Main
+
+### New Features
+
+- Support custom attribute definition via ohsome filter query for the Attribute Completeness indicator ([#848])
+
+[#848]: https://github.com/GIScience/ohsome-quality-api/pull/848
 
 
 ## Release  1.7.0

ohsome_quality_api/api/api.py

@@ -25,7 +25,8 @@
     oqt,
 )
 from ohsome_quality_api.api.request_models import (
-    AttributeCompletenessRequest,
+    AttributeCompletenessFilterRequest,
+    AttributeCompletenessKeyRequest,
     IndicatorDataRequest,
     IndicatorRequest,
 )
@@ -268,13 +269,15 @@ async def post_indicator_ms(parameters: IndicatorDataRequest) -> CustomJSONRespo
 )
 async def post_attribute_completeness(
     request: Request,
-    parameters: AttributeCompletenessRequest,
+    parameters: AttributeCompletenessKeyRequest | AttributeCompletenessFilterRequest,
 ) -> Any:
     """Request the Attribute Completeness indicator for your area of interest."""
-    for attribute in parameters.attribute_keys:
-        validate_attribute_topic_combination(
-            attribute.value, parameters.topic_key.value
-        )
+    if isinstance(parameters, AttributeCompletenessKeyRequest):
+        for attribute in parameters.attribute_keys:
+            validate_attribute_topic_combination(
+                attribute,
+                parameters.topic,
+            )
 
     return await _post_indicator(request, "attribute-completeness", parameters)
 
@@ -306,19 +309,12 @@ async def post_indicator(
 
 
 async def _post_indicator(
-    request: Request, key: str, parameters: IndicatorRequest
+    request: Request,
+    key: str,
+    parameters: IndicatorRequest,
 ) -> Any:
-    validate_indicator_topic_combination(key, parameters.topic_key.value)
-    attribute_keys = getattr(parameters, "attribute_keys", None)
-    if attribute_keys:
-        attribute_keys = [attribute.value for attribute in attribute_keys]
-    indicators = await oqt.create_indicator(
-        key=key,
-        bpolys=parameters.bpolys,
-        topic=get_topic_preset(parameters.topic_key.value),
-        include_figure=parameters.include_figure,
-        attribute_keys=attribute_keys,
-    )
+    validate_indicator_topic_combination(key, parameters.topic)
+    indicators = await oqt.create_indicator(key=key, **dict(parameters))
 
     if request.headers["accept"] == MEDIA_TYPE_JSON:
         return {
@@ -339,10 +335,12 @@ async def _post_indicator(
         }
     else:
         detail = "Content-Type needs to be either {0} or {1}".format(
-            MEDIA_TYPE_JSON, MEDIA_TYPE_GEOJSON
+            MEDIA_TYPE_JSON,
+            MEDIA_TYPE_GEOJSON,
         )
         raise HTTPException(
-            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail=detail
+            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+            detail=detail,
         )
 
 

ohsome_quality_api/api/request_models.py

@@ -2,11 +2,16 @@
 
 import geojson
 from geojson_pydantic import Feature, FeatureCollection, MultiPolygon, Polygon
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+)
 
 from ohsome_quality_api.attributes.definitions import AttributeEnum
-from ohsome_quality_api.topics.definitions import TopicEnum
-from ohsome_quality_api.topics.models import TopicData
+from ohsome_quality_api.topics.definitions import TopicEnum, get_topic_preset
+from ohsome_quality_api.topics.models import TopicData, TopicDefinition
 from ohsome_quality_api.utils.helper import snake_to_lower_camel
 
 
@@ -49,29 +54,54 @@ class BaseBpolys(BaseConfig):
 
     @field_validator("bpolys")
     @classmethod
-    def transform(cls, value) -> geojson.FeatureCollection:
+    def transform_bpolys(cls, value) -> geojson.FeatureCollection:
         # NOTE: `geojson_pydantic` library is used only for validation and openAPI-spec
         # generation. To avoid refactoring all code the FeatureCollection object of
         # the `geojson` library is still used every else.
         return geojson.loads(value.model_dump_json())
 
 
 class IndicatorRequest(BaseBpolys):
-    topic_key: TopicEnum = Field(
+    topic: TopicEnum = Field(
         ...,
         title="Topic Key",
         alias="topic",
     )
     include_figure: bool = True
 
+    @field_validator("topic")
+    @classmethod
+    def transform_topic(cls, value) -> TopicDefinition:
+        return get_topic_preset(value.value)
+
 
-class AttributeCompletenessRequest(IndicatorRequest):
+class AttributeCompletenessKeyRequest(IndicatorRequest):
     attribute_keys: List[AttributeEnum] = Field(
         ...,
         title="Attribute Keys",
         alias="attributes",
     )
 
+    @field_validator("attribute_keys")
+    @classmethod
+    def transform_attributes(cls, value) -> list[str]:
+        return [attribute.value for attribute in value]
+
+
+class AttributeCompletenessFilterRequest(IndicatorRequest):
+    attribute_filter: str = Field(
+        ...,
+        title="Attribute Filter",
+        description="ohsome filter query representing custom attributes.",
+    )
+    attribute_title: str = Field(
+        ...,
+        title="Attribute Title",
+        description=(
+            "Title describing the attributes represented by the Attribute Filter."
+        ),
+    )
+
 
 class IndicatorDataRequest(BaseBpolys):
     """Model for the `/indicators/mapping-saturation/data` endpoint.

ohsome_quality_api/indicators/attribute_completeness/indicator.py

@@ -1,6 +1,5 @@
 import logging
 from string import Template
-from typing import List
 
 import dateutil.parser
 import plotly.graph_objects as go
@@ -24,8 +23,12 @@ class AttributeCompleteness(BaseIndicator):
 
     Terminology:
         topic: Category of map features. Translates to a ohsome filter.
-        attribute: Additional (expected) tag(s) describing a map feature. Translates to
-            a ohsome filter.
+        attribute: Additional (expected) tag(s) describing a map feature.
+            attribute_keys: a set of predefined attributes wich will be
+                translated to an ohsome filter
+            attribute_filter: ohsome filter query representing custom attributes
+            attribute_title:  Title describing the attributes represented by
+                the Attribute Filter
 
     Example: How many buildings (topic) have height information (attribute)?
 
@@ -40,23 +43,37 @@ def __init__(
         self,
         topic: Topic,
         feature: Feature,
-        attribute_keys: List[str] = None,
+        attribute_keys: list[str] | None = None,
+        attribute_filter: str | None = None,
+        attribute_title: str | None = None,
     ) -> None:
         super().__init__(topic=topic, feature=feature)
         self.threshold_yellow = 0.75
         self.threshold_red = 0.25
         self.attribute_keys = attribute_keys
+        self.attribute_filter = attribute_filter
+        self.attribute_title = attribute_title
         self.absolute_value_1 = None
         self.absolute_value_2 = None
         self.description = None
+        if self.attribute_keys:
+            self.attribute_filter = build_attribute_filter(
+                self.attribute_keys,
+                self.topic.key,
+            )
+            self.attribute_title = ", ".join(
+                [
+                    get_attribute(self.topic.key, k).name.lower()
+                    for k in self.attribute_keys
+                ]
+            )
 
     async def preprocess(self) -> None:
-        attribute = build_attribute_filter(self.attribute_keys, self.topic.key)
         # Get attribute filter
         response = await ohsome_client.query(
             self.topic,
             self.feature,
-            attribute_filter=attribute,
+            attribute_filter=self.attribute_filter,
         )
         timestamp = response["ratioResult"][0]["timestamp"]
         self.result.timestamp_osm = dateutil.parser.isoparse(timestamp)
@@ -90,19 +107,21 @@ def calculate(self) -> None:
             )
 
     def create_description(self):
-        attribute_names = [
-            get_attribute(self.topic.key, attribute_key).name.lower()
-            for attribute_key in self.attribute_keys
-        ]
+        if self.result.value is None:
+            raise TypeError("Result value should not be None.")
+        else:
+            result = round(self.result.value * 100, 1)
+        if self.attribute_title is None:
+            raise TypeError("Attribute title should not be None.")
+        else:
+            tags = "attributes " + self.attribute_title
         all, matched = self.compute_units_for_all_and_matched()
         self.description = Template(self.templates.result_description).substitute(
-            result=round(self.result.value * 100, 1),
+            result=result,
             all=all,
             matched=matched,
             topic=self.topic.name.lower(),
-            tags="attributes " + ", ".join(attribute_names)
-            if len(attribute_names) > 1
-            else "attribute " + attribute_names[0],
+            tags=tags,
         )
 
     def create_figure(self) -> None:

ohsome_quality_api/oqt.py

@@ -1,7 +1,7 @@
 """Controller for computing Indicators."""
 
 import logging
-from typing import Coroutine, List
+from typing import Coroutine
 
 from geojson import Feature, FeatureCollection
 
@@ -18,7 +18,8 @@ async def create_indicator(
     bpolys: FeatureCollection,
     topic: TopicData | TopicDefinition,
     include_figure: bool = True,
-    attribute_keys: List[str] = None,
+    *args,
+    **kwargs,
 ) -> list[Indicator]:
     """Create indicator(s) for features of a GeoJSON FeatureCollection.
 
@@ -40,7 +41,14 @@ async def create_indicator(
         ]:
             validate_area(feature)
         tasks.append(
-            _create_indicator(key, feature, topic, include_figure, attribute_keys)
+            _create_indicator(
+                key,
+                feature,
+                topic,
+                include_figure,
+                *args,
+                **kwargs,
+            )
         )
     return await gather_with_semaphore(tasks)
 
@@ -50,7 +58,8 @@ async def _create_indicator(
     feature: Feature,
     topic: Topic,
     include_figure: bool = True,
-    attribute_keys: List[str] = None,
+    *args,
+    **kwargs,
 ) -> Indicator:
     """Create an indicator from scratch."""
 
@@ -59,13 +68,16 @@ async def _create_indicator(
     logging.info("Feature id:     {0:4}".format(feature.get("id", "None")))
 
     indicator_class = get_class_from_key(class_type="indicator", key=key)
-    if key == "attribute-completeness":
-        indicator = indicator_class(topic, feature, attribute_keys)
-    else:
-        indicator = indicator_class(topic, feature)
+    indicator = indicator_class(
+        topic,
+        feature,
+        *args,
+        **kwargs,
+    )
 
     logging.info("Run preprocessing")
     await indicator.preprocess()
+
     logging.info("Run calculation")
     indicator.calculate()
 

ohsome_quality_api/utils/validators.py

@@ -7,7 +7,7 @@
 )
 from ohsome_quality_api.config import get_config_value
 from ohsome_quality_api.indicators.definitions import get_valid_indicators
-from ohsome_quality_api.topics.definitions import TopicEnum
+from ohsome_quality_api.topics.models import BaseTopic
 from ohsome_quality_api.utils.exceptions import (
     AttributeTopicCombinationError,
     GeoJSONError,
@@ -20,19 +20,23 @@
 from ohsome_quality_api.utils.helper_geo import calculate_area
 
 
-def validate_attribute_topic_combination(attribute: AttributeEnum, topic: TopicEnum):
+def validate_attribute_topic_combination(attribute: AttributeEnum, topic: BaseTopic):
     """As attributes are only meaningful for a certain topic,
     we need to check if the given combination is valid."""
 
-    valid_attributes_for_topic = get_attributes()[topic]
+    valid_attributes_for_topic = get_attributes()[topic.key]
     valid_attribute_names = [attribute for attribute in valid_attributes_for_topic]
 
     if attribute not in valid_attributes_for_topic:
-        raise AttributeTopicCombinationError(attribute, topic, valid_attribute_names)
+        raise AttributeTopicCombinationError(
+            attribute,
+            topic.key,
+            valid_attribute_names,
+        )
 
 
-def validate_indicator_topic_combination(indicator: str, topic: str):
-    if indicator not in get_valid_indicators(topic):
+def validate_indicator_topic_combination(indicator: str, topic: BaseTopic):
+    if indicator not in get_valid_indicators(topic.key):
         raise IndicatorTopicCombinationError(indicator, topic)
 
 

tests/conftest.py

@@ -136,15 +136,27 @@ def attribute() -> Attribute:
 
 
 @pytest.fixture(scope="class")
-def attribute_key() -> str:
+def attribute_key() -> list[str]:
     return ["height"]
 
 
 @pytest.fixture(scope="class")
-def attribute_key_multiple() -> str:
+def attribute_key_multiple() -> list[str]:
     return ["height", "house-number"]
 
 
+@pytest.fixture
+def attribute_filter() -> str:
+    """Custom attribute filter."""
+    return "height=* or building:levels=*"
+
+
+@pytest.fixture
+def attribute_title() -> str:
+    """Attributes title belonging to custom attribute filter (`attribute_filter)`."""
+    return "Height"
+
+
 @pytest.fixture(scope="class")
 def feature_germany_heidelberg() -> Feature:
     path = os.path.join(