Merge pull request #57 from Deltares/partial-imports-to-make-bundling…

…-easier

Remove fiat core from dependencies, cleanup imports

fiat_toolbox/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.9"
+__version__ = "0.1.10"

fiat_toolbox/equity/equity.py

@@ -5,7 +5,8 @@
 import numpy as np
 import pandas as pd
 import parse
-from fiat.models.calc import calc_rp_coef
+
+from fiat_toolbox.equity.fiat_functions import calc_rp_coef
 
 
 class Equity:

fiat_toolbox/equity/fiat_functions.py

@@ -0,0 +1,76 @@
+import math
+
+# Instead of having fiat core as a dependency for this one function that doesnt use any other functions from fiat core,
+# the function was copied here (26/9/2024, 661e0f2b2d6396346140316412c5957bc10eb03b) from https://github.com/Deltares/Delft-FIAT/blob/master/src/fiat/models/calc.py 
+
+def calc_rp_coef(
+    rp: list | tuple,
+):
+    """
+    Calculates coefficients used to compute the EAD as a linear function of
+    the known damages.
+
+    Parameters
+    ----------
+    rp : list or tuple of int
+        Return periods T1 … Tn for which damages are known.
+
+    Returns
+    -------
+    list of float
+        Coefficients a1, …, an used to compute the EAD as a linear function of the known damages.
+
+    Notes
+    -----
+    In which D(f) is the damage, D, as a function of the frequency of exceedance, f.
+    In order to compute this EAD, function D(f) needs to be known for the entire range of frequencies.
+    Instead, D(f) is only given for the n frequencies as mentioned in the table above. 
+    So, in order to compute the integral above, some assumptions need to be made for function D(f):
+    (i)   For f > f1 the damage is assumed to be equal to 0.
+    (ii)  For f < fn, the damage is assumed to be equal to Dn.
+    (iii) For all other frequencies, the damage is estimated from log-linear interpolation 
+          between the known damages and frequencies.
+    """
+    # Step 1: Compute frequencies associated with T-values.
+    _rp = sorted(rp)
+    idxs = [_rp.index(n) for n in rp]
+    rp_u = sorted(rp)
+    rp_l = len(rp_u)
+
+    f = [1 / n for n in rp_u]
+    lf = [math.log(1 / n) for n in rp_u]
+
+    if rp_l == 1:
+        return f
+
+    # Step 2:
+    c = [(1 / (lf[idx] - lf[idx + 1])) for idx in range(rp_l - 1)]
+
+    # Step 3:
+    G = [(f[idx] * lf[idx] - f[idx]) for idx in range(rp_l)]
+
+    # Step 4:
+    a = [
+        (
+            (1 + c[idx] * lf[idx + 1]) * (f[idx] - f[idx + 1])
+            + c[idx] * (G[idx + 1] - G[idx])
+        )
+        for idx in range(rp_l - 1)
+    ]
+    b = [
+        (c[idx] * (G[idx] - G[idx + 1] + lf[idx + 1] * (f[idx + 1] - f[idx])))
+        for idx in range(rp_l - 1)
+    ]
+
+    # Step 5:
+    alpha = [
+        b[0]
+        if idx == 0
+        else f[idx] + a[idx - 1]
+        if idx == rp_l - 1
+        else a[idx - 1] + b[idx]
+        for idx in range(rp_l)
+    ]
+
+    return [alpha[idx] for idx in idxs]
+

fiat_toolbox/infographics/infographics.py

@@ -2,18 +2,15 @@
 from pathlib import Path
 from typing import Dict, List, Union
 
-import plotly.graph_objects as go
 import tomli
 import validators
 from PIL import Image
-from plotly.graph_objects import Figure
+from plotly.graph_objects import Bar, Figure, Pie
 from plotly.subplots import make_subplots
 
 from fiat_toolbox.infographics.infographics_interface import IInfographicsParser
 from fiat_toolbox.metrics_writer.fiat_read_metrics_file import MetricsFileReader
 
-logging.basicConfig(level=logging.INFO)
-
 
 class InfographicsParser(IInfographicsParser):
     """Class for creating the infographic"""
@@ -24,6 +21,7 @@ def __init__(
         metrics_full_path: Union[Path, str],
         config_base_path: Union[Path, str],
         output_base_path: Union[Path, str],
+        logger: logging.Logger = logging.getLogger(__name__),
     ) -> None:
         """Initialize the InfographicsParser
 
@@ -57,6 +55,8 @@ def __init__(
             output_base_path = Path(output_base_path)
         self.output_base_path = output_base_path
 
+        self.logger = logger
+
     def _get_impact_metrics(self) -> Dict:
         """Get the impact metrics for a scenario
 
