apply function on time window

skais/ais/ais_trajectory.py

@@ -193,6 +193,17 @@ def angle_dispersion(dat, radius):
 
     return disp
 
+@jit(nopython=True)
+def apply_func_on_window(dat, func, radius, on_edge='copy'):
+    result = np.zeros(dat.shape)
+    if on_edge == 'copy':
+        dat = np.concatenate([np.full(radius, dat[0]), dat, np.full(radius, dat[-1])])
+        for i in range(radius, dat.shape[0] - radius):
+            data = dat[i - radius:i + radius + 1]
+            result[i - radius] = func(data)
+        return result
+    else:
+        raise ValueError(f"{on_edge} not recognised, options are: ['copy']")
 
 class AISTrajectory:
     def __init__(self, df, interpolation_time=None):
@@ -388,3 +399,12 @@ class AISTrajectory:
             index += i
 
         return result
+
+    def apply_func_on_time_window(self, func, radius, column, new_column=None):
+        dat = self.df[column]
+        result = apply_func_on_window(dat, func, radius, on_edge='copy')
+
+        if new_column is None:
+            self.df[column] = result
+        else:
+            self.df[new_column] = result
\ No newline at end of file

skais/process/basic_features_operations.py

+import cmath
+
 import numpy as np
 
 
@@ -31,3 +33,10 @@ def angular_mean(angles):
 def angular_std(angles):
     return 1 - angular_dispersion(angles)
 
+
+def angular_time_derivative(angle1, angle2, ts1, ts2):
+    z1 = np.cos(angle1) + np.sin(angle1) * 1j
+    z2 = np.cos(angle2) + np.sin(angle2) * 1j
+
+    z = z2 / z1
+    return cmath.polar(z)[1] / (ts2 - ts1)

skais/tests/process/test_basic_features_operations.py

@@ -2,7 +2,7 @@ import unittest
 
 import numpy as np
 
-from skais.process.basic_features_operations import angular_mean, angular_std
+from skais.process.basic_features_operations import angular_mean, angular_std, angular_time_derivative
 
 
 class TestAngular(unittest.TestCase):
@@ -46,7 +46,17 @@ class TestAngular(unittest.TestCase):
 
         self.assertEqual(0, angular_std(x))
 
-    def test_angular_std(self):
+    def test_angular_std_2(self):
         x = np.radians(np.array([0, 90, 180, 270]))
 
         self.assertAlmostEqual(1, angular_std(x))
+
+    def test_angular_time_derivative(self):
+        result = angular_time_derivative(0, np.pi, 0, 1)
+
+        self.assertAlmostEqual(np.pi, result)
+
+    def test_angular_time_derivative_2(self):
+        result = angular_time_derivative(7*np.pi/4, np.pi/4, 0, 1)
+
+        self.assertAlmostEqual(np.pi/2, result)
\ No newline at end of file