Climate trace ghg emission - manifest file correction

statvar_imports/climate_trace/README.md

@@ -40,15 +40,31 @@ The script will create separate CSV files for each gas type in the `input_files/
 
 ### Processing Data
 
-After downloading, you can process the generated CSV files using the `stat_var_processor.py` tool. Here's an example:
+After downloading, you can process the generated CSV files using the `stat_var_processor.py` tool.
+
+**For sectors:**
 
 ```bash
- ../../tools/statvar_importer/stat_var_processor.py \
-  --input_data=input_files/all_countries_co2.csv \
+../../tools/statvar_importer/stat_var_processor.py \
+  --input_data=input_files/all*.csv \
   --pv_map=climate_pvmap_sectors.csv \
   --config_file=common_metadata.csv \
-  --output_path=output/sectors_CO2_output \
-  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf
+  --output_path=output/output_sectors \
+  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \
+  --output_counters=counters/sectors_counters.csv
+```
+
+**For subsectors:**
+
+```bash
+../../tools/statvar_importer/stat_var_processor.py \
+  --input_data=input_files/all*.csv \
+  --pv_map=climate_pvmap_subsectors.csv \
+  --config_file=common_metadata.csv \
+  --output_path=output/output_subsectors \
+  --statvar_dcid_remap_csv=statvar_remap.csv \
+  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \
+  --output_counters=counters/subsectors_counters.csv
 ```
 
 ## Project Structure
@@ -58,4 +74,6 @@ After downloading, you can process the generated CSV files using the `stat_var_p
 *   `common_metadata.csv`: Defines common metadata parameters for data processing.
 *   `statvar_remap.csv`: Contains the mapping from internal statistical variable names to Data Commons IDs.
 *   `input_files/`: Directory containing the segregated CSV files (e.g., `all_countries_co2.csv`) after running the main script.
+*   `counters/`: Directory containing operational counters generated by `stat_var_processor.py`.
 *   `output/`: Directory containing further processed and standardized output files (e.g., TMCF, MCF files).
+

statvar_imports/climate_trace/climate_pvmap_subsectors.csv

@@ -72,7 +72,7 @@ non-residential-onsite-fuel-usage,emissionSource,FuelCombustionForNonResidential
 forest-land-clearing,emissionSource,ForestClearing,,,,,,,,,,
 forest-land-degradation,emissionSource,ForestLandDegredation,,,,,,,,,,
 forest-land-fires,emissionSource,ForestLandFire,,,,,,,,,,
-removals,emissionSource,Removals,,,,,,,,,,
+removals,emissionSource,EmissionRemovals,,,,,,,,,,
 shrubgrass-fires,emissionSource,ShrublandFire,,,,,,,,,,
 water-reservoirs,emissionSource,WaterReservoirs,,,,,,,,,,
 wetland-fires,emissionSource,WetlandFire,,,,,,,,,,

statvar_imports/climate_trace/download_and_segregate_by_gas.py

@@ -6,22 +6,43 @@
 import os
 import logging
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
-def download_and_process_zip(url, country_iso, gas):
+
+def get_retry_session(retries=3, backoff_factor=1):
+    """Creates a requests.Session with connection pooling and retry backoff."""
+    session = requests.Session()
+    retry = Retry(
+        total=retries,
+        backoff_factor=backoff_factor,
+        status_forcelist=[429, 500, 502, 503, 504],
+        allowed_methods=["GET"],
+        raise_on_status=False,
+    )
+    adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
+
+def download_and_process_zip(url, country_iso, gas, session=None):
     """
     Downloads a single zip file and processes it in memory, returning a DataFrame.
     """
     try:
         logging.info(f"  Downloading: {country_iso} for {gas}...")
-        response = requests.get(url)
+        client = session if session is not None else requests
+        response = client.get(url, timeout=60)
         response.raise_for_status()
 
         with zipfile.ZipFile(io.BytesIO(response.content)) as zip_ref:
             csv_files_info = [
                 f for f in zip_ref.infolist()
-                if "country" in f.filename.lower() and f.filename.endswith('.csv') and not f.is_dir()
+                if ("country" in f.filename.lower() and f.filename.endswith('.csv') and
+                    not f.is_dir())
             ]
 
             df_list = []
@@ -33,88 +54,135 @@ def download_and_process_zip(url, country_iso, gas):
             if df_list:
                 return pd.concat(df_list, ignore_index=True)
             else:
+                logging.warning(
+                    f"    -> No relevant CSV files found in zip for {country_iso} ({gas})"
+                )
                 return None
