Testing coverage improvement

study_lyte/adjustments.py

@@ -67,7 +67,7 @@ def get_normalized_at_border(series: pd.Series, fractional_basis: float = 0.01,
 
 def merge_on_to_time(df_list, final_time):
     """
-    Reindex the df fram the list onto a final time stamp
+    Reindex the df from the list onto a final time stamp
     """
     # Build dummy result in case no data is passed
     result = pd.DataFrame()
@@ -213,7 +213,6 @@ def aggregate_by_depth(df, new_depth, df_depth_col='depth', agg_method='mean'):
 
 
 def assume_no_upward_motion(series, method='nanmean', max_wind_frac=0.15):
-    from .plotting import plot_ts
 
     i = 1
     result = series.copy()
@@ -234,26 +233,12 @@ def assume_no_upward_motion(series, method='nanmean', max_wind_frac=0.15):
             # grab last index, assign values
             result.iloc[i-1:new_i] = new
             ind = result.iloc[:new_i] <= new
-            # Find only continuous areas where condition is true
-            #continuous = (ind).astype(int).diff().abs().cumsum() == 0
 
             result.iloc[:new_i][ind] = new
-            # ax = plot_ts(series, alpha=0.5, show=False, features=[i, new_i])
-            # ax = plot_ts(result, ax=ax)
-            # Watch out for mid values less than the new value
-            #new_val_idx = np.where(ind)[0][0] + (i-1)
-            #ind = result.iloc[:new_val_idx] < new
-            #result.iloc[:new_val_idx][ind] = new
-            #from .plotting import plot_ts
-
-            #plot_ts(result, features=[i, new_i])
-
             i = new_i
 
         else:
             i += 1
-    #from .plotting import plot_ts
-    #result = result.rolling(window=max_n, center=True, closed='both', min_periods=1).mean()
     return result
 
 def convert_force_to_pressure(force, tip_diameter_m, geom_adj=1):
@@ -282,9 +267,7 @@ def zfilter(series, fraction):
     filter_coefficients = np.ones(window) / window
 
     # Apply the filter forward
-    filtered_signal = lfilter(filter_coefficients, 1, series)
-
-    # Apply the filter backward
-    filtered_signal = lfilter(filter_coefficients, 1, filtered_signal[::-1])[::-1]
-
-    return filtered_signal
+    zi = np.zeros(filter_coefficients.shape[0]-1) #lfilter_zi(filter_coefficients, 1)
+    filtered, zf = lfilter(filter_coefficients, 1, series, zi=zi)
+    filtered = lfilter(filter_coefficients, 1, filtered[::-1], zi=zf)[0][::-1]
+    return filtered

study_lyte/plotting.py

@@ -1,78 +1,4 @@
 import matplotlib.pyplot as plt
-from enum import Enum
-
-
-class EventStyle(Enum):
-    START = 'g', '--', 1
-    STOP = 'r', '--',  1
-    SURFACE = 'lightsteelblue', '--', 1
-    ERROR = 'orangered', 'dotted',   1
-    UNKNOWN = 'k', '--', 1
-
-    @classmethod
-    def from_name(cls, name):
-        result = cls.UNKNOWN
-        for e in cls:
-            if e.name == name.upper():
-                result = e
-                break
-        return result
-
-    @property
-    def color(self):
-        return self.value[0]
-
-    @property
-    def linestyle(self):
-        return self.value[1]
-
-    @property
-    def linewidth(self):
-        return self.value[2]
-
-    @property
-    def label(self):
-        return self.name.title()
-
-class SensorStyle(Enum):
-    """
-    Enum to handle plotting titles and preferred colors
-    """
-    # Df column name, plot title, color
-    RAW_FORCE = 'Sensor1', 'Raw Force', 'black'
-    RAW_AMBIENT_NIR = 'Sensor2', 'Ambient', 'darkorange'
-    RAW_ACTIVE_NIR = 'Sensor3', 'Raw Active', 'crimson'
-    ACTIVE_NIR = 'nir', 'NIR', 'crimson'
-    ACC_X_AXIS = 'X-Axis', 'X-Axis', 'darkslategrey'
-    ACC_Y_AXIS = 'Y-Axis', 'Y-Axis', 'darkgreen'
-    ACC_Z_AXIS = 'Z-Axis', 'Z-Axis', 'darkorange'
-    ACCELERATION = 'acceleration', 'Acc. Magn.', 'darkgreen'
-    FUSED = 'fused', 'Fused', 'magenta'
-    CONSTRAINED_BAROMETER = 'barometer', 'Constr. Baro.', 'navy'
-    RAW_BARO = 'filtereddepth', 'Raw Baro.', 'Brown'
-    UNKNOWN = 'UNKNOWN', 'UNKNOWN', None
-
-    @property
-    def column(self):
-        return self.value[0]
-
-    @property
-    def label(self):
-        return self.value[1].title()
-
-    @property
-    def color(self):
-        return self.value[2]
-
-    @classmethod
-    def from_column(cls, column):
-        result = cls.UNKNOWN
-        for e in cls:
-            if e.column.upper() == column.upper():
-                result = e
-                break
-        return result
-
 
 def plot_events(ax, profile_events, plot_type='normal', event_alpha=0.6):
     """

study_lyte/profile.py

@@ -223,7 +223,7 @@ def barometer(self):
                     if self.metadata['ZPFO'] < 50:
                         LOG.info('Filtering barometer data...')
                         # TODO: make this more intelligent
-                        baro = zfilter(self.raw['filtereddepth'], 0.1)
+                        baro = zfilter(self.raw['filtereddepth'].values, 0.4)
                         baro = pd.DataFrame.from_dict({'baro':baro, 'time': self.raw['time']})
                         baro = baro.set_index('time')['baro']
 

study_lyte/styles.py

@@ -0,0 +1,77 @@
+from enum import Enum
+
+
+class EventStyle(Enum):
+    """
+    Styles for plotting events in a timeseries, enums defined by
+    color, line style and line width
+    """
+    START = 'g', '--', 1
+    STOP = 'r', '--',  1
+    SURFACE = 'lightsteelblue', '--', 1
+    ERROR = 'orangered', 'dotted',   1
+    UNKNOWN = 'k', '--', 1
+
+    @classmethod
+    def from_name(cls, name):
+        result = cls.UNKNOWN
+        for e in cls:
+            if e.name == name.upper():
+                result = e
+                break
+        return result
+
+    @property
+    def color(self):
+        return self.value[0]
+
+    @property
+    def linestyle(self):
+        return self.value[1]
+
+    @property
+    def linewidth(self):
+        return self.value[2]
+
+    @property
+    def label(self):
+        return self.name.title()
+
+class SensorStyle(Enum):
+    """
+    Enum to handle plotting titles and preferred colors
+    """
+    # Df column name, plot title, color
+    RAW_FORCE = 'Sensor1', 'Raw Force', 'black'
+    RAW_AMBIENT_NIR = 'Sensor2', 'Ambient', 'darkorange'
+    RAW_ACTIVE_NIR = 'Sensor3', 'Raw Active', 'crimson'
+    ACTIVE_NIR = 'nir', 'NIR', 'crimson'
+    ACC_X_AXIS = 'X-Axis', 'X-Axis', 'darkslategrey'
+    ACC_Y_AXIS = 'Y-Axis', 'Y-Axis', 'darkgreen'
+    ACC_Z_AXIS = 'Z-Axis', 'Z-Axis', 'darkorange'
+    ACCELERATION = 'acceleration', 'Acc. Magn.', 'darkgreen'
+    FUSED = 'fused', 'Fused', 'magenta'
+    CONSTRAINED_BAROMETER = 'barometer', 'Constr. Baro.', 'navy'
+    RAW_BARO = 'filtereddepth', 'Raw Baro.', 'Brown'
+    UNKNOWN = 'UNKNOWN', 'UNKNOWN', None
+
+    @property
+    def column(self):
+        return self.value[0]
+
+    @property
+    def label(self):
+        return self.value[1].title()
+
+    @property
+    def color(self):
+        return self.value[2]
+
+    @classmethod
+    def from_column(cls, column):
+        result = cls.UNKNOWN
+        for e in cls:
+            if e.column.upper() == column.upper():
+                result = e
+                break
+        return result

tests/test_adjustments.py

@@ -1,7 +1,7 @@
 from study_lyte.adjustments import (get_directional_mean, get_neutral_bias_at_border, get_normalized_at_border, \
                                     merge_time_series, remove_ambient, apply_calibration,
                                     aggregate_by_depth, get_points_from_fraction, assume_no_upward_motion,
-                                    convert_force_to_pressure)
+                                    convert_force_to_pressure, merge_on_to_time, zfilter)
 import pytest
 import pandas as pd
 import numpy as np
@@ -65,6 +65,20 @@ def test_get_normalized_at_border(data, fractional_basis, direction, ideal_norm_
     result = get_normalized_at_border(df['data'], fractional_basis=fractional_basis, direction=direction)
     assert result.iloc[ideal_norm_index] == 1
 
+@pytest.mark.parametrize('data1_hz, data2_hz, desired_hz', [
+    (10, 5, 20)
+])
+def test_merge_on_to_time(data1_hz, data2_hz, desired_hz):
+    df1 = pd.DataFrame({'data1':np.arange(1,stop=data1_hz+1), 'time': np.arange(0, 1, 1 / data1_hz)})
+    df2 = pd.DataFrame({'data2':np.arange(100,stop=data2_hz+100), 'time': np.arange(0, 1, 1 / data2_hz)})
+    desired =  np.arange(0, 1, 1 / desired_hz)
+
+    final = merge_on_to_time([df1, df2], desired)
+    # Ensure we have essentially the same timestep
+    tsteps = np.unique(np.round(final['time'].diff(), 6))
+    tsteps = tsteps[~np.isnan(tsteps)]
+    # Assert only a nan and a real number exist for timesteps
+    assert tsteps == np.round(1/desired_hz, 6)
 
 @pytest.mark.parametrize('data_list, expected', [
     # Typical use, low sample to high res
@@ -184,9 +198,6 @@ def test_assume_no_upward_motion(data, method, expected):
 def test_assume_no_upward_motion_real(raw_df, fname, column, method, expected_depth):
     result = assume_no_upward_motion(raw_df[column], method=method)
     delta_d = abs(result.max() - result.min())
-    from study_lyte.plotting import plot_ts
-    ax = plot_ts(raw_df[column] - raw_df[column].max(), alpha=0.5, show=False)
-    ax = plot_ts(result - result.max(), ax=ax, show=True)
     assert pytest.approx(delta_d, abs=3) == expected_depth
 
 
@@ -198,3 +209,11 @@ def test_convert_force_to_pressure(force, tip_diameter, adj, expected):
     expected = pd.Series(np.array(expected).astype(float), index=range(0, len(expected)))
     result = convert_force_to_pressure(force_series, tip_diameter, adj)
     pd.testing.assert_series_equal(result, expected)
+
+@pytest.mark.parametrize('data, fraction, expected', [
+    # Test a simple noise data situation
+    ([0, 10, 0, 20, 0, 30], 0.4, [2.5, 5., 7.5, 10., 12.5, 22.5]),
+])
+def test_zfilter(data, fraction, expected):
+    result = zfilter(pd.Series(data), fraction)
+    np.testing.assert_equal(result, expected)

tests/test_profile.py

@@ -155,6 +155,14 @@ def test_recursion_exceedance_on_depth(self, profile, filename, depth_method):
         except Exception:
             pytest.fail("Unable to invoke profile.force, likely an recursion issue...")
 
+    @pytest.mark.parametrize('filename, depth_method, total_depth', [
+        # Is filtered
+        ('egrip.csv', 'fused', 199),
+        # Not filtered
+        ('kaslo.csv','fused', 116),
+    ])
+    def test_barometer_is_filtered(self, profile, filename, depth_method, total_depth):
+        assert pytest.approx(profile.barometer.distance_traveled, abs=1) == total_depth
 
 class TestLegacyProfile():
     @pytest.fixture()
@@ -163,6 +171,7 @@ def profile(self, data_dir):
         p = join(data_dir, f)
         profile = LyteProfileV6(p)
         return profile
+
     def test_stop_wo_accel(self, profile):
         """
         Test profile is able to compute surface and stop from older

tests/test_styles.py

@@ -0,0 +1,43 @@
+"""
+Place to store common styling of plots as well as streamlining common tasks
+like labeling and coloring of events and sensors.
+"""
+
+from study_lyte.styles import EventStyle, SensorStyle
+import pytest
+
+class TestEventStyle:
+    @pytest.mark.parametrize("name, expected", [
+        ('error', EventStyle.ERROR),
+        ('blarg', EventStyle.UNKNOWN)
+
+    ])
+    def test_from_name(self, name, expected):
+        style = EventStyle.from_name(name)
+        assert style == expected
+
+    @pytest.mark.parametrize("property, expected",[
+        ('color', 'g'),
+        ('linestyle', '--'),
+        ('linewidth', 1),
+        ('label', 'Start'),
+    ])
+    def test_property(self, property, expected):
+        assert getattr(EventStyle.START, property) == expected
+
+
+class TestSensorStyle:
+    @pytest.mark.parametrize("name, expected", [
+        ('Sensor1', SensorStyle.RAW_FORCE),
+    ])
+    def test_from_column(self, name, expected):
+        style = SensorStyle.from_column(name)
+        assert style == expected
+
+    @pytest.mark.parametrize("property, expected",[
+        ('column', 'fused'),
+        ('label', 'Fused'),
+        ('color', 'magenta'),
+    ])
+    def test_property(self, property, expected):
+        assert getattr(SensorStyle.FUSED, property) == expected