Spaces:
Runtime error
Runtime error
File size: 3,785 Bytes
d5dce88 |
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 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
"""
Helper functionality for interoperability with stdlib `logging`.
"""
import logging
import sys
from contextlib import contextmanager
try:
from typing import Iterator, List, Optional, Type # noqa: F401
except ImportError:
pass
from ..std import tqdm as std_tqdm
class _TqdmLoggingHandler(logging.StreamHandler):
def __init__(
self,
tqdm_class=std_tqdm # type: Type[std_tqdm]
):
super(_TqdmLoggingHandler, self).__init__()
self.tqdm_class = tqdm_class
def emit(self, record):
try:
msg = self.format(record)
self.tqdm_class.write(msg, file=self.stream)
self.flush()
except (KeyboardInterrupt, SystemExit):
raise
except: # noqa pylint: disable=bare-except
self.handleError(record)
def _is_console_logging_handler(handler):
return (isinstance(handler, logging.StreamHandler)
and handler.stream in {sys.stdout, sys.stderr})
def _get_first_found_console_logging_handler(handlers):
for handler in handlers:
if _is_console_logging_handler(handler):
return handler
@contextmanager
def logging_redirect_tqdm(
loggers=None, # type: Optional[List[logging.Logger]],
tqdm_class=std_tqdm # type: Type[std_tqdm]
):
# type: (...) -> Iterator[None]
"""
Context manager redirecting console logging to `tqdm.write()`, leaving
other logging handlers (e.g. log files) unaffected.
Parameters
----------
loggers : list, optional
Which handlers to redirect (default: [logging.root]).
tqdm_class : optional
Example
-------
```python
import logging
from tqdm import trange
from tqdm.contrib.logging import logging_redirect_tqdm
LOG = logging.getLogger(__name__)
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
with logging_redirect_tqdm():
for i in trange(9):
if i == 4:
LOG.info("console logging redirected to `tqdm.write()`")
# logging restored
```
"""
if loggers is None:
loggers = [logging.root]
original_handlers_list = [logger.handlers for logger in loggers]
try:
for logger in loggers:
tqdm_handler = _TqdmLoggingHandler(tqdm_class)
orig_handler = _get_first_found_console_logging_handler(logger.handlers)
if orig_handler is not None:
tqdm_handler.setFormatter(orig_handler.formatter)
tqdm_handler.stream = orig_handler.stream
logger.handlers = [
handler for handler in logger.handlers
if not _is_console_logging_handler(handler)] + [tqdm_handler]
yield
finally:
for logger, original_handlers in zip(loggers, original_handlers_list):
logger.handlers = original_handlers
@contextmanager
def tqdm_logging_redirect(
*args,
# loggers=None, # type: Optional[List[logging.Logger]]
# tqdm=None, # type: Optional[Type[tqdm.tqdm]]
**kwargs
):
# type: (...) -> Iterator[None]
"""
Convenience shortcut for:
```python
with tqdm_class(*args, **tqdm_kwargs) as pbar:
with logging_redirect_tqdm(loggers=loggers, tqdm_class=tqdm_class):
yield pbar
```
Parameters
----------
tqdm_class : optional, (default: tqdm.std.tqdm).
loggers : optional, list.
**tqdm_kwargs : passed to `tqdm_class`.
"""
tqdm_kwargs = kwargs.copy()
loggers = tqdm_kwargs.pop('loggers', None)
tqdm_class = tqdm_kwargs.pop('tqdm_class', std_tqdm)
with tqdm_class(*args, **tqdm_kwargs) as pbar:
with logging_redirect_tqdm(loggers=loggers, tqdm_class=tqdm_class):
yield pbar
|