Added initial test for write_data_unit_tiles_to_database

tests/conftest.py

@@ -29,7 +29,8 @@ load_dotenv(Path(".env").resolve())
 def test_db():
     """A test database simulating to contain the following tables:
     - obm_built_area_assessments (of the OBM Tiles database)
-    - aggregated_sources (of the GDE Tiles database).
+    - aggregated_sources (of the GDE Tiles database)
+    - data_unit_tiles (of the GDE Tiles database).
     """
 
     init_test_db()
@@ -37,8 +38,8 @@ def test_db():
 
 
 def init_test_db():
-    """Populates the test database that simulates to contain obm_built_area_assessments and
-    aggregated_sources with a basic schema and data.
+    """Populates the test database that simulates to contain obm_built_area_assessments,
+    aggregated_sources and data_unit_tiles with a basic schema and data.
     """
 
     if "GDEIMPORTER_DB_HOST" in os.environ:  # When running the CI pipeline
@@ -66,4 +67,10 @@ def init_test_db():
                 if command != "\n":
                     db.cursor.execute(command)
 
+        # Create columns and populate the aggregated_sources table
+        with open("tests/data/test_database_data_unit_tiles.sql", "r") as file:
+            for command in file.read().split(";"):
+                if command != "\n":
+                    db.cursor.execute(command)
+
         db.close_connection()

tests/data/expected_results_test_get_data_unit_tiles.csv

-quadkey,type,lon_w,lat_s,lon_e,lat_n,contained,size_data_unit_tile_area,fraction_data_unit_area,size_data_unit_tile_built_up_area
-120222331000133202,MultiPolygon,4.9994,42.0026,5.000153,42.0032,False,2784.421,0.068256627491064,393.720042735683
-120222331000133203,MultiPolygon,5.000153,42.0026,5.001526,42.0032,False,5154.434,0.126354556823582,1355.42193743533
-120222331000133212,Polygon,5.001526,42.002366,5.002899,42.0032,False,7544.606,0.184946659039292,4331.24437901172
-120222331000133213,Polygon,5.002899,42.002366,5.0037,42.0032,False,6144.446,0.150623473160473,2828.98618958169
-120222331000133220,Polygon,4.9994,42.0015,5.000153,42.0018,False,2077.655,0.050931135553836,333.852842868529
-120222331000133221,Polygon,5.000153,42.0015,5.001526,42.0018,False,3791.219,0.092937031799445,360.852144117784
-120222331000133230,Polygon,5.001526,42.0015,5.002899,42.002366,False,6913.142,0.169467102240224,1566.1305332532
-120222331000133231,Polygon,5.002899,42.0015,5.0037,42.002366,False,6383.493,0.156483413892085,2101.1230827478
+quadkey,type,lon_w,lat_s,lon_e,lat_n,contained,size_data_unit_tile_area,fraction_data_unit_area,size_data_unit_tile_built_up_area,fraction_data_unit_built_up_area
+120222331000133202,MultiPolygon,4.9994,42.0026,5.000153,42.0032,False,2784.421,0.068256627491064,393.720042735683,0.091
+120222331000133203,MultiPolygon,5.000153,42.0026,5.001526,42.0032,False,5154.434,0.126354556823582,1355.42193743533,0.082
+120222331000133212,Polygon,5.001526,42.002366,5.002899,42.0032,False,7544.606,0.184946659039292,4331.24437901172,0.237
+120222331000133213,Polygon,5.002899,42.002366,5.0037,42.0032,False,6144.446,0.150623473160473,2828.98618958169,0.141
+120222331000133220,Polygon,4.9994,42.0015,5.000153,42.0018,False,2077.655,0.050931135553836,333.852842868529,0.022
+120222331000133221,Polygon,5.000153,42.0015,5.001526,42.0018,False,3791.219,0.092937031799445,360.852144117784,0.067
+120222331000133230,Polygon,5.001526,42.0015,5.002899,42.002366,False,6913.142,0.169467102240224,1566.1305332532,0.083
+120222331000133231,Polygon,5.002899,42.0015,5.0037,42.002366,False,6383.493,0.156483413892085,2101.1230827478,0.037

tests/data/test_database_data_unit_tiles.sql

+DROP TABLE IF EXISTS data_unit_tiles;
+CREATE TYPE occupancycase AS ENUM ('residential', 'commercial', 'industrial');
+CREATE TABLE data_unit_tiles
+(
+    quadkey char(18), 
+    aggregated_source_id SMALLINT,
+    occupancy_case occupancycase,
+    exposure_entity char(3),
+    data_unit_id varchar,
+    size_data_unit_tile_area FLOAT,
+    size_data_unit_tile_built_up_area FLOAT,
+    fraction_data_unit_area FLOAT,
+    fraction_data_unit_built_up_area FLOAT,
+
+    PRIMARY KEY (quadkey, aggregated_source_id, occupancy_case, data_unit_id)
+);
+INSERT INTO data_unit_tiles(quadkey, 
+                            aggregated_source_id,
+                            occupancy_case,
+                            exposure_entity,
+                            data_unit_id,
+                            size_data_unit_tile_area,
+                            size_data_unit_tile_built_up_area,
+                            fraction_data_unit_area,
+                            fraction_data_unit_built_up_area)
+VALUES ('120222331000133202',2,'residential','GRC','GRC_39821',2532.671,287.720,0.0572566,0.086);

tests/test_data_unit_tiles.py

@@ -24,8 +24,11 @@ import pandas
 import shapely
 import pytest
 from copy import deepcopy
+from pyproj import CRS
 from gdeimporter.tools.data_unit_tiles import DataUnitTilesHelper
 from gdeimporter.configuration import Configuration
+from gdeimporter.tools.database import Database
+from gdeimporter.dataunit import DataUnit
 
 logger = logging.getLogger()
 
@@ -419,3 +422,79 @@ def test_retrieve_built_up_area(test_db):
     expected_built_up_areas = numpy.array([7512.38, 3449.81, 1266.45, 5016.22, 0.0])
 
     numpy.testing.assert_array_equal(returned_built_up_areas, expected_built_up_areas)
+
+
+def test_write_data_unit_tiles_to_database(test_db):
+    # Database credentials (the Configuration class will define the credentials based on whether
+    # the code is running in the CI or locally)
+    config = Configuration(
+        os.path.join(os.path.dirname(__file__), "data", "config_for_testing_dummy.yml")
+    )
+
+    # Create a DataUnit object with data-unit tiles that will be written to the database
+    target_column_name = "ID"
+    unit_ids = ["39821"]
+    geometries = [shapely.geometry.Polygon([(22.5, 13.7), (28.3, 14.2), (25.1, 16.7)])]
+    geometry = geopandas.GeoSeries(geometries)
+    aux_data = {target_column_name: unit_ids, "geometry": geometry}
+    geometries_table = geopandas.GeoDataFrame(aux_data)
+    geometries_table.crs = CRS("epsg:4326")
+    example_data_unit = DataUnit(unit_ids[0], geometries_table, target_column_name)
+    example_data_unit.data_unit_tiles = pandas.read_csv(
+        os.path.join(
+            os.path.dirname(__file__),
+            "data",
+            "expected_results_test_get_data_unit_tiles.csv",
+        ),
+        usecols=[
+            "quadkey",
+            "size_data_unit_tile_area",
+            "fraction_data_unit_area",
+            "size_data_unit_tile_built_up_area",
+            "fraction_data_unit_built_up_area",
+        ],
+        dtype={
+            "quadkey": str,
+            "size_data_unit_tile_area": float,
+            "fraction_data_unit_area": float,
+            "size_data_unit_tile_built_up_area": float,
+            "fraction_data_unit_built_up_area": float,
+        },
+    )
+
+    exposure_entity_code = "GRC"
+    for aggregated_source_id in [2, 3]:
+        for occupancy_case in ["residential", "commercial", "industrial"]:
+            DataUnitTilesHelper.write_data_unit_tiles_to_database(
+                config.database_gde_tiles,
+                "data_unit_tiles",
+                aggregated_source_id,
+                occupancy_case,
+                exposure_entity_code,
+                example_data_unit,
+            )
+
+    # Query the database to verify that the data-unit tiles have been written to the table
+    db_test = Database(**config.database_gde_tiles)
+    db_test.create_connection_and_cursor()
+
+    sql_command = "SELECT COUNT(*) FROM data_unit_tiles"
+    sql_command += " WHERE (quadkey='%s' AND aggregated_source_id='%s'"
+    sql_command += " AND occupancy_case='%s' AND data_unit_id='%s');"
+
+    for aggregated_source_id in [2, 3]:
+        for occupancy_case in ["residential", "commercial", "industrial"]:
+            for i in range(example_data_unit.data_unit_tiles.shape[0]):
+                db_test.cursor.execute(
+                    sql_command
+                    % (
+                        example_data_unit.data_unit_tiles["quadkey"].values[i],
+                        str(aggregated_source_id),
+                        occupancy_case,
+                        "%s_%s" % (exposure_entity_code, example_data_unit.id),
+                    )
+                )
+                exec_result = db_test.cursor.fetchall()
+                assert exec_result[0][0] == 1
+
+    db_test.close_connection()