I work on a project with time series data. So there are samples (\$y\$), and each sample has a timestamp (\$x\$). The data will be visualized, but often there are time series which contain samples that do not change over several timestamps. In a visualization (a \$y(x)\$ plot) those repeated samples can be removed, since the visualization will just show a horizontal line. (For example, if a constant value 3.14 is sampled 1 million times in one second, the plot would have the same visual appearance as if the value had been sampled just twice in that period.) I wrote an algorithm that removes those repeated samples in the order to reduce the amount of data and speed up visualization. My algorithm works, but I know that it can be done better.
Here is the algorithm (in the file compress_2D_signal.py):
def compress_2D_signal(x, y): end = len(x) end_1 = len(x) - 1 x1 =  y1 =  x1append = x1.append y1append = y1.append for ind, elements in enumerate(zip(x, y)): if ind == end or ind >= end_1 or y[ind - 1] != y[ind] or y[ind + 1] != y[ind] or ind == 0: x1append(elements) y1append(elements) return x1, y1
and here are the tests:
from unittest import TestCase import time from compress_2D_signal import compress_2D_signal class Test2DSignal(TestCase): def test_store_sample_array_one_signal_parameter_returns_same_when_each_sample_different(self): x = [1, 2, 3, 4] y = [0, 1, 2, 3] x1, y1 = compress_2D_signal(x, y) self.assertEqual(x, x1) self.assertEqual(y, y1) def test_store_sample_array_one_signal_parameter(self): x = [1, 2, 3, 4, 5] y = [0, 2, 2, 2, 0] x1, y1 = compress_2D_signal(x, y) self.assertEqual([1, 2, 4, 5], x1) self.assertEqual([0, 2, 2, 0], y1) def test_store_sample_array_one_signal_parameter_2(self): x = [1, 2, 3, 4, 5, 6, 7, 8, 9] y = [0, 2, 2, 2, 0, 0, 1, 1, 1] x1, y1 = compress_2D_signal(x, y) self.assertEqual([1, 2, 4, 5, 6, 7, 9], x1) self.assertEqual([0, 2, 2, 0, 0, 1, 1], y1) def test_time_self(self): length = int(1e6) x = range(length) y = range(length) start = time.time() compress_2D_signal(x, y) stop = time.time() print('Time elapsed for %s points : %s this are %s points/second' % (length, stop - start, length / (stop - start)))