Source code for aw_transform.chunk_events_by_key

import logging
from typing import List, Dict
from datetime import timedelta

from aw_core.models import Event

logger = logging.getLogger(__name__)


[docs]def chunk_events_by_key( events: List[Event], key: str, pulsetime: float = 5.0 ) -> List[Event]: chunked_events: List[Event] = [] for event in events: if key not in event.data: break timediff = timedelta(seconds=999999999) # FIXME: ugly but works if len(chunked_events) > 0: timediff = event.timestamp - (events[-1].timestamp + events[-1].duration) if ( len(chunked_events) > 0 and chunked_events[-1].data[key] == event.data[key] and timediff < timedelta(seconds=pulsetime) ): chunked_event = chunked_events[-1] chunked_event.duration += event.duration chunked_event.data["subevents"].append(event) else: data = {key: event.data[key], "subevents": [event]} chunked_event = Event( timestamp=event.timestamp, duration=event.duration, data=data ) chunked_events.append(chunked_event) return chunked_events