updates

pyproject.toml

@@ -19,20 +19,21 @@ classifiers = [
   "Programming Language :: Python :: Implementation :: CPython",
 ]
 dependencies = [
-    "numpy>=2.1.1",
-    "pandas>=2.2.2",
+    "numpy>=2.1.2",
+    "pandas>=2.2.3",
     "matplotlib>=3.8.3",
-    "pyarrow>=3.9.2",
+    "pyarrow>=17.0.0",
     "garmin-fit-sdk>=21.141.0",
-    "plotly>=5.24.0",
+    "plotly>=5.24.1",
 ]
 [tool.uv]
 dev-dependencies = [
-    "coverage>=7.6.1",
-    "hatch>=1.12.0",
-    "mypy>=1.11.2",
+    "coverage>=7.6.3",
+    "hatch>=1.13.0",
+    "mypy>=1.12.0",
     "pytest-cov>=5.0.0",
     "pytest>=8.3.3",
+    "notebook>=7.2.2",
 ]
 [project.urls]
 Documentation = "https://github.com/vincentdavis/cycling-dynamics#readme"
@@ -60,7 +61,7 @@ cov = [
 ]
 
 [[tool.hatch.envs.all.matrix]]
-python = ["3.11", "3.12"]
+python = ["3.12", "3.13"]
 
 [tool.hatch.envs.types]
 dependencies = [

ruff.toml

@@ -1,20 +1,27 @@
 # Ruff configuration file
-target-version = "py311"
+target-version = "py312"
 line-length = 120
 
 [lint]
 select = [
     # flake8-bugbear
     "B",
+    #pydoclint
+    "DOC",
     # pydocstyle
     "D",
     # pycodestyle
-    "E", "W",
+    # - ERRORS
+    "E",
+    # - Warning
+    "W",
     # Pyflakes
     "F",
     "FLY",
     # isort
     "I",
+    # Fast API
+    "FAST",
     # pyupgrade
     "UP",
     # flake8-simplify
@@ -23,8 +30,12 @@ select = [
     "DJ",
     # Ruff
     "RUF",
+    # NumPy-specific rules (NPY)
+    "NPY",
+    # Perflint (PERF)
+    "PERF"
 ]
-ignore = []
+ignore = ["G004", "D107"]
 # Allow fix for all enabled rules (when `--fix`) is provided.
 fixable = ["ALL"]
 unfixable = []
@@ -39,4 +50,18 @@ indent-style = "space"
 # Like Black, respect magic trailing commas.
 skip-magic-trailing-comma = false
 # Like Black, automatically detect the appropriate line ending.
-line-ending = "auto"
+line-ending = "auto"
+
+# Enable auto-formatting of code examples in docstrings. Markdown,
+# reStructuredText code/literal blocks and doctests are all supported.
+#
+# This is currently disabled by default, but it is planned for this
+# to be opt-out in the future.
+docstring-code-format = true
+
+# Set the line length limit used when formatting code snippets in
+# docstrings.
+#
+# This only has an effect when the `docstring-code-format` setting is
+# enabled.
+docstring-code-line-length = "dynamic"

src/cycling_dynamics/calc.py

@@ -3,6 +3,7 @@
 import logging
 from math import asin, atan, cos, radians, sin, sqrt, tan
 
+import numpy as np
 import pandas as pd
 
 
@@ -57,8 +58,9 @@ def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> fl
 
 
 def angle_type(a, b, c):
-    """Calculate the cosine of the angle using Law of Cosines and determine the angle type
-    Is the angle between sides a and b acute, obtuse or 90 degrees (right angle)
+    """Calculate the cosine of the angle using Law of Cosines and determine the angle type.
+
+    If the angle between sides a and b acute, obtuse or 90 degrees (right angle)
      a: The original control point to next point on path
      b: The original control point to point on other path
      c: Second point on path the other path point.
@@ -91,18 +93,75 @@ def add_metrics(df: pd.DataFrame, rolling_window: int = 30, ftp: int | None = No
     # power
     df["np"] = (df["power"] ** 4).rolling(window=30).mean() ** 0.25
     if ftp is not None:
-        df["IF"] = df["np"] / ftp
-        df["TSS"] = (df["power"] * df["IF"] * df["seconds"] / ftp / 3600).cumsum()
+        df = intensity_factor(df, ftp=ftp)
+        df = total_training_stress(df, ftp=ftp)
+        logging.info(f"Total Training Stress (TSS): {df['TSS'].iloc[-1]:0.2f} kcal/hr (FTP: {ftp})")
+    return df
+
+
+def total_training_stress(df: pd.DataFrame, ftp: int) -> pd.DataFrame:
+    """Calculat Total Training Stress."""
+    df["TSS"] = (df["power"] * df["IF"] * df["seconds"] / ftp / 3600).cumsum()
     return df
 
 
 def normalized_power(df: pd.DataFrame) -> pd.DataFrame:
     """Calculate the normalized power of a ride."""
     df["np"] = (df["power"] ** 4).rolling(window=30).mean() ** 0.25
-    return df["np"]
+    return df
 
 
 def intensity_factor(df: pd.DataFrame, ftp: int) -> pd.DataFrame:
     """Calculate the intensity factor of a ride."""
     df["IF"] = ((df["power"] ** 4).rolling(window=30).mean() ** 0.25) / ftp
     return df
+
+
+def vam(df: pd.DataFrame) -> pd.DataFrame:
+    """Calculate vertical ascent rate."""
+    # TODO Improve calculation by using 3 points
+    df["vam"] = (df["altitude"].diff() / df.seconds.diff()) * 3600
+    return df
+
+
+def slope(df: pd.DataFrame) -> pd.DataFrame:
+    """Calculate the slope."""
+    # TODO Improve calculation by using 3 points
+    df["slope"] = df["altitude"].diff() / df.distance.diff()
+    df["slope_3sec"] = df["slope"].rolling(window=3, center=True).mean()
+    df["slope_3sec"] = df["slope_3sec"].fillna(df["slope_3sec"])
+    return df
+
+
+def zero_seconds(df: pd.DataFrame) -> pd.DataFrame:
+    """Create or reset seconds callumn starting at zero."""
+    # TODO: make sure it is sorted
+    df["seconds"] = df["timestamp"].sub(df["timestamp"].min()).dt.total_seconds()
+    # if "seconds" not in df.columns:
+    #     df["seconds"] = pd.to_datetime(df.index, unit="s", origin="unix").astype(int) // 10**9
+    # df["seconds"] = df["seconds"] - df.seconds.min()
+    return df
+
+
+def speed(df: pd.DataFrame) -> pd.DataFrame:
+    """Calculate speed if missing."""
+    # TODO Improve calculation by using 3 points
+    df["speed"] = df.distance.diff() / df.time.diff()
+
+def speed_3sec(df: pd.DataFrame) -> pd.DataFrame:
+    df["speed_3sec"] = df["speed"].rolling(window=3, center=True).mean()
+    df['speed_3sec'] = df['speed_moving_avg'].fillna(df['speed'])
+    return df
+
+
+def air_density(df: pd.DataFrame, temperature: float = 30) -> pd.DataFrame:
+    """Calculate air density."""
+    if temperature not in df.columns:
+        logging.info("Using default temperature of 30 degrees C")
+        df["temperature"] = 30
+    df["air_density"] = (
+        (101325 / (287.05 * 273.15))
+        * (273.15 / (df["temperature"] + 273.15))
+        * np.exp((-101325 / (287.05 * 273.15)) * 9.8067 * (df["altitude"] / 101325))
+    )
+    return df

src/cycling_dynamics/dynamics.py

@@ -0,0 +1,63 @@
+"""Dynamic simulations."""
+
+import numpy as np
+import pandas as pd
+
+
+def simulator(
+    df: pd.DataFrame,
+    smoothing: int = 3,
+    rider_weight: float = 65.0,
+    bike_weight: float = 5.0,
+    wind_speed: float = 0,
+    wind_direction: int = 0,
+    temperature: float = 30,
+    drag_coefficient: float = 0.8,
+    frontal_area: float = 0.565,
+    rolling_resistance: float = 0.005,
+    efficiency_loss: float = 0.04,
+) -> pd.DataFrame:
+    """Calculate the components of power loss at each point and estimated total power needed.
+
+    df: Usually from a FIT file or other GPS file. Required columns [distance, speed and/or time, altitude]
+    """
+    try:
+        assert all([c in df.columns for c in ["distance", "altitude"]])
+    except AssertionError:
+        raise AssertionError("Missing columns in dataframe. Must have 'seconds', 'distance', and 'altitude'")
+
+    CdA = drag_coefficient * frontal_area
+
+    df["effective_wind_speed"] = np.cos(np.radians(wind_direction)) * wind_speed
+
+    # Components of power, watts
+    df["air_drag_watts"] = (
+        0.5 * CdA * df["air_density"] * np.square(df["speed"] + df["effective_wind_speed"]) * df["speed"]
+    )
+    df["climbing_watts"] = (bike_weight + rider_weight) * 9.8067 * np.sin(np.arctan(df["slope"])) * df["speed"]
+    df["rolling_watts"] = (
+        np.cos(np.arctan(df["slope"])) * 9.8067 * (bike_weight + rider_weight) * rolling_resistance * df["speed"]
+    )
+    df["acceleration_watts"] = (bike_weight + rider_weight) * (df["speed"].diff() / df["seconds"].diff()) * df["speed"]
+    df["est_power_no_loss"] = df[["air_drag_watts", "climbing_watts", "rolling_watts", "acceleration_watts"]].sum(
+        axis="columns"
+    )
+    df["est_power"] = df["est_power_no_loss"] / (1 - efficiency_loss)
+    df["efficiency_loss_watts"] = df["est_power_no_loss"] - df["est_power"]
+    df["est_power_no_acceleration"] = (df["est_power_no_loss"] - df["acceleration_watts"]) / (1 - efficiency_loss)
+
+    if smoothing > 0:
+        df["speed_smoothed"] = df["speed"].rolling(window=smoothing, center=True).mean()
+        df["slope_smoothed"] = df["slope"].rolling(window=smoothing, center=True).mean()
+        df["power_smoothed"] = df["power"].rolling(window=smoothing, center=True).mean()
+        df["air_drag_watts_smoothed"] = df["air_drag_watts"].rolling(window=smoothing, center=True).mean()
+        df["climbing_watts_smoothed"] = df["climbing_watts"].rolling(window=smoothing, center=True).mean()
+        df["rolling_watts_smoothed"] = df["rolling_watts"].rolling(window=smoothing, center=True).mean()
+        df["est_power_smoothed"] = df["est_power"].rolling(window=smoothing, center=True).mean()
+        df["efficiency_loss_watts_smoothed"] = df["efficiency_loss_watts"].rolling(window=smoothing, center=True).mean()
+        df["acceleration_watts_smoothed"] = df["acceleration_watts"].rolling(window=smoothing, center=True).mean()
+        df["est_power_no_acceleration_smoothed"] = (
+            df["est_power_no_acceleration"].rolling(window=smoothing, center=True).mean()
+        )
+
+    return df

src/cycling_dynamics/load_data.py

@@ -3,23 +3,27 @@
 import pandas as pd
 from garmin_fit_sdk import Decoder, Stream
 
+from cycling_dynamics import calc
+from cycling_dynamics.calc import slope
+
 logging.basicConfig(level=logging.INFO)
 
 
-def load_fit_file(file_path: str, add_metrics: bool = True, rolling_window: int = 30) -> pd.DataFrame:
-    """
-    Load a FIT file and return its data as a pandas DataFrame.
+def load_fit_file(
+    file_path: str, add_fields: tuple[str] = ("seconds", "slope", "vam", "speed", "normalized_power", "air_density")
+) -> pd.DataFrame:
+    """Load a FIT file and return its data as a pandas DataFrame.
 
     Args:
         file_path (str): Path to the FIT file.
-        add_metrics (bool, optional): Flag to add additional metrics. Defaults to True.
-        rolling_window (int, optional): Window size for rolling calculations. Defaults to 30.
+        add_fields (bool, optional): Flag to add additional metrics. Defaults to True.
 
     Returns:
         pd.DataFrame: DataFrame containing the FIT file data.
 
     Raises:
         ValueError: If the file is not a valid FIT file or if there are decoding errors.
+
     """
     logging.info("Loading FIT file")
 
@@ -43,12 +47,31 @@ def load_fit_file(file_path: str, add_metrics: bool = True, rolling_window: int
         if col in df.columns:
             df[col] = df[col] / 1e7
 
-    if "enhanced_speed" in df.columns and "speed" in df.columns:
-        logging.info("Using enhanced speed")
-        df.drop(columns=["speed"], inplace=True)
-    df.rename(columns={"enhanced_speed": "speed"}, inplace=True)
-    if "enhanced_altitude" in df.columns and "altitude" in df.columns:
-        logging.info("Using enhanced altitude")
-        df.drop(columns=["altitude"], inplace=True)
-    df.rename(columns={"enhanced_altitude": "altitude"}, inplace=True)
+    enhanced_cols = ["enhanced_distance", "enhanced_speed", "enhanced_altitude"]
+    for col in enhanced_cols:
+        if col in df.columns:
+            logging.info(f"Using {col}")
+            if col.split("_")[1] in df.columns:
+                df.drop(columns=[col.split("_")[1]], inplace=True)
+            df.rename(columns={col: col.split("_")[1]}, inplace=True)
+
+    if "seconds" in add_fields and "seconds" not in df.columns:
+        logging.info("Using calculated seconds")
+        df = calc.zero_seconds(df)
+    if "speed" in add_fields and "speed" not in df.columns:
+        logging.info("Using calculated speed")
+        df = calc.speed(df)
+    if "slope" in add_fields and "slope" not in df.columns:
+        logging.info("Using calculated slope")
+        df = slope(df)
+    if "vam" in add_fields and "vam" not in df.columns:
+        logging.info("Using calculated vertical acceleration magnitude")
+        df = calc.vam(df)
+    if "normalized_power" in add_fields and "normalized_power" not in df.columns:
+        logging.info("Using calculated normalized power")
+        df = calc.normalized_power(df)
+    if "air_density" in add_fields and "air_density" not in df.columns:
+        logging.info("Using calculated air density")
+        df = calc.air_density(df)
+
     return df