wip

Author: matthiasschaub

workers/ohsome_quality_analyst/api/api.py

@@ -32,7 +32,6 @@
 from ohsome_quality_analyst.config import configure_logging
 from ohsome_quality_analyst.definitions import (
     ATTRIBUTION_URL,
-    INDICATOR_LAYER,
     get_attribution,
     get_dataset_names,
     get_fid_fields,
@@ -280,14 +279,6 @@ async def get_available_regions(asGeoJSON: bool = False):
         return response
 
 
-@app.get("/indicator-layer-combinations")
-async def get_indicator_layer_combinations():
-    """Get names of available indicator-layer combinations."""
-    response = empty_api_response()
-    response["result"] = INDICATOR_LAYER
-    return response
-
-
 @app.get("/indicators")
 async def indicator_names():
     """Get names of available indicators."""

workers/ohsome_quality_analyst/api/request_models.py

@@ -8,20 +8,20 @@
 """
 
 from enum import Enum
-from typing import Optional, Union
+from typing import Optional, Tuple, Union
 
 import pydantic
 from geojson import Feature, FeatureCollection
 from pydantic import BaseModel
 
 from ohsome_quality_analyst.base.layer import LayerData
 from ohsome_quality_analyst.definitions import (
-    INDICATOR_LAYER,
     get_dataset_names,
     get_fid_fields,
     get_indicator_names,
     get_layer_keys,
     get_report_names,
+    get_valid_layers,
 )
 from ohsome_quality_analyst.utils.helper import loads_geojson, snake_to_lower_camel
 
@@ -36,6 +36,14 @@ class BaseIndicator(BaseModel):
     name: IndicatorEnum = pydantic.Field(
         ..., title="Indicator Name", example="GhsPopComparisonBuildings"
     )
+    threshholds: Optional[
+        Tuple[
+            Union[float, str],
+            Union[str, float],
+            Union[str, float],
+            Union[str, float],
+        ]
+    ] = None
     include_svg: bool = False
     include_html: bool = False
     include_data: bool = False
@@ -48,6 +56,17 @@ class Config:
         allow_mutation = False
         extra = "forbid"
 
+    @pydantic.root_validator
+    @classmethod
+    def validate_thresholds(cls, values):
+        if values["threshholds"] is not None and values["name"] != "Currentness":
+            raise ValueError(
+                "Setting custom threshholds is only supported for the Currentness "
+                + "Indicator.",
+            )
+        else:
+            return values
+
 
 class BaseReport(BaseModel):
     name: ReportEnum = pydantic.Field(
@@ -131,13 +150,13 @@ class IndicatorBpolys(BaseIndicator, BaseLayerName, BaseBpolys):
     @pydantic.root_validator
     @classmethod
     def validate_indicator_layer(cls, values):
-        try:
-            indicator_layer = (values["name"].value, values["layer_key"].value)
-        except KeyError:
-            raise ValueError("An issue with the layer or indicator name occurred.")
-        if indicator_layer not in INDICATOR_LAYER:
+        indicator_key = values["name"].value
+        layer_key = values["layer_key"].value
+        if layer_key not in get_valid_layers(indicator_key):
             raise ValueError(
-                "Indicator layer combination is invalid: " + str(indicator_layer)
+                "Layer ({0}) is not available for indicator ({1})".format(
+                    layer_key, indicator_key
+                )
             )
         else:
             return values
@@ -147,13 +166,13 @@ class IndicatorDatabase(BaseIndicator, BaseLayerName, BaseDatabase):
     @pydantic.root_validator
     @classmethod
     def validate_indicator_layer(cls, values):
-        try:
-            indicator_layer = (values["name"].value, values["layer_key"].value)
-        except KeyError:
-            raise ValueError("An issue with the layer or indicator name occurred.")
-        if indicator_layer not in INDICATOR_LAYER:
+        indicator_key = values["name"].value
+        layer_key = values["layer_key"].value
+        if layer_key not in get_valid_layers(indicator_key):
             raise ValueError(
-                "Indicator layer combination is invalid: " + str(indicator_layer)
+                "Layer ({0}) is not available for indicator ({1})".format(
+                    layer_key, indicator_key
+                )
             )
         else:
             return values

workers/ohsome_quality_analyst/base/indicator.py

@@ -5,7 +5,7 @@
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
 from io import StringIO
-from typing import Dict, Literal, Optional
+from typing import Dict, Literal, Optional, Tuple
 
 import matplotlib.pyplot as plt
 from dacite import from_dict
@@ -42,8 +42,8 @@ class Result:
             value is determined by the result classes
         value (float): The result value
         class_ (int): The result class. An integer between 1 and 5. It maps to the
-            result labels. This value is used by the reports to determine an overall
-            result.
+            result labels (1 -> red; 5 -> green). This value is used by the reports to
+            determine an overall result.
         description (str): The result description.
         svg (str): Figure of the result as SVG
     """
@@ -63,17 +63,32 @@ def label(self) -> Literal["green", "yellow", "red", "undefined"]:
 
 
 class BaseIndicator(metaclass=ABCMeta):
-    """The base class of every indicator."""
+    """The base class of every indicator.
+
+    Attributes:
+        thresholds (tuple): A tuple with four float values representing the thresholds
+            between the result classes. The first element is the threshold between the
+            result class 1 and 2, the second element is the threshold between the result
+            class 2 and 3 and so on.
+    """
 
     def __init__(
         self,
         layer: Layer,
         feature: Feature,
+        thresholds: Optional[Tuple[float, float, float, float]] = None,
     ) -> None:
         self.layer: Layer = layer
         self.feature: Feature = feature
+
         # setattr(object, key, value) could be used instead of relying on from_dict.
         metadata = get_metadata("indicators", type(self).__name__)
+        layer_thresholds = metadata.pop("layer-thresholds")
+        if thresholds is None:
+            # TODO: filter layer_thresholds
+            self.thresholds = layer_thresholds[layer.key]
+        else:
+            self.thresholds = thresholds
         self.metadata: Metadata = from_dict(data_class=Metadata, data=metadata)
         self.result: Result = Result(
             description=self.metadata.label_description["undefined"],

workers/ohsome_quality_analyst/cli/cli.py

@@ -15,11 +15,7 @@
 )
 from ohsome_quality_analyst.cli import options
 from ohsome_quality_analyst.config import configure_logging, get_config_value
-from ohsome_quality_analyst.definitions import (
-    INDICATOR_LAYER,
-    load_layer_definitions,
-    load_metadata,
-)
+from ohsome_quality_analyst.definitions import load_layer_definitions, load_metadata
 from ohsome_quality_analyst.geodatabase import client as db_client
 from ohsome_quality_analyst.utils.helper import json_serialize, write_geojson
 
@@ -93,13 +89,6 @@ def get_available_regions():
         click.echo(format_row.format(region["ogc_fid"], region["name"]))
 
 
-@cli.command("list-indicator-layer-combination")
-def get_indicator_layer_combination():
-    """List all possible indicator-layer-combinations."""
-    for combination in INDICATOR_LAYER:
-        click.echo(combination)
-
-
 @cli.command("create-indicator")
 @cli_option(options.indicator_name)
 @cli_option(options.layer_key)

workers/ohsome_quality_analyst/definitions.py

@@ -1,10 +1,12 @@
 """Global Variables and Functions."""
+from __future__ import annotations
+
 import glob
 import logging
 import os
 from dataclasses import dataclass
 from types import MappingProxyType
-from typing import Dict, List, Optional
+from typing import Dict, List, Literal, Optional, Tuple
 
 import yaml
 
@@ -57,60 +59,6 @@ class RasterDataset:
     ),
 )
 
