Performance test (beta). Tiny improvements.

PyEMD/EMD.py

@@ -71,7 +71,6 @@ def __init__(self, spline_kind='cubic', nbsym=2, **kwargs):
         self.range_thr = 0.001
 
         self.nbsym = nbsym
-        self.reduce_scale = 1.
         self.scale_factor = 1.
 
         self.spline_kind = spline_kind
@@ -655,7 +654,7 @@ def end_condition(self, S, IMF):
             Is this the end?
         """
         # When to stop EMD
-        tmp = S.copy() - np.sum(IMF, axis=0)
+        tmp = S - np.sum(IMF, axis=0)
 
 #       # Power is enough
 #       if np.log10(np.abs(tmp).sum()/np.abs(Res).sum()) < self.power_thr:
@@ -736,11 +735,9 @@ def emd(self, S, T=None, max_imf=None):
         # Make sure same types are dealt
         S, T = self._common_dtype(S, T)
         self.DTYPE = S.dtype
-        scale = 1.
         N = len(S)
 
         Res = S.astype(self.DTYPE)
-        Res, scaledS = Res/scale, S/scale
         imf = np.zeros(len(S), dtype=self.DTYPE)
         imf_old = np.nan
 
@@ -756,7 +753,7 @@ def emd(self, S, T=None, max_imf=None):
         while(notFinish):
             self.logger.debug('IMF -- '+str(imfNo))
 
-            Res = scaledS - np.sum(IMF[:imfNo], axis=0)
+            Res[:] = S - np.sum(IMF[:imfNo], axis=0)
             imf = Res.copy()
             mean = np.zeros(len(S), dtype=self.DTYPE)
 
@@ -781,10 +778,10 @@ def emd(self, S, T=None, max_imf=None):
                 if extNo > 2:
 
                     max_env, min_env, eMax, eMin = self.extract_max_min_spline(T, imf)
-                    mean = 0.5*(max_env+min_env)
+                    mean[:] = 0.5*(max_env+min_env)
 
                     imf_old = imf.copy()
-                    imf = imf - self.reduce_scale*mean
+                    imf[:] = imf - mean
 
                     # Fix number of iterations
                     if self.FIXE:
@@ -836,17 +833,15 @@ def emd(self, S, T=None, max_imf=None):
             IMF = np.vstack((IMF, imf.copy()))
             imfNo += 1
 
-            if self.end_condition(scaledS, IMF) or imfNo==max_imf:
+            if self.end_condition(S, IMF) or imfNo==max_imf:
                 notFinish = False
                 break
 
         # Saving residuum
-        Res = scaledS - np.sum(IMF,axis=0)
+        Res = S - np.sum(IMF,axis=0)
         if not np.allclose(Res,0):
             IMF = np.vstack((IMF, Res))
 
-        IMF = IMF*scale
-
         return IMF
 
 ###################################################

PyEMD/tests/test_performance.py

@@ -0,0 +1,60 @@
+#!/usr/bin/python
+# Coding: UTF-8
+
+from __future__ import print_function
+
+import logging
+import numpy as np
+import time
+import unittest
+
+from PyEMD import EMD
+
+class PerformanceTest(unittest.TestCase):
+
+    logger = logging.getLogger(__name__)
+
+    def _timeit(self, fn, signal, T, N=10):
+        avg_t = 0
+        for n in range(N):
+            t0 = time.time()
+            IMF = fn(signal, T)
+            t1 = time.time()
+            avg_t += t1-t0
+        return avg_t/N
+
+    def test_EMD_max_execution_time(self):
+        t_min, t_max = 0, 1
+        N = 100
+        T = np.linspace(t_min, t_max, N)
+        all_test_signals = []
+
+        # These are local values. I'd be veeerrry surprised if everyone had such performance.
+        # In case your test is failing: ignore and carry on.
+        expected_times = [0.012, 0.039, 0.038, 0.050, 0.038, 0.049, 0.048, 0.050, 0.027, 0.050]
+        received_times = [0]*len(expected_times)
+
+        all_w = np.arange(10,20)
+        for w in all_w:
+            signal = np.sin(w*2*np.pi*T)
+            signal[:] = signal[:] + 2*np.cos(5*2*np.pi*T)
+            all_test_signals.append(signal)
+
+        emd = EMD()
+        emd.FIXE = 10
+
+        for idx, signal in enumerate(all_test_signals):
+            avg_t = self._timeit(emd.emd, signal, T, N=10)
+
+            self.logger.info("{}. t = {:.4} (exp. {})".format(idx, avg_t, expected_times[idx]))
+            received_times[idx] = avg_t
+
+       # allclose = np.allclose(received_times, expected_times, atol=1e-2)
+       # self.assertTrue(allclose)
+
+        less_than = received_times <= expected_times
+        self.assertTrue(np.all(less_than))
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    unittest.main()