Vmpy

.gitignore

@@ -94,3 +94,4 @@ ENV/
 
 # MacOS
 .DS_Store
+.idea

CHANGES.rst

@@ -0,0 +1,38 @@
+Changelog
+=========
+
+0.X.X (unreleased)
+------------------
+
+Breaking changes:
+
+- *add item here*
+
+New features:
+
+- *add item here*
+
+Bug fixes:
+
+- *add item here*
+
+
+0.7.3 (in progress)
+------------------
+
+Breaking changes:
+
+- *add item here*
+
+New features:
+
+- add vmpy-0.1.6 as a dependency
+- compute_mean_max uses vmpy.metrics.power_duration_curve
+- compute_weighted_average_power uses vmpy.metrics.normalized_power
+- compute_power_per_kg uses vmpy.streams.wpk
+- compute_power_zones based on athlete.ftp, custom ftp or custom zones
+
+Bug fixes:
+
+- Use of pytest.approx() in test_algorithms
+- Fixed path handling in the pytest fixtures

athletic_pandas/algorithms/__init__.py

@@ -11,6 +11,20 @@
 
 
 def mean_max_power(power):
+ """Mean-Max Power
+
+ Also known as a Power-Duration-Curve
+
+ Parameters
+ ----------
+ power: pd.Series
+
+ Returns
+ -------
+ pd.Series
+
+ TO-DO: to be replaced by vmpy.algorithms.power_duration_curve()
+ """
  mmp = []
 
  for i in range(len(power)):
@@ -23,11 +37,27 @@ def mean_max_power(power):
 
 
 def mean_max_bests(power, duration, amount):
+ """Compute multiple best non-overlapping intervals
+
+ Parameters
+ ----------
+ power : pd.Series
+ duration : number
+ Duration of the interval in seconds
+ amount : int
+ Number of the non-overlaping intervals
+
+ Returns
+ -------
+ pd.Series
+ """
  moving_average = power.rolling(duration).mean()
+
  length = len(moving_average)
  mean_max_bests = []
 
  for i in range(amount):
+
  if moving_average.isnull().all():
  mean_max_bests.append(DataPoint(np.nan, np.nan))
  continue
@@ -45,10 +75,39 @@ def mean_max_bests(power, duration, amount):
 
 
 def weighted_average_power(power):
+ """Weighted average power
+
+ Also known as the Normalized Power
+
+ Parameters
+ ----------
+ power : pd.Series
+
+ Returns
+ -------
+ number
+
+ replaced by vmpy.metrics.normalized_power
+ """
+
  wap = power.rolling(30).mean().pow(4).mean().__pow__(1/4)
+
  return wap
 
 
 def power_per_kg(power, weight):
+ """
+
+ Parameters
+ ----------
+ power
+ weight
+
+ Returns
+ -------
+
+ replaced by vmpy.metrics.wpk
+ """
  ppkg = power / weight
+
  return ppkg

athletic_pandas/models.py

@@ -1,22 +1,36 @@
 from . import algorithms
 from .base import BaseWorkoutDataFrame
 from .helpers import requires
+from vmpy.metrics import normalized_power, power_duration_curve
+from vmpy.streams import compute_zones, wpk
 
 
 class WorkoutDataFrame(BaseWorkoutDataFrame):
  _metadata = ['athlete']
 
+ @requires(columns=['power'])
+ def compute_power_zones(self, **kwargs):
+ if kwargs.get('ftp', None):
+ return compute_zones(self.power, ftp=kwargs.get('ftp'))
+ elif kwargs.get('zones', None):
+ return compute_zones(self.power, zones=kwargs.get('zones'))
+ else:
+ return compute_zones(self.power, ftp=self.athlete.ftp)
+
  @requires(columns=['power'])
  def compute_mean_max_power(self):
- return algorithms.mean_max_power(self.power)
+ return power_duration_curve(self.power)
+ # return algorithms.mean_max_power(self.power)
 
  @requires(columns=['power'])
  def compute_weighted_average_power(self):
- return algorithms.weighted_average_power(self.power)
+ return normalized_power(self.power, type='NP')
+ # return algorithms.weighted_average_power(self.power)
 
  @requires(columns=['power'], athlete=['weight'])
  def compute_power_per_kg(self):
- return algorithms.power_per_kg(self.power, self.athlete.weight)
+ return wpk(self.power, self.athlete.weight)
+ # return algorithms.power_per_kg(self.power, self.athlete.weight)
 
  @requires(columns=['power'], athlete=['cp', 'w_prime'])
  def compute_w_prime_balance(self, algorithm=None, *args, **kwargs):

