add interpolation padding method (#2)

* fix warning of incorrect escape

* add padding interpolation

* add interpolation mode to the calculation options

* add changelog and bump package version

---------

Co-authored-by: Julien Hericher <Julien.Hericher@ngi.no>

Author: Julien76

CHANGES.md

@@ -0,0 +1,10 @@
+# Changes
+
+## Version 0.1.5:
+
+**date: 2024-09-24**
+
+- Add a new interpolation mode for missing values:
+  - you can now choose between:
+    - "padding" -> propagate last known value in a DataFrame column downward for NaN, only work downward (NaN values at start won't be filled);
+    - "linear" -> linearly interpolate missing values based on the last two known, it works both downward and updward.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ngi-calculations"
-version = "0.1.4"
+version = "0.1.5"
 description = "CPT correlations including commonly used empirical correlations"
 authors = [
     "Julien Hericher <Julien.Hericher@ngi.no>", 

src/ngi_calculations/cpt_correlations/definitions/geo.py

@@ -230,7 +230,7 @@ class GeoParameters:
         label="normalized cone resistance",
         unit="MPa",
         symbol="qn",
-        equation="$q_{net} = 1000 * q_{t} - \sigma_v^{total}$",
+        equation=r"$q_{net} = 1000 * q_{t} - \sigma_v^{total}$",
         value_range=(0, None),
     )
 
@@ -320,7 +320,7 @@ class GeoParameters:
         label="Effective vertical stress",
         unit="kPa",
         symbol="sig_v eff",
-        equation="$\sigma_v^{eff} = \max( \sigma_v^{total} - u_0 ; 0 )$",
+        equation=r"$\sigma_v^{eff} = \max( \sigma_v^{total} - u_0 ; 0 )$",
         value_range=(0, None),
         legend_="σ v,eff",
     )
@@ -354,7 +354,7 @@ class GeoParameters:
         label="normalized cone resistance",
         unit="-",
         symbol="Qt",
-        equation="$Qt = ( 1000 * qt - \sigma_v^{total} ) / \sigma_v^{eff}$",
+        equation=r"$Qt = ( 1000 * qt - \sigma_v^{total} ) / \sigma_v^{eff}$",
         value_range=(0, None),
     )
 
@@ -364,7 +364,7 @@ class GeoParameters:
         # label="normalized pressure",
         unit="-",
         symbol="Bq",
-        equation="$Bq = ( u_2 - u_0 ) / ( 1000 * qt - \sigma_v^{total} )$",
+        equation=r"$Bq = ( u_2 - u_0 ) / ( 1000 * qt - \sigma_v^{total} )$",
         value_range=(0, None),
     )
 
@@ -382,7 +382,7 @@ class GeoParameters:
         label="normalized friction ratio",
         unit="%",
         symbol="Fr",
-        equation="$Fr = fs / ( 1000 * qt - \sigma_v^{total} ) * 100\%$",
+        equation=r"$Fr = fs / ( 1000 * qt - \sigma_v^{total} ) * 100\%$",
         value_range=(0, None),
     )
 
@@ -391,23 +391,23 @@ class GeoParameters:
         label="soil behavior index",
         unit="-",
         symbol="Ic",
-        equation="$Ic = \sqrt{( 3.47 - \log Qt)^2 + (\log Fr + 1.22)^2 }$",
+        equation=r"$Ic = \sqrt{( 3.47 - \log Qt)^2 + (\log Fr + 1.22)^2 }$",
     )
 
     n = GeoParameter(
         key="n",
         label="exponent for normalized soil behavior index",
         unit="-",
         symbol="n",
-        equation="$n = \min \left( 0.381 * Ic + 0.05 * (\sigma_v^{eff} / p_{Atm}) - 0.15 ), 1.0\right)$",
+        equation=r"$n = \min \left( 0.381 * Ic + 0.05 * (\sigma_v^{eff} / p_{Atm}) - 0.15 ), 1.0\right)$",
     )
 
     Qtn = GeoParameter(
         key="Qtn",
         label="normalized cone resistance by n exponent",
         unit="-",
         symbol="Qtn",
-        equation="$Qtn = Qt * \left( \frac{p_{Atm}}{\sigma_v^{eff}}\right)^{n - 1}$",
+        equation=r"$Qtn = Qt * \left( \frac{p_{Atm}}{\sigma_v^{eff}}\right)^{n - 1}$",
     )
 
     Icn = GeoParameter(
@@ -416,7 +416,7 @@ class GeoParameters:
         axis_title="SBT",
         unit="-",
         symbol="Icn",
-        equation="$Icn =  \sqrt{\left( 3.47 - \\frac{\log Qtn}{\log 10} \\right)^2 + \left(\\frac{\log Fr}{\log 10} + 1.22 \\right)^2 }$",
+        equation=r"$Icn =  \sqrt{\left( 3.47 - \\frac{\log Qtn}{\log 10} \\right)^2 + \left(\\frac{\log Fr}{\log 10} + 1.22 \\right)^2 }$",
         legend_="Icn (Robertson, 2009)",
         value_range=(0, 4.5),
     )

