Implement several SQL queries from Initializer as functions

exposurelib/database.py

@@ -23,6 +23,8 @@ import glob
 import shutil
 import os.path
 from abc import abstractmethod, ABC
+from typing import Generator
+
 from exposurelib.utils import get_geom_of_quadkey, batched
 from taxonomylib.taxonomylib import Taxonomy
 from databaselib.database import AbstractDatabase, SpatiaLiteDatabase, PostGISDatabase
@@ -739,16 +741,58 @@ class AbstractExposure(AbstractDatabase, ABC):
         self.cursor.execute(sql_statement)
         return self.cursor.fetchall()
 
+    def get_country_assets_simple(self, country_iso_code: str) -> list:
+        """
+        Retrieves the country assets of one country without the normalized values.
+
+        Args:
+            country_iso_code (str):
+                ISO 3166-1 alpha-3 code of the country.
+
+        Returns:
+            List of all country assets of the given country.
+        """
+
+        sql_statement = f"""
+            SELECT taxonomy_id, number, structural, structural_per_sqm, night, night_per_sqm
+            FROM AssetCountry
+            WHERE country_iso_code = '{country_iso_code}'
+            """
+        self.cursor.execute(sql_statement)
+        return self.cursor.fetchall()
+
+    def insert_boundary(self, boundary_id: str, geometry_wkt: str):
+        """
+        Inserts a boundary into the `Boundaries` or `Boundary` table.
+
+        Args:
+            boundary_id (str):
+                ID of the boundary.
+            geometry_wkt (str):
+                Geometry of the boundary as WKT.
+        """
+
+        sql_statement = f"""
+            INSERT INTO {self.boundary_table}
+            ({self.boundary_id_field}, {self.geometry_field})
+            VALUES
+            (
+                '{boundary_id}',
+                CastToMultipolygon(GeomFromText('{geometry_wkt}', 4326))
+            )
+            """
+        self.cursor.execute(sql_statement)
+
     def boundary_exists(self, boundary_id):
         """
         Checks if the boundary with the given ID exists in the boundary table.
 
         Args:
             boundary_id (str):
-                ID of the boundary
+                ID of the boundary.
 
         Returns:
-            `True` if the boundary exists, `False` otherwise
+            `True` if the boundary exists, `False` otherwise.
 
         """
 
@@ -763,6 +807,28 @@ class AbstractExposure(AbstractDatabase, ABC):
         self.cursor.execute(sql_statement)
         return self.cursor.fetchone()[0]
 
+    def get_district_boundaries_generator(self, country_iso_code: str) -> Generator:
+        """
+        Provides a generator over all district boundaries of a country.
+
+        Args:
+            country_iso_code (str):
+                ISO 3166-1 alpha-3 code of the country.
+
+        Returns:
+            Generator over all district boundaries of a country.
+        """
+
+        sql_statement = f"""
+            SELECT {self.boundary_id_field}, ST_AsText({self.geometry_field})
+            FROM {self.boundary_table}
+            WHERE {self.boundary_id_field} LIKE '{country_iso_code}%'
+            """
+        self.cursor.execute(sql_statement)
+
+        for boundary_id, geom_wkt in self.cursor:
+            yield boundary_id, geom_wkt
+
     def get_boundary_geometry(self, boundary_id, buffer=None):
         """
         Retrieve the geometry of a boundary for a given boundary ID. This boundary
@@ -888,6 +954,28 @@ class AbstractExposure(AbstractDatabase, ABC):
         tiles = [quadkey for quadkey, in self.cursor]
         return tiles
 
+    def get_country_tiles_generator(self, country_iso_code: str) -> Generator:
+        """
+        Provides a generator over all tiles of a country.
+
+        Args:
+            country_iso_code (str):
+                ISO 3166-1 alpha-3 code of the country.
+
+        Returns:
+            Generator over all buildings within a country.
+        """
+
+        sql_statement = f"""
+            SELECT quadkey, ST_AsText(geometry), built_area_size, built_up_ratio, completeness
+            FROM {self.tile_view}
+            WHERE country_iso_code = '{country_iso_code}'
+            """
+        self.cursor.execute(sql_statement)
+
+        for quadkey, geometry_wkt, built_area_size, built_up_ratio, completeness in self.cursor:
+            yield quadkey, geometry_wkt, built_area_size, built_up_ratio, completeness
+
     def get_country_entities(self, country_iso_code):
         """
         Retrieve IDs of all entities within a country.
@@ -955,6 +1043,53 @@ class AbstractExposure(AbstractDatabase, ABC):
         tiles = [query_result[0] for query_result in self.cursor.fetchall()]
         return tiles
 