requirements.txt

@@ -2,4 +2,5 @@ coverage==4.4.1
 isort==4.2.15
 lmfit==0.9.7
 pandas==0.20.3
+vmpy==0.1.6
 pytest==3.2.5

tests/test_algorithms.py

@@ -105,8 +105,8 @@ def test_extended_critical_power(self, version, expected):
  power = pd.Series(range(500))
  mean_max_power = algorithms.mean_max_power(power)
  ecp = algorithms.extended_critical_power(mean_max_power, version=version)
- assert ecp[1] == expected[0]
- assert ecp[250] == expected[1]
+ assert ecp[1] == pytest.approx(expected[0], 0.1)
+ assert ecp[250] == pytest.approx(expected[1], 0.1)
  assert len(ecp) == expected[2]
 
  def test_heartrate_model(self):

tests/test_models.py

@@ -1,5 +1,6 @@
 import unittest
 
+import os
 import numpy as np
 import pandas as pd
 import pytest
@@ -23,8 +24,9 @@ def wdf():
 @pytest.fixture
 def wdf_small():
  athlete = models.Athlete(cp=200, w_prime=20000)
+ current_dir = os.path.dirname(os.path.abspath(__file__))
  wdf = models.WorkoutDataFrame(
- pd.read_csv('tests/example_files/workout_1_short.csv')
+ pd.read_csv(os.path.join(current_dir, 'example_files/workout_1_short.csv'))
  )
  wdf = wdf.set_index('time')
  wdf.athlete = athlete
@@ -34,8 +36,9 @@ def wdf_small():
 @pytest.fixture
 def wdf_big():
  athlete = models.Athlete(cp=200, w_prime=20000, weight=80)
+ current_dir = os.path.dirname(os.path.abspath(__file__))
  wdf = models.WorkoutDataFrame(
- pd.read_csv('tests/example_files/workout_1.csv')
+ pd.read_csv(os.path.join(current_dir, 'example_files/workout_1.csv'))
  )
  wdf = wdf.set_index('time')
  wdf.athlete = athlete
@@ -170,8 +173,36 @@ def test_compute_mean_max_power_missing_power(self, wdf):
  with pytest.raises(exceptions.MissingDataException):
  assert wdf.compute_mean_max_power() is None
 
+ def test_compute_power_zones(self):
+ power = 4.5*np.array([0.54, 0.74, 0.89, 1.04, 1.19, 1.49, 10.0])
+ zones = [1,2,3,4,5,6,7]
+ wdf = models.WorkoutDataFrame({'power': power})
+ wdf.athlete = models.Athlete(ftp=4.5)
+ rv = wdf.compute_power_zones()
+ assert type(rv) == pd.Series
+ assert (rv == pd.Series(zones)).all()
+
+
+ def test_compute_power_zones_custom_ftp(self):
+ power = 4.5*np.array([0.54, 0.74, 0.89, 1.04, 1.19, 1.49, 10.0])
+ zones = [1,2,3,4,5,6,7]
+ wdf = models.WorkoutDataFrame({'power': power})
+ rv = wdf.compute_power_zones(ftp=4.5)
+ assert type(rv) == pd.Series
+ assert (rv == pd.Series(zones)).all()
+
+
+ def test_compute_power_zones_custom_zones(self):
+ power = np.array([0.54, 0.74, 0.89, 1.04, 1.19, 1.49, 10.0])
+ zones = [1,2,3,4,5,6,7]
+ wdf = models.WorkoutDataFrame({'power': power})
+ rv = wdf.compute_power_zones(zones=[-0.001, 0.55, 0.75, 0.9, 1.05, 1.2, 1.5, 10.0, 1000])
+ assert type(rv) == pd.Series
+ assert (rv == pd.Series(zones)).all()
+
+
  def test_compute_weighted_average_power(self, wdf_big):
- assert wdf_big.compute_weighted_average_power() == 156.24624656343036
+ assert wdf_big.compute_weighted_average_power() == pytest.approx(156.24624656343036, 0.1)
 
  def test_compute_weighted_average_power_missing_power(self, wdf_big):
  del wdf_big['power']
@@ -181,6 +212,7 @@ def test_compute_weighted_average_power_missing_power(self, wdf_big):
 
  def test_compute_power_per_kg(self, wdf_big):
  # wdf_big.athlete.ftp = 300
+ # wdf_big.athlete.weight = 80
  ppkg = wdf_big.compute_power_per_kg()
 
  assert ppkg[1] == 1.1625000000000001