diff --git a/statvar_imports/climate_trace/README.md b/statvar_imports/climate_trace/README.md index 43505ca56c..bdf3b3f4c9 100644 --- a/statvar_imports/climate_trace/README.md +++ b/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). + diff --git a/statvar_imports/climate_trace/climate_pvmap_subsectors.csv b/statvar_imports/climate_trace/climate_pvmap_subsectors.csv index 47b00422aa..91d772cad2 100644 --- a/statvar_imports/climate_trace/climate_pvmap_subsectors.csv +++ b/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,,,,,,,,,, diff --git a/statvar_imports/climate_trace/download_and_segregate_by_gas.py b/statvar_imports/climate_trace/download_and_segregate_by_gas.py index bc067a9e2e..0e020c86e9 100644 --- a/statvar_imports/climate_trace/download_and_segregate_by_gas.py +++ b/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,15 +54,34 @@ 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(): @@ -49,72 +89,100 @@ 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() diff --git a/statvar_imports/climate_trace/manifest.json b/statvar_imports/climate_trace/manifest.json index 0a847c95de..6e779e9f3a 100644 --- a/statvar_imports/climate_trace/manifest.json +++ b/statvar_imports/climate_trace/manifest.json @@ -1,7 +1,7 @@ { "import_specifications": [ { - "import_name": "ClimateTrace", + "import_name": "ClimateTrace_GHG_Emissions", "curator_emails": [ "support@datacommons.org" ], @@ -9,19 +9,21 @@ "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 * *" } ] -} \ No newline at end of file +} diff --git a/statvar_imports/climate_trace/statvar_remap.csv b/statvar_imports/climate_trace/statvar_remap.csv index 67438b2247..5e88a66f4a 100644 --- a/statvar_imports/climate_trace/statvar_remap.csv +++ b/statvar_imports/climate_trace/statvar_remap.csv @@ -20,7 +20,7 @@ Annual_Amount_Emissions_ManureLeftOnPasture_CarbonDioxide,Annual_Emissions_Carbo Annual_Amount_Emissions_ManureManagementOperation_CarbonDioxide,Annual_Emissions_CarbonDioxide_ManureManagementOperation Annual_Amount_Emissions_NonBroadcastingVessels_CarbonDioxide,Annual_Emissions_CarbonDioxide_NonBroadcastingVessels Annual_Amount_Emissions_OilAndGasTransport_CarbonDioxide,Annual_Emissions_CarbonDioxide_OilAndGasTransport -Annual_Amount_Emissions_Removals_CarbonDioxide,Annual_Emissions_CarbonDioxide_Removals +Annual_Amount_Emissions_EmissionRemovals_CarbonDioxide,Annual_Emissions_CarbonDioxide_EmissionRemovals Annual_Amount_Emissions_TextileLeatherApparel_CarbonDioxide,Annual_Emissions_CarbonDioxide_TextileLeatherApparel Annual_Amount_Emissions_WaterReservoirs_CarbonDioxide,Annual_Emissions_CarbonDioxide_WaterReservoirs Annual_Amount_Emissions_WoodProducts_CarbonDioxide,Annual_Emissions_CarbonDioxide_WoodProducts @@ -45,7 +45,7 @@ Annual_Amount_Emissions_ManureLeftOnPasture_Methane,Annual_Emissions_Methane_Man Annual_Amount_Emissions_ManureManagementOperation_Methane,Annual_Emissions_Methane_ManureManagementOperation Annual_Amount_Emissions_NonBroadcastingVessels_Methane,Annual_Emissions_Methane_NonBroadcastingVessels Annual_Amount_Emissions_OilAndGasTransport_Methane,Annual_Emissions_Methane_OilAndGasTransport -Annual_Amount_Emissions_Removals_Methane,Annual_Emissions_Methane_Removals +Annual_Amount_Emissions_EmissionRemovals_Methane,Annual_Emissions_Methane_EmissionRemovals Annual_Amount_Emissions_TextileLeatherApparel_Methane,Annual_Emissions_Methane_TextileLeatherApparel Annual_Amount_Emissions_WaterReservoirs_Methane,Annual_Emissions_Methane_WaterReservoirs Annual_Amount_Emissions_WoodProducts_Methane,Annual_Emissions_Methane_WoodProducts @@ -70,7 +70,7 @@ Annual_Amount_Emissions_ManureLeftOnPasture_NitrousOxide,Annual_Emissions_Nitrou Annual_Amount_Emissions_ManureManagementOperation_NitrousOxide,Annual_Emissions_NitrousOxide_ManureManagementOperation Annual_Amount_Emissions_NonBroadcastingVessels_NitrousOxide,Annual_Emissions_NitrousOxide_NonBroadcastingVessels Annual_Amount_Emissions_OilAndGasTransport_NitrousOxide,Annual_Emissions_NitrousOxide_OilAndGasTransport -Annual_Amount_Emissions_Removals_NitrousOxide,Annual_Emissions_NitrousOxide_Removals +Annual_Amount_Emissions_EmissionRemovals_NitrousOxide,Annual_Emissions_NitrousOxide_EmissionRemovals Annual_Amount_Emissions_TextileLeatherApparel_NitrousOxide,Annual_Emissions_NitrousOxide_TextileLeatherApparel Annual_Amount_Emissions_WaterReservoirs_NitrousOxide,Annual_Emissions_NitrousOxide_WaterReservoirs Annual_Amount_Emissions_WoodProducts_NitrousOxide,Annual_Emissions_NitrousOxide_WoodProducts @@ -95,7 +95,7 @@ Annual_Amount_Emissions_ManureLeftOnPasture_CarbonDioxideEquivalent20YearGlobalW Annual_Amount_Emissions_ManureManagementOperation_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_ManureManagementOperation Annual_Amount_Emissions_NonBroadcastingVessels_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_NonBroadcastingVessels Annual_Amount_Emissions_OilAndGasTransport_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_OilAndGasTransport -Annual_Amount_Emissions_Removals_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_Removals +Annual_Amount_Emissions_EmissionRemovals_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_EmissionRemovals Annual_Amount_Emissions_TextileLeatherApparel_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_TextileLeatherApparel Annual_Amount_Emissions_WaterReservoirs_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_WaterReservoirs Annual_Amount_Emissions_WoodProducts_CarbonDioxideEquivalent20YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent20YearGlobalWarmingPotential_WoodProducts @@ -120,7 +120,7 @@ Annual_Amount_Emissions_ManureLeftOnPasture_CarbonDioxideEquivalent100YearGlobal Annual_Amount_Emissions_ManureManagementOperation_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_ManureManagementOperation Annual_Amount_Emissions_NonBroadcastingVessels_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_NonBroadcastingVessels Annual_Amount_Emissions_OilAndGasTransport_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_OilAndGasTransport -Annual_Amount_Emissions_Removals_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_Removals +Annual_Amount_Emissions_EmissionRemovals_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_EmissionRemovals Annual_Amount_Emissions_TextileLeatherApparel_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_TextileLeatherApparel Annual_Amount_Emissions_WaterReservoirs_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_WaterReservoirs -Annual_Amount_Emissions_WoodProducts_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_WoodProducts \ No newline at end of file +Annual_Amount_Emissions_WoodProducts_CarbonDioxideEquivalent100YearGlobalWarmingPotential,Annual_Emissions_CarbonDioxideEquivalent100YearGlobalWarmingPotential_WoodProducts