@@ -311,7 +311,6 @@ def _check_image_source(
             Union[str, Image, None]
                 The image source or None if the image source is not a url or a local path
         """
-
         # Check if the image is a url. If so, add the image to the pie chart
         if validators.url(img):
             # Add the pie chart image
@@ -440,7 +439,7 @@ def _get_pie_chart_figure(data: Dict, **kwargs):
 
         Returns
         -------
-            go.Figure
+            Figure
                 The pie chart figure
         """
 
@@ -479,7 +478,7 @@ def _get_pie_chart_figure(data: Dict, **kwargs):
         # Add the pie chart to the figure
         for idx, (key, value) in enumerate(data.items()):
             # Create single pie chart
-            trace = go.Pie(
+            trace = Pie(
                 values=value["Values"],
                 labels=value["Labels"],
                 hole=0.6,
@@ -614,7 +613,7 @@ def _get_bar_chart_figure(data: Dict, **kwargs):
 
         Returns
         -------
-        go.Figure
+        Figure
             The bar chart figure
         """
 
@@ -653,7 +652,7 @@ def _get_bar_chart_figure(data: Dict, **kwargs):
             ):
                 # Add bar to the figure
                 fig.add_trace(
-                    go.Bar(
+                    Bar(
                         x=[label],
                         y=[int(value)],
                         marker={
@@ -760,12 +759,12 @@ def _get_bar_chart_figure(data: Dict, **kwargs):
 
     def _get_infographics(
         self,
-    ) -> go.Figure:
+    ) -> Figure:
         """Get the infographic for a scenario
 
         Returns
         -------
-        go.Figure
+        Figure
             The infographic for the scenario
 
         """
@@ -804,7 +803,7 @@ def _get_infographics(
             )
             return_fig.append(charts_fig)
         except FileNotFoundError:
-            logging.warning("No charts configuration file found")
+            self.logger.warning("No charts configuration file found")
 
         # Get the pie chart dictionaries from the configuration for people
         try:
@@ -831,7 +830,7 @@ def _get_infographics(
             )
             return_fig.append(people_fig)
         except FileNotFoundError:
-            logging.warning("No people configuration file found")
+            self.logger.warning("No people configuration file found")
 
         # Get the bar chart dictionaries from the configuration for roads
         try:
@@ -853,17 +852,17 @@ def _get_infographics(
             )
             return_fig.append(roads_fig)
         except FileNotFoundError:
-            logging.warning("No roads configuration file found")
+            self.logger.warning("No roads configuration file found")
 
         # Return the figure
         return return_fig
 
-    def get_infographics(self) -> Union[List[go.Figure], go.Figure]:
+    def get_infographics(self) -> Union[List[Figure], Figure]:
         """Get the infographic for a scenario
 
         Returns
         -------
-        Union[List[go.Figure], go.Figure]
+        Union[List[Figure], Figure]
             The infographic for the scenario as a list of figures or a single figure
         """
 

fiat_toolbox/infographics/infographics_interface.py

@@ -1,8 +1,22 @@
+import logging
 from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Union
 
 
 class IInfographicsParser(ABC):
     """Interface for creating the infographic"""
+    logger: logging.Logger
+
+    @abstractmethod
+    def __init__(
+        self,
+        scenario_name: str,
+        metrics_full_path: Union[Path, str],
+        config_base_path: Union[Path, str],
+        output_base_path: Union[Path, str],
+        logger: logging.Logger = logging.getLogger(__name__),
+    ) -> None: ...
 
     @abstractmethod
     def get_infographics(self) -> str:

fiat_toolbox/infographics/risk_infographics.py

@@ -1,9 +1,9 @@
 import base64
-import os
+import logging
 from pathlib import Path
 from typing import Dict, List, Union
 
-import plotly.graph_objects as go
+from plotly.graph_objects import Figure
 
 from fiat_toolbox.infographics.infographics import InfographicsParser
 from fiat_toolbox.infographics.infographics_interface import IInfographicsParser
@@ -14,13 +14,15 @@
 
 class RiskInfographicsParser(IInfographicsParser):
     """Class for creating the infographic"""
+    logger: logging.Logger = logging.getLogger(__name__),
 
     def __init__(
         self,
         scenario_name: str,
         metrics_full_path: Union[Path, str],
         config_base_path: Union[Path, str],
         output_base_path: Union[Path, str],
+        logger: logging.Logger = logging.getLogger(__name__),
     ) -> None:
         """Initialize the InfographicsParser
 
@@ -53,6 +55,8 @@ def __init__(
         if isinstance(output_base_path, str):
             output_base_path = Path(output_base_path)
         self.output_base_path = output_base_path
+        self.logger = logger
+
 
     def _get_impact_metrics(self) -> Dict:
         """Get the impact metrics for a scenario
@@ -93,26 +97,27 @@ def _encode_image_from_path(image_path: str) -> str:
         str
             The base64 encoded image string
         """
-        if image_path is None:
+        if not Path.exists(image_path):
+            RiskInfographicsParser.logger.error(f"Image not found at {image_path}")
             return
-        if os.path.isfile(image_path):
-            with open(image_path, "rb") as image_file:
-                encoded_string = base64.b64encode(image_file.read()).decode()
-            return f'data:image/png;base64,{encoded_string}'
+        with open(image_path, "rb") as image_file:
+            encoded_string = base64.b64encode(image_file.read()).decode()
+
+        return f'data:image/png;base64,{encoded_string}'
 
     @staticmethod
     def _figures_list_to_html(
-        rp_fig: go.Figure,
+        rp_fig: Figure,
         metrics: Dict,
         charts: Dict,
         file_path: Union[str, Path] = "infographics.html",
-        image_path: Union[str, Path] = None,
+        image_folder_path: Union[str, Path] = None,
     ):
         """Save a list of plotly figures in an HTML file
 
         Parameters
         ----------
-            rp_fig : go.Figure
+            rp_fig : Figure
                 The plotly figure consisting of the pie charts for multiple return periods
             metrics : Dict
                 The impact metrics for the scenario
@@ -121,7 +126,6 @@ def _figures_list_to_html(
             image_path : Union[str, Path], optional
                 Path to the image folder, by default None
         """
-
         # Convert the file_path to a Path object
         if isinstance(file_path, str):
             file_path = Path(file_path)
@@ -139,14 +143,14 @@ def _figures_list_to_html(
             file_path.parent.mkdir(parents=True)
 
         # Check if the image_path exists
-        expected_damage_path = InfographicsParser._check_image_source(charts['Other']['Expected_Damages']['image'], image_path, return_image=False)
-        flooded_path = InfographicsParser._check_image_source(charts['Other']['Flooded']['image'], image_path, return_image=False)
+        expected_damage_path = InfographicsParser._check_image_source(charts['Other']['Expected_Damages']['image'], image_folder_path, return_image=False)
+        flooded_path = InfographicsParser._check_image_source(charts['Other']['Flooded']['image'], image_folder_path, return_image=False)
 
         # Div height is the max of the chart heights
         div_height = max(charts['Other']['Expected_Damages']['height'], charts['Other']['Flooded']['height'], charts['Other']['Return_Periods']['plot_height'])
 
         # Write the html to the file
-        with open(file_path, "w", encoding="utf-8") as infographics:
+        with open(file_path, mode="w", encoding="utf-8") as infographics:
             rp_charts = rp_fig.to_html(config={'displayModeBar': False}).split("<body>")[1].split("</body>")[0]
 
             infographics.write(
@@ -229,12 +233,12 @@ def _figures_list_to_html(
 
     def _get_infographics(
         self,
-    ) -> Union[Dict, Dict, go.Figure]:
+    ) -> Union[Dict, Dict, Figure]:
         """Get the infographic for a scenario
 
         Returns
         -------
-        go.Figure
+        Figure
             The infographic for the scenario
 
         """
@@ -278,12 +282,12 @@ def _get_infographics(
         # Return the figure
         return metrics, charts, charts_fig
 
-    def get_infographics(self) -> Union[List[go.Figure], go.Figure]:
+    def get_infographics(self) -> Union[List[Figure], Figure]:
         """Get the infographic for a scenario
 
         Returns
         -------
-        Union[List[go.Figure], go.Figure]
+        Union[List[Figure], Figure]
             The infographic for the scenario as a list of figures or a single figure
         """
 
@@ -309,7 +313,7 @@ def write_infographics_to_file(self) -> str:
 
         # Check if the infographic already exists. If so, return the path
         if Path.exists(infographic_html):
-            # TODO: Print logging message
+            RiskInfographicsParser.logger.info(f"Infographic already exists, skipping creation. Path: {infographic_html}")
             return str(infographic_html)
 
         # Get the infographic