Compressing time series by removing repeated samples

Question

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)))

Answer 1

You want to keep points where the value for \$y\$ is different from either the preceding or succeeding value. So in NumPy you can write:

import numpy as np

def compress_2D_signal(x, y):
    """Compress signal y(x) by omitting repeated values for y.

    Takes a signal y(x) as two array-likes x and y. Returns the
    compressed signal y1(x1) as the tuple x1, y1, where y1 contains
    the first and last values of y, and values of y that are different
    from the preceeding or succeeding value, and x1 contains the
    corresponding values from x.

    """
    x, y = np.asarray(x), np.asarray(y)
    keep = np.empty_like(x, dtype=bool)
    if len(x) > 0:
        keep[0] = keep[-1] = True
        keep[1:-1] = (y[1:-1] != y[:-2]) | (y[1:-1] != y[2:])
    return x[keep], y[keep]

Notes on this code:

1. This returns NumPy arrays (not Python lists), so you'll need to revise your test cases to use numpy.testing.assert_array_equal instead of unittest.TestCase.assertEqual.

2. If you pass it Python lists, then much of the runtime will be in the numpy.asarray call (converting the input to NumPy arrays) and not in the compression code. So for a fair test I think it makes sense to update test_time_self so that it calls numpy.arange instead of range.

   With this change, I find that the revised code is about 30 times faster than the original code.

A couple of other review points:

1. There's no docstring for compress_2D_signal. It's hard to use and maintain code when there's no documentation.

2. The test case code is very repetitive — it would be simpler to loop over a list of cases, like this:

   from numpy.testing import assert_array_equal
   
   class Test2DSignal(TestCase):
       _CASES = [
           # x, y, x1, y1
           ([1, 2, 3, 4], [0, 1, 2, 3],
            [1, 2, 3, 4], [0, 1, 2, 3]),
           ([1, 2, 3, 4, 5], [0, 2, 2, 2, 0],
            [1, 2, 4, 5], [0, 2, 2, 0]),
           ([1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 2, 2, 2, 0, 0, 1, 1, 1], 
            [1, 2, 4, 5, 6, 7, 9], [0, 2, 2, 0, 0, 1, 1]),
       ]
   
       def test_2D_signal(self):
           for x, y, x1_expected, y1_expected in self._CASES:
               x1_found, y1_found = compress_2D_signal(x, y)
               assert_array_equal(x1_expected, x1_found)
               assert_array_equal(y1_expected, y1_found)
   

   This makes it easier to add new cases. In particular, I would add tests for edge cases such as empty and single-element inputs, like this:

       _CASES = [
           # x, y, x1, y1
           ([], [], [], []),
           ([1], [1], [1], [1]),
           ([1, 2], [1, 1], [1, 2], [1, 1]),
           ([1, 2, 3], [1, 1, 1], [1, 3], [1, 1]),
           # etc.