+    except requests.exceptions.HTTPError as e:
+        if e.response is not None and e.response.status_code == 404:
+            logging.warning(f"    -> Not found (404) for {country_iso} ({gas}) at {url}")
+            return None
+        status_code = (
+            e.response.status_code if getattr(e, 'response', None) is not None else "N/A"
+        )
+        logging.error(
+            f"    -> HTTP Error for {country_iso} ({gas}) at {url} (Status: {status_code}): {e}"
+        )
+        raise
     except requests.exceptions.RequestException as e:
-        logging.error(f"    -> Failed to download {url}: {e}")
+        status_code = (
+            e.response.status_code if getattr(e, 'response', None) is not None else "N/A"
+        )
+        logging.error(
+            f"    -> Request failed for {country_iso} ({gas}) at {url} (Status: {status_code}): {e}"
+        )
         raise
-    except zipfile.BadZipFile:
-        logging.error(f"    -> Bad zip file for {url}")
+    except zipfile.BadZipFile as e:
+        logging.error(f"    -> Bad zip file for {country_iso} ({gas}) at {url}: {e}")
         raise
     except Exception as e:
-        logging.error(f"    -> An unexpected error occurred for {url}: {e}")
+        logging.error(f"    -> Unexpected error for {country_iso} ({gas}) at {url}: {e}")
         raise
 
 def download_and_segregate_by_gas():
     """
     Generates a fresh list of country download URLs and then downloads all
     data, saving a separate concatenated CSV for each gas.
     """
-    logging.info("--- Step 1: Generating Country URL List ---")
+    failed_downloads = []
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    session = get_retry_session()
+
+    logging.info("--- Step 1: Generating Country List ---")
     api_country_codes = set()
+    countries_url = "https://api.climatetrace.org/v7/admins?level=0"
+    logging.info(f"Fetching country list from API: {countries_url}")
     try:
-        countries_url = "https://api.climatetrace.org/v7/admins?level=0"
-        response = requests.get(countries_url)
+        response = session.get(countries_url, timeout=60)
         response.raise_for_status()
         countries = response.json()
         api_country_codes = {country['id'] for country in countries}
-        logging.info(f"Successfully fetched {len(api_country_codes)} countries from API.")
+        logging.info(
+            f"Successfully fetched {len(api_country_codes)} countries from API ({countries_url}). "
+            f"Status: {response.status_code}."
+        )
     except requests.exceptions.RequestException as e:
-        logging.error(f"Error: Could not fetch country list from API: {e}")
-        raise
+        status_code = (
+            e.response.status_code if getattr(e, 'response', None) is not None else "N/A"
+        )
+        response_text = (
+            e.response.text if getattr(e, 'response', None) is not None else "No response body"
+        )
+        logging.error(
+            f"Error: Could not fetch country list from API ({countries_url}). "
+            f"Status: {status_code}, Response: {response_text}, Error: {e}"
+        )
+        # Not raising here, we might still have check_country.csv
 
     local_country_codes = set()
     try:
-        with open('check_country.csv', 'r') as f:
+        with open(os.path.join(script_dir, 'check_country.csv'), 'r') as f:
             local_country_codes = {line.strip() for line in f if line.strip()}
         logging.info(f"Read {len(local_country_codes)} countries from 'check_country.csv'.")
     except FileNotFoundError:
         logging.warning("Warning: 'check_country.csv' not found. Will only use countries from API.")
 
     combined_codes = sorted(list(api_country_codes.union(local_country_codes)))
+    if not combined_codes:
+        logging.error("No countries to process. Exiting.")
+        raise RuntimeError("No country codes found from API or check_country.csv.")
+
     logging.info(f"Total unique countries to process: {len(combined_codes)}")
 
     gases = ["co2", "ch4", "n2o", "co2e_20yr", "co2e_100yr"]
     base_url = "https://downloads.climatetrace.org/latest/country_packages"
-    country_urls = {}
-    for iso in combined_codes:
-        country_urls[iso] = {}
-        for gas in gases:
-            url = f"{base_url}/{gas}/{iso}.zip"
-            country_urls[iso][gas] = url
-
-    logging.info("--- Step 1 Complete: URL List Generated ---\n")
+
     logging.info("--- Step 2: Downloading and Processing Data ---")
 
-    output_dir = "input_files"
+    output_dir = os.path.join(script_dir, "input_files")
     os.makedirs(output_dir, exist_ok=True)
 
-    all_gases = sorted(list(set(gas for gases in country_urls.values() for gas in gases)))
-    logging.info(f"Found data for the following gas types: {', '.join(all_gases)}\n")
+    logging.info(f"Found data for the following gas types: {', '.join(gases)}\n")
 
-    for gas in all_gases:
+    for gas in gases:
         logging.info(f"--- Starting processing for gas: {gas} ---")
 
-        gas_specific_urls = []
-        for iso, gas_data in country_urls.items():
-            if gas in gas_data:
-                gas_specific_urls.append({"iso": iso, "url": gas_data[gas]})
-
         gas_dataframes = []
