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
|
from __future__ import annotations
import functools
from typing import Any, Callable, TypeVar
from sphinx.locale import __
from sphinx.util import logging
from sphinx.util.console import bold # type: ignore[attr-defined]
if False:
from collections.abc import Iterable, Iterator
from types import TracebackType
logger = logging.getLogger(__name__)
def display_chunk(chunk: Any) -> str:
if isinstance(chunk, (list, tuple)):
if len(chunk) == 1:
return str(chunk[0])
return f'{chunk[0]} .. {chunk[-1]}'
return str(chunk)
T = TypeVar('T')
def status_iterator(
iterable: Iterable[T],
summary: str,
color: str = 'darkgreen',
length: int = 0,
verbosity: int = 0,
stringify_func: Callable[[Any], str] = display_chunk,
) -> Iterator[T]:
single_line = verbosity < 1
bold_summary = bold(summary)
if length == 0:
logger.info(bold_summary, nonl=True)
for item in iterable:
logger.info(stringify_func(item) + ' ', nonl=True, color=color)
yield item
else:
for i, item in enumerate(iterable, start=1):
if single_line:
# clear the entire line ('Erase in Line')
logger.info('\x1b[2K', nonl=True)
logger.info(f'{bold_summary}[{i / length: >4.0%}] ', nonl=True) # NoQA: G004
# Emit the string representation of ``item``
logger.info(stringify_func(item), nonl=True, color=color)
# If in single-line mode, emit a carriage return to move the cursor
# to the start of the line.
# If not, emit a newline to move the cursor to the next line.
logger.info('\r' * single_line, nonl=single_line)
yield item
logger.info('')
class SkipProgressMessage(Exception):
pass
class progress_message:
def __init__(self, message: str) -> None:
self.message = message
def __enter__(self) -> None:
logger.info(bold(self.message + '... '), nonl=True)
def __exit__(
self,
typ: type[BaseException] | None,
val: BaseException | None,
tb: TracebackType | None,
) -> bool:
if isinstance(val, SkipProgressMessage):
logger.info(__('skipped'))
if val.args:
logger.info(*val.args)
return True
elif val:
logger.info(__('failed'))
else:
logger.info(__('done'))
return False
def __call__(self, f: Callable) -> Callable:
@functools.wraps(f)
def wrapper(*args: Any, **kwargs: Any) -> Any:
with self:
return f(*args, **kwargs)
return wrapper
|
