feat:add osm ref to results (wip)

Author: ofr1tz

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -1,5 +1,6 @@
 import csv
 import io
+import json
 from typing import List, Tuple
 
 import dateutil.parser
@@ -269,6 +270,10 @@ def results_to_file(
 
             if type(result_data["results"]) is dict:
                 for taskId, result in result_data["results"].items():
+
+                    ref_data = result_data.get("ref", {}).get(taskId, {})
+                    ref_json = json.dumps(ref_data) if ref_data else None
+
                     if result_type == "geometry":
                         result = geojson.dumps(geojson.GeometryCollection(result))
                     w.writerow(
@@ -283,6 +288,7 @@ def results_to_file(
                             result,
                             app_version,
                             client_type,
+                            ref_json,
                         ]
                     )
             elif type(result_data["results"]) is list:
@@ -292,6 +298,10 @@ def results_to_file(
                 # if first key (list index) is 5
                 # list indicies 0-4 will have value None
                 for taskId, result in enumerate(result_data["results"]):
+
+                    ref_data = result_data.get("ref", {}).get(taskId, {})
+                    ref_json = json.dumps(ref_data) if ref_data else None
+
                     if result is None:
                         continue
                     else:
@@ -309,6 +319,7 @@ def results_to_file(
                                 result,
                                 app_version,
                                 client_type,
+                                ref_json,
                             ]
                         )
             else:
@@ -369,6 +380,7 @@ def save_results_to_postgres(
         "result",
         "app_version",
         "client_type",
+        "ref",
     ]
     p_con.copy_from(results_file, result_temp_table, columns)
     results_file.close()
@@ -439,7 +451,8 @@ def save_results_to_postgres(
             SELECT
                 ms.mapping_session_id,
                 r.task_id,
-                {result_sql}
+                {result_sql},
+                r.ref
             FROM {result_temp_table} r
             JOIN mapping_sessions ms ON
                 ms.project_id = r.project_id

mapswipe_workers/mapswipe_workers/generate_stats/project_stats.py

@@ -98,6 +98,7 @@ def get_results(
     Parse timestamp as datetime object and add attribute "day" for each result.
     Return None if there are no results for this project.
     Otherwise, return dataframe.
+    Include the 'ref' JSON field in integer results if it exists.
 
     Parameters
     ----------
@@ -108,7 +109,10 @@ def get_results(
     if result_table == "mapping_sessions_results_geometry":
         result_sql = "ST_AsGeoJSON(msr.result) as result"
     else:
-        result_sql = "msr.result"
+        result_sql = """
+            (msr.result->>'result')::int as result,
+            msr.result->'ref' as ref
+        """
 
     sql_query = sql.SQL(
         f"""
@@ -504,6 +508,44 @@ def get_statistics_for_geometry_result_project(project_id: str):
         return project_stats_dict
 
 
+def unify_refs(ref_list):
+    if not ref_list:
+        return None
+    first_ref = json.dumps(ref_list[0], sort_keys=True)
+    for r in ref_list[1:]:
+        if json.dumps(r, sort_keys=True) != first_ref:
+            return "multiple"
+    return ref_list[0]
+
+
+def add_ref_to_agg_results(
+    results_df: pd.DataFrame, agg_results_df: pd.DataFrame
+) -> pd.DataFrame:
+    """
+    Add a 'ref' column to agg_results_df.
+    If all user refs for a task are identical, use that ref.
+    If refs differ, set ref to 'multiple'.
+    """
+
+    # collect refs per task
+    refs_per_task = (
+        results_df.groupby(["project_id", "group_id", "task_id"])["ref"]
+        .apply(list)
+        .reset_index()
+    )
+
+    refs_per_task["ref"] = refs_per_task["ref"].apply(unify_refs)
+
+    # merge into agg_results_df
+    agg_results_df = agg_results_df.merge(
+        refs_per_task[["project_id", "group_id", "task_id", "ref"]],
+        on=["project_id", "group_id", "task_id"],
+        how="left",
+    )
+
+    return agg_results_df
+
+
 def get_statistics_for_integer_result_project(
     project_id: str, project_info: pd.Series, generate_hot_tm_geometries: bool
 ) -> dict:
@@ -550,6 +592,9 @@ def get_statistics_for_integer_result_project(
             tasks_df,
             project_info["custom_options"],
         )
+
+        agg_results_df = add_ref_to_agg_results(results_df, agg_results_df)
+
         agg_results_df.to_csv(agg_results_filename, index_label="idx")
 
         geojson_functions.gzipped_csv_to_gzipped_geojson(

mapswipe_workers/requirements.txt

@@ -6,6 +6,7 @@ flake8==3.8.3
 geojson==3.0.1
 mapswipe-workers==3.0
 pandas==1.5.2
+numpy==1.26.4
 pre-commit==2.9.2
 psycopg2-binary==2.9.3
 python-dateutil==2.8.1

postgres/scripts/add_ref_to_results_for_conflation.sql

@@ -0,0 +1,2 @@
+ALTER TABLE mapping_sessions_results
+ADD COLUMN ref jsonb;