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