from threading import Event, Lock from typing import Optional class DblEvent: """ Overview: A double event object, can open and close. Bases on 2 event objects """ def __init__(self, opened: bool = False): """ Overview: Constructor of `DblEvent` Arguments: - opened (:obj:`bool`): Initial status (`True` means open, `False` means close, default is `False`) """ self.__open_event = Event() self.__close_event = Event() self.__lock = Lock() if opened: self.__open_event.set() else: self.__close_event.set() def wait_for_open(self, timeout: Optional[float] = None): """ Overview: Wait until the event is opened Arguments: - timeout (:obj:`Optional[float]`): Waiting time out in seconds """ self.__open_event.wait(timeout=timeout) def wait_for_close(self, timeout: Optional[float] = None): """ Overview: Wait until the event is closed Arguments: - timeout (:obj:`Optional[float]`): Waiting time out in seconds """ self.__close_event.wait(timeout=timeout) def open(self): """ Overview: Open this event """ with self.__lock: self.__open_event.set() self.__close_event.clear() def close(self): """ Overview: Close this event """ with self.__lock: self.__close_event.set() self.__open_event.clear() def is_open(self) -> bool: """ Overview: Get if the event is opened Returns: - opened (:obj:`bool`): The event is opened or not """ with self.__lock: return self.__open_event.is_set() def is_close(self) -> bool: """ Overview: Get if the event is closed Returns: - opened (:obj:`bool`): The event is closed or not """ with self.__lock: return self.__close_event.is_set()