-# Possible indicator layer combinations
-INDICATOR_LAYER = (
-    ("BuildingCompleteness", "building_area"),
-    ("GhsPopComparisonBuildings", "building_count"),
-    ("GhsPopComparisonRoads", "jrc_road_length"),
-    ("GhsPopComparisonRoads", "major_roads_length"),
-    ("MappingSaturation", "building_count"),
-    ("MappingSaturation", "major_roads_length"),
-    ("MappingSaturation", "amenities"),
-    ("MappingSaturation", "jrc_health_count"),
-    ("MappingSaturation", "jrc_mass_gathering_sites_count"),
-    ("MappingSaturation", "jrc_railway_length"),
-    ("MappingSaturation", "jrc_road_length"),
-    ("MappingSaturation", "jrc_education_count"),
-    ("MappingSaturation", "mapaction_settlements_count"),
-    ("MappingSaturation", "mapaction_major_roads_length"),
-    ("MappingSaturation", "mapaction_rail_length"),
-    ("MappingSaturation", "mapaction_lakes_area"),
-    ("MappingSaturation", "mapaction_rivers_length"),
-    ("MappingSaturation", "ideal_vgi_infrastructure"),
-    ("MappingSaturation", "poi"),
-    ("MappingSaturation", "lulc"),
-    ("Currentness", "major_roads_count"),
-    ("Currentness", "building_count"),
-    ("Currentness", "amenities"),
-    ("Currentness", "jrc_health_count"),
-    ("Currentness", "jrc_education_count"),
-    ("Currentness", "jrc_road_count"),
-    ("Currentness", "jrc_railway_count"),
-    ("Currentness", "jrc_airport_count"),
-    ("Currentness", "jrc_water_treatment_plant_count"),
-    ("Currentness", "jrc_power_generation_plant_count"),
-    ("Currentness", "jrc_cultural_heritage_site_count"),
-    ("Currentness", "jrc_bridge_count"),
-    ("Currentness", "jrc_mass_gathering_sites_count"),
-    ("Currentness", "mapaction_settlements_count"),
-    ("Currentness", "mapaction_major_roads_length"),
-    ("Currentness", "mapaction_rail_length"),
-    ("Currentness", "mapaction_lakes_count"),
-    ("Currentness", "mapaction_rivers_length"),
-    ("PoiDensity", "poi"),
-    ("TagsRatio", "building_count"),
-    ("TagsRatio", "major_roads_length"),
-    ("TagsRatio", "jrc_health_count"),
-    ("TagsRatio", "jrc_education_count"),
-    ("TagsRatio", "jrc_road_length"),
-    ("TagsRatio", "jrc_airport_count"),
-    ("TagsRatio", "jrc_power_generation_plant_count"),
-    ("TagsRatio", "jrc_cultural_heritage_site_count"),
-    ("TagsRatio", "jrc_bridge_count"),
-    ("TagsRatio", "jrc_mass_gathering_sites_count"),
-    ("Minimal", "minimal"),
-)
-
 ATTRIBUTION_TEXTS = MappingProxyType(
     {
         "OSM": "© OpenStreetMap contributors",
@@ -125,17 +73,16 @@ class RasterDataset:
 )
 
 
-def load_metadata(module_name: str) -> Dict:
-    """Read metadata of all indicators or reports from YAML files.
+def load_metadata(module_name: Literal["indicators", "reports"]) -> Dict:
+    """Load metadata of all indicators or reports from YAML files.
 
-    Those text files are located in the directory of each indicator/report.
+    The YAML files are located in the directory of each individual indicator or report.
 
-    Args:
-        module_name: Either indicators or reports.
     Returns:
-        A Dict with the class names of the indicators/reports
-        as keys and metadata as values.
+        A dictionary with the indicator or report keys as directory keys and the content
+        of the YAML file (metadata) as values.
     """
+    # TODO: Is this check needed if Literal is used in func declaration?
     if module_name != "indicators" and module_name != "reports":
         raise ValueError("module name value can only be 'indicators' or 'reports'.")
 
@@ -272,11 +219,15 @@ def get_attribution(data_keys: list) -> str:
     return "; ".join([str(v) for v in filtered.values()])
 
 
+# TODO
 def get_valid_layers(indcator_name: str) -> tuple:
     """Get valid Indicator/Layer combination of an Indicator."""
-    return tuple([tup[1] for tup in INDICATOR_LAYER if tup[0] == indcator_name])
+    return tuple(
+        [tup[1] for tup in INDICATOR_LAYER_THRESHOLDS if tup[0] == indcator_name]
+    )
 
 
+# TODO
 def get_valid_indicators(layer_key: str) -> tuple:
     """Get valid Indicator/Layer combination of a Layer."""
-    return tuple([tup[0] for tup in INDICATOR_LAYER if tup[1] == layer_key])
+    return tuple([tup[0] for tup in INDICATOR_LAYER_THRESHOLDS if tup[1] == layer_key])

workers/ohsome_quality_analyst/indicators/building_completeness/indicator.py

@@ -2,6 +2,7 @@
 import os
 from io import StringIO
 from string import Template
+from typing import Optional, Tuple
 
 import dateutil.parser
 import geojson
@@ -58,10 +59,14 @@ def __init__(
         self,
         layer: Layer,
         feature: Feature,
+        thresholds: Optional[Tuple[float, float, float, float]],
     ) -> None:
+        if thresholds is None:
+            thresholds = (0.2, 0.5, 0.8, 0.9)
         super().__init__(
             layer=layer,
             feature=feature,
+            thresholds=thresholds,
         )
         self.model_name: str = "Random Forest Regressor"
         # Lists of elements per hexagonal cell

workers/ohsome_quality_analyst/indicators/building_completeness/metadata.yaml

@@ -23,3 +23,5 @@ BuildingCompleteness:
     average of the ratios per hex-cell between the building area mapped in OSM and the
     predicted building area is $completeness_ratio %. The weight is the
     predicted building area.
+  layer-thresholds:
+    - { layer: building_area, thresholds: [0.2, 0.5, 0.8, 0.9] }

workers/ohsome_quality_analyst/indicators/currentness/metadata.yaml

@@ -26,3 +26,7 @@ Currentness:
       bad or if there is just nothing to map here.
   result_description: |
       Over 50% of the $elements features ($layer_name) were edited $years.
+  layer-thresholds:
+    - { layer: amenities, thresholds: [ 0.2, null, 0.6, null] }
+    - { layer: building_count, thresholds: [ 0.2, null, 0.6, null] }
+    - { layer: major_roads_count, thresholds: [ 0.2, null, 0.6, null] }

workers/ohsome_quality_analyst/indicators/ghs_pop_comparison_buildings/metadata.yaml

@@ -23,4 +23,6 @@ GhsPopComparisonBuildings:
     $pop_count people living in an area of 
     $area sqkm, which results in a population density 
     $pop_count_per_sqkm of people per sqkm.
-    $feature_count_per_sqkm buildings per sqkm mapped. 
+    $feature_count_per_sqkm buildings per sqkm mapped.
+  layer-thresholds:
+    - { layer: building_count, thresholds: [{ a:0.75 }, null, { a: 5.0 }, null] }

workers/ohsome_quality_analyst/indicators/ghs_pop_comparison_roads/metadata.yaml

@@ -24,3 +24,5 @@ GhsPopComparisonRoads:
     $area sqkm, which results in a population density 
     $pop_count_per_sqkm of people per sqkm.
     $feature_length_per_sqkm km of roads per sqkm mapped.
+  layer-thresholds:
+    - { layer: major_roads_length, thresholds: [{ a: 1000 }, null, { a: 500 }, null] }

workers/ohsome_quality_analyst/indicators/mapping_saturation/metadata.yaml

@@ -15,3 +15,11 @@ MappingSaturation:
       Saturation could not be calculated.
   result_description: |
     The saturation of the last 3 years is $saturation%.
+  layer-thresholds:
+    - { layer: amenities, thresholds: [0.3, null, 0.97, null] }
+    - { layer: amenities, thresholds: [0.3, null, 0.97, null] }
+    - { layer: building_count, thresholds: [0.3, null, 0.97, null] }
+    - { layer: ideal_vgi_infrastructure, thresholds: [0.3, null, 0.97, null] }
+    - { layer: lulc, thresholds: [0.3, null, 0.97, null] }
+    - { layer: major_roads_length, thresholds: [0.3, null, 0.97, null] }
+    - { layer: poi, thresholds: [0.3, null, 0.97, null] }

workers/ohsome_quality_analyst/indicators/minimal/indicator.py

@@ -10,8 +10,8 @@
 
 
 class Minimal(BaseIndicator):
-    def __init__(self, layer: Layer, feature: Feature) -> None:
-        super().__init__(layer=layer, feature=feature)
+    def __init__(self, layer: Layer, feature: Feature, thresholds: tuple) -> None:
+        super().__init__(layer=layer, feature=feature, thresholds=thresholds)
         self.count = 0
 
     async def preprocess(self) -> None: