Hopping window

[Psykopear]

Added a HoppingWindowConfig object.

I also refactored the Windower trait to autoimplement some methods since it was equal on both the WindowConfigs. Trait methods can still be overridden if we need a different behavior in the future.

I also added a new macro, add_pymethods, which alleviates the problem of having to copy paste a lot of code for the infamuous egregious hack. I also have a separate branch where I use the macro everywhere else in the codebase, but I'd like to merge this first (or revert it if we don't like the macro).

Fixes #172

[whoahbot]

LGTM

[whoahbot]

I rewrote this test to make it a little easier (for me) to satisfy myself that the window was doing what I expected.

def test_hopping_window():
    start_at = datetime(2022, 1, 1, tzinfo=timezone.utc)

    flow = Dataflow()

    def gen():
        yield ("ALL", {"time": start_at + timedelta(seconds=1), "val": "a"})
        yield ("ALL", {"time": start_at + timedelta(seconds=4), "val": "b"})
        yield ("ALL", {"time": start_at + timedelta(seconds=8), "val": "c"})
        yield ("ALL", {"time": start_at + timedelta(seconds=12), "val": "d"})
        yield ("ALL", {"time": start_at + timedelta(seconds=13), "val": "e"})
        yield ("ALL", {"time": start_at + timedelta(seconds=14), "val": "f"})
        yield ("ALL", {"time": start_at + timedelta(seconds=16), "val": "g"})
        # This is late, and should be ignored
        yield ("ALL", {"time": start_at + timedelta(seconds=1), "val": "h"})

    flow.input("inp", TestingBuilderInputConfig(gen))

    clock_config = EventClockConfig(
        lambda e: e["time"], wait_for_system_duration=timedelta(0)
    )
    window_config = HoppingWindowConfig(
        length=timedelta(seconds=10), start_at=start_at, offset=timedelta(seconds=5)
    )

    def add(acc, x):
        acc.append(x["val"])
        return acc

    flow.fold_window("sum", clock_config, window_config, list, add)

    out = []
    flow.capture(TestingOutputConfig(out))

    run_main(flow)
    print(out)

    assert sorted(out) == sorted(
        [
            ("ALL", ["a", "b", "c"]),
            ("ALL", ["c", "d", "e", "f"]),
            ("ALL", ["d", "e", "f", "g"]),
            ("ALL", ["g"]),
        ]
    )

[davidselassie]

I do think #196 (comment) is a little more readable. Not for this PR, but probably all the window tests should look like this.

[Psykopear]

Yes that's much more clear, I'm changing it

[davidselassie]

Great! Thanks.

[davidselassie]

I do think #196 (comment) is a little more readable. Not for this PR, but probably all the window tests should look like this.

[davidselassie]

My thinking with keeping these all abstract was that "sessionization" windowers might not have a fixed hash of closed times and would need a different data structure. Although there might be a way to keep using this. I'm happy for you to merge this and once we get to session windows we can figure out if it still works.

[Psykopear]

Yes I was thinking about this too when reading about other window types. This way I tied the data structure to the trait, which it wasn't before, so I'm probably going to revert/rethink this before merging

[Psykopear]

So, I also realized that in the end a monad is a monoid in the category of endofunctors a TumblingWindow is just a HoppingWindow with length==offset, so I refactored TumblingWindowConfig to be:

impl WindowBuilder for TumblingWindowConfig {
    fn build(&self, _py: Python) -> StringResult<Builder> {
        Ok(Box::new(super::hopping_window::HoppingWindower::builder(
            self.length,
            self.length,
            self.start_at.unwrap_or_else(Utc::now),
        )))
    }
}