from_iterable_with_interval.py

from rx import Observable, AnonymousObservable
from rx.concurrency import current_thread_scheduler
from rx.internal import extensionclassmethod


@extensionclassmethod(Observable)
def from_iterable_with_interval(cls, period, iterable, scheduler=None):
    """Creates an Observable that polls iterator made from iterable every period millis."""
    iterator = iter(iterable)
    return Observable.interval(period, scheduler=scheduler).map(lambda v: next(iterator))

...
reed_switch_sensor = \
    Observable.from_iterable_with_interval(500, source, scheduler=aio_scheduler). \
    distinct_until_changed(). \
    timestamp()