Find a way to prevent flapping status messages #97

New issue

Open

opened 2025-02-20 07:58:41 +00:00 by thomasjsn · 0 comments

thomasjsn commented

2025-02-20 07:58:41 +00:00

(Migrated from git.homelab.no)

Keep track of messages sent the last X minutes, and take a break if it gets too many.

from collections import deque
import time

class TimedCounter:
    def __init__(self, seconds: int):
        self.seconds = seconds
        self.timestamps = deque()

    def add_event(self):
        now = time.time()
        self.timestamps.append(now)

        # Remove expired timestamps
        while self.timestamps and now - self.timestamps[0] > self.seconds:
            self.timestamps.popleft()

    def get_count(self):
        return len(self.timestamps)


# Usage
status_msg_counter = TimedCounter(5)
status_msg_counter.add_event()
time.sleep(4)
status_msg_counter.add_event()
print(status_msg_counter.get_count())  # Number of events in last 5 minutes

Keep track of messages sent the last X minutes, and take a break if it gets too many. ```python from collections import deque import time class TimedCounter: def __init__(self, seconds: int): self.seconds = seconds self.timestamps = deque() def add_event(self): now = time.time() self.timestamps.append(now) # Remove expired timestamps while self.timestamps and now - self.timestamps[0] > self.seconds: self.timestamps.popleft() def get_count(self): return len(self.timestamps) # Usage status_msg_counter = TimedCounter(5) status_msg_counter.add_event() time.sleep(4) status_msg_counter.add_event() print(status_msg_counter.get_count()) # Number of events in last 5 minutes ```