67 adapt carbon module to new timba data structure

c_module/cli/cli.py

@@ -6,7 +6,7 @@
 
 @click.command()
 @click.option('-ADD_ON', '--add_on_activated', "add_on_activated",
-              default=user_input[ParamNames.add_on_activated.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.add_on_activated.value], show_default=True, required=True, is_flag=True,
               help="Flag to use the carbon module as a standalone module or as a TiMBA add-on.")
 @click.option('-SC', '--sc_num', "sc_num",
               default=user_input[ParamNames.sc_num.value], show_default=True, required=True, type=int,
@@ -18,31 +18,38 @@
               show_default=True, required=True, type=int,
               help="End year of carbon calculations.")
 @click.option('-CF_AGB', '--calc_c_forest_agb', "calc_c_forest_agb",
-              default=user_input[ParamNames.calc_c_forest_agb.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.calc_c_forest_agb.value], show_default=True, required=True, is_flag=True,
               help="Flag to activate carbon calculation for aboveground forest biomass.")
 @click.option('-CF_BGB', '--calc_c_forest_bgb', "calc_c_forest_bgb",
-              default=user_input[ParamNames.calc_c_forest_bgb.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.calc_c_forest_bgb.value], show_default=True, required=True, is_flag=True,
               help="Flag to activate carbon calculation for belowground forest biomass.")
 @click.option('-CF_S', '--calc_c_forest_soil', "calc_c_forest_soil",
-              default=user_input[ParamNames.calc_c_forest_soil.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.calc_c_forest_soil.value], show_default=True, required=True, is_flag=True,
               help="Flag to activate carbon calculation for forest soil.")
 @click.option('-CF_DWL', '--calc_c_forest_dwl', "calc_c_forest_dwl",
-              default=user_input[ParamNames.calc_c_forest_dwl.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.calc_c_forest_dwl.value], show_default=True, required=True, is_flag=True,
               help="Flag to activate carbon calculation for dead wood and litter.")
 @click.option('-C_HWP', '--calc_c_hwp', "calc_c_hwp",
-              default=user_input[ParamNames.calc_c_hwp.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.calc_c_hwp.value], show_default=True, required=True, is_flag=True,
               help="Flag to activate carbon calculation for harvested wood products.")
 @click.option('-C_HWP_A', '--c_hwp_accounting_approach', "c_hwp_accounting_approach",
               default=user_input[ParamNames.c_hwp_accounting_approach.value], show_default=True, required=True,
               type=str, help="Flag to select the accounting approach for carbon in harvested wood products.")
 @click.option('-R', '--read_in_pkl', "read_in_pkl",
-              default=user_input[ParamNames.read_in_pkl.value], show_default=True, required=True, type=bool,
+              default=user_input[ParamNames.read_in_pkl.value], show_default=True, required=True, is_flag=True,
               help="Flag to control if pkl- or csv-files are read; reads in if True.")
 @click.option('-SD', '--show_carbon_dashboard', 'show_carbon_dashboard',
-              default=user_input[ParamNames.show_carbon_dashboard.value], show_default=True, required=False, type=bool,
-              help="Flag to launch carbon dashboard.")
+              default=user_input[ParamNames.show_carbon_dashboard.value], show_default=True, required=False,
+              is_flag=True, help="Flag to launch carbon dashboard.")
+@click.option('-UD', '--fao_data_update', 'fao_data_update',
+              default=user_input[ParamNames.fao_data_update.value], show_default=True, required=False, is_flag=True,
+              help="Flag to update FAO data.")
+@click.option('-FP', '--folderpath', 'folderpath', default=user_input[ParamNames.folderpath.value],
+              show_default=True, required=False, type=str, help="Path to directory with Input/Output folder.")
+
 def cli(add_on_activated, sc_num, start_year, end_year, calc_c_forest_agb, calc_c_forest_bgb, calc_c_forest_soil,
-        calc_c_forest_dwl, calc_c_hwp, c_hwp_accounting_approach, read_in_pkl, show_carbon_dashboard):
+        calc_c_forest_dwl, calc_c_hwp, c_hwp_accounting_approach, read_in_pkl, show_carbon_dashboard, fao_data_update,
+        folderpath):
 
     user_input_cli = {
         ParamNames.add_on_activated.value: add_on_activated,
@@ -57,6 +64,8 @@ def cli(add_on_activated, sc_num, start_year, end_year, calc_c_forest_agb, calc_
         ParamNames.calc_c_hwp.value: calc_c_hwp,
         ParamNames.c_hwp_accounting_approach.value: c_hwp_accounting_approach,
         ParamNames.show_carbon_dashboard.value: show_carbon_dashboard,
+        ParamNames.fao_data_update.value: fao_data_update,
+        ParamNames.folderpath.value: folderpath,
         # Adavanced settings not available via CLI
         ParamNames.historical_c_hwp.value: user_input[ParamNames.historical_c_hwp.value],
         ParamNames.hist_hwp_start_year.value: user_input[ParamNames.hist_hwp_start_year.value],

c_module/data_management/data_manager.py

@@ -1,13 +1,11 @@
-from c_module.parameters.paths import (INPUT_FOLDER, TIMBADIR_INPUT, ADD_INFO_CARBON_PATH, ADD_INFO_COUNTRY,
-                                       FAOSTAT_DATA, FRA_DATA, OUTPUT_FOLDER, TIMBADIR_OUTPUT, FAOSTAT_URL, FAO_DIR,
-                                       FRA_URL)
 from c_module.parameters.paths import cmodule_is_standalone, extract_scenarios
-from c_module.parameters.defines import (VarNames, ParamNames, CountryConstants)
+from c_module.parameters.defines import (VarNames, ParamNames, CountryConstants, FolderNames, PathNames)
 from c_module.user_io.default_parameters import user_input
 import pandas as pd
 from tqdm import tqdm
 from pathlib import Path
 import requests
+from io import BytesIO
 import zipfile
 import io
 import time
@@ -18,9 +16,12 @@ class DataManager:
 
     @staticmethod
     def set_sc_paths(self):
-        if user_input[ParamNames.add_on_activated.value] or not cmodule_is_standalone():
-            # input paths for add-on c-module
+        TIMBADIR_INPUT = self.paths[PathNames.TIMBADIR_INPUT.value]
+        TIMBADIR_OUTPUT = self.paths[PathNames.TIMBADIR_OUTPUT.value]
+        INPUT_FOLDER = self.paths[PathNames.INPUT_FOLDER.value]
 
+        if user_input[ParamNames.add_on_activated.value] or not cmodule_is_standalone(debug=False):
+            # input paths for add-on c-module
             scenarios = extract_scenarios(input_folder=TIMBADIR_INPUT,
                                           output_folder=TIMBADIR_OUTPUT,
                                           sc_num=user_input[ParamNames.sc_num.value])
@@ -35,6 +36,136 @@ def set_sc_paths(self):
 
         self.sc_path = PKL_RESULTS_INPUT
 
+    @staticmethod
+    def check_input_data(self):
+        """
+        Checks input data for the C-Module in two steps. First, the input data structure is checked. After, the content
+        of each input data folder is checked.
+        """
+        self.logger.info(f"C-Module - Check input data for carbon module")
+        DataManager.check_input_data_structure(self)
+        DataManager.check_input_data_content(self)
+
+    @staticmethod
+    def check_input_data_structure(self):
+        """
+        Checks the input data structure. If input data folder are missing, the missing folder is generated.
+        """
+        INPUT_FOLDER = self.paths[PathNames.INPUT_FOLDER.value]
+        INPUT_FOLDER.mkdir(parents=True, exist_ok=True)
+
+        if cmodule_is_standalone(debug=False):
+            required = {FolderNames.additional_info.value, FolderNames.projection_data.value}
+        else:
+            required = {FolderNames.additional_info.value}
+        existing = {p.name for p in Path(INPUT_FOLDER).iterdir() if p.is_dir()}
+        missing = list(required - existing)
+        if len(missing) > 0:
+            for missing_folder in missing:
+                NEW_FOLDER = INPUT_FOLDER / Path(missing_folder)
+                NEW_FOLDER.mkdir(parents=True, exist_ok=True)
+
+    @staticmethod
+    def check_input_data_content(self):
+        """
+        Checks if input data folder content corresponds to folder content from the C-Module main branch on GitHub.
+        Missing input data is downloaded automatically.
+        :param self: C-Module object
+        """
+        INPUT_FOLDER = self.paths[PathNames.INPUT_FOLDER.value]
+        CMODULE_ZIP_URL = self.paths[PathNames.CMODULE_ZIP_URL.value]
+        ADD_INFO_DIR = self.paths[PathNames.ADD_INFO_DIR.value]
+        DEFAULT_PROJECTION_DIR = self.paths[PathNames.DEFAULT_PROJECTION_DIR.value]
+
+        subfolders = [p.name for p in INPUT_FOLDER.iterdir() if p.is_dir()]
+        for folder in subfolders:
+            if (folder == FolderNames.additional_info.value) or (folder == FolderNames.projection_data.value):
+                if folder == FolderNames.additional_info.value:
+                    # download additional info data
+                    GIT_DATA_DIR = ADD_INFO_DIR
+
+                if folder == FolderNames.projection_data.value:
+                    # download projection data
+                    GIT_DATA_DIR = DEFAULT_PROJECTION_DIR
+
+                folder_path = INPUT_FOLDER / Path(folder)
+                missing_files = DataManager.compare_local_and_remote(local_folder_path=folder_path,
+                                                                     repo_zip_url=CMODULE_ZIP_URL,
+                                                                     target_subdir=GIT_DATA_DIR)
+
+                for missing_file in list(missing_files):
+                    DataManager.download_carbon_data_from_github(self=self,
+                                                                 repo_zip_url=CMODULE_ZIP_URL,
+                                                                 target_subdir=GIT_DATA_DIR,
+                                                                 folder_path=folder_path,
+                                                                 missing_file=missing_file)
+
+    @staticmethod
+    def compare_local_and_remote(local_folder_path: Path, repo_zip_url: str, target_subdir: str):
+        """
+        Compares local and remote input data folder and returns missing files.
+        :param local_folder_path: Local input data folder
+        :param repo_zip_url: Remote input data zip url
+        :param target_subdir: Target subdirectory of remote input data folder
+        :return: Missing files in local folder
+        """
+        response = requests.get(repo_zip_url, timeout=60)
+        response.raise_for_status()
+
+        with zipfile.ZipFile(BytesIO(response.content)) as zip_file:
+            zip_files = zip_file.namelist()
+
+        # GitHub files
+        github_filenames = {
+            Path(f).name
+            for f in zip_files
+            if f.startswith(target_subdir) and not f.endswith("/")
+        }
+
+        # Local files
+        local_filenames = {
+            p.name for p in local_folder_path.iterdir() if p.is_file()
+        }
+
+        missing_local = github_filenames - local_filenames
+
+        return missing_local
+
+    @staticmethod
+    def download_carbon_data_from_github(self, repo_zip_url: str, target_subdir: str, folder_path: Path,
+                                         missing_file: str):
+        """
+        Downloads missing input data from GitHub.
+        :param self: C-Module object
+        :param repo_zip_url: Remote input data zip url
+        :param target_subdir: Target subdirectory of remote input data folder
+        :param folder_path: Local input data folder
+        :param missing_file: Input data files missing in local folder
+        """
+        response = requests.get(repo_zip_url, timeout=60)
+        response.raise_for_status()
+
+        with zipfile.ZipFile(BytesIO(response.content)) as zip_file:
+            zip_files = zip_file.namelist()
+
+            target_path = None
+            for f in zip_files:
+                if f.startswith(target_subdir) and f.endswith(missing_file):
+                    target_path = f
+                    break
+
+            if not target_path:
+                raise FileNotFoundError(
+                    f"{missing_file} not found in GitHub folder {target_subdir}"
+                )
+
+            self.logger.info(f"C-Module - Download {missing_file} from GitHub")
+
+            with zip_file.open(target_path) as zf:
+                out_file = folder_path / missing_file
+                with open(out_file, "wb") as f:
+                    f.write(zf.read())
+
     @staticmethod
     def load_data(filepath, table_name, input_source):
         if input_source.lower() == "excel":
@@ -101,6 +232,7 @@ def load_timba_data(self):
 
     @staticmethod
     def save_data(self):
+        OUTPUT_FOLDER = self.paths[PathNames.OUTPUT_FOLDER.value]
         for sc in self.sc_list:
             carbon_data_ext = DataManager.flattening_data(data=self.carbon_data[sc])
             carbon_data_ext = DataManager.add_additional_info(self, data=carbon_data_ext, sc=sc)
@@ -109,12 +241,11 @@ def save_data(self):
             if not self.UserInput[ParamNames.add_on_activated.value]:
                 DataManager.serialize_to_pickle(self.timba_data[sc], OUTPUT_FOLDER / Path(f"{sc}.pkl"))
             else:
-                DataManager.serialize_to_pickle(
-                    self.carbon_data[sc], OUTPUT_FOLDER / Path(f"{self.time_stamp}_{sc}.pkl"))
+                DataManager.serialize_to_pickle(self.carbon_data[sc], OUTPUT_FOLDER / Path(f"{sc}.pkl"))
 
             for df_key in self.carbon_data[sc].keys():
                 carbon_data = self.carbon_data[sc][df_key]
-                carbon_data_path = OUTPUT_FOLDER / Path(f"{df_key}_D{self.time_stamp}_{sc}")
+                carbon_data_path = OUTPUT_FOLDER / Path(f"{df_key}_{sc}")
                 carbon_data.to_csv(f"{carbon_data_path}.csv", index=False)
 
     @staticmethod
@@ -129,6 +260,7 @@ def merge_sc_data(self):
 
     @staticmethod
     def load_additional_data(self):
+        ADD_INFO_COUNTRY = self.paths[PathNames.ADD_INFO_COUNTRY.value]
         self.add_data["country_data"] = DataManager.load_data(
             f"{ADD_INFO_COUNTRY}.csv", ADD_INFO_COUNTRY, "csv")
 
@@ -139,7 +271,6 @@ def retrieve_commodity_num(self):
         commodity_dict = VarNames.commodity_dict.value
         commodity_code = VarNames.commodity_code.value
         commodity_num_name = VarNames.commodity_num.value
-
         commodity_num = len(self.timba_data[self.sc_list[0]][timba_data_all][commodity_code].unique())
         self.add_data[commodity_dict] = {}
         self.add_data[commodity_dict][commodity_num_name] = commodity_num
@@ -162,6 +293,8 @@ def load_additional_data_carbon(self):
         Additional information for projections of carbon removals and emissions are readin
         :param self: object of class C-Module
         """
+        ADD_INFO_CARBON_PATH = self.paths[PathNames.ADD_INFO_CARBON_PATH.value]
+
         commodity_code = VarNames.commodity_code.value
         for sheet_name in pd.ExcelFile(f"{ADD_INFO_CARBON_PATH}.xlsx").sheet_names:
             if "CarbonHWP_" in sheet_name:
@@ -186,6 +319,8 @@ def load_faostat_data(self, update_data: bool):
         :param self: object of class C-Module
         :param update_data: Flag to update FAOSTAT data even if max cache age is not reached
         """
+        FAOSTAT_DATA = self.paths[PathNames.FAOSTAT_DATA.value]
+        FAO_DIR = self.paths[PathNames.FAO_DIR.value]
         CSV_FILE = Path(f"{FAOSTAT_DATA}.csv")
 
         CACHE_MAX_AGE = 2 * 30 * 24 * 60 * 60  # 2 months
@@ -216,6 +351,11 @@ def download_fao_api_data(self, database: str):
         :param database: Database name
         :return: FAOSTAT data as DataFrame
         """
+        FRA_URL = self.paths[PathNames.FRA_URL.value]
+        FAOSTAT_URL = self.paths[PathNames.FAOSTAT_URL.value]
+        FAOSTAT_DATA = self.paths[PathNames.FAOSTAT_DATA.value]
+        FRA_DATA = self.paths[PathNames.FRA_DATA.value]
+
         self.logger.info(f"C-Module - Download {database} data from API")
         if database == "FRA":
             database_url = FRA_URL
@@ -387,6 +527,8 @@ def load_fra_data(self, update_data: bool):
         :param update_data: Flag to update FAOSTAT data even if max cache age is not reached
         """
         # Paths
+        FRA_DATA = self.paths[PathNames.FRA_DATA.value]
+        FAO_DIR = self.paths[PathNames.FAO_DIR.value]
         CSV_FILE = Path(f"{FRA_DATA}.csv")
 
         CACHE_MAX_AGE = 2 * 30 * 24 * 60 * 60  # 2 months

c_module/data_management/process_manager.py

@@ -1,13 +1,13 @@
 from c_module.data_management.data_manager import DataManager
-from c_module.parameters.paths import (FAOSTAT_DATA, FRA_DATA)
-from c_module.parameters.defines import (VarNames, ParamNames)
+from c_module.parameters.defines import (VarNames, ParamNames, PathNames)
 from pathlib import Path
 from c_module.logic.visualisation import Carbon_DashboardPlotter
 
 
 class ProcessManager:
     @staticmethod
     def run_readin_process(self):
+        DataManager.check_input_data(self)
         DataManager.set_sc_paths(self)
         ProcessManager.readin_add_data_process(self)
         ProcessManager.readin_timba_process(self)
@@ -37,6 +37,7 @@ def readin_carbon_process(self):
     @staticmethod
     def readin_faostat_process(self):
         self.logger.info("C-Module - Reading in FAOSTAT data")
+        FAOSTAT_DATA = self.paths[PathNames.FAOSTAT_DATA.value]
         DataManager.load_faostat_data(self, update_data=self.UserInput[ParamNames.fao_data_update.value])
         if not Path(f"{FAOSTAT_DATA}_processed.pkl").is_file():
             DataManager.prep_faostat_data(self)
@@ -48,6 +49,7 @@ def readin_faostat_process(self):
     @staticmethod
     def readin_fra_process(self):
         self.logger.info("C-Module - Reading in FRA data")
+        FRA_DATA = self.paths[PathNames.FRA_DATA.value]
         # TODO implement fra processing steps
         DataManager.load_fra_data(self, update_data=self.UserInput[ParamNames.fao_data_update.value])
         if not Path(f"{FRA_DATA}_processed.pkl").is_file():

c_module/logic/base_logger.py

@@ -3,18 +3,17 @@
 from pathlib import Path
 import datetime as dt
 import logging
-from c_module.parameters.paths import LOGGING_OUTPUT_FOLDER
 
 
-def get_logger(user_path: Union[str, Path, None], add_on_activated: bool):
+def get_logger(user_path: Union[str, Path, None], add_on_activated: bool, logging_folder):
     current_dt = dt.datetime.now().strftime("%Y%m%d")
     if add_on_activated:
         filename = f"{current_dt}_TiMBA.log"
     else:
         filename = rf"{current_dt}_C_Module.log"
 
     if user_path is None:
-        filepath = os.path.join(LOGGING_OUTPUT_FOLDER, filename)
+        filepath = os.path.join(logging_folder, filename)
     else:
         filepath = os.path.join(user_path, "output", filename)
     if not os.path.exists(filepath):