Skip to content

Instantly share code, notes, and snippets.

@mfcabrera

mfcabrera/stream_sample.py

Forked from anonymous/stream_sample.py
Last active August 29, 2015 14:27
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save mfcabrera/d4f2189723ef1119e616 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save mfcabrera/d4f2189723ef1119e616 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. Miguel Cabrera revised this gist Aug 13, 2015. 1 changed file with 63 additions and 24 deletions.
    87 changes: 63 additions & 24 deletions stream_sample.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,49 +1,88 @@
    from builtins import range
    """
    Programming task
    ================
    Implement the method iter_sample below to make the Unit test pass. iter_sample
    is supposed to peek at the first n elements of an iterator, and determine the
    minimum and maximum values (using their comparison operators) found in that
    sample. To make it more interesting, the method is supposed to return an
    iterator which will return the same exact elements that the original one would
    have yielded, i.e. the first n elements can't be missing.
    Create your solution as a private fork, and send us the URL.
    """

    from itertools import count
    import unittest

    def stream_sample(iterable, n):
    def iter_sample(it, n):
    """
    Peek at the first n elements of an iterable, and determine the min and max
    values. Preserve all elements in the iterable!
    Return a tuple of minimum, maximum (in sample), and an iterator that yields all
    Peek at the first n elements of an iterator, and determine the min and max
    values. Preserve all elements in the iterator!
    @param it: Iterator, potentially infinite
    @param n: Number of elements to peek off the iterator
    @return: Tuple of minimum, maximum (in sample), and an iterator that yields
    all elements that would have been yielded by the original iterator.
    """

    pass
    mx = it.next()
    mn = mx
    buff = []

    def iter_func():
    for x in buff:
    yield x
    for x in it:
    yield x

    buff.append(mx)

    for w in range(n-1):
    try:
    e = it.next()
    buff.append(e)
    if mx < e:
    mx = e
    if mn > e:
    mn = e
    except StopIteration:
    break

    return (mn, mx, iter_func())

    class StreamSampleTestCase(unittest.TestCase):

    def test_stream_sample(self):
    def test_smoke(self):

    # sample only the first 10 elements of a range of length 100

    stream = range(100)

    min_val, max_val, new_stream = stream_sample(iter(stream), 10)


    it = iter(range(100))
    min_val, max_val, new_it = iter_sample(it, 10)

    self.assertEqual(0, min_val)
    self.assertEqual(9, max_val)
    # all elements are still there:
    self.assertEqual(list(range(100)), list(new_stream))
    self.assertEqual(list(range(100)), list(new_it))

    def test_sample_all(self):

    # sample more elements than there are - no error raised
    # now we now the global maximum!

    stream = range(100)

    min_val, max_val, new_stream = stream_sample(iter(stream), 1000)


    it = iter(range(100))
    min_val, max_val, new_it = iter_sample(it, 1000)

    self.assertEqual(0, min_val)
    self.assertEqual(99, max_val)
    self.assertEqual(list(range(100)), list(new_stream))
    self.assertEqual(list(range(100)), list(new_it))

    def test_infinite_stream(self):

    # and guess what - it also works with infinite iterators

    stream = count(0)
    it = count(0)
    min_val, max_val, _ = iter_sample(it, 10)

    min_val, max_val, new_stream = stream_sample(iter(stream), 10)

    self.assertEqual(0, min_val)
    self.assertEqual(9, max_val)

  2. @invalid-email-address Anonymous created this gist Aug 13, 2015.
    51 changes: 51 additions & 0 deletions stream_sample.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,51 @@
    from builtins import range
    from itertools import count
    import unittest

    def stream_sample(iterable, n):
    """
    Peek at the first n elements of an iterable, and determine the min and max
    values. Preserve all elements in the iterable!
    Return a tuple of minimum, maximum (in sample), and an iterator that yields all
    """

    pass

    class StreamSampleTestCase(unittest.TestCase):

    def test_stream_sample(self):

    # sample only the first 10 elements of a range of length 100

    stream = range(100)

    min_val, max_val, new_stream = stream_sample(iter(stream), 10)

    self.assertEqual(0, min_val)
    self.assertEqual(9, max_val)
    # all elements are still there:
    self.assertEqual(list(range(100)), list(new_stream))

    # sample more elements than there are - no error raised
    # now we now the global maximum!

    stream = range(100)

    min_val, max_val, new_stream = stream_sample(iter(stream), 1000)

    self.assertEqual(0, min_val)
    self.assertEqual(99, max_val)
    self.assertEqual(list(range(100)), list(new_stream))

    # and guess what - it also works with infinite iterators

    stream = count(0)

    min_val, max_val, new_stream = stream_sample(iter(stream), 10)

    self.assertEqual(0, min_val)
    self.assertEqual(9, max_val)

    if __name__ == "__main__":
    unittest.main()