Source code for retentioneering.eventstream.helpers.filter_events_helper

from __future__ import annotations

from typing import Callable, Optional

from pandas import DataFrame, Series

from retentioneering.backend.tracker import (
    collect_data_performance,
    time_performance,
    track,
)
from retentioneering.utils.doc_substitution import docstrings

from ..types import EventstreamSchemaType, EventstreamType


class FilterEventsHelperMixin:
[docs] @docstrings.with_indent(12) @time_performance( # type: ignore scope="filter_events", event_name="helper", event_value="combine", ) def filter_events( self: EventstreamType, func: Callable[[DataFrame, Optional[EventstreamSchemaType]], Series] ) -> EventstreamType: """ A method of ``Eventstream`` class that filters input ``eventstream`` based on custom conditions. Parameters ---------- %(FilterEvents.parameters)s Returns ------- Eventstream The filtered ``eventstream``. """ calling_params = { "func": func, } # avoid circular import from retentioneering.data_processors_lib import FilterEvents, FilterEventsParams from retentioneering.preprocessing_graph import PreprocessingGraph from retentioneering.preprocessing_graph.nodes import EventsNode p = PreprocessingGraph(source_stream=self) # type: ignore node = EventsNode(processor=FilterEvents(params=FilterEventsParams(func=func))) # type: ignore p.add_node(node=node, parents=[p.root]) result = p.combine(node) del p collect_data_performance( scope="filter_events", event_name="metadata", called_params=calling_params, performance_data={}, eventstream_index=self._eventstream_index, parent_eventstream_index=self._eventstream_index, child_eventstream_index=result._eventstream_index, ) return result