Merge pull request #23 from parkin/absoluteChangeStrategy

Absolute change strategy implemented

src/pypore/strategies/absolute_change_threshold_strategy.pyx

@@ -10,8 +10,13 @@ cdef class AbsoluteChangeThresholdStrategy(ThresholdStrategy):
         self.change_end = change_end
 
     cdef double compute_starting_threshold_c(self, double baseline, double variance):
-        # TODO figure out how to handle both positive and negative changes...
-        raise NotImplementedError
+        """
+        :returns: change_start from strategy initialization. 'baseline' and 'variance' parameters have no effect.
+        """
+        return self.change_start
 
     cdef double compute_ending_threshold_c(self, double baseline, double variance):
-        raise NotImplementedError
+        """
+        :returns: change_end from strategy initialization. 'baseline' and 'variance' parameters have no effect.
+        """
+        return self.change_end

src/pypore/strategies/tests/__init__.py

@@ -0,0 +1 @@
+__author__ = 'parkin'

src/pypore/strategies/tests/test_absolute_change_threshold_strategy.py

@@ -0,0 +1,28 @@
+import unittest
+
+from pypore.strategies.absolute_change_threshold_strategy import AbsoluteChangeThresholdStrategy
+
+
+class TestAbsoluteChangeThresholdStrategy(unittest.TestCase):
+    def test_compute_starting_threshold(self):
+        """
+        Test that the starting and ending thresholds do not depend on the baseline or variance.
+        """
+        start_changes = [-1.e9, -1, -1.e-9, 0., 1, 1.e-9, 1.e9]
+        end_changes = [-1.e9, -1, -1.e-9, 0., 1, 1.e-9, 1.e9]
+        baselines = [-1.e9, -1., -1.e-9, 0., 1, 1.e-9, 1.e9]
+        variances = [-1.e9, -1., -1.e-9, 0., 1, 1.e-9, 1.e9]
+        for i in xrange(len(start_changes)):
+            strategy = AbsoluteChangeThresholdStrategy(start_changes[i], end_changes[i])
+
+            for j in xrange(len(variances)):
+                start_thresh = strategy.compute_starting_threshold(baselines[j], variances[j])
+                end_thresh = strategy.compute_ending_threshold(baselines[j], variances[j])
+
+                self.assertEqual(start_thresh, start_changes[i],
+                                 "Unexpected starting threshold. Should be {0}, was {1}.".format(start_changes[i],
+                                                                                                 start_thresh))
+                self.assertEqual(end_thresh, end_changes[i],
+                                 "Unexpected ending threshold. Should be {0}, was {1}.".format(end_changes[i],
+                                                                                               end_thresh))
+

src/pypore/tests/test_event_finder.py

@@ -11,6 +11,7 @@
 import pypore.filetypes.event_database as ed
 
 import pypore.sampledata.testing_files as tf
+from pypore.tests.util import _test_file_manager
 
 
 class TestEventFinder(unittest.TestCase):
@@ -302,6 +303,80 @@ def test_debug_option(self):
         os.remove(event_databases[0])
 
 
+DIRECTORY = os.path.dirname(os.path.abspath(__file__))
+
+from pypore.event_finder import Parameters
+from pypore.strategies.absolute_change_threshold_strategy import AbsoluteChangeThresholdStrategy
+
+
+class TestEventFinderAbsoluteChangeThresholdStrategy(unittest.TestCase):
+    @_test_file_manager(DIRECTORY)
+    def test_too_large_start_threshold(self, filename):
+        """
+        Tests that we don't find events when the starting threshold is too large.
+        """
+        data_file = tf.get_abs_path('chimera_1event.log')
+
+        parameters = Parameters(threshold_strategy=AbsoluteChangeThresholdStrategy(100., 1.))
+
+        event_databases = find_events([data_file], parameters=parameters,
+                                      save_file_names=[filename], debug=True)
+
+        h5file = ed.open_file(filename, mode='r')
+
+        event_count = h5file.get_event_count()
+        self.assertEqual(event_count, 0, "Unexpected event count. Should be {0}, was {1}.".format(event_count, 0))
+
+        h5file.close()
+
+    @_test_file_manager(DIRECTORY)
+    def test_good_thresholds(self, filename):
+        """
+        Tests that we find the correct number of events when the starting and ending thresholds are appropriate.
+        """
+        data_file = tf.get_abs_path('chimera_1event.log')
+
+        parameters = Parameters(threshold_strategy=AbsoluteChangeThresholdStrategy(2., 1.))
+
+        event_databases = find_events([data_file], parameters=parameters,
+                                      save_file_names=[filename], debug=True)
+
+        h5file = ed.open_file(filename, mode='r')
+
+        #Check the number of events
+        event_count = h5file.get_event_count()
+        self.assertEqual(event_count, 1, "Unexpected event count. Should be {0}, was {1}.".format(event_count, 0))
+
+        #Check the event length
+        sample_rate = h5file.get_sample_rate()
+        event_length = h5file.get_event_row(0)['event_length']/sample_rate
+        event_length_should_be = 0.00024
+        percent_diff = abs(event_length - event_length_should_be) / event_length_should_be
+        self.assertLessEqual(percent_diff, 0.05,
+                             "Unexpected event length. Should be {0}, was {1}.".format(event_length_should_be,
+                                                                                       event_length))
+
+        h5file.close()
+
+    @_test_file_manager(DIRECTORY)
+    def test_too_large_end_threshold(self, filename):
+        """
+        Tests that we don't find events when the ending threshold is too large.
+        """
+        data_file = tf.get_abs_path('chimera_1event.log')
+
+        parameters = Parameters(threshold_strategy=AbsoluteChangeThresholdStrategy(2., 1000.))
+
+        event_databases = find_events([data_file], parameters=parameters,
+                                      save_file_names=[filename], debug=True)
+
+        h5file = ed.open_file(filename, mode='r')
+
+        event_count = h5file.get_event_count()
+        self.assertEqual(event_count, 0, "Unexpected event count. Should be {0}, was {1}.".format(event_count, 0))
+
+        h5file.close()
+
 
 if __name__ == "__main__":
     # import sys;sys.argv = ['', 'Test.testName']