fix(conflation): ref column in agg_results

Author: ofr1tz

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -271,7 +271,7 @@ 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_data = result_data.get("reference", {}).get(taskId, {})
                     ref_json = json.dumps(ref_data) if ref_data else r"\N"
 
                     if result_type == "geometry":
@@ -299,7 +299,7 @@ def results_to_file(
                 # 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_data = result_data.get("reference", {}).get(taskId, {})
                     ref_json = json.dumps(ref_data) if ref_data else r"\N"
 
                     if result is None:

mapswipe_workers/mapswipe_workers/generate_stats/project_stats.py

@@ -5,6 +5,7 @@
 import os
 import tempfile
 import typing
+import csv
 
 import pandas as pd
 from pandas.api.types import is_numeric_dtype
@@ -109,10 +110,7 @@ def get_results(
     if result_table == "mapping_sessions_results_geometry":
         result_sql = "ST_AsGeoJSON(msr.result) as result"
     else:
-        result_sql = """
-            (msr.result->>'result')::int as result,
-            msr.result->'ref' as ref
-        """
+        result_sql = "msr.result as result, msr.ref as ref"
 
     sql_query = sql.SQL(
         f"""
@@ -431,6 +429,9 @@ def get_agg_results_by_task_id(
         :, ~agg_results_df.columns.str.contains("Unnamed")
     ]
 
+    # Add ref column
+    agg_results_df = add_ref_to_agg_results(results_df, agg_results_df)
+
     return agg_results_df
 
 
@@ -508,41 +509,26 @@ 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:
+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'.
+    Adds a 'ref' column to agg_results_df for writing to CSV
     """
 
-    # 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)
+    refs_per_task = results_df.groupby("task_id")["ref"].apply(list)
 
-    # 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",
-    )
+    ref_values = {}
+    for task_id, refs in refs_per_task.items():
+        # Filter out None or empty dicts
+        refs = [r for r in refs if r not in (None, {}, "") and not pd.isna(r)]
+        if not refs:
+            continue
+        elif all(r == refs[0] for r in refs):
+            ref_values[task_id] = refs[0]
+        else:
+            ref_values[task_id] = refs
 
+    if ref_values:
+        agg_results_df["ref"] = agg_results_df["task_id"].map(ref_values).fillna("")
     return agg_results_df
 
 
@@ -593,9 +579,7 @@ def get_statistics_for_integer_result_project(
             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")
+        agg_results_df.to_csv(agg_results_filename, index_label="idx", quotechar='"', quoting=csv.QUOTE_MINIMAL)
 
         geojson_functions.gzipped_csv_to_gzipped_geojson(
             filename=agg_results_filename,

mapswipe_workers/tests/unittests/test_project_stats.py

@@ -3,6 +3,7 @@
 import pandas as pd
 
 from mapswipe_workers.generate_stats.project_stats import (
+    add_ref_to_agg_results,
     add_missing_result_columns,
     calc_agreement,
     calc_count,
@@ -172,6 +173,31 @@ def test_calc_parent_option_count(self):
             assert list(compared["other"].index) == updated_index
             assert list(compared["other"]) == updated_value
 
+    def test_add_ref_single_ref(self):
+        # All results have the same ref
+        results_df = pd.DataFrame({
+            "task_id": ["t1", "t1"],
+            "ref": [{"osmId": 123, "osmType": "ways_poly"}, {"osmId": 123, "osmType": "ways_poly"}]
+        })
+        agg_results_df = pd.DataFrame({"task_id": ["t1"]})
+        updated_df = add_ref_to_agg_results(results_df, agg_results_df.copy())
+        self.assertIn("ref", updated_df.columns)
+        self.assertEqual(updated_df["ref"].iloc[0], {"osmId": 123, "osmType": "ways_poly"})
+
+    def test_add_ref_multiple_refs(self):
+        # Different refs for same task
+        results_df = pd.DataFrame({
+            "task_id": ["t1", "t1"],
+            "ref": [{"osmId": 123}, {"osmId": 456}]
+        })
+        agg_results_df = pd.DataFrame({"task_id": ["t1"]})
+        updated_df = add_ref_to_agg_results(results_df, agg_results_df.copy())
+        self.assertIn("ref", updated_df.columns)
+        self.assertEqual(
+            updated_df["ref"].iloc[0],
+            [{"osmId": 123}, {"osmId": 456}]
+        )
+
 
 if __name__ == "__main__":
     unittest.main()