+        critical_errors = []
         with ThreadPoolExecutor(max_workers=10) as executor:
-            future_to_url = {
-                executor.submit(download_and_process_zip, item['url'], item['iso'], gas): item
-                for item in gas_specific_urls
+            future_to_iso = {
+                executor.submit(
+                    download_and_process_zip,
+                    f"{base_url}/{gas}/{iso}.zip",
+                    iso,
+                    gas,
+                    session
+                ): iso
+                for iso in combined_codes
             }
-            for future in as_completed(future_to_url):
-                result_df = future.result()
-                if result_df is not None and not result_df.empty:
-                    gas_dataframes.append(result_df)
+            for future in as_completed(future_to_iso):
+                iso = future_to_iso[future]
+                try:
+                    result_df = future.result()
+                    if result_df is not None and not result_df.empty:
+                        gas_dataframes.append(result_df)
+                    else:
+                        failed_downloads.append(f"{iso} ({gas})")
+                except Exception as e:
+                    logging.error(f"Critical error for {iso} ({gas}): {e}")
+                    critical_errors.append(f"{iso} ({gas}) - Error: {e}")
+
+        if critical_errors:
+            raise RuntimeError(
+                f"Critical download failures for {gas}:\n" + "\n".join(critical_errors)
+            )
 
         if not gas_dataframes:
             logging.info(f"No data was downloaded for {gas}. The output file will not be created.")
             continue
 
         output_filename = os.path.join(output_dir, f"all_countries_{gas}.csv")
+        temp_filename = f"{output_filename}.tmp"
         logging.info(f"\n  -> All downloads for {gas} complete. Concatenating...")
 
         try:
@@ -128,13 +196,32 @@ def download_and_segregate_by_gas():
                 )
 
             logging.info(f"  -> Saving combined data to {output_filename}...")
-            final_df.to_csv(output_filename, index=False)
-            logging.info(f"  -> Successfully created {output_filename} with {len(final_df)} rows.\n")
+            final_df.to_csv(temp_filename, index=False)
+            os.replace(temp_filename, output_filename)
+            logging.info(
+                f"  -> Successfully created {output_filename} with {len(final_df)} rows.\n"
+            )
         except Exception as e:
+            if os.path.exists(temp_filename):
+                try:
+                    os.remove(temp_filename)
+                except OSError:
+                    pass
             logging.error(f"  -> An error occurred during the final processing for {gas}: {e}\n")
             raise
 
+    session.close()
     logging.info("--- All processing complete. ---")
+    if failed_downloads:
+        logging.warning(
+            f"The following {len(failed_downloads)} downloads were not found (404) "
+            "or contained no relevant CSV files:"
+        )
+        for failure in sorted(failed_downloads):
+            logging.warning(f"  - {failure}")
+    else:
+        logging.info("All downloads succeeded!")
+
 
 if __name__ == '__main__':
     download_and_segregate_by_gas()

statvar_imports/climate_trace/manifest.json

@@ -1,27 +1,29 @@
 {
   "import_specifications": [
     {
-      "import_name": "ClimateTrace",
+      "import_name": "ClimateTrace_GHG_Emissions",
       "curator_emails": [
         "support@datacommons.org"
       ],
       "provenance_url": "https://climatetrace.org/",
       "provenance_description": "Climate Trace data provides comprehensive, granular, and up-to-date greenhouse gas emissions data using satellite imagery and other public data sources.",
       "scripts": [
         "download_and_segregate_by_gas.py",
-        "../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/all*.csv --pv_map=climate_pvmap_sectors.csv --config_file=common_metadata.csv --output_path=output/output_sectors --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf",
-        "../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/all*.csv --pv_map=climate_pvmap_subsectors.csv --config_file=common_metadata.csv --output_path=output/output_subsectors --statvar_dcid_remap_csv=statvar_remap.csv  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf"
+        "../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/all*.csv --pv_map=climate_pvmap_sectors.csv --config_file=common_metadata.csv --output_path=output/output_sectors --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_counters=counters/sectors_counters.csv",
+        "../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/all*.csv --pv_map=climate_pvmap_subsectors.csv --config_file=common_metadata.csv --output_path=output/output_subsectors --statvar_dcid_remap_csv=statvar_remap.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_counters=counters/subsectors_counters.csv"
       ],
       "import_inputs": [
         {
           "template_mcf": "common_output.tmcf",
-          "cleaned_csv": "output/*.csv"
+          "cleaned_csv": "output/*.csv",
+          "node_mcf": "output/*.mcf"
         }
       ],
-      "source_file": [
-        "input_files/*"
+      "source_files": [
+        "input_files/*",
+        "counters/*.csv"
       ],
       "cron_schedule": "30 05 7,24 * *"
     }
   ]
-}
+}