Add run_length

docs/api.rst

@@ -125,6 +125,7 @@ These tools return summarized or aggregated data from an iterable.
 .. autofunction:: unique_to_each
 .. autofunction:: locate
 .. autofunction:: consecutive_groups
+.. autoclass:: run_length
 
 ----
 

more_itertools/more.py

@@ -50,6 +50,7 @@
     'padded',
     'peekable',
     'rstrip',
+    'run_length',
     'side_effect',
     'sliced',
     'sort_together',
@@ -1587,3 +1588,32 @@ def consecutive_groups(iterable, ordering=lambda x: x):
         enumerate(iterable), key=lambda x: x[0] - ordering(x[1])
     ):
         yield map(itemgetter(1), g)
+
+
+class run_length(object):
+    """
+    :func:`run_length.encode` compresses an iterable with run-length encoding.
+    It yields groups of repeated items with the count of how many times they
+    were repeated:
+
+        >>> uncompressed = 'abbcccdddd'
+        >>> list(run_length.encode(uncompressed))
+        [('a', 1), ('b', 2), ('c', 3), ('d', 4)]
+
+    :func:`run_length.decode` decompresses an iterable that was previously
+    compressed with run-length encoding. It yields the items of the
+    decompressed iterable:
+
+        >>> compressed = [('a', 1), ('b', 2), ('c', 3), ('d', 4)]
+        >>> list(run_length.decode(compressed))
+        ['a', 'b', 'b', 'c', 'c', 'c', 'd', 'd', 'd', 'd']
+
+    """
+
+    @staticmethod
+    def encode(iterable):
+        return ((k, ilen(g)) for k, g in groupby(iterable))
+
+    @staticmethod
+    def decode(iterable):
+        return chain.from_iterable(repeat(k, n) for k, n in iterable)

more_itertools/tests/test_more.py

@@ -1472,3 +1472,17 @@ def test_exotic_ordering(self):
             [('d', 'b', 'c', 'a'), ('d', 'c', 'a', 'b')],
         ]
         self.assertEqual(actual, expected)
+
+
+class RunLengthTest(TestCase):
+    def test_encode(self):
+        iterable = (int(str(n)[0]) for n in count(800))
+        actual = mi.take(4, mi.run_length.encode(iterable))
+        expected = [(8, 100), (9, 100), (1, 1000), (2, 1000)]
+        self.assertEqual(actual, expected)
+
+    def test_decode(self):
+        iterable = [('d', 4), ('c', 3), ('b', 2), ('a', 1)]
+        actual = ''.join(mi.run_length.decode(iterable))
+        expected = 'ddddcccbba'
+        self.assertEqual(actual, expected)