feat(attribute-completeness): use sql query

Optionally use SQL based query instead of ohsome API query given a
feature flag.

Author: Gigaszi

ohsome_quality_api/indicators/attribute_completeness/indicator.py

@@ -1,7 +1,9 @@
+import json
 import logging
+import os
 from string import Template
 
-import dateutil.parser
+import dateutil
 import plotly.graph_objects as go
 from geojson import Feature
 
@@ -12,6 +14,10 @@
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client
 from ohsome_quality_api.topics.models import BaseTopic as Topic
+from ohsome_quality_api.trino import client as trino_client
+from ohsome_quality_api.utils.helper_geo import get_bounding_box
+
+WORKING_DIR = os.path.dirname(os.path.abspath(__file__))
 
 
 class AttributeCompleteness(BaseIndicator):
@@ -46,8 +52,10 @@ def __init__(
         attribute_keys: list[str] | None = None,
         attribute_filter: str | None = None,
         attribute_title: str | None = None,
+        feature_flag_sql: bool = False,  # Feature flag to use SQL instead of ohsome API queries
     ) -> None:
         super().__init__(topic=topic, feature=feature)
+        self.feature_flag_sql = feature_flag_sql
         self.threshold_yellow = 0.75
         self.threshold_red = 0.25
         self.attribute_keys = attribute_keys
