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[0])
y1append(elements[1])
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)))