src/ngi_calculations/cpt_correlations/methods/cpt_process/calculations.py

@@ -170,7 +170,7 @@ def _integrate_lab_profile(self):
         _df.set_index(depth, drop=False, inplace=True)
 
         # Interpolate the missing values.
-        _df = interpolate_missing_values(_df, key_col=depth, col_list=lab_cols)
+        _df = interpolate_missing_values(_df, key_col=depth, col_list=lab_cols, mode=self.options.interpolation_mode)
 
         # # Prevent values for certain lab profiles to go below zero
         # _df[GEO.u0.key] = _df[GEO.u0.key].clip(lower=0.0)

src/ngi_calculations/cpt_correlations/methods/cpt_process/options.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from pydantic.main import BaseModel
 
 
@@ -6,3 +8,4 @@ class CptProcessOptions(BaseModel):
     shift_depth: float = 0.0  # Shift CPT depth (in m, negative value shifts upwards)
     adjust_depth_tilt: bool = False  # Adjust depth due to tilt
     compensate_atm_pressure: bool = False
+    interpolation_mode: Literal["linear", "padding"] = "linear"

src/ngi_calculations/cpt_correlations/utils/interpolation.py

@@ -1,16 +1,36 @@
-from typing import List
+from typing import Literal
 
 import numpy as np
 import pandas as pd
 from scipy.interpolate import interp1d
 
 
-def interpolate_missing_values(df: pd.DataFrame, key_col: str, inplace: bool = False, col_list: List[str] = False):
+def set_replace_columns(df: pd.DataFrame, col_list: list[str]) -> list[str]:
+    # Get the columns where interpolation should be performed (any column with NaN)
+    return col_list if col_list else df.columns[df.isna().any()].tolist()
+
+
+def interpolate_missing_values__padding_method(df: pd.DataFrame, inplace: bool = False, col_list: list[str] = []):
     # Whether to mutate the provided DataFrame or not
     _df = df if inplace else df.copy()
 
-    # Get the columns where interpolation should be performed (any column with NaN)
-    _cols: List[str] = col_list if col_list else _df.columns[df.isna().any()].tolist()
+    # Get the columns where interpolation should be performed
+    _cols = set_replace_columns(_df, col_list)
+
+    for col in _cols:
+        _df[col] = _df[col].interpolate(method="pad")
+
+    return _df
+
+
+def interpolate_missing_values__linear_method(
+    df: pd.DataFrame, key_col: str, inplace: bool = False, col_list: list[str] = []
+):
+    # Whether to mutate the provided DataFrame or not
+    _df = df if inplace else df.copy()
+
+    # Get the columns where interpolation should be performed
+    _cols = set_replace_columns(_df, col_list)
 
     # Store the key col values as a numpy array
     key_values = np.array(_df[key_col].values.tolist())
@@ -30,3 +50,17 @@ def interpolate_missing_values(df: pd.DataFrame, key_col: str, inplace: bool = F
             _df[col] = f(key_values)
 
     return _df
+
+
+def interpolate_missing_values(
+    df: pd.DataFrame,
+    key_col: str = "depth",
+    inplace: bool = False,
+    col_list: list[str] = [],
+    mode: Literal["linear", "padding"] = "linear",
+):
+    return (
+        interpolate_missing_values__linear_method(df, key_col, inplace, col_list)
+        if mode == "linear"
+        else interpolate_missing_values__padding_method(df, inplace, col_list)
+    )

tests/utils/test_interpolation.py

@@ -14,15 +14,29 @@
 )
 
 
-def test_interpolate_missing_values() -> None:
-    expected = pd.DataFrame(
-        {
-            "depth": [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0],
-            "A": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
-            "B": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
-            "C": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
-            "D": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
-        }
-    )
-    result = interpolate_missing_values(df, key_col="depth")
-    pd.testing.assert_frame_equal(result, expected)
+class Test_Interpolation:
+    def test_interpolate_missing_values__linear_method(self) -> None:
+        expected = pd.DataFrame(
+            {
+                "depth": [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0],
+                "A": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
+                "B": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
+                "C": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
+                "D": [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0],
+            }
+        )
+        result = interpolate_missing_values(df, key_col="depth", mode="linear")
+        pd.testing.assert_frame_equal(result, expected)
+
+    def test_interpolate_missing_values__padding_method(self) -> None:
+        expected = pd.DataFrame(
+            {
+                "depth": [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0],
+                "A": [np.nan, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 90.0, 90.0],
+                "B": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 90.0, 90.0],
+                "C": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 100.0],
+                "D": [np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, 80.0, 80.0, 100.0],
+            }
+        )
+        result = interpolate_missing_values(df, mode="padding")
+        pd.testing.assert_frame_equal(result, expected)