Эквивалент Python для chunk_time Руби?

Question

В моем Ruby-коде есть несколько интересных примеров использования, которые я хочу преобразовать в Python.Я думаю, что мы можем использовать любую библиотеку, я в основном использую pandas и numpy.

Например, представьте, что у вас есть массив временных событий, эти события имеют временную метку и другие свойства (объект или кортеж).

Мне нужен список / массив групп, где эти группы являются «непрерывными» событиями с льготным периодом g единиц (в данном случае, единиц времени).

Вмой код на Ruby я использую примерно так:

grouped_events = events.chunk_while do |previous_event, next_event|
   next_event.timestamp <= previous_event.timestamp + grace_period
end

Поскольку я использую не только синхронизированные события, но и все, что я могу сортировать (так что они как-то сравнимы), я спрашиваю: естьуниверсальный способ, или известная библиотека, которая делает это?

Answer 1

Python не имеет эквивалентной функции.Вам придется написать свой собственный.

Вот моя реализация, использующая итератор и оператор yield :

def chunk_while(iterable, predicate):
    itr = iter(iterable)

    try:
        prev_value = next(itr)
    except StopIteration:
        # if the iterable is empty, yield nothing
        return

    chunk = [prev_value]
    for value in itr:
        # if the predicate returns False, start a new chunk
        if not predicate(prev_value, value):
            yield chunk
            chunk = []

        chunk.append(value)
        prev_value = value

    # don't forget to yield the final chunk
    if chunk:
        yield chunk

Который можно использовать так:

>>> list(chunk_while([1, 3, 2, 5, 5], lambda prev, next_: next_ <= prev + 2))
[[1, 3, 2], [5, 5]]

Answer 2

Я не знаю готового решения для этого, но написать его с нуля не так уж сложно.Выполните итерацию по элементам последовательности, проверяя каждую пару элементов на соответствие вашим критериям и выберите, добавлять ли ее в существующую группу или создать новую группу.

import collections

def pairs(seq):
    """yields (previous, current) pairs from the given iterable."""
    no_item = object()
    previous = no_item
    for item in seq:
        if previous is not no_item:
            yield (previous, item)
        previous = item

def chunk_contiguous(seq, criteria):
    cur_group = []
    for previous, current in pairs(seq):
        if criteria(previous, current):
            cur_group.append(current)
        else:
            yield cur_group
            cur_group = [current]
    if cur_group: 
        yield cur_group

Event = collections.namedtuple("Event", ["name", "timestamp"])

events = [
    Event("foo", 0),
    Event("bar", 1),
    Event("baz", 10),
    Event("qux", 12),
    Event("Larry", 17),
    Event("Curly", 21),
    Event("Moe", 25),

]

g = 4

for group in chunk_contiguous(events, lambda previous, current: current.timestamp <= previous.timestamp + g):
    print(group)

Результат:

[Event(name='bar', timestamp=1)]
[Event(name='baz', timestamp=10), Event(name='qux', timestamp=12)]
[Event(name='Larry', timestamp=17), Event(name='Curly', timestamp=21), Event(name='Moe', timestamp=25)]