Disable the given `tqdm_instance` upon `write()` or `flush()` errors.
| 181 | |
| 182 | |
| 183 | class DisableOnWriteError(ObjectWrapper): |
| 184 | """ |
| 185 | Disable the given `tqdm_instance` upon `write()` or `flush()` errors. |
| 186 | """ |
| 187 | @staticmethod |
| 188 | def disable_on_exception(tqdm_instance, func): |
| 189 | """ |
| 190 | Quietly set `tqdm_instance.miniters=inf` if `func` raises `errno=5`. |
| 191 | """ |
| 192 | tqdm_instance = proxy(tqdm_instance) |
| 193 | |
| 194 | def inner(*args, **kwargs): |
| 195 | try: |
| 196 | return func(*args, **kwargs) |
| 197 | except OSError as e: |
| 198 | if e.errno != 5: |
| 199 | raise |
| 200 | try: |
| 201 | tqdm_instance.miniters = float('inf') |
| 202 | except ReferenceError: |
| 203 | pass |
| 204 | except ValueError as e: |
| 205 | if 'closed' not in str(e): |
| 206 | raise |
| 207 | try: |
| 208 | tqdm_instance.miniters = float('inf') |
| 209 | except ReferenceError: |
| 210 | pass |
| 211 | return inner |
| 212 | |
| 213 | def __init__(self, wrapped, tqdm_instance): # noqa: B042 |
| 214 | super().__init__(wrapped) |
| 215 | if hasattr(wrapped, 'write'): |
| 216 | self.wrapper_setattr( |
| 217 | 'write', self.disable_on_exception(tqdm_instance, wrapped.write)) |
| 218 | if hasattr(wrapped, 'flush'): |
| 219 | self.wrapper_setattr( |
| 220 | 'flush', self.disable_on_exception(tqdm_instance, wrapped.flush)) |
| 221 | |
| 222 | def __eq__(self, other): |
| 223 | return self._wrapped == getattr(other, '_wrapped', other) |
| 224 | |
| 225 | |
| 226 | class CallbackIOWrapper(ObjectWrapper): |