code: add logging to file

Author: finozzifa

Snakefile

@@ -2,10 +2,10 @@
 #
 # SPDX-License-Identifier: GPL-3.0-only
 
+import pathlib
 
 configfile: "config.yaml"
 
-
 rule compile_cost_assumptions:
     input:
         inflation_rate="inputs/prc_hicp_aind__custom_9928419_spreadsheet.xlsx",
@@ -32,6 +32,8 @@ rule compile_cost_assumptions:
         mem=500,
     conda:
         "environment.yaml"
+    log:
+        pathlib.Path("logs", "compile_cost_assumptions.log"),
     script:
         "scripts/compile_cost_assumptions.py"
 
@@ -54,6 +56,8 @@ rule compile_cost_assumptions_usa:
         mem=500,
     conda:
         "environment.yaml"
+    log:
+        pathlib.Path("logs", "compile_cost_assumptions_usa.log"),
     script:
         "scripts/compile_cost_assumptions_usa.py"
 

scripts/_helpers.py

@@ -10,6 +10,9 @@
 import numpy as np
 import pandas as pd
 
+import logging
+import sys
+
 
 class Dict(dict):
     """
@@ -251,3 +254,55 @@ def get_factor(inflation_rate_df, ref_year, eur_year_val):
     )
 
     return costs
+
+
+def configure_logging(snakemake, skip_handlers=False):
+    """
+    Configure the basic behaviour for the logging module.
+
+    Note: Must only be called once from the __main__ section of a script.
+
+    The setup includes printing log messages to STDERR and to a log file defined
+    by either (in priority order): snakemake.log.python, snakemake.log[0] or "logs/{rulename}.log".
+    Additional keywords from logging.basicConfig are accepted via the snakemake configuration
+    file under snakemake.config.logging.
+
+    Parameters
+    ----------
+    snakemake : snakemake object
+        Your snakemake object containing a snakemake.config and snakemake.log.
+    skip_handlers : True | False (default)
+        Do (not) skip the default handlers created for redirecting output to STDERR and file.
+    """
+
+    kwargs = snakemake.config.get("logging", dict()).copy()
+    kwargs.setdefault("level", "INFO")
+
+    if skip_handlers is False:
+        fallback_path = Path(__file__).parent.joinpath(
+            "..", "logs", f"{snakemake.rule}.log"
+        )
+        logfile = snakemake.log.get(
+            "python", snakemake.log[0] if snakemake.log else fallback_path
+        )
+        kwargs.update(
+            {
+                "handlers": [
+                    # Prefer the 'python' log, otherwise take the first log for each
+                    # Snakemake rule
+                    logging.FileHandler(logfile),
+                    logging.StreamHandler(),
+                ]
+            }
+        )
+    logging.basicConfig(**kwargs)
+
+    # Setup a function to handle uncaught exceptions and include them with their stacktrace into logfiles
+    def handle_exception(exc_type, exc_value, exc_traceback):
+        # Log the exception
+        logger = logging.getLogger()
+        logger.error(
+            "Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)
+        )
+
+    sys.excepthook = handle_exception

scripts/compile_cost_assumptions.py

@@ -33,11 +33,10 @@
 
 import numpy as np
 import pandas as pd
-from _helpers import adjust_for_inflation
+from _helpers import adjust_for_inflation, configure_logging, mock_snakemake
 from currency_converter import ECB_URL, CurrencyConverter
 from scipy import interpolate
 
-logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 try:
@@ -3943,11 +3942,10 @@ def prepare_inflation_rate(fn: str) -> pd.DataFrame:
 #  ---------- MAIN ------------------------------------------------------------
 if __name__ == "__main__":
     if "snakemake" not in globals():
-        from _helpers import mock_snakemake
-
-        # os.chdir(os.path.join(os.getcwd(), "scripts"))
         snakemake = mock_snakemake("compile_cost_assumptions")
 
+    configure_logging(snakemake)
+
     years_list = list(snakemake.config["years"])
     inflation_rate = prepare_inflation_rate(snakemake.input.inflation_rate)
 

scripts/compile_cost_assumptions_usa.py

@@ -10,12 +10,14 @@
 """
 
 import pathlib
-
+import logging
 import numpy as np
 import pandas as pd
-from _helpers import adjust_for_inflation, mock_snakemake
+from _helpers import adjust_for_inflation, configure_logging, mock_snakemake
 from compile_cost_assumptions import prepare_inflation_rate
 
+logger = logging.getLogger(__name__)
+
 
 def get_conversion_dictionary(flag: str) -> dict:
     """
@@ -904,6 +906,8 @@ def duplicate_fuel_cost(input_file_path: str, list_of_years: list) -> pd.DataFra
     if "snakemake" not in globals():
         snakemake = mock_snakemake("compile_cost_assumptions_usa")
 
+    configure_logging(snakemake)
+
     year_list = sorted(snakemake.config["years"])
     num_digits = snakemake.config["ndigits"]
     eur_reference_year = snakemake.config["eur_year"]
@@ -922,6 +926,8 @@ def duplicate_fuel_cost(input_file_path: str, list_of_years: list) -> pd.DataFra
         "nrel_atb_technology_to_remove"
     ]
 
+    logger.info("Configuration variables are set")
+
     if len(set(snakemake.config["years"])) < len(snakemake.config["years"]):
         raise Exception(
             "Please verify the list of cost files. It may contain duplicates."
@@ -935,9 +941,13 @@ def duplicate_fuel_cost(input_file_path: str, list_of_years: list) -> pd.DataFra
     # get the discount rate values for the US
     discount_rate_df = pd.read_csv(input_file_discount_rate)
 
+    logger.info("discount_rate file for the US has been read in")
+
     # get the fuel costs values for the US
     fuel_costs_df = duplicate_fuel_cost(input_file_fuel_costs, year_list)
 
+    logger.info("fuel_cost file for the US has been read in")
+
     for year_val in year_list:
         # get the cost file to modify
         input_cost_path = [
@@ -956,6 +966,8 @@ def duplicate_fuel_cost(input_file_path: str, list_of_years: list) -> pd.DataFra
         else:
             raise Exception(f"{year_val} is not a considered year")
 
+        logger.info("The file {} is used for year {}".format(input_atb_path, year_val))
+
         manual_input_usa_df = pre_process_manual_input_usa(
             input_file_manual_input_usa,
             input_file_eur_inflation_rate,
@@ -1060,6 +1072,7 @@ def duplicate_fuel_cost(input_file_path: str, list_of_years: list) -> pd.DataFra
         if len(output_cost_path_list) == 1:
             output_cost_path = output_cost_path_list[0]
             updated_cost_df.to_csv(output_cost_path, index=False)
+            logger.info("The cost assumptions file for the US has been compiled for year {}".format(year_val))
         else:
             raise Exception(
                 "Please verify the list of cost files. It may contain duplicates."