summaryrefslogtreecommitdiffstats
path: root/sphinx/util/display.py
diff options
context:
space:
mode:
Diffstat (limited to 'sphinx/util/display.py')
-rw-r--r--sphinx/util/display.py94
1 files changed, 94 insertions, 0 deletions
diff --git a/sphinx/util/display.py b/sphinx/util/display.py
new file mode 100644
index 0000000..199119c
--- /dev/null
+++ b/sphinx/util/display.py
@@ -0,0 +1,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