How to extract blocks form a source observable?

[juxeii]

I have [1,2,3,2,4,5,0,2,1,3,4,5,2,2,1,4,5]

What I want as output is

[[1,2,3,2,4,5],[1,3,4,5],[1,4,5]]

This means I want to extract blocks from the input sequence with some start marker, here 1, and some end marker, here 5. Elements in between these blocks are ignored.

Is there a readable combination of operators for achieving this?

[matiboy]

Hi @juxeii

If I understand your question, assuming that you expect an observable of lists, not a list of lists, I think you're looking for window_toggle.

The below should work. Not sure it's very readable. Basically you need to make your input hot and shared, then use it as shown inside the create function in the below test.

def test_issue_706():
    scheduler = TestScheduler()
    messages = [
        on_next(200 + (i + 1) * 10, x)
        for i, x in enumerate([1, 2, 3, 2, 4, 5, 0, 2, 1, 3, 4, 5, 2, 2, 1, 4, 5])
    ]
    print(messages)
    numbers = scheduler.create_hot_observable(messages)

    numbers_shared = numbers.pipe(operators.share())

    def create():
        return numbers_shared.pipe(
            operators.window_toggle(
                numbers_shared.pipe(
                    operators.filter(lambda x: x == 1),
                ),
                lambda _: numbers_shared.pipe(
                    operators.filter(lambda x: x == 5),
                ),
            ),
            operators.flat_map(lambda window: window.pipe(operators.to_list())),
        )

    results = scheduler.start(create=create)

    assert results.messages == [
        on_next(260, [1, 2, 3, 2, 4, 5]),
        on_next(320, [1, 3, 4, 5]),
        on_next(370, [1, 4, 5]),
    ]

[juxeii]

Works like it should!
Thank you very much!

[juxeii]

Hi,
there is something weird going on with run and subscribe in your example.
Running

nums = [1, 2, 3, 2, 4, 5, 0, 2, 1, 3, 4, 5, 2, 2, 1, 4, 5]
numbers = rx.from_(nums)
numbers_shared = numbers.pipe(ops.share())

numbers_shared.pipe(
    ops.window_toggle(
        numbers_shared.pipe(
            ops.filter(lambda x: x == 1),
        ),
        lambda _: numbers_shared.pipe(
            ops.filter(lambda x: x == 5),
        ),
    ),
    ops.flat_map(lambda window: window.pipe(ops.to_list())),
    ops.to_list()
).subscribe(print)

gives as result
[[1, 2, 3, 2, 4, 5], [1, 3, 4, 5], [1, 4, 5]] which is correct.

But running

nums = [1, 2, 3, 2, 4, 5, 0, 2, 1, 3, 4, 5, 2, 2, 1, 4, 5]
numbers = rx.from_(nums)
numbers_shared = numbers.pipe(ops.share())

r = numbers_shared.pipe(
    ops.window_toggle(
        numbers_shared.pipe(
            ops.filter(lambda x: x == 1),
        ),
        lambda _: numbers_shared.pipe(
            ops.filter(lambda x: x == 5),
        ),
    ),
    ops.flat_map(lambda window: window.pipe(ops.to_list())),
    ops.to_list()
)

print(r.run())

gives [[1, 2, 3, 2, 4, 5], [2, 1, 3, 4, 5], [2, 1, 4, 5]] which is wrong.

Do you know what makes run different in terms of sequence logic?

[matiboy]

Hi @juxeii

The difference between the two is the scheduler.
run creates a NewThreadScheduler whereas the .subscribe has None for scheduler.
TL;DR: a new thread is therefore created on each iteration and it creates a potential gap and therefore those "incorrect" emissions.

In depth:
The result of NewThreadScheduler is that when, inside window_toggle, empty()'s .subscribe is called, it happens inside a new thread (each time) for run, but inside main thread for your .subscribe(print). The new thread is the one closing the "right" side of group_join which is what's being used under the hood. So sometimes the "right" side emits before it closes, resulting in extra emissions at the beginning of windows.

You cannot override the scheduler for run; whereas you could provide any scheduler to .subscribe if you don't want to block this thread.

Overall I think my solution wasn't ideal for the use case after all; personally I'd write a custom operator. I can share that if you need.

[juxeii]

Hi,

yes please, a custom operator would be ideal.

The reason why I detected this behavior is that I need to return a list of lists. But

return numbers_shared.pipe(
    ops.window_toggle(
        numbers_shared.pipe(
            ops.filter(lambda x: x == 1),
        ),
        lambda _: numbers_shared.pipe(
            ops.filter(lambda x: x == 5),
        ),
    ),
    ops.flat_map(lambda window: window.pipe(ops.to_list())),
    ops.to_list()
).run()

returns the wrong values.
Therefore, I ended up with

