A class to send timer-based events. Args: event_target: The object which will receive the timer events. interval: The time between timer events, in seconds. name: A name to assign the event (for debugging). callback: An optional callback to invoke when the event
| 30 | |
| 31 | @rich_repr |
| 32 | class Timer: |
| 33 | """A class to send timer-based events. |
| 34 | |
| 35 | Args: |
| 36 | event_target: The object which will receive the timer events. |
| 37 | interval: The time between timer events, in seconds. |
| 38 | name: A name to assign the event (for debugging). |
| 39 | callback: An optional callback to invoke when the event is handled. |
| 40 | repeat: The number of times to repeat the timer, or None to repeat forever. |
| 41 | skip: Enable skipping of scheduled events that couldn't be sent in time. |
| 42 | pause: Start the timer paused. |
| 43 | """ |
| 44 | |
| 45 | _timer_count: int = 1 |
| 46 | |
| 47 | def __init__( |
| 48 | self, |
| 49 | event_target: MessageTarget, |
| 50 | interval: float, |
| 51 | *, |
| 52 | name: str | None = None, |
| 53 | callback: TimerCallback | None = None, |
| 54 | repeat: int | None = None, |
| 55 | skip: bool = True, |
| 56 | pause: bool = False, |
| 57 | ) -> None: |
| 58 | self._target_repr = repr(event_target) |
| 59 | self._target = weakref.ref(event_target) |
| 60 | self._interval = interval |
| 61 | self.name = f"Timer#{self._timer_count}" if name is None else name |
| 62 | self._timer_count += 1 |
| 63 | self._callback = callback |
| 64 | self._repeat = repeat |
| 65 | self._skip = skip |
| 66 | self._task: Task | None = None |
| 67 | self._reset: bool = False |
| 68 | self._original_pause = pause |
| 69 | |
| 70 | @cached_property |
| 71 | def _active(self) -> Event: |
| 72 | event = Event() |
| 73 | if not self._original_pause: |
| 74 | event.set() |
| 75 | return event |
| 76 | |
| 77 | def __rich_repr__(self) -> Result: |
| 78 | yield self._interval |
| 79 | yield "name", self.name |
| 80 | yield "repeat", self._repeat, None |
| 81 | |
| 82 | @property |
| 83 | def target(self) -> MessageTarget: |
| 84 | target = self._target() |
| 85 | if target is None: |
| 86 | raise EventTargetGone() |
| 87 | return target |
| 88 | |
| 89 | def _start(self) -> None: |
no outgoing calls
no test coverage detected
searching dependent graphs…