File size: 2,163 Bytes
079c32c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
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()
|