diff --git a/src/utils/event_counter.py b/src/utils/event_counter.py new file mode 100644 index 0000000..5b5a596 --- /dev/null +++ b/src/utils/event_counter.py @@ -0,0 +1,110 @@ +from datetime import datetime +import re + +class Event_Counter: + """A class to track and count events, storing the event timestamps and allowing queries + + for the number of events in specific time windows (e.g., last 5 minutes, 10 minutes).""" + REGEX_PATTERN = r'(\d+)\s*(sec|second|seconds|s|min|minute|minutes|m|h|hour|hours|d|day|days)' + TIME_MULTIPLIERS = { + 'sec': 1, 'second': 1, 'seconds': 1, 's': 1, + 'min': 60, 'minute': 60, 'minutes': 60, 'm': 60, + 'h': 3600, 'hour': 3600, 'hours': 3600, + 'd': 86400, 'day': 86400, 'days': 86400 + } + + def __init__(self, retention_duration:int = 900) -> None: + """ + Initializes the Event_Counter object with a specified retention duration. + + Args: + retention_duration (int, optional): + The time period (in seconds) for which event data should be retained. Defaults to 900 (15 minutes). + + uwu + """ + self.__events:list[float] = [] + self.__total_events = 0 + self.__retention_duration = retention_duration + + def cleanup(self): + """Cleans up events that are older than the retention duration. + + Removes all events that occurred outside the defined retention window.""" + offset_time = datetime.now().timestamp() - self.__retention_duration + while len(self.__events) > 0: + event_time = self.__events[0] + if event_time < offset_time: + self.__events.pop(0) + else: + break + + @classmethod + def duration_to_seconds(cls, duration_str:str) -> int: + """Converts a human-readable duration string (e.g., '5min 2sec') into its equivalent in seconds. + + Args: + duration_str (str): The string representing the duration (e.g., "5min 2sec"). + + Returns: + int: The total duration in seconds derived from the input string.""" + matches = re.findall(cls.REGEX_PATTERN, duration_str, flags=re.IGNORECASE) + + total_seconds = 0 + for value, unit in matches: + unit = unit.lower() + total_seconds += int(value) * cls.TIME_MULTIPLIERS[unit] + + return total_seconds + + def increment(self, count:int = 1, skip_cleanup:bool = False): + """Increments the event count by a specified number of events. + + Args: + count (int, optional): The number of events to add. Defaults to 1. + skip_cleanup (bool, optional): If True, skips the cleanup process. Defaults to False.""" + timestamp = datetime.now().timestamp() + self.__total_events += count + self.__events.extend([timestamp] * count) + + if skip_cleanup: + return + self.cleanup() + + def get_count(self, time_window:str = "5M") -> int: + """Returns the number of events that occurred within a specified time window. + + Args: + time_window (str, optional): + The time window to query, e.g., "5M" for 5 minutes. Defaults to "5M". + + Returns: + int: The count of events that occurred within the specified time window.""" + self.cleanup() + + offset_time = datetime.now().timestamp() - self.duration_to_seconds(time_window) + counter = 0 + for event_time in self.__events: + if event_time > offset_time: + counter += 1 + else: + break + + return counter + + def get_total_events(self) -> int: + """Returns the total number of events tracked by the Event_Counter. + + Returns: + int: The total number of events that have been recorded.""" + return self.__total_events + +if __name__ == "__main__": + from time import sleep + counter = Event_Counter(1) + counter.increment() + sleep(1) + counter.increment() + print(counter.get_count("5sEc")) + sleep(0) + print(counter.get_count("5sEc")) \ No newline at end of file