Merge pull request #133 from Open-Earth-Foundation/on-5287-argentina-waste

feat: pipeline for waste argentina

Author: amanda-eames

cc-mage/data_exporters/load_indec_activity.sql

@@ -0,0 +1,32 @@
+WITH activity_data AS (
+    SELECT DISTINCT
+        MD5(CONCAT_WS(
+        '-',
+        activity_name,
+        activity_subcategory_type::TEXT,
+        method_id::TEXT
+    ))::UUID AS activity_id,
+        activity_name,
+        activity_units,
+        method_id,
+        activity_subcategory_type
+    FROM raw_data.arg_solid_waste_indec
+)
+INSERT INTO modelled.activity_subcategory (
+    activity_id, 
+    activity_name, 
+    activity_units, 
+    gpcmethod_id, 
+    activity_subcategory_type
+)
+SELECT  activity_id,
+        activity_name,
+        activity_units,
+        method_id as gpcmethod_id,
+        activity_subcategory_type::json
+FROM activity_data
+ON CONFLICT (activity_id) DO UPDATE SET
+    activity_name = EXCLUDED.activity_name,
+    activity_units = EXCLUDED.activity_units,
+    gpcmethod_id = EXCLUDED.gpcmethod_id,
+    activity_subcategory_type = EXCLUDED.activity_subcategory_type;

cc-mage/data_exporters/load_indec_ef.sql

@@ -0,0 +1,38 @@
+INSERT INTO modelled.emissions_factor (
+    emissionfactor_id, 
+    gas_name, 
+    emissionfactor_value,
+    unit_denominator,
+    activity_id,
+    datasource_name,
+    active_from,
+    active_to,
+    actor_id,
+    publisher_id,
+    dataset_id
+)
+SELECT emissionfactor_id, 
+    gas_name,
+    avg(emissionfactor_value) as emissionfactor_value, -- have to take the average because ef changes each month
+    emissions_factor_units AS unit_denominator,  
+    activity_id,
+    datasource_name,  
+    DATE(emissions_year || '-01-01')  AS active_from,      
+    DATE(emissions_year || '-12-31') AS active_to,        
+    actor_id,
+    publisher_id,
+    dataset_id
+FROM raw_data.ct_staging_v2025
+GROUP BY emissionfactor_id, gas_name, emissions_factor_units, activity_id, datasource_name, emissions_year, actor_id, publisher_id, dataset_id
+ON CONFLICT (emissionfactor_id)
+DO UPDATE SET
+    gas_name = EXCLUDED.gas_name,
+    emissionfactor_value = EXCLUDED.emissionfactor_value,
+    unit_denominator = EXCLUDED.unit_denominator,
+    activity_id = EXCLUDED.activity_id,
+    datasource_name = COALESCE(EXCLUDED.datasource_name, emissions_factor.datasource_name),
+    active_from = COALESCE(EXCLUDED.active_from, emissions_factor.active_from),
+    active_to = COALESCE(EXCLUDED.active_to, emissions_factor.active_to),
+    actor_id = EXCLUDED.actor_id,
+    publisher_id = EXCLUDED.publisher_id,
+    dataset_id = EXCLUDED.dataset_id;

cc-mage/data_exporters/load_indec_emissions.sql

@@ -0,0 +1,55 @@
+INSERT INTO modelled.emissions (
+    emissions_id,
+    datasource_name,
+    actor_id,
+    city_id,
+    gpc_reference_number,
+    emissions_value,
+    emissions_year,
+    emissions_units,
+    gpcmethod_id,
+    gas_name,
+    emissionfactor_id,
+    activity_id,
+    activity_value,
+    spatial_granularity,
+    geometry_type,
+    geometry,
+    geometry_id
+)
+ SELECT 
+    emissions_id,
+    datasource_name,
+    actor_id,
+    city_id,
+    gpc_reference_number,
+    emissions_value::numeric as emissions_value,
+    emissions_year,
+    'kg' as emissions_units,
+    method_id as gpcmethod_id,
+    upper(gas_name) as gas_name,
+    emissionfactor_id,
+    activity_id,
+    activity_value,
+    'city' as spatial_granularity,
+    geometry_type,
+    geometry,
+    geometry_id
+FROM raw_data.arg_solid_waste_indec
+ON CONFLICT (emissions_id) DO UPDATE SET
+    datasource_name = EXCLUDED.datasource_name,
+    actor_id = EXCLUDED.actor_id,
+    city_id = EXCLUDED.city_id,
+    gpc_reference_number = EXCLUDED.gpc_reference_number,
+    emissions_value = EXCLUDED.emissions_value,
+    emissions_year = EXCLUDED.emissions_year,
+    emissions_units = EXCLUDED.emissions_units,
+    gpcmethod_id = EXCLUDED.gpcmethod_id,
+    gas_name = EXCLUDED.gas_name,
+    emissionfactor_id = EXCLUDED.emissionfactor_id,
+    activity_id = EXCLUDED.activity_id,
+    activity_value = EXCLUDED.activity_value,
+    spatial_granularity = EXCLUDED.spatial_granularity,
+    geometry_type = EXCLUDED.geometry_type,
+    geometry = EXCLUDED.geometry,
+    geometry_id = EXCLUDED.geometry_id;

