feat(attribute-completeness): support custom filter

support custom attribute ohsome filter

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,10 @@ 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: a custom ohsome filter
 
     Example: How many buildings (topic) have height information (attribute)?
 
@@ -40,23 +41,45 @@ def __init__(
         self,
         topic: Topic,
         feature: Feature,
-        attribute_keys: List[str] = None,
+        attribute_keys: list[str] | None = None,
+        attribute_filter: str | None = None,
+        attribute_names: list[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_names = attribute_names
         self.absolute_value_1 = None
         self.absolute_value_2 = None
         self.description = None
+        # fmt: off
+        # TODO: Remove once validated by pydantic request model
+        if (
+            all(v is None for v in (attribute_keys, attribute_filter)) or
+            all(v is not None for v in (attribute_keys, attribute_filter))
+        ):
+            raise TypeError(
+                "Either `attribute_keys` or `attribute_filter` needs to be given"
+            )
+        # fmt: on
+        if self.attribute_keys:
+            self.attribute_filter = build_attribute_filter(
+                self.attribute_keys,
+                self.topic.key,
+            )
+            self.attribute_names = [
+                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 +113,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
-        ]
         all, matched = self.compute_units_for_all_and_matched()
+        if self.result.value is None:
+            raise TypeError("result value should not be None")
+        else:
+            result = round(self.result.value * 100, 1)
+        if len(self.attribute_names) > 1:
+            tags = "attributes " + ", ".join(self.attribute_names)
+        else:
+            tags = "attribute " + self.attribute_names[0]
         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:

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_names() -> list[str]:
+    """Attributes names belonging to custom attribute filter (`attribute_filter)`."""
+    return ["Height"]
+
+
 @pytest.fixture(scope="class")
 def feature_germany_heidelberg() -> Feature:
     path = os.path.join(

tests/integrationtests/fixtures/approved/test_attribute_completeness-TestFigure-test_create_figure-indicator0 (setup).approved.txt

@@ -0,0 +1 @@
+28.8% of all "building count" features (all: 29936 elements) in your area of interest have the selected additional attribute height of buildings (matched: 8630 elements). 

tests/integrationtests/fixtures/approved/test_attribute_completeness-TestFigure-test_create_figure-indicator1 (setup).approved.txt

@@ -0,0 +1 @@
+59.7% of all "building count" features (all: 30237 elements) in your area of interest have the selected additional attribute Height (matched: 18057 elements). 

tests/integrationtests/fixtures/approved/test_attribute_completeness-test_create_description_attribute_filter.approved.txt

@@ -0,0 +1 @@
+20.0% of all "building count" features (all: 10 elements) in your area of interest have the selected additional attribute Height (matched: 2 elements). 

tests/integrationtests/fixtures/approved/test_attribute_completeness-test_create_description_attribute_keys_multiple.approved.txt

@@ -0,0 +1 @@
+20.0% of all "building count" features (all: 10 elements) in your area of interest have the selected additional attributes height of buildings, house number, street address (matched: 2 elements). 

tests/integrationtests/fixtures/approved/test_attribute_completeness-test_create_description_attribute_keys_single.approved.txt

@@ -0,0 +1 @@
+20.0% of all "building count" features (all: 10 elements) in your area of interest have the selected additional attribute height of buildings (matched: 2 elements). 

tests/integrationtests/indicators/test_attribute_completeness.py

@@ -6,57 +6,116 @@
 import plotly.graph_objects as pgo
 import plotly.io as pio
 import pytest
+from approvaltests import verify
 
 from ohsome_quality_api.indicators.attribute_completeness.indicator import (
     AttributeCompleteness,
 )
-from tests.integrationtests.utils import get_topic_fixture, oqapi_vcr
+from tests.integrationtests.utils import PytestNamer, get_topic_fixture, oqapi_vcr
+
+
+class TestInit:
+    @oqapi_vcr.use_cassette
+    def test_preprocess_missing_parameter(
+        self, topic_building_count, feature_germany_heidelberg
+    ):
+        with pytest.raises(TypeError):
+            AttributeCompleteness(
+                topic_building_count,
+                feature_germany_heidelberg,
+            )
+
+    def test_preprocess_too_many_parameter(
+        self,
+        topic_building_count,
+        feature_germany_heidelberg,
+        attribute_key,
+        attribute_filter,
+    ):
+        with pytest.raises(TypeError):
+            AttributeCompleteness(
+                topic_building_count,
+                feature_germany_heidelberg,
+                attribute_key,
+                attribute_filter,
+            )
 
 
 class TestPreprocess:
     @oqapi_vcr.use_cassette
-    def test_preprocess(
+    def test_preprocess_attribute_keys_single(
         self, topic_building_count, feature_germany_heidelberg, attribute_key
     ):
         indicator = AttributeCompleteness(
             topic_building_count,
             feature_germany_heidelberg,
-            attribute_key,
+            attribute_keys=attribute_key,
         )
         asyncio.run(indicator.preprocess())
         assert indicator.result.value is not None
         assert isinstance(indicator.result.timestamp, datetime)
         assert isinstance(indicator.result.timestamp_osm, datetime)
 
     @oqapi_vcr.use_cassette
-    def test_preprocess_multiple_attribute_keys(
+    def test_preprocess_attribute_keys_multiple(
         self, topic_building_count, feature_germany_heidelberg, attribute_key_multiple
     ):
         indicator = AttributeCompleteness(
             topic_building_count,
             feature_germany_heidelberg,
-            attribute_key_multiple,
+            attribute_keys=attribute_key_multiple,
         )
         asyncio.run(indicator.preprocess())
         assert indicator.result.value is not None
         assert isinstance(indicator.result.timestamp, datetime)
         assert isinstance(indicator.result.timestamp_osm, datetime)
 
-
-class TestCalculation:
-    @pytest.fixture(scope="class")
     @oqapi_vcr.use_cassette
-    def indicator(
+    def test_preprocess_attribute_filter(
         self,
         topic_building_count,
         feature_germany_heidelberg,
-        attribute_key,
+        attribute_filter,
+        attribute_names,
     ):
-        i = AttributeCompleteness(
+        indicator = AttributeCompleteness(
             topic_building_count,
             feature_germany_heidelberg,
-            attribute_key,
+            attribute_filter=attribute_filter,
+            attribute_names=attribute_names,
         )
+        asyncio.run(indicator.preprocess())
+        assert indicator.result.value is not None
+        assert isinstance(indicator.result.timestamp, datetime)
+        assert isinstance(indicator.result.timestamp_osm, datetime)
+
+
+class TestCalculation:
+    @pytest.fixture(
+        scope="class",
+        params=(
+            ["height"],
+            {
+                "attribute_filter": "height=* or building:levels=*",
+                "attribute_names": ["Height"],
+            },
+        ),
+    )
+    @oqapi_vcr.use_cassette
+    def indicator(self, request, topic_building_count, feature_germany_heidelberg):
+        if isinstance(request.param, list):
+            i = AttributeCompleteness(
+                topic_building_count,
+                feature_germany_heidelberg,
+                attribute_keys=request.param,
+            )
+        else:
+            i = AttributeCompleteness(
+                topic_building_count,
+                feature_germany_heidelberg,
+                attribute_filter=request.param["attribute_filter"],
+                attribute_names=request.param["attribute_names"],
+            )
         asyncio.run(i.preprocess())
         i.calculate()
         return i
@@ -69,8 +128,8 @@ def test_calculate(self, indicator):
         assert isinstance(indicator.result.timestamp, datetime)
         assert isinstance(indicator.result.timestamp_osm, datetime)
 
-    @oqapi_vcr.use_cassette()
-    def test_no_features(self, attribute):
+    @oqapi_vcr.use_cassette
+    def test_no_features(self):
         """Test area with no features"""
         infile = os.path.join(
             os.path.dirname(os.path.abspath(__file__)),
@@ -95,26 +154,37 @@ def test_no_features(self, attribute):
 
 
 class TestFigure:
-    @pytest.fixture(scope="class")
+    @pytest.fixture(
+        scope="class",
+        params=(
+            ["height"],
+            {
+                "attribute_filter": "height=* or building:levels=*",
+                "attribute_names": ["Height"],
+            },
+        ),
+    )
     @oqapi_vcr.use_cassette
     def indicator(
         self,
+        request,
         topic_building_count,
         feature_germany_heidelberg,
-        attribute_key,
     ):
-        indicator = AttributeCompleteness(
-            topic_building_count,
-            feature_germany_heidelberg,
-            attribute_key,
-        )
+        if isinstance(request.param, list):
+            indicator = AttributeCompleteness(
+                topic_building_count, feature_germany_heidelberg, request.param
+            )
+        else:
+            indicator = AttributeCompleteness(
+                topic_building_count,
+                feature_germany_heidelberg,
+                attribute_filter=request.param["attribute_filter"],
+                attribute_names=request.param["attribute_names"],
+            )
         asyncio.run(indicator.preprocess())
         indicator.calculate()
-        assert indicator.description == (
-            '28.8% of all "building count" features (all: 29936 elements)'
-            " in your area of interest have the selected additional attribute"
-            " height of buildings (matched: 8630 elements). "
-        )
+        verify(indicator.description, namer=PytestNamer())
         return indicator
 
     # comment out for manual test
@@ -129,7 +199,7 @@ def test_create_figure(self, indicator):
         pgo.Figure(indicator.result.figure)  # test for valid Plotly figure
 
 
-def test_create_description():
+def test_create_description_attribute_keys_single():
     indicator = AttributeCompleteness(
         get_topic_fixture("building-count"),
         "foo",
@@ -139,14 +209,10 @@ def test_create_description():
     indicator.absolute_value_1 = 10
     indicator.absolute_value_2 = 2
     indicator.create_description()
-    assert indicator.description == (
-        '20.0% of all "building count" features (all: 10 elements) in your area of '
-        "interest have the selected additional attribute height of buildings "
-        "(matched: 2 elements). "
-    )
+    verify(indicator.description, namer=PytestNamer())
 
 
-def test_create_description_multiple_attributes():
+def test_create_description_attribute_keys_multiple():
     indicator = AttributeCompleteness(
         get_topic_fixture("building-count"),
         "foo",
@@ -156,11 +222,21 @@ def test_create_description_multiple_attributes():
     indicator.absolute_value_1 = 10
     indicator.absolute_value_2 = 2
     indicator.create_description()
-    assert indicator.description == (
-        '20.0% of all "building count" features (all: 10 elements) in your area of '
-        "interest have the selected additional attributes height of buildings, house "
-        "number, street address (matched: 2 elements). "
+    verify(indicator.description, namer=PytestNamer())
+
+
+def test_create_description_attribute_filter(attribute_filter, attribute_names):
+    indicator = AttributeCompleteness(
+        get_topic_fixture("building-count"),
+        "foo",
+        attribute_filter=attribute_filter,
+        attribute_names=attribute_names,
     )
+    indicator.result.value = 0.2
+    indicator.absolute_value_1 = 10
+    indicator.absolute_value_2 = 2
+    indicator.create_description()
+    verify(indicator.description, namer=PytestNamer())
 
 
 @pytest.mark.parametrize(