96 lines
2.5 KiB
Python
96 lines
2.5 KiB
Python
import threading
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Callable
|
|
|
|
|
|
class Clock:
|
|
|
|
millis : int
|
|
increment : int
|
|
|
|
on_update : Callable[[str], None]
|
|
on_terminated : Callable[[], None]
|
|
|
|
def __init__(self, value: int, increment: int) -> None:
|
|
self.millis = value * 1000
|
|
self.increment = increment * 1000
|
|
|
|
self._run_event = threading.Event()
|
|
self._terminate_event = threading.Event()
|
|
self._lock = threading.Lock()
|
|
|
|
self.thread = threading.Thread(target=self._update_clock, daemon=True)
|
|
|
|
def _compute_seconds(self) -> int:
|
|
return (self.millis // 1000) % 60
|
|
|
|
def _compute_minutes(self) -> int:
|
|
return self.millis // 60000
|
|
|
|
def _add_increment(self) -> None:
|
|
self.millis += self.increment
|
|
|
|
def _update_clock(self):
|
|
while not self._terminate_event.is_set():
|
|
self._run_event.wait()
|
|
|
|
with self._lock:
|
|
if self.millis <= 0:
|
|
self.millis = 0
|
|
self._run_event.clear()
|
|
self.__notify_update()
|
|
break
|
|
|
|
self.millis -= 1
|
|
if self.millis % 1000 == 0:
|
|
self.__notify_update()
|
|
|
|
time.sleep(0.001)
|
|
|
|
def __notify_update(self):
|
|
print(self.clock_to_str())
|
|
self.on_update(self.clock_to_str())
|
|
|
|
def __notify_terminated(self):
|
|
self.on_terminated()
|
|
|
|
def set_on_update(self, callback : Callable[[str], None]):
|
|
self.on_update = callback
|
|
|
|
def set_on_terminated(self, callback: Callable[[], None]):
|
|
self.on_terminated = callback
|
|
|
|
def is_running(self) -> bool:
|
|
return self._run_event.is_set()
|
|
|
|
def start(self) -> None:
|
|
self._run_event.set()
|
|
if not self.thread.is_alive():
|
|
self.thread = threading.Thread(target=self._update_clock, daemon=True)
|
|
self.thread.start()
|
|
|
|
def stop(self) -> None:
|
|
if self._run_event.is_set():
|
|
with self._lock:
|
|
self._add_increment()
|
|
self._run_event.clear()
|
|
|
|
def terminate(self) -> None:
|
|
self._terminate_event.set()
|
|
self._run_event.set()
|
|
self.thread.join()
|
|
|
|
def clock_to_str(self) -> str:
|
|
with self._lock:
|
|
minutes = self._compute_minutes()
|
|
seconds = self._compute_seconds()
|
|
return f"{minutes:02d}:{seconds:02d}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
clock = Clock(600, 0)
|
|
clock.start()
|
|
while True:
|
|
time.sleep(1)
|
|
print(clock.clock_to_str()) |