+    def insert_tile(
+        self,
+        quadkey: str,
+        country_iso_code: str,
+        geometry_wkt: str,
+        built_area_size: float,
+        built_up_ratio: float,
+        completeness: int,
+    ):
+        """
+        Inserts a tile into the tile table.
+
+        Args:
+            quadkey (str):
+                Quadkey of the tile.
+            country_iso_code (str):
+                ISO 3166-1 alpha-3 code of the country.
+            geometry_wkt (str):
+                Tile geometry in WKT.
+            built_area_size (float):
+                Size of the built area in the tile.
+            built_up_ratio (float):
+                Ratio of the area covered by building footprints compared to `built_area_size`.
+            completeness (int):
+                Building-completeness assessment (0: incomplete, 1: complete).
+        """
+
+        sql_statement = f"""
+            INSERT INTO {self.tile_table} (
+                quadkey,
+                country_iso_code,
+                built_area_size,
+                built_up_ratio,
+                completeness,
+                {self.geometry_field}
+            )
+            VALUES (
+                '{quadkey}',
+                '{country_iso_code}',
+                {built_area_size if built_area_size else 'NULL'},
+                {built_up_ratio if built_up_ratio else 'NULL'},
+                {completeness if completeness else 'NULL'},
+                CastToMultipolygon(GeomFromText('{geometry_wkt}', 4326))
+            )
+            """
+        self.cursor.execute(sql_statement)
+
     def get_tiles_within_super_tile(self, super_quadkey, sorted_by_latitude=False):
         """
         Retrieve the Quadkeys of all tiles within a super tile specified by its Quadkey.
@@ -1027,6 +1162,65 @@ class AbstractExposure(AbstractDatabase, ABC):
                 total_built_area += built_area
         return total_built_area
 
+    def get_district_tiles_with_built_area(self, boundary_id: str) -> list:
+        """
+        Retrieve all tiles within a district that contain built area. The district is specified
+        by its boundary ID.
+
+        Args:
+            boundary_id:
+                ID of the district boundary.
+
+        Returns:
+            List of all tiles containing the Quadkey and the built area.
+        """
+
+        sql_statement = f"""
+            SELECT quadkey, built_area_size
+            FROM {self.tile_view}
+            INNER JOIN {self.boundary_table}
+                ON ST_Contains(
+                        {self.boundary_table}.{self.geometry_field},
+                        ST_Centroid({self.tile_view}.{self.geometry_field})
+                    )
+            WHERE {self.boundary_table}.{self.boundary_id_field} = '{boundary_id}'
+                AND built_area_size IS NOT NULL
+            """
+        self.cursor.execute(sql_statement)
+        return self.cursor.fetchall()
+
+    def get_district_built_area(self, boundary_id):
+        """
+        Calculates the total built area of a district. The district is specified by its boundary
+        ID.
+
+        Args:
+            boundary_id:
+                ID of the district boundary.
+
+        Returns:
+            Total built area or `None` if the boundary geometry does not cover any area.
+        """
+
+        # Calculate the total built area of the district.
+        sql_statement = f"""
+            SELECT SUM(built_area_size)
+            FROM {self.tile_view}
+            INNER JOIN {self.boundary_table}
+                ON ST_Contains(
+                        {self.boundary_table}.{self.geometry_field},
+                        ST_Centroid({self.tile_view}.{self.geometry_field})
+                    )
+            WHERE {self.boundary_table}.{self.boundary_id_field} = '{boundary_id}'
+                AND built_area_size IS NOT NULL
+            """
+        self.cursor.execute(sql_statement)
+        total_built_area = self.cursor.fetchone()
+        if total_built_area is not None:
+            return total_built_area[0]
+        else:
+            return None
+
     def get_country_built_area(self, country_iso_code):
         """
         Retrieves the total built-area size of all tiles in a country.
@@ -3136,9 +3330,9 @@ class SpatiaLiteExposure(SpatiaLiteDatabase, AbstractExposure):
             VALUES (
                 {quadkey},
                 '{country_iso_code}',
-                {built_area_size},
-                {built_up_ratio},
-                {completeness},
+                {built_area_size if built_area_size else 'NULL'},
+                {built_up_ratio if built_up_ratio else 'NULL'},
+                {completeness if completeness else 'NULL'},
                 ST_Multi(GeomFromText('{geom_wkt}', 4326))
             )
             """
@@ -4269,3 +4463,66 @@ class PostGISExposure(PostGISDatabase, AbstractExposure):
             """
         self.cursor.execute(sql_statement)
         self.connection.commit()
+
+    def get_district_tiles_with_built_area(self, boundary_id: str) -> list:
+        """
+        Retrieve all tiles within a district that contain built area. The district is specified
+        by its boundary ID.
+
+        Args:
+            boundary_id:
+                ID of the district boundary.
+
+        Returns:
+            List of all tiles containing the Quadkey and the built area.
+        """
+
+        sql_statement = f"""
+            SELECT quadkey, built_area_size
+            FROM {self.tile_view}
+            INNER JOIN {self.boundary_table}
+                ON ST_Contains(
+                        {self.boundary_table}.{self.geometry_field},
+                        ST_Centroid({self.tile_view}.{self.geometry_field})
+                    )
+            WHERE {self.boundary_table}.{self.boundary_id_field} = '{boundary_id}'
+                AND {self.tile_view}.{self.geometry_field}
+                    && {self.boundary_table}.{self.geometry_field}
+                AND built_area_size IS NOT NULL
+            """
+        self.cursor.execute(sql_statement)
+        return self.cursor.fetchall()
+
+    def get_district_built_area(self, boundary_id):
+        """
+        Calculates the total built area of a district. The district is specified by its boundary
+        ID.
+
+        Args:
+            boundary_id:
+                ID of the district boundary.
+
+        Returns:
+            Total built area or `None` if the boundary geometry does not cover any area.
+        """
+
+        # Calculate the total built area of the district.
+        sql_statement = f"""
+            SELECT SUM(built_area_size)
+            FROM {self.tile_view}
+            INNER JOIN {self.boundary_table}
+                ON ST_Contains(
+                        {self.boundary_table}.{self.geometry_field},
+                        ST_Centroid({self.tile_view}.{self.geometry_field})
+                    )
+            WHERE {self.boundary_table}.{self.boundary_id_field} = '{boundary_id}'
+                AND {self.tile_view}.{self.geometry_field}
+                    && {self.boundary_table}.{self.geometry_field}
+                AND built_area_size IS NOT NULL
+            """
+        self.cursor.execute(sql_statement)
+        total_built_area = self.cursor.fetchone()
+        if total_built_area is not None:
+            return total_built_area[0]
+        else:
+            return None