@@ -56,7 +64,9 @@ def __init__(
         self.absolute_value_1 = None
         self.absolute_value_2 = None
         self.description = None
-        if self.attribute_keys:
+        if self.feature_flag_sql:
+            self.attribute_filter = attribute_filter
+        elif self.attribute_keys:
             self.attribute_filter = build_attribute_filter(
                 self.attribute_keys,
                 self.topic.key,
@@ -74,17 +84,50 @@ def __init__(
             )
 
     async def preprocess(self) -> None:
-        # Get attribute filter
-        response = await ohsome_client.query(
-            self.topic,
-            self.feature,
-            attribute_filter=self.attribute_filter,
-        )
-        timestamp = response["ratioResult"][0]["timestamp"]
-        self.result.timestamp_osm = dateutil.parser.isoparse(timestamp)
-        self.result.value = response["ratioResult"][0]["ratio"]
-        self.absolute_value_1 = response["ratioResult"][0]["value"]
-        self.absolute_value_2 = response["ratioResult"][0]["value2"]
+        if self.feature_flag_sql:
+            filter = self.topic.sql_filter
+            file_path = os.path.join(WORKING_DIR, "query.sql")
+            with open(file_path, "r") as file:
+                template = file.read()
+
+            bounding_box = get_bounding_box(self.feature)
+            geometry = json.dumps(self.feature["geometry"])
+
+            sql = template.format(
+                bounding_box=bounding_box,
+                geometry=geometry,
+                filter=filter,
+            )
+            query = await trino_client.query(sql)
+            results = await trino_client.fetch(query)
+            # TODO: Check for None
+            self.absolute_value_1 = results[0][0]
+
+            filter = self.topic.sql_filter + " AND " + self.attribute_filter
+            sql = template.format(
+                bounding_box=bounding_box,
+                geometry=geometry,
+                filter=filter,
+            )
+            query = await trino_client.query(sql)
+            results = await trino_client.fetch(query)
+            self.absolute_value_2 = results[0][0]
+
+            # timestamp = response["ratioResult"][0]["timestamp"]
+            # self.result.timestamp_osm = dateutil.parser.isoparse(timestamp)
+            self.result.value = self.absolute_value_2 / self.absolute_value_1
+        else:
+            # Get attribute filter
+            response = await ohsome_client.query(
+                self.topic,
+                self.feature,
+                attribute_filter=self.attribute_filter,
+            )
+            timestamp = response["ratioResult"][0]["timestamp"]
+            self.result.timestamp_osm = dateutil.parser.isoparse(timestamp)
+            self.result.value = response["ratioResult"][0]["ratio"]
+            self.absolute_value_1 = response["ratioResult"][0]["value"]
+            self.absolute_value_2 = response["ratioResult"][0]["value2"]
 
     def calculate(self) -> None:
         # result (ratio) can be NaN if no features matching filter1

ohsome_quality_api/indicators/attribute_completeness/query.sql

@@ -0,0 +1,20 @@
+WITH bpoly AS (
+    SELECT
+        to_geometry (from_geojson_geometry ('{geometry}')) AS geometry
+)
+SELECT
+    Sum(
+        CASE WHEN ST_Within (ST_GeometryFromText (contributions.geometry), bpoly.geometry) THEN
+            length
+        ELSE
+            Cast(ST_Length (ST_Intersection (ST_GeometryFromText (contributions.geometry), bpoly.geometry)) AS integer)
+        END) AS length
+FROM
+    bpoly,
+    sotm2024_iceberg.geo_sort.contributions AS contributions
+WHERE
+    status = 'latest'
+    AND ST_Intersects (bpoly.geometry, ST_GeometryFromText (contributions.geometry))
+    AND {filter}
+    AND (bbox.xmax <= {bounding_box.lon_max} AND bbox.xmin >= {bounding_box.lon_min})
+    AND (bbox.ymax <= {bounding_box.lat_max} AND bbox.ymin >= {bounding_box.lat_min})

ohsome_quality_api/topics/models.py

@@ -36,6 +36,7 @@ class TopicDefinition(BaseTopic):
     projects: list[ProjectEnum]
     source: str | None = None
     ratio_filter: str | None = None
+    sql_filter: str | None = None
 
 
 class TopicData(BaseTopic):

ohsome_quality_api/topics/presets.yaml

@@ -63,6 +63,13 @@ car-roads:
   aggregation_type: length
   filter: >-
     highway in (motorway, trunk, primary, secondary, tertiary, residential, service, living_street, trunk_link, motorway_link, primary_link, secondary_link, tertiary_link) and type:way
+  sql_filter: >-
+    osm_type = 'way'
+    AND element_at (contributions.tags, 'highway') IS NOT NULL
+    AND contributions.tags['highway'] IN ('motorway', 'trunk',
+    'motorway_link', 'trunk_link', 'primary', 'primary_link', 'secondary',
+    'secondary_link', 'tertiary', 'tertiary_link', 'unclassified',
+    'residential')
   indicators:
     - mapping-saturation
     - attribute-completeness

tests/conftest.py

@@ -58,6 +58,11 @@ def topic_building_count(topic_key_building_count) -> TopicDefinition:
     return get_topic_preset(topic_key_building_count)
 
 
+@pytest.fixture(scope="class")
+def topic_car_roads() -> TopicDefinition:
+    return get_topic_preset("car-roads")
+
+
 @pytest.fixture(scope="class")
 def topic_building_area() -> TopicDefinition:
     return get_topic_preset("building-area")

tests/integrationtests/indicators/test_attribute_completeness.py

@@ -177,6 +177,24 @@ def test_create_figure(self, indicator):
         pgo.Figure(indicator.result.figure)  # test for valid Plotly figure
 
 
+class TestFeatureFlagSql:
+    # TODO: use VCR cassette
+    def test_preprocess_attribute_sql_filter(
+        self,
+        topic_car_roads,
+        feature_germany_heidelberg,
+    ):
+        attribute_filter = "element_at (contributions.tags, 'name') IS NOT NULL"
+        indicator = AttributeCompleteness(
+            topic_car_roads,
+            feature_germany_heidelberg,
+            attribute_filter=attribute_filter,
+            feature_flag_sql=True,
+        )
+        asyncio.run(indicator.preprocess())
+        assert indicator.result.value is not None
+
+
 def test_create_description_attribute_keys_single():
     indicator = AttributeCompleteness(
         get_topic_fixture("building-count"),