feat: get data fro Attriute Completeness, Building Comparison and Land Cover Completeness from ohsomedb

ohsome_quality_api/attributes/definitions.py

@@ -52,7 +52,31 @@ def get_attribute_preset(topic_key: str) -> List[Attribute]:
         ) from error
 
 
-def build_attribute_filter(attribute_key: List[str] | str, topic_key: str) -> str:
+def build_attribute_filter_ohsomedb(
+    attribute_key: List[str] | str, topic_key: str
+) -> str:
+    """Build attribute filter for ohsome API query."""
+    attributes = get_attributes()
+    try:
+        if isinstance(attribute_key, str):
+            return attribute_key
+        else:
+            attribute_filter = ""
+            for i, key in enumerate(attribute_key):
+                if i == 0:
+                    attribute_filter = "(" + attributes[topic_key][key].filter + ")"
+                else:
+                    attribute_filter += (
+                        " and (" + attributes[topic_key][key].filter + ")"
+                    )
+            return attribute_filter
+    except KeyError as error:
+        raise KeyError("Invalid topic or attribute key(s).") from error
+
+
+def build_attribute_filter_ohsomeapi(
+    attribute_key: List[str] | str, topic_key: str
+) -> str:
     """Build attribute filter for ohsome API query."""
     attributes = get_attributes()
     try:

ohsome_quality_api/indicators/attribute_completeness/indicator.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import datetime, timezone
 from string import Template
 
 import dateutil.parser
@@ -7,10 +8,13 @@
 from fastapi_i18n import _, get_locale
 from geojson import Feature
 
+from ohsome_quality_api import ohsomedb
 from ohsome_quality_api.attributes.definitions import (
-    build_attribute_filter,
+    build_attribute_filter_ohsomeapi,
+    build_attribute_filter_ohsomedb,
     get_attribute,
 )
+from ohsome_quality_api.config import get_config_value
 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 Topic
@@ -60,24 +64,63 @@ def __init__(
         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
-                ]
-            )
+        ohsomedb_enabled = get_config_value("ohsomedb_enabled")
+        if ohsomedb_enabled is True or ohsomedb_enabled in ("True", "true"):
+            if self.attribute_keys:
+                self.attribute_filter = build_attribute_filter_ohsomedb(
+                    self.attribute_keys,
+                    self.topic.key,
+                )
+                self.attribute_title = ", ".join(
+                    [
+                        get_attribute(self.topic.key, k).name.lower()
+                        for k in self.attribute_keys
+                    ]
+                )
+            else:
+                self.attribute_filter = build_attribute_filter_ohsomedb(
+                    self.attribute_filter,
+                    self.topic.key,
+                )
         else:
-            self.attribute_filter = build_attribute_filter(
-                self.attribute_filter,
-                self.topic.key,
-            )
+            if self.attribute_keys:
+                self.attribute_filter = build_attribute_filter_ohsomeapi(
+                    self.attribute_keys,
+                    self.topic.key,
+                )
+                self.attribute_title = ", ".join(
+                    [
+                        get_attribute(self.topic.key, k).name.lower()
+                        for k in self.attribute_keys
+                    ]
+                )
+            else:
+                self.attribute_filter = build_attribute_filter_ohsomeapi(
+                    self.attribute_filter,
+                    self.topic.key,
+                )
+
+    async def preprocess(self):
+        ohsomedb_enabled = get_config_value("ohsomedb_enabled")
+        if ohsomedb_enabled is True or ohsomedb_enabled in ("True", "true"):
+            await self.preprocess_ohsomedb()
+        else:
+            await self.preprocess_ohsomeapi()
+
+    async def preprocess_ohsomedb(self) -> None:
+        # Get attribute filter
+        result = await ohsomedb.attribute_completeness(
+            aggregation=self.topic.aggregation_type,
+            bpolys=self.feature.geometry,
+            filter_=self.topic.filter,
+            attribute_filter_=self.attribute_filter,
+        )
+        self.result.timestamp_osm = datetime.now(timezone.utc)
+        self.result.value = result[0]["attribute_completeness"]
+        self.absolute_value_1 = result[0]["total_aggregation"]
+        self.absolute_value_2 = result[0]["aggregation_with_attribute"]
 
-    async def preprocess(self) -> None:
+    async def preprocess_ohsomeapi(self) -> None:
         # Get attribute filter
         response = await ohsome_client.query(
             self.topic,
@@ -92,7 +135,7 @@ async def preprocess(self) -> None:
 
     def calculate(self) -> None:
         # result (ratio) can be NaN if no features matching filter1
-        if self.result.value == "NaN":
+        if self.result.value == "NaN" or self.absolute_value_1 == 0:
             self.result.value = None
         if self.result.value is None:
             self.result.description += _(" No features in this region")

ohsome_quality_api/indicators/building_comparison/indicator.py

@@ -13,6 +13,8 @@
 from geojson import Feature
 from numpy import mean
 
+from ohsome_quality_api import ohsomedb
+from ohsome_quality_api.config import get_config_value
 from ohsome_quality_api.definitions import Color, get_attribution
 from ohsome_quality_api.geodatabase import client as db_client
 from ohsome_quality_api.indicators.base import BaseIndicator
@@ -71,7 +73,14 @@ async def coverage(cls, inverse=False) -> list[Feature]:
     def attribution(cls) -> str:
         return get_attribution(["OSM", "EUBUCCO", "Microsoft Buildings"])
 
-    async def preprocess(self) -> None:
+    async def preprocess(self):
+        ohsomedb_enabled = get_config_value("ohsomedb_enabled")
+        if ohsomedb_enabled is True or ohsomedb_enabled in ("True", "true"):
+            await self.preprocess_ohsomedb()
+        else:
+            await self.preprocess_ohsomeapi()
+
+    async def preprocess_ohsomeapi(self) -> None:
         for key, val in self.data_ref.items():
             # get coverage [%]
             self.area_cov[key] = await db_client.get_intersection_area(
@@ -100,6 +109,38 @@ async def preprocess(self) -> None:
             timestamp = result["result"][0]["timestamp"]
             self.result.timestamp_osm = parser.isoparse(timestamp)
 
+    async def preprocess_ohsomedb(self) -> None:
+        for key, val in self.data_ref.items():
+            # get coverage [%]
+            self.area_cov[key] = await db_client.get_intersection_area(
+                self.feature,
+                val["coverage"]["simple"],
+            )
+            if self.check_major_edge_cases(key) != "":
+                continue
+
+            # clip input geom with coverage of reference dataset
+            feature = await db_client.get_intersection_geom(
+                self.feature,
+                val["coverage"]["simple"],
+            )
+
+            # get reference building area
+            result = await get_reference_building_area(
+                geojson.dumps(feature), val["table_name"]
+            )
+            self.area_ref[key] = result / (1000 * 1000)
+
+            # get osm building area
+            result = await ohsomedb.single_snapshot_aggregation(
+                aggregation=self.topic.aggregation_type,
+                bpolys=self.feature.geometry,
+                filter_=self.topic.filter,
+            )
+            value = float(result[0]["value"]) or 0.0  # if None
+            self.area_osm[key] = value / (1000 * 1000)
+            self.result.timestamp_osm = result[0]["snapshot_ts"]
+
     def calculate(self) -> None:
         major_edge_case: bool = False
         result_description: str = ""

ohsome_quality_api/indicators/land_cover_completeness/indicator.py

@@ -7,6 +7,8 @@
 from fastapi_i18n import _, get_locale
 from geojson import Feature
 
+from ohsome_quality_api import ohsomedb
+from ohsome_quality_api.config import get_config_value
 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 Topic
@@ -26,15 +28,31 @@ def __init__(
         self.th_low = 0.50  # Above or equal to this value label should be yellow
 
     async def preprocess(self):
-        # get osm building area
+        ohsomedb_enabled = get_config_value("ohsomedb_enabled")
+        if ohsomedb_enabled is True or ohsomedb_enabled in ("True", "true"):
+            await self.preprocess_ohsomedb()
+        else:
+            await self.preprocess_ohsomeapi()
 
+    async def preprocess_ohsomeapi(self):
         result = await ohsome_client.query(self.topic, self.feature, density=True)
         self.osm_area_ratio = result["result"][0]["value"] or 0.0  # if None
         timestamp = result["result"][0]["timestamp"]
         self.result.timestamp_osm = parser.isoparse(timestamp)
 
+    async def preprocess_ohsomedb(self):
+        # get osm building area
+
+        result = await ohsomedb.density(
+            aggregation=self.topic.aggregation_type,
+            bpolys=self.feature.geometry,
+            filter_=self.topic.filter,
+        )
+        self.osm_area_ratio = result[0]["value"] or 0.0  # if None
+        self.result.timestamp_osm = result[0]["snapshot_ts"]
+
     def calculate(self):
-        self.osm_area_ratio /= 1000000
+        #        self.osm_area_ratio /= 1000000
         self.result.value = round(self.osm_area_ratio, 2)
         if self.result.value >= self.th_high:
             self.result.class_ = 5

ohsome_quality_api/ohsomedb/__init__.py

@@ -1,7 +1,17 @@
 from .requests import (
+    attribute_completeness,
     contributions,
+    density,
     elements,
+    single_snapshot_aggregation,
     users,
 )
 
-__all__ = ("contributions", "elements", "users")
+__all__ = (
+    "attribute_completeness",
+    "contributions",
+    "density",
+    "elements",
+    "single_snapshot_aggregation",
+    "users",
+)

ohsome_quality_api/ohsomedb/requests.py

@@ -89,3 +89,84 @@ async def elements(
         bpolys.model_dump_json(),
         database="ohsomedb",
     )
+
+
+@validate_call
+async def attribute_completeness(
+    *,
+    aggregation: Literal["count", "length", "area"],
+    bpolys: Polygon | MultiPolygon,
+    filter_: OhsomeFilter,
+    attribute_filter_: OhsomeFilter,
+):
+    sql_filter, sql_filter_args = ohsome_filter_to_sql(filter_)
+    attribute_sql_filter, attribute_sql_filter_args = ohsome_filter_to_sql(
+        attribute_filter_, args_shift=len(sql_filter_args)
+    )
+    template = ENV.get_template("attribute_completeness.sql")
+    query = template.render(
+        **{
+            "aggregation": aggregation,
+            "contributions": get_config_value("ohsomedb_contributions_table"),
+            "geom": len(attribute_sql_filter_args) + len(sql_filter_args) + 1,
+            "filter": sql_filter,
+            "attribute_filter": attribute_sql_filter,
+        }
+    )
+    total_sql_filter_args = sql_filter_args + attribute_sql_filter_args
+    return await client.fetch(
+        query,
+        *total_sql_filter_args,
+        bpolys.model_dump_json(),
+        database="ohsomedb",
+    )
+
+
+@validate_call
+async def single_snapshot_aggregation(
+    *,
+    aggregation: Literal["count", "length", "area"],
+    bpolys: Polygon | MultiPolygon,
+    filter_: OhsomeFilter,
+):
+    sql_filter, sql_filter_args = ohsome_filter_to_sql(filter_)
+    template = ENV.get_template("single_snapshot_aggregation.sql")
+    query = template.render(
+        **{
+            "aggregation": aggregation,
+            "contributions": get_config_value("ohsomedb_contributions_table"),
+            "geom": len(sql_filter_args) + 1,
+            "filter": sql_filter,
+        }
+    )
+    return await client.fetch(
+        query,
+        *sql_filter_args,
+        bpolys.model_dump_json(),
+        database="ohsomedb",
+    )
+
+
+@validate_call
+async def density(
+    *,
+    aggregation: Literal["area"],
+    bpolys: Polygon | MultiPolygon,
+    filter_: OhsomeFilter,
+):
+    sql_filter, sql_filter_args = ohsome_filter_to_sql(filter_)
+    template = ENV.get_template("density.sql")
+    query = template.render(
+        **{
+            "aggregation": aggregation,
+            "contributions": get_config_value("ohsomedb_contributions_table"),
+            "geom": len(sql_filter_args) + 1,
+            "filter": sql_filter,
+        }
+    )
+    return await client.fetch(
+        query,
+        *sql_filter_args,
+        bpolys.model_dump_json(),
+        database="ohsomedb",
+    )