(self, key, timeout=1)
| 102 | |
| 103 | # TODO: this should be implemented in params |
| 104 | def _read_param(self, key, timeout=1): |
| 105 | ret = None |
| 106 | start_time = time.monotonic() |
| 107 | while ret is None: |
| 108 | ret = self.params.get(key) |
| 109 | if time.monotonic() - start_time > timeout: |
| 110 | break |
| 111 | time.sleep(0.01) |
| 112 | return ret |
| 113 | |
| 114 | def _wait_for_update(self, timeout=30, clear_param=False): |
| 115 | if clear_param: |
no test coverage detected