blocks =[]
numbers_shared.pipe(
    ops.window_toggle(
        numbers_shared.pipe(
            ops.filter(lambda x: x == 1),
        ),
        lambda _: numbers_shared.pipe(
            ops.filter(lambda x: x == 5),
        ),
    ),
    ops.flat_map(lambda window: window.pipe(ops.to_list())),
).subscribe(lambda b: blocks.append(b))

return blocks

But having blocks defined outside the reactive context is not nice. Is there no other way returning the list of lists?

[matiboy]

Hi @juxeii

I personally see nothing wrong with having blocks defined as you did, if you are already forced into a "synchronous" form of code with run or the subscribe you propose.

Generally though, if you have an external variable like this, you should be able to replace it with a operators.scan.
Below is a solution I thought about after reading what you mentioned above:

def handle(acc, x):
    if x == 1 or len(acc[1]) > 0:
        return None, acc[1] + [x] # None as first item means "don't emit" via the below `filter` operator; second item is our window values
    if x == 5:
        return acc[1] + [x], [] # First item is what we'll emit since we reached a number 5; second value is basically a reset
    return None, [] # not one, and not in a window, don't emit and don't keep values

outcome = messages.pipe(
    operators.scan(
        handle, (None, []),
    ),
    operators.map(lambda x: x[0]),
    operators.filter(lambda x: x is not None),
    operators.to_list(),
).run()

This can also be written as a much less readable - but valid - lambda:

outcome = messages.pipe(
    operators.scan(
        lambda acc, x: (acc[1] + [x], [])
        if x == 5
        else ((None, acc[1] + [x]) if (x == 1 or len(acc[1]) > 0) else (None, [])),
        (None, []),
    ),
    operators.filter(lambda x: x[0] is not None),
    operators.map(lambda x: x[0]),
    operators.to_list(),
).run()

And finally a fully custom operator which is more verbose code-wise but gives you more flexibility imo:

# This is the actual operator
def list_from_to(start_number, end_number):
    lock = threading.Lock()
    current_values = []
    is_on = False

    def operator(source):
        def subscribe(observer, scheduler=None):
            def on_next(value):
                nonlocal is_on
                with lock:
                    if value == start_number:
                        is_on = True
                    if is_on:
                        current_values.append(value)
                    if value == end_number:
                        is_on = False
                        observer.on_next(list(current_values))
                        current_values.clear()

            return source.subscribe(
                on_next=on_next,
                on_completed=observer.on_completed,
                on_error=observer.on_error,
                scheduler=scheduler,
            )

        return Observable(subscribe)

    return operator

messages = reactivex.from_list(
    [1, 2, 3, 2, 4, 5, 0, 3, 4, 2, 1, 3, 4, 5, 2, 2, 1, 4, 5]
)

outcome = messages.pipe(list_from_to(1, 5), operators.to_list()).run()

[juxeii]

Thx for your effort.
Tbh, the operator solution I like the most, since it is readable from a client perspective.
Funny enough, I had the operator solution in a non-reactive code already and this I wanted to hide with rxpy :)

The example I have given is with numbers, in real code it is with other type(s), so I would need to rewrite the operator for a generic use anyway.
But why not using window_toggle as you suggested? I like that one too.

BTW: the documentation for run under https://rxpy.readthedocs.io/en/latest/reference_observable.html does not mention any scheduler or threading.
Quote: Subscribes to the observable source. Then blocks and waits for the observable source to either complete or error
I concluded that this is semantically equivalent to a simple subscribe. Was not clear to me.

[juxeii]

Generic operator is now

def list_from_to(start_compare: Callable[[T], bool],
                 end_compare: Callable[[T], bool]) -> Observable[list[T]]:
    lock = threading.Lock()
    current_values = []
    is_on = False

    def operator(source):
        def subscribe(observer, scheduler=None):
            def on_next(value):
                nonlocal is_on
                with lock:
                    if start_compare(value):
                        is_on = True
                    if is_on:
                        current_values.append(value)
                    if end_compare(value):
                        is_on = False
                        observer.on_next(list(current_values))
                        current_values.clear()

            return source.subscribe(
                on_next=on_next,
                on_completed=observer.on_completed,
                on_error=observer.on_error,
                scheduler=scheduler,
            )

        return Observable(subscribe)

    return operator

Will be used something like

numbers.pipe(
    utils.list_from_to(lambda x: x == 1, lambda x: x == 5),
    ops.to_list()
).run()

[matiboy]

Very nice to have made it more generic with a start_compare/end_compare callables.

In the end I didn't fancy my initial window_toggle solution because of that operator's internal implementation: it uses a group_join with "right side timer" being empty() which seems... awkward, and the scheduler isn't passed to that empty(), which can lead to unexpected behaviour as you thankfully spotted.

[juxeii]

Yes, the interplay between run and window_toggle is not optimal and a bit confusing, regarding the docs.
Time for refactoring? :-)

Thanks for your great support!