Merge pull request #453 from mapswipe/zip-files

compress files for export with gzip #451

Author: Hagellach37

mapswipe_workers/mapswipe_workers/generate_stats/project_stats.py

@@ -1,6 +1,8 @@
 import datetime
 import os
 from typing import List
+import gzip
+import tempfile
 
 import pandas as pd
 from psycopg2 import sql
@@ -26,15 +28,21 @@ def add_metadata_to_csv(filename: str):
     logger.info(f"added metadata to {filename}.")
 
 
-def write_sql_to_csv(filename: str, sql_query: sql.SQL):
+def write_sql_to_gzipped_csv(filename: str, sql_query: sql.SQL):
     """
     Use the copy statement to write data from postgres to a csv file.
     """
 
+    # generate temporary file which will be automatically deleted at the end
+    tmp_csv_file = os.path.join(tempfile._get_default_tempdir(), 'tmp.csv')
     pg_db = auth.postgresDB()
-    with open(filename, "w") as f:
+    with open(tmp_csv_file, "w") as f:
         pg_db.copy_expert(sql_query, f)
-    logger.info(f"wrote csv file from sql: {filename}")
+
+    with open(tmp_csv_file, 'rb') as f_in, gzip.open(filename, 'wb') as f_out:
+        f_out.writelines(f_in)
+
+    logger.info(f"wrote gzipped csv file from sql: {filename}")
 
 
 def load_df_from_csv(filename: str) -> pd.DataFrame:
@@ -44,7 +52,11 @@ def load_df_from_csv(filename: str) -> pd.DataFrame:
     """
     dtype_dict = {"project_id": str, "group_id": str, "task_id": str}
 
-    df = pd.read_csv(filename, dtype=dtype_dict)
+    df = pd.read_csv(
+        filename,
+        dtype=dtype_dict,
+        compression="gzip"
+    )
     logger.info(f"loaded pandas df from {filename}")
     return df
 
@@ -73,7 +85,7 @@ def get_results(filename: str, project_id: str) -> pd.DataFrame:
         ) TO STDOUT WITH CSV HEADER
         """
     ).format(sql.Literal(project_id))
-    write_sql_to_csv(filename, sql_query)
+    write_sql_to_gzipped_csv(filename, sql_query)
 
     df = load_df_from_csv(filename)
 
@@ -117,7 +129,7 @@ def get_tasks(filename: str, project_id: str) -> pd.DataFrame:
             ) TO STDOUT WITH CSV HEADER
             """
         ).format(sql.Literal(project_id))
-        write_sql_to_csv(filename, sql_query)
+        write_sql_to_gzipped_csv(filename, sql_query)
 
     df = load_df_from_csv(filename)
     return df
@@ -152,7 +164,7 @@ def get_groups(filename: str, project_id: str) -> pd.DataFrame:
             ) TO STDOUT WITH CSV HEADER
             """
         ).format(sql.Literal(project_id))
-        write_sql_to_csv(filename, sql_query)
+        write_sql_to_gzipped_csv(filename, sql_query)
 
     df = load_df_from_csv(filename)
     return df
@@ -322,11 +334,11 @@ def get_per_project_statistics(project_id: str, project_info: pd.Series) -> dict
     """
 
     # set filenames
-    results_filename = f"{DATA_PATH}/api/results/results_{project_id}.csv"
-    tasks_filename = f"{DATA_PATH}/api/tasks/tasks_{project_id}.csv"
-    groups_filename = f"{DATA_PATH}/api/groups/groups_{project_id}.csv"
-    agg_results_filename = f"{DATA_PATH}/api/agg_results/agg_results_{project_id}.csv"
-    agg_results_by_user_id_filename = f"{DATA_PATH}/api/users/users_{project_id}.csv"
+    results_filename = f"{DATA_PATH}/api/results/results_{project_id}.csv.gz"
+    tasks_filename = f"{DATA_PATH}/api/tasks/tasks_{project_id}.csv.gz"
+    groups_filename = f"{DATA_PATH}/api/groups/groups_{project_id}.csv.gz"
+    agg_results_filename = f"{DATA_PATH}/api/agg_results/agg_results_{project_id}.csv.gz"
+    agg_results_by_user_id_filename = f"{DATA_PATH}/api/users/users_{project_id}.csv.gz"
     project_stats_by_date_filename = f"{DATA_PATH}/api/history/history_{project_id}.csv"
 
     # load data from postgres or local storage if already downloaded
@@ -339,11 +351,22 @@ def get_per_project_statistics(project_id: str, project_info: pd.Series) -> dict
         groups_df = get_groups(groups_filename, project_id)
         tasks_df = get_tasks(tasks_filename, project_id)
 
+        if any("maxar" in s for s in project_info["tile_server_names"]):
+            add_metadata = True
+
         # aggregate results by task id
         agg_results_df = get_agg_results_by_task_id(results_df, tasks_df)
-        agg_results_df.to_csv(agg_results_filename, index_label="idx")
+        agg_results_df.to_csv(
+            agg_results_filename,
+            index_label="idx"
+        )
+
+        geojson_functions.gzipped_csv_to_gzipped_geojson(
+            filename=agg_results_filename,
+            geometry_field="geom",
+            add_metadata=add_metadata
+        )
         logger.info(f"saved agg results for {project_id}: {agg_results_filename}")
-        geojson_functions.csv_to_geojson(agg_results_filename, "geom")
 
         # aggregate results by user id
         # TODO: solve memory issue for agg results by user id
@@ -352,7 +375,8 @@ def get_per_project_statistics(project_id: str, project_info: pd.Series) -> dict
                 results_df, agg_results_df
             )
             agg_results_by_user_id_df.to_csv(
-                agg_results_by_user_id_filename, index_label="idx"
+                agg_results_by_user_id_filename,
+                index_label="idx"
             )
             logger.info(
                 f"saved agg results for {project_id}: {agg_results_by_user_id_filename}"
@@ -361,10 +385,6 @@ def get_per_project_statistics(project_id: str, project_info: pd.Series) -> dict
             sentry.capture_exception()
             logger.info(f"failed to agg results by user id for {project_id}")
 
-        if any("maxar" in s for s in project_info["tile_server_names"]):
-            add_metadata_to_csv(agg_results_filename)
-            geojson_functions.add_metadata_to_geojson(agg_results_filename)
-
         project_stats_by_date_df = project_stats_by_date.get_project_history(
             results_df, groups_df
         )
@@ -380,7 +400,10 @@ def get_per_project_statistics(project_id: str, project_info: pd.Series) -> dict
             # do not do this for ArbitraryGeometry / BuildingFootprint projects
             logger.info(f"do NOT generate tasking manager geometries for {project_id}")
         else:
-            tasking_manager_geometries.generate_tasking_manager_geometries(project_id)
+            tasking_manager_geometries.generate_tasking_manager_geometries(
+                project_id=project_id,
+                agg_results_filename=agg_results_filename
+            )
 
         # prepare output of function
         project_stats_dict = {

mapswipe_workers/mapswipe_workers/generate_stats/tasking_manager_geometries.py

@@ -1,4 +1,5 @@
 import csv
+import gzip
 import threading
 from queue import Queue
 
@@ -8,15 +9,15 @@
 from mapswipe_workers.utils import geojson_functions, tile_functions
 
 
-def load_data(project_id: str, csv_file: str) -> list:
+def load_data(project_id: str, gzipped_csv_file: str) -> list:
     """
     This will load the aggregated results csv file into a list of dictionaries.
     For further steps we currently rely on task_x, task_y, task_z and yes_share and
     maybe_share and wkt
     """
 
     project_data = []
-    with open(csv_file, "r") as f:
+    with gzip.open(gzipped_csv_file, mode="rt") as f:
         reader = csv.reader(f, delimiter=",")
 
         for i, row in enumerate(reader):
@@ -416,7 +417,7 @@ def dissolve_project_data(project_data):
     return dissolved_geometry
 
 
-def generate_tasking_manager_geometries(project_id: str):
+def generate_tasking_manager_geometries(project_id: str, agg_results_filename):
     """
     This functions runs the workflow to create a GeoJSON file ready to be used in the
     HOT Tasking Manager.
@@ -428,14 +429,13 @@ def generate_tasking_manager_geometries(project_id: str):
     Finally, both data sets are saved into GeoJSON files.
     """
 
-    raw_data_filename = f"{DATA_PATH}/api/agg_results/agg_results_{project_id}.csv"
     filtered_data_filename = f"{DATA_PATH}/api/yes_maybe/yes_maybe_{project_id}.geojson"
     tasking_manager_data_filename = (
         f"{DATA_PATH}/api/hot_tm/hot_tm_{project_id}.geojson"
     )
 
     # load project data from existing files
-    results = load_data(project_id, raw_data_filename)
+    results = load_data(project_id, agg_results_filename)
 
     # filter yes and maybe results
     filtered_results = filter_data(results)

mapswipe_workers/mapswipe_workers/utils/geojson_functions.py

@@ -1,12 +1,67 @@
 import json
 import os
+import gzip
+import shutil
 import subprocess
+import tempfile
 
 from osgeo import ogr, osr
 
 from mapswipe_workers.definitions import logger
 
 
+def gzipped_csv_to_gzipped_geojson(
+        filename: str,
+        geometry_field: str = "geom",
+        add_metadata: bool = False
+):
+    """Convert gzipped csv file to gzipped GeoJSON.
+
+    First the gzipped files are unzipped and stored in temporary csv and geojson files.
+    Then the unzipped csv file is converted into a geojson file with ogr2ogr.
+    Last, the generated geojson file is again compressed using gzip.
+    """
+    # generate temporary files which will be automatically deleted at the end
+    tmp_csv_file = os.path.join(tempfile._get_default_tempdir(), 'tmp.csv')
+    tmp_geojson_file = os.path.join(tempfile._get_default_tempdir(), 'tmp.geojson')
+
+    outfile = filename.replace(".csv", f"_{geometry_field}.geojson")
+
+    # uncompress content of zipped csv file and save to csv file
+    with gzip.open(filename, 'rb') as f_in:
+        with open(tmp_csv_file, "wb") as f_out:
+            shutil.copyfileobj(f_in, f_out)
+
+    # use ogr2ogr to transform csv file into geojson file
+    # TODO: remove geom column from normal attributes in sql query
+    subprocess.run(
+        [
+            "ogr2ogr",
+            "-f",
+            "GeoJSON",
+            tmp_geojson_file,
+            tmp_csv_file,
+            "-sql",
+            f'SELECT *, CAST({geometry_field} as geometry) FROM "tmp"',  # noqa E501
+        ],
+        check=True,
+    )
+
+    if add_metadata:
+        add_metadata_to_geojson(tmp_geojson_file)
+
+    cast_datatypes_for_geojson(tmp_geojson_file)
+
+    # compress geojson file with gzip
+    with open(tmp_geojson_file, "r") as f:
+        json_data = json.load(f)
+
+    with gzip.open(outfile, 'wt') as fout:
+        json.dump(json_data, fout)
+
+    logger.info(f"converted {filename} to {outfile} with ogr2ogr.")
+
+
 def csv_to_geojson(filename: str, geometry_field: str = "geom"):
     """
     Use ogr2ogr to convert csv file to GeoJSON