cc-mage/data_exporters/load_indec_methodology.sql

@@ -0,0 +1,20 @@
+WITH methodology AS (
+    SELECT DISTINCT
+        method_id,
+        methodology_name,
+        null as methodology_description,
+        gpc_reference_number
+    FROM raw_data.arg_solid_waste_indec
+)
+INSERT INTO modelled.ghgi_methodology 
+    (method_id, methodology_name, methodology_description, gpc_reference_number)
+SELECT
+    method_id,
+    methodology_name,
+    methodology_description,
+    gpc_reference_number
+FROM methodology
+ON CONFLICT (method_id) DO UPDATE SET
+    methodology_name = EXCLUDED.methodology_name,
+    methodology_description = EXCLUDED.methodology_description,
+    gpc_reference_number = EXCLUDED.gpc_reference_number;

cc-mage/data_exporters/load_solid_waste_indec_db.py

@@ -0,0 +1,31 @@
+from mage_ai.settings.repo import get_repo_path
+from mage_ai.io.config import ConfigFileLoader
+from mage_ai.io.postgres import Postgres
+from pandas import DataFrame
+from os import path
+
+if 'data_exporter' not in globals():
+    from mage_ai.data_preparation.decorators import data_exporter
+
+
+@data_exporter
+def export_data_to_postgres(df: DataFrame, **kwargs) -> None:
+    """
+    Template for exporting data to a PostgreSQL database.
+    Specify your configuration settings in 'io_config.yaml'.
+
+    Docs: https://docs.mage.ai/design/data-loading#postgresql
+    """
+    schema_name = 'raw_data'  # Specify the name of the schema to export data to
+    table_name = 'arg_solid_waste_staging'  # Specify the name of the table to export data to
+    config_path = path.join(get_repo_path(), 'io_config.yaml')
+    config_profile = 'default'
+
+    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:
+        loader.export(
+            df,
+            schema_name,
+            table_name,
+            index=False,  # Specifies whether to include index in exported table
+            if_exists='replace',  # Specify resolution policy if table name already exists
+        )

cc-mage/data_loaders/load_solid_waste_indec.py

@@ -0,0 +1,36 @@
+from mage_ai.settings.repo import get_repo_path
+from mage_ai.io.config import ConfigFileLoader
+from mage_ai.io.s3 import S3
+from os import path
+if 'data_loader' not in globals():
+    from mage_ai.data_preparation.decorators import data_loader
+if 'test' not in globals():
+    from mage_ai.data_preparation.decorators import test
+
+
+@data_loader
+def load_from_s3_bucket(*args, **kwargs):
+    """
+    Template for loading data from a S3 bucket.
+    Specify your configuration settings in 'io_config.yaml'.
+
+    Docs: https://docs.mage.ai/design/data-loading#s3
+    """
+    config_path = path.join(get_repo_path(), 'io_config.yaml')
+    config_profile = 'default'
+
+    bucket_name = kwargs['bucket_name']
+    object_key = 'files/local/argentina/indec/arg_indec.csv'
+
+    return S3.with_config(ConfigFileLoader(config_path, config_profile)).load(
+        bucket_name,
+        object_key,
+    )
+
+
+@test
+def test_output(output, *args) -> None:
+    """
+    Template code for testing the output of the block.
+    """
+    assert output is not None, 'The output is undefined'