Add harmonization of beneficiary data in credit processing functions

Author: andersy005

offsets_db_data/apx.py

@@ -11,6 +11,7 @@
     load_registry_project_column_mapping,
 )
 from offsets_db_data.credits import *  # noqa: F403
+from offsets_db_data.credits import harmonize_beneficiary_data
 from offsets_db_data.models import credit_without_id_schema, project_schema
 from offsets_db_data.projects import *  # noqa: F403
 
@@ -44,7 +45,12 @@ def determine_transaction_type(df: pd.DataFrame, *, download_type: str) -> pd.Da
 
 @pf.register_dataframe_method
 def process_apx_credits(
-    df: pd.DataFrame, *, download_type: str, registry_name: str, arb: pd.DataFrame | None = None
+    df: pd.DataFrame,
+    *,
+    download_type: str,
+    registry_name: str,
+    arb: pd.DataFrame | None = None,
+    harmonize_beneficiary_info: bool = False,
 ) -> pd.DataFrame:
     """
     Process APX credits data by setting registry, determining transaction types, renaming columns,
@@ -94,6 +100,9 @@ def process_apx_credits(
     )
     if arb is not None and not arb.empty:
         data = data.merge_with_arb(arb=arb)
+
+    if harmonize_beneficiary_info:
+        data = data.pipe(harmonize_beneficiary_data)
     return data
 
 

offsets_db_data/credits.py

@@ -1,3 +1,7 @@
+import pathlib
+import subprocess
+import tempfile
+
 import janitor  # noqa: F401
 import pandas as pd
 import pandas_flavor as pf
@@ -122,3 +126,29 @@ def merge_with_arb(credits: pd.DataFrame, *, arb: pd.DataFrame) -> pd.DataFrame:
 
     df = pd.concat([df, arb], ignore_index=True)
     return df
+
+
+def harmonize_beneficiary_data(credits: pd.DataFrame) -> pd.DataFrame:
+    """
+    Harmonize the beneficiary information by removing the 'beneficiary_id' column and renaming the 'beneficiary_name' column to 'beneficiary'.
+
+    Parameters
+    ----------
+    credits : pd.DataFrame
+        Input DataFrame containing credit data.
+    """
+
+    tempdir = tempfile.gettempdir()
+    temp_path = pathlib.Path(tempdir) / 'credits.csv'
+    credits.to_csv(temp_path, index=False)
+
+    try:
+        result = subprocess.run(
+            ['offsets-db-data-orcli', 'run', 'list'],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        print(result.stdout)
+    except Exception as e:
+        raise e

offsets_db_data/gld.py

@@ -10,21 +10,18 @@
     load_protocol_mapping,
     load_registry_project_column_mapping,
 )
-from offsets_db_data.credits import (
-    aggregate_issuance_transactions,  # noqa: F401
-    filter_and_merge_transactions,  # noqa: F401
-    merge_with_arb,  # noqa: F401
-)
+from offsets_db_data.credits import aggregate_issuance_transactions  # noqa: F401
+from offsets_db_data.credits import filter_and_merge_transactions  # noqa: F401
+from offsets_db_data.credits import merge_with_arb  # noqa: F401
+from offsets_db_data.credits import harmonize_beneficiary_data
 from offsets_db_data.models import credit_without_id_schema, project_schema
-from offsets_db_data.projects import (
-    harmonize_country_names,  # noqa: F401
-    add_category,  # noqa: F401
-    add_is_compliance_flag,  # noqa: F401
-    map_protocol,  # noqa: F401
-    harmonize_status_codes,  # noqa: F401
-    add_first_issuance_and_retirement_dates,  # noqa: F401
-    add_retired_and_issued_totals,  # noqa: F401
-)
+from offsets_db_data.projects import add_category  # noqa: F401
+from offsets_db_data.projects import add_first_issuance_and_retirement_dates  # noqa: F401
+from offsets_db_data.projects import add_is_compliance_flag  # noqa: F401
+from offsets_db_data.projects import add_retired_and_issued_totals  # noqa: F401
+from offsets_db_data.projects import harmonize_country_names  # noqa: F401
+from offsets_db_data.projects import harmonize_status_codes  # noqa: F401
+from offsets_db_data.projects import map_protocol  # noqa: F401
 
 
 @pf.register_dataframe_method
@@ -80,6 +77,7 @@ def process_gld_credits(
     registry_name: str = 'gold-standard',
     prefix: str = 'GLD',
     arb: pd.DataFrame | None = None,
+    harmonize_beneficiary_info: bool = False,
 ) -> pd.DataFrame:
     """
     Process Gold Standard credits data by renaming columns, setting registry, determining transaction types,
@@ -142,6 +140,9 @@ def process_gld_credits(
             .validate(schema=credit_without_id_schema)
         )
 
+    if harmonize_beneficiary_info:
+        data = data.pipe(harmonize_beneficiary_data)
+
     return data
 
 

offsets_db_data/vcs.py

@@ -11,6 +11,7 @@
     load_registry_project_column_mapping,
 )
 from offsets_db_data.credits import *  # noqa: F403
+from offsets_db_data.credits import harmonize_beneficiary_data
 from offsets_db_data.models import credit_without_id_schema, project_schema
 from offsets_db_data.projects import *  # noqa: F403
 
@@ -175,6 +176,7 @@ def process_vcs_credits(
     registry_name: str = 'verra',
     prefix: str = 'VCS',
     arb: pd.DataFrame | None = None,
+    harmonize_beneficiary_info: bool = False,
 ) -> pd.DataFrame:
     """
     Process Verra credits data, including generation of project IDs, determination of transaction types,
@@ -235,6 +237,9 @@ def process_vcs_credits(
     if arb is not None and not arb.empty:
         data = data.merge_with_arb(arb=arb)
 
+    if harmonize_beneficiary_info:
+        data = data.pipe(harmonize_beneficiary_data)
+
     return data