pyrox.dev
1from typing import Callable, Optional
2import time
3
4from .logger import rootlog
5
6
7class PollingConditionFailed(Exception):
8 pass
9
10
11class PollingCondition:
12 condition: Callable[[], bool]
13 seconds_interval: float
14 description: Optional[str]
15
16 last_called: float
17 entered: bool
18
19 def __init__(
20 self,
21 condition: Callable[[], Optional[bool]],
22 seconds_interval: float = 2.0,
23 description: Optional[str] = None,
24 ):
25 self.condition = condition # type: ignore
26 self.seconds_interval = seconds_interval
27
28 if description is None:
29 if condition.__doc__:
30 self.description = condition.__doc__
31 else:
32 self.description = condition.__name__
33 else:
34 self.description = str(description)
35
36 self.last_called = float("-inf")
37 self.entered = False
38
39 def check(self) -> bool:
40 if self.entered or not self.overdue:
41 return True
42
43 with self, rootlog.nested(self.nested_message):
44 rootlog.info(f"Time since last: {time.monotonic() - self.last_called:.2f}s")
45 try:
46 res = self.condition() # type: ignore
47 except Exception:
48 res = False
49 res = res is None or res
50 rootlog.info(self.status_message(res))
51 return res
52
53 def maybe_raise(self) -> None:
54 if not self.check():
55 raise PollingConditionFailed(self.status_message(False))
56
57 def status_message(self, status: bool) -> str:
58 return f"Polling condition {'succeeded' if status else 'failed'}: {self.description}"
59
60 @property
61 def nested_message(self) -> str:
62 nested_message = ["Checking polling condition"]
63 if self.description is not None:
64 nested_message.append(repr(self.description))
65
66 return " ".join(nested_message)
67
68 @property
69 def overdue(self) -> bool:
70 return self.last_called + self.seconds_interval < time.monotonic()
71
72 def __enter__(self) -> None:
73 self.entered = True
74
75 def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore
76 self.entered = False
77 self.last_called = time.monotonic()