evt-timeline.py

from datetime import datetime, timedelta
from collections import defaultdict
"""
Based on the work done on this paper
https://www.conviva.com/wp-content/uploads/2023/01/Raising-the-Level-of-Abstraction-for-Time-State-Analytics.pdf
Neat work C3!
"""

class Event:
    """
    Represents an event with a timestamp and a state.
    """
    def __init__(self, timestamp: datetime, state: str):
        self.timestamp = timestamp  # The time the event occurred
        self.state = state  # The state associated with the event

class Timeline:
    """
    Represents a timeline of events and provides methods to build the timeline and calculate durations.
    """
    def __init__(self):
        self.events = []  # List to store events
        self.timeline = []  # List to store state durations

    def add_event(self, event: Event):
        """
        Adds an event to the timeline and keeps events sorted by time.
        """
        self.events.append(event)
        self.events.sort(key=lambda x: x.timestamp)  # Keep events sorted by their timestamp

    def build_timeline(self):
        """
        Builds the timeline by calculating the duration each state was active.
        """
        current_state = None
        prev_timestamp = None
        for event in self.events:
            if current_state and prev_timestamp:
                # Append the current state and its duration to the timeline
                self.timeline.append((current_state, event.timestamp - prev_timestamp))
            # Update the current state and previous timestamp
            current_state = event.state
            prev_timestamp = event.timestamp

    def get_state_durations(self):
        """
        Returns a dictionary with the total duration for each state.
        """
        state_durations = defaultdict(timedelta)
        for state, duration in self.timeline:
            state_durations[state] += duration  # Accumulate the duration for each state
        return state_durations

    def total_duration(self, state: str):
        """
        Returns the total duration for a specific state.
        """
        durations = self.get_state_durations()
        return durations[state]

    def average_duration(self, state: str):
        """
        Returns the average duration for a specific state.
        """
        durations = self.get_state_durations()
        count = sum(1 for s, d in self.timeline if s == state)  # Count the number of occurrences of the state
        return durations[state] / count if count else timedelta(0)  # Avoid division by zero

# Example Usage
events = [
    Event(datetime(2023, 1, 1, 10, 0, 0), 'buffering'),
    Event(datetime(2023, 1, 1, 10, 0, 5), 'playing'),
    Event(datetime(2023, 1, 1, 10, 1, 0), 'buffering'),
    Event(datetime(2023, 1, 1, 10, 1, 10), 'playing')
]

timeline = Timeline()
for event in events:
    timeline.add_event(event)  # Add each event to the timeline

timeline.build_timeline()  # Build the timeline to calculate state durations

total_buffering_time = timeline.total_duration('buffering')  # Calculate total buffering time
average_buffering_time = timeline.average_duration('buffering')  # Calculate average buffering time

print(f"Total Buffering Time: {total_buffering_time}")
print(f"Average Buffering Time: {average_buffering_time}")