import sys IS_WINDOWS: bool = sys.platform == "win32" if IS_WINDOWS: import win32api else: import signal class WhileStopper: is_registered: bool stop_requested: bool def __init__(self) -> None: self.is_registered = False self.stop_requested = False def is_stop_requested(self) -> bool: if not self.is_registered: raise RuntimeError('unexpected stopper status') return self.stop_requested def register(self): if self.is_registered: raise RuntimeError('unexpected stopper status') self.stop_requested = False if IS_WINDOWS: win32api.SetConsoleCtrlHandler(self.__win_handler, True) else: signal.signal(signal.SIGINT, self.__posix_handler) self.is_registered = True def unregister(self): if not self.is_registered: raise RuntimeError('unexpected stopper status') if IS_WINDOWS: win32api.SetConsoleCtrlHandler(None, False) else: signal.signal(signal.SIGINT, signal.SIG_DFL) self.is_registered = False def __win_handler(self, ctrl_type: int) -> bool: self.stop_requested = True return True def __posix_handler(self, signal, frame) -> None: self.stop_requested = True