Coverage for yuio / io.py: 94%
1048 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-05 11:41 +0000
1# Yuio project, MIT license.
2#
3# https://github.com/taminomara/yuio/
4#
5# You're free to copy this file to your project and edit it for your needs,
6# just keep this copyright line please :3
8"""
9This module implements user-friendly console input and output.
11Configuration
12-------------
14Yuio configures itself upon import using environment variables:
16- ``FORCE_NO_COLOR``: disable colored output,
17- ``FORCE_COLOR``: enable colored output.
19The only thing it doesn't do automatically is wrapping :data:`sys.stdout`
20and :data:`sys.stderr` into safe proxies. The :mod:`yuio.app` CLI builder
21will do it for you, though, so you don't need to worry about it.
23.. autofunction:: setup
25To introspect the current state of Yuio's initialization, use the following functions:
27.. autofunction:: get_term
29.. autofunction:: get_theme
31.. autofunction:: wrap_streams
33.. autofunction:: restore_streams
35.. autofunction:: streams_wrapped
37.. autofunction:: orig_stderr
39.. autofunction:: orig_stdout
42Printing messages
43-----------------
45To print messages for the user, there are functions with an interface similar
46to the one from :mod:`logging`. Messages are highlighted using
47:ref:`color tags <color-tags>` and formatted using either
48:ref:`%-formatting or template strings <percent-format>`.
50.. autofunction:: info
52.. autofunction:: warning
54.. autofunction:: success
56.. autofunction:: failure
58.. autofunction:: failure_with_tb
60.. autofunction:: error
62.. autofunction:: error_with_tb
64.. autofunction:: heading
66.. autofunction:: md
68.. autofunction:: hl
70.. autofunction:: br
72.. autofunction:: hr
74.. autofunction:: raw
77.. _percent-format:
79Formatting the output
80---------------------
82Yuio supports `printf-style formatting`__, similar to :mod:`logging`. If you're using
83Python 3.14 or later, you can also use `template strings`__.
85__ https://docs.python.org/3/library/stdtypes.html#printf-style-string-formatting
86__ https://docs.python.org/3/library/string.html#template-strings
88.. invisible-code-block: python
90 config = ...
92.. tab-set::
93 :sync-group: formatting-method
95 .. tab-item:: Printf-style formatting
96 :sync: printf
98 ``%s`` and ``%r`` specifiers are handled to respect colors and `rich repr protocol`__.
99 Additionally, they allow specifying flags to control whether rendered values should
100 be highlighted, and should they be rendered in multiple lines:
102 __ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
104 - ``#`` enables colors in repr (i.e. ``%#r``);
105 - ``+`` splits repr into multiple lines (i.e. ``%+r``, ``%#+r``).
107 .. code-block:: python
109 yuio.io.info("Loaded config: %#+r", config)
111 .. tab-item:: Template strings
112 :sync: template
114 When formatting template strings, default format specification is extended
115 to respect colors and `rich repr protocol`__. Additionally, it allows
116 specifying flags to control whether rendered values should be highlighted,
117 and should they be rendered in multiple lines:
119 __ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
121 - ``#`` enables colors in repr (i.e. ``{var:#}``);
122 - ``+`` splits repr into multiple lines (i.e. ``{var:+}``, ``{var:#+}``);
123 - unless explicit conversion is given (i.e. ``!s``, ``!r``, or ``!a``),
124 this format specification applies to objects that don't define
125 custom ``__format__`` method;
126 - full format specification is available :ref:`here <t-string-spec>`.
128 .. code-block:: python
130 yuio.io.info(t"Loaded config: {config!r:#+}")
132 .. note::
134 The formatting algorithm is as follows:
136 1. if formatting conversion is specified (i.e. ``!s``, ``!r``, or ``!a``),
137 the object is passed to
138 :meth:`ReprContext.convert() <yuio.string.ReprContext.convert>`;
139 2. otherwise, if object defines custom ``__format__`` method,
140 this method is used;
141 3. otherwise, we fall back to
142 :meth:`ReprContext.convert() <yuio.string.ReprContext.convert>`
143 with assumed conversion flag ``"s"``.
145To support highlighted formatting, define ``__colorized_str__``
146or ``__colorized_repr__`` on your class. See :ref:`pretty-protocol` for implementation
147details.
149To support rich repr protocol, define function ``__rich_repr__`` on your class.
150This method should return an iterable of tuples, as described in Rich__ documentation.
152__ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
155.. _color-tags:
157Coloring the output
158-------------------
160By default, all messages are colored according to their level (i.e. which function
161you use to print them).
163If you need inline colors, you can use special tags in your log messages:
165.. code-block:: python
167 yuio.io.info("Using the <c code>code</c> tag.")
169You can combine multiple colors in the same tag:
171.. code-block:: python
173 yuio.io.info("<c bold green>Success!</c>")
175Only tags that appear in the message itself are processed:
177.. code-block:: python
179 yuio.io.info("Tags in this message --> %s are printed as-is", "<c color>")
181For highlighting inline code, Yuio supports parsing CommonMark's backticks:
183.. code-block:: python
185 yuio.io.info("Using the `backticks`.")
186 yuio.io.info("Using the `` nested `backticks` ``")
188List of all tags that are available by default:
190- ``code``, ``note``, ``path``: highlights,
191- ``bold``, ``b``, ``dim``, ``d``, ``italic``, ``i``,
192 ``underline``, ``u``, ``inverse``: font style,
193- ``normal``, ``muted``, ``red``, ``green``, ``yellow``, ``blue``,
194 ``magenta``, ``cyan``: colors.
196You can add more with :doc:`themes </internals/theme>`.
199Formatting utilities
200--------------------
202There are several :ref:`formatting utilities <formatting-utilities>` defined
203in :mod:`yuio.string` and re-exported in :mod:`yuio.io`. These utilities
204perform various formatting tasks when converted to strings, allowing you to lazily
205build more complex messages.
208Indicating progress
209-------------------
211You can use the :class:`Task` class to indicate status and progress
212of some task:
214.. autoclass:: Task
216 .. automethod:: progress
218 .. automethod:: progress_size
220 .. automethod:: progress_scale
222 .. automethod:: iter
224 .. automethod:: comment
226 .. automethod:: done
228 .. automethod:: error
230 .. automethod:: subtask
233Querying user input
234-------------------
236If you need to get something from the user, :func:`ask` is the way to do it.
238.. autofunction:: ask
240.. autofunction:: wait_for_user
242You can also prompt the user to edit something with the :func:`edit` function:
244.. autofunction:: edit
246.. autofunction:: detect_editor
248All of these functions throw an error if something goes wrong:
250.. autoclass:: UserIoError
253Suspending the output
254---------------------
256You can temporarily disable printing of tasks and messages
257using the :class:`SuspendOutput` context manager.
259.. autoclass:: SuspendOutput
261 .. automethod:: resume
263 .. automethod:: info
265 .. automethod:: warning
267 .. automethod:: success
269 .. automethod:: failure
271 .. automethod:: failure_with_tb
273 .. automethod:: error
275 .. automethod:: error_with_tb
277 .. automethod:: heading
279 .. automethod:: md
281 .. automethod:: hl
283 .. automethod:: br
285 .. automethod:: hr
287 .. automethod:: raw
290Python's `logging` and yuio
291---------------------------
293If you want to direct messages from the :mod:`logging` to Yuio,
294you can add a :class:`Handler`:
296.. autoclass:: Handler
298.. autoclass:: Formatter
301Helper types
302------------
304.. type:: ExcInfo
305 :canonical: tuple[type[BaseException] | None, BaseException | None, types.TracebackType | None]
307 Exception information as returned by :func:`sys.exc_info`.
310Re-imports
311----------
313.. type:: And
314 :no-index:
316 Alias of :obj:`yuio.string.And`.
318.. type:: ColorizedString
319 :no-index:
321 Alias of :obj:`yuio.string.ColorizedString`.
323.. type:: Format
324 :no-index:
326 Alias of :obj:`yuio.string.Format`.
328.. type:: Hl
329 :no-index:
331 Alias of :obj:`yuio.string.Hl`.
333.. type:: Hr
334 :no-index:
336 Alias of :obj:`yuio.string.Hr`.
338.. type:: Indent
339 :no-index:
341 Alias of :obj:`yuio.string.Indent`.
343.. type:: JoinRepr
344 :no-index:
346 Alias of :obj:`yuio.string.JoinRepr`.
348.. type:: JoinStr
349 :no-index:
351 Alias of :obj:`yuio.string.JoinStr`.
353.. type:: Link
354 :no-index:
356 Alias of :obj:`yuio.string.Link`.
358.. type:: Md
359 :no-index:
361 Alias of :obj:`yuio.string.Md`.
363.. type:: Or
364 :no-index:
366 Alias of :obj:`yuio.string.Or`.
368.. type:: Repr
369 :no-index:
371 Alias of :obj:`yuio.string.Repr`.
373.. type:: Stack
374 :no-index:
376 Alias of :obj:`yuio.string.Stack`.
378.. type:: TypeRepr
379 :no-index:
381 Alias of :obj:`yuio.string.TypeRepr`.
383.. type:: WithBaseColor
384 :no-index:
386 Alias of :obj:`yuio.string.WithBaseColor`.
388.. type:: Wrap
389 :no-index:
391 Alias of :obj:`yuio.string.Wrap`.
394"""
396from __future__ import annotations
398import abc
399import atexit
400import enum
401import functools
402import logging
403import math
404import os
405import re
406import shutil
407import string
408import subprocess
409import sys
410import tempfile
411import threading
412import time
413import traceback
414import types
415from logging import LogRecord
417import yuio.color
418import yuio.md
419import yuio.parse
420import yuio.string
421import yuio.term
422import yuio.theme
423import yuio.widget
424from yuio.string import (
425 And,
426 ColorizedString,
427 Format,
428 Hl,
429 Hr,
430 Indent,
431 JoinRepr,
432 JoinStr,
433 Link,
434 Md,
435 Or,
436 Repr,
437 Stack,
438 TypeRepr,
439 WithBaseColor,
440 Wrap,
441)
442from yuio.util import dedent as _dedent
444import yuio._typing_ext as _tx
445from typing import TYPE_CHECKING
447if TYPE_CHECKING:
448 import typing_extensions as _t
449else:
450 from yuio import _typing as _t
452__all__ = [
453 "And",
454 "ColorizedString",
455 "ExcInfo",
456 "Format",
457 "Formatter",
458 "Handler",
459 "Hl",
460 "Hr",
461 "Indent",
462 "JoinRepr",
463 "JoinStr",
464 "Link",
465 "Md",
466 "Or",
467 "Repr",
468 "Stack",
469 "SuspendOutput",
470 "Task",
471 "TypeRepr",
472 "UserIoError",
473 "WithBaseColor",
474 "Wrap",
475 "ask",
476 "br",
477 "detect_editor",
478 "edit",
479 "error",
480 "error_with_tb",
481 "failure",
482 "failure_with_tb",
483 "get_term",
484 "get_theme",
485 "heading",
486 "hl",
487 "hr",
488 "info",
489 "make_repr_context",
490 "md",
491 "orig_stderr",
492 "orig_stdout",
493 "raw",
494 "restore_streams",
495 "setup",
496 "streams_wrapped",
497 "success",
498 "wait_for_user",
499 "warning",
500 "wrap_streams",
501]
503T = _t.TypeVar("T")
504M = _t.TypeVar("M", default=_t.Never)
505S = _t.TypeVar("S", default=str)
507ExcInfo: _t.TypeAlias = tuple[
508 type[BaseException] | None,
509 BaseException | None,
510 types.TracebackType | None,
511]
512"""
513Exception information as returned by :func:`sys.exc_info`.
515"""
518_IO_LOCK = threading.Lock()
519_IO_MANAGER: _IoManager | None = None
520_STREAMS_WRAPPED: bool = False
521_ORIG_STDERR: _t.TextIO | None = None
522_ORIG_STDOUT: _t.TextIO | None = None
525def _manager() -> _IoManager:
526 global _IO_MANAGER
528 if _IO_MANAGER is None:
529 with _IO_LOCK:
530 if _IO_MANAGER is None:
531 _IO_MANAGER = _IoManager()
532 return _IO_MANAGER
535class UserIoError(yuio.PrettyException, IOError):
536 """
537 Raised when interaction with user fails.
539 """
542def setup(
543 *,
544 term: yuio.term.Term | None = None,
545 theme: (
546 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
547 ) = None,
548 wrap_stdio: bool = True,
549):
550 """
551 Initial setup of the logging facilities.
553 :param term:
554 terminal that will be used for output.
556 If not passed, the global terminal is not re-configured;
557 the default is to use a term attached to :data:`sys.stderr`.
558 :param theme:
559 either a theme that will be used for output, or a theme constructor that takes
560 a :class:`~yuio.term.Term` and returns a theme.
562 If not passed, the global theme is not re-configured; the default is to use
563 :class:`yuio.theme.DefaultTheme` then.
564 :param wrap_stdio:
565 if set to :data:`True`, wraps :data:`sys.stdout` and :data:`sys.stderr`
566 in a special wrapper that ensures better interaction
567 with Yuio's progress bars and widgets.
569 .. note::
571 If you're working with some other library that wraps :data:`sys.stdout`
572 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
574 .. _colorama: https://github.com/tartley/colorama
576 .. warning::
578 This operation is not atomic. Call this function before creating new threads
579 that use :mod:`yuio.io` or output streams to avoid race conditions.
581 """
583 global _IO_MANAGER
585 if not (manager := _IO_MANAGER):
586 with _IO_LOCK:
587 if not (manager := _IO_MANAGER):
588 _IO_MANAGER = _IoManager(term, theme)
589 if manager is not None:
590 manager.setup(term, theme)
592 if wrap_stdio:
593 wrap_streams()
596def get_term() -> yuio.term.Term:
597 """
598 Get the global instance of :class:`~yuio.term.Term` that is used
599 with :mod:`yuio.io`.
601 If global setup wasn't performed, this function implicitly performs it.
603 :returns:
604 Instance of :class:`~yuio.term.Term` that's used to print messages and tasks.
606 """
608 return _manager().term
611def get_theme() -> yuio.theme.Theme:
612 """
613 Get the global instance of :class:`~yuio.theme.Theme`
614 that is used with :mod:`yuio.io`.
616 If global setup wasn't performed, this function implicitly performs it.
618 :returns:
619 Instance of :class:`~yuio.theme.Theme` that's used to format messages and tasks.
621 """
623 return _manager().theme
626def make_repr_context(
627 *,
628 term: yuio.term.Term | None = None,
629 file: _t.TextIO | None = None,
630 to_stdout: bool = False,
631 to_stderr: bool = False,
632 theme: yuio.theme.Theme | None = None,
633 multiline: bool | None = None,
634 highlighted: bool | None = None,
635 max_depth: int | None = None,
636 width: int | None = None,
637) -> yuio.string.ReprContext:
638 """
639 Create new :class:`~yuio.string.ReprContext` for the given term and theme.
641 :param term:
642 terminal where to print this message. If not given, terminal from
643 :func:`get_term` is used.
644 :param file:
645 shortcut for creating non-interactive `term` for a file.
646 :param to_stdout:
647 shortcut for setting `term` to ``stdout``.
648 :param to_stderr:
649 shortcut for setting `term` to ``stderr``.
650 :param theme:
651 theme used to format the message. If not given, theme from
652 :func:`get_theme` is used.
653 :param multiline:
654 sets initial value for
655 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
656 Default is :data:`False`.
657 :param highlighted:
658 sets initial value for
659 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
660 Default is :data:`False`.
661 :param max_depth:
662 sets initial value for
663 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
664 Default is :data:`False`.
665 :param width:
666 sets initial value for
667 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
668 If not given, uses current terminal width or :attr:`Theme.fallback_width`
669 depending on whether `term` is attached to a TTY device and whether colors
670 are supported by the target terminal.
672 """
674 if (term is not None) + to_stdout + to_stderr > 1:
675 raise TypeError("term, to_stdout, to_stderr can't be given together")
677 manager = _manager()
679 theme = manager.theme
680 if term is None:
681 if file is not None:
682 term = yuio.term.Term(
683 file, sys.stdin, is_unicode=yuio.term.stream_is_unicode(file)
684 )
685 elif to_stdout:
686 term = manager.out_term
687 elif to_stderr:
688 term = manager.err_term
689 else:
690 term = manager.term
691 if width is None and (term.ostream_is_tty or term.supports_colors):
692 width = manager.rc.canvas_width
694 return yuio.string.ReprContext(
695 term=term,
696 theme=theme,
697 multiline=multiline,
698 highlighted=highlighted,
699 max_depth=max_depth,
700 width=width,
701 )
704def wrap_streams():
705 """
706 Wrap :data:`sys.stdout` and :data:`sys.stderr` so that they honor
707 Yuio tasks and widgets. If strings are already wrapped, this function
708 has no effect.
710 .. note::
712 If you're working with some other library that wraps :data:`sys.stdout`
713 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
715 .. seealso::
717 :func:`setup`.
719 .. _colorama: https://github.com/tartley/colorama
721 """
723 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
725 if _STREAMS_WRAPPED:
726 return
728 with _IO_LOCK:
729 if _STREAMS_WRAPPED: # pragma: no cover
730 return
732 if yuio.term._output_is_tty(sys.stdout):
733 _ORIG_STDOUT, sys.stdout = sys.stdout, _WrappedOutput(sys.stdout)
734 if yuio.term._output_is_tty(sys.stderr):
735 _ORIG_STDERR, sys.stderr = sys.stderr, _WrappedOutput(sys.stderr)
736 _STREAMS_WRAPPED = True
738 atexit.register(restore_streams)
741def restore_streams():
742 """
743 Restore wrapped streams. If streams weren't wrapped, this function
744 has no effect.
746 .. seealso::
748 :func:`wrap_streams`, :func:`setup`
750 """
752 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
754 if not _STREAMS_WRAPPED:
755 return
757 with _IO_LOCK:
758 if not _STREAMS_WRAPPED: # pragma: no cover
759 return
761 if _ORIG_STDOUT is not None:
762 sys.stdout = _ORIG_STDOUT
763 _ORIG_STDOUT = None
764 if _ORIG_STDERR is not None:
765 sys.stderr = _ORIG_STDERR
766 _ORIG_STDERR = None
767 _STREAMS_WRAPPED = False
770def streams_wrapped() -> bool:
771 """
772 Check if :data:`sys.stdout` and :data:`sys.stderr` are wrapped.
773 See :func:`setup`.
775 :returns:
776 :data:`True` is streams are currently wrapped, :data:`False` otherwise.
778 """
780 return _STREAMS_WRAPPED
783def orig_stderr() -> _t.TextIO:
784 """
785 Return the original :data:`sys.stderr` before wrapping.
787 """
789 return _ORIG_STDERR or sys.stderr
792def orig_stdout() -> _t.TextIO:
793 """
794 Return the original :data:`sys.stdout` before wrapping.
796 """
798 return _ORIG_STDOUT or sys.stdout
801@_t.overload
802def info(msg: _t.LiteralString, /, *args, **kwargs): ...
803@_t.overload
804def info(msg: yuio.string.ToColorable, /, **kwargs): ...
805def info(msg: yuio.string.ToColorable, /, *args, **kwargs):
806 """info(msg: typing.LiteralString, /, *args, **kwargs)
807 info(msg: ~string.templatelib.Template, /, **kwargs) ->
808 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
810 Print an info message.
812 :param msg:
813 message to print.
814 :param args:
815 arguments for ``%``\\ -formatting the message.
816 :param kwargs:
817 any additional keyword arguments will be passed to :func:`raw`.
819 """
821 msg_colorable = yuio.string._to_colorable(msg, args)
822 kwargs.setdefault("tag", "info")
823 kwargs.setdefault("wrap", True)
824 kwargs.setdefault("add_newline", True)
825 raw(msg_colorable, **kwargs)
828@_t.overload
829def warning(msg: _t.LiteralString, /, *args, **kwargs): ...
830@_t.overload
831def warning(msg: yuio.string.ToColorable, /, **kwargs): ...
832def warning(msg: yuio.string.ToColorable, /, *args, **kwargs):
833 """warning(msg: typing.LiteralString, /, *args, **kwargs)
834 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
835 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
837 Print a warning message.
839 :param msg:
840 message to print.
841 :param args:
842 arguments for ``%``\\ -formatting the message.
843 :param kwargs:
844 any additional keyword arguments will be passed to :func:`raw`.
846 """
848 msg_colorable = yuio.string._to_colorable(msg, args)
849 kwargs.setdefault("tag", "warning")
850 kwargs.setdefault("wrap", True)
851 kwargs.setdefault("add_newline", True)
852 raw(msg_colorable, **kwargs)
855@_t.overload
856def success(msg: _t.LiteralString, /, *args, **kwargs): ...
857@_t.overload
858def success(msg: yuio.string.ToColorable, /, **kwargs): ...
859def success(msg: yuio.string.ToColorable, /, *args, **kwargs):
860 """success(msg: typing.LiteralString, /, *args, **kwargs)
861 success(msg: ~string.templatelib.Template, /, **kwargs) ->
862 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
864 Print a success message.
866 :param msg:
867 message to print.
868 :param args:
869 arguments for ``%``\\ -formatting the message.
870 :param kwargs:
871 any additional keyword arguments will be passed to :func:`raw`.
873 """
875 msg_colorable = yuio.string._to_colorable(msg, args)
876 kwargs.setdefault("tag", "success")
877 kwargs.setdefault("wrap", True)
878 kwargs.setdefault("add_newline", True)
879 raw(msg_colorable, **kwargs)
882@_t.overload
883def error(msg: _t.LiteralString, /, *args, **kwargs): ...
884@_t.overload
885def error(msg: yuio.string.ToColorable, /, **kwargs): ...
886def error(msg: yuio.string.ToColorable, /, *args, **kwargs):
887 """error(msg: typing.LiteralString, /, *args, **kwargs)
888 error(msg: ~string.templatelib.Template, /, **kwargs) ->
889 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
891 Print an error message.
893 :param msg:
894 message to print.
895 :param args:
896 arguments for ``%``\\ -formatting the message.
897 :param kwargs:
898 any additional keyword arguments will be passed to :func:`raw`.
900 """
902 msg_colorable = yuio.string._to_colorable(msg, args)
903 kwargs.setdefault("tag", "error")
904 kwargs.setdefault("wrap", True)
905 kwargs.setdefault("add_newline", True)
906 raw(msg_colorable, **kwargs)
909@_t.overload
910def error_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
911@_t.overload
912def error_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
913def error_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
914 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
915 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
916 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
918 Print an error message and capture the current exception.
920 Call this function in the ``except`` clause of a ``try`` block
921 or in an ``__exit__`` function of a context manager to attach
922 current exception details to the log message.
924 :param msg:
925 message to print.
926 :param args:
927 arguments for ``%``\\ -formatting the message.
928 :param kwargs:
929 any additional keyword arguments will be passed to :func:`raw`.
931 """
933 msg_colorable = yuio.string._to_colorable(msg, args)
934 kwargs.setdefault("tag", "error")
935 kwargs.setdefault("wrap", True)
936 kwargs.setdefault("add_newline", True)
937 kwargs.setdefault("exc_info", True)
938 raw(msg_colorable, **kwargs)
941@_t.overload
942def failure(msg: _t.LiteralString, /, *args, **kwargs): ...
943@_t.overload
944def failure(msg: yuio.string.ToColorable, /, **kwargs): ...
945def failure(msg: yuio.string.ToColorable, /, *args, **kwargs):
946 """failure(msg: typing.LiteralString, /, *args, **kwargs)
947 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
948 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
950 Print a failure message.
952 :param msg:
953 message to print.
954 :param args:
955 arguments for ``%``\\ -formatting the message.
956 :param kwargs:
957 any additional keyword arguments will be passed to :func:`raw`.
959 """
961 msg_colorable = yuio.string._to_colorable(msg, args)
962 kwargs.setdefault("tag", "failure")
963 kwargs.setdefault("wrap", True)
964 kwargs.setdefault("add_newline", True)
965 raw(msg_colorable, **kwargs)
968@_t.overload
969def failure_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
970@_t.overload
971def failure_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
972def failure_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
973 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
974 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
975 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
977 Print a failure message and capture the current exception.
979 Call this function in the ``except`` clause of a ``try`` block
980 or in an ``__exit__`` function of a context manager to attach
981 current exception details to the log message.
983 :param msg:
984 message to print.
985 :param args:
986 arguments for ``%``\\ -formatting the message.
987 :param kwargs:
988 any additional keyword arguments will be passed to :func:`raw`.
990 """
992 msg_colorable = yuio.string._to_colorable(msg, args)
993 kwargs.setdefault("tag", "failure")
994 kwargs.setdefault("wrap", True)
995 kwargs.setdefault("add_newline", True)
996 kwargs.setdefault("exc_info", True)
997 raw(msg_colorable, **kwargs)
1000@_t.overload
1001def heading(msg: _t.LiteralString, /, *args, level: int = 1, **kwargs): ...
1002@_t.overload
1003def heading(msg: yuio.string.ToColorable, /, *, level: int = 1, **kwargs): ...
1004def heading(msg: yuio.string.ToColorable, /, *args, level: int = 1, **kwargs):
1005 """heading(msg: typing.LiteralString, /, *args, level: int = 1, **kwargs)
1006 heading(msg: ~string.templatelib.Template, /, *, level: int = 1, **kwargs) ->
1007 heading(msg: ~yuio.string.ToColorable, /, *, level: int = 1, **kwargs) ->
1009 Print a heading message.
1011 :param msg:
1012 message to print.
1013 :param args:
1014 arguments for ``%``\\ -formatting the message.
1015 :param level:
1016 level of the heading.
1017 :param kwargs:
1018 any additional keyword arguments will be passed to :func:`raw`.
1020 """
1022 msg_colorable = yuio.string._to_colorable(msg, args)
1023 level = kwargs.pop("level", 1)
1024 kwargs.setdefault("heading", True)
1025 kwargs.setdefault("tag", f"heading/{level}")
1026 kwargs.setdefault("wrap", True)
1027 kwargs.setdefault("add_newline", True)
1028 raw(msg_colorable, **kwargs)
1031@_t.overload
1032def md(
1033 msg: _t.LiteralString,
1034 /,
1035 *args,
1036 dedent: bool = True,
1037 allow_headings: bool = True,
1038 **kwargs,
1039): ...
1040@_t.overload
1041def md(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs): ...
1042def md(msg: str, /, *args, dedent: bool = True, allow_headings: bool = True, **kwargs):
1043 """md(msg: typing.LiteralString, /, *args, dedent: bool = True, allow_headings: bool = True, **kwargs)
1044 md(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs) ->
1046 Print a markdown-formatted text.
1048 Yuio supports all CommonMark block markup except tables. Inline markup is limited
1049 to backticks and color tags. See :mod:`yuio.md` for more info.
1051 :param msg:
1052 message to print.
1053 :param args:
1054 arguments for ``%``\\ -formatting the message.
1055 :param dedent:
1056 whether to remove leading indent from markdown.
1057 :param allow_headings:
1058 whether to render headings as actual headings or as paragraphs.
1059 :param kwargs:
1060 any additional keyword arguments will be passed to :func:`raw`.
1062 """
1064 info(
1065 yuio.string.Md(msg, *args, dedent=dedent, allow_headings=allow_headings),
1066 **kwargs,
1067 )
1070def br(**kwargs):
1071 """
1072 Print an empty string.
1074 :param kwargs:
1075 any additional keyword arguments will be passed to :func:`raw`.
1077 """
1079 raw("\n", **kwargs)
1082@_t.overload
1083def hr(msg: _t.LiteralString = "", /, *args, weight: int | str = 1, **kwargs): ...
1084@_t.overload
1085def hr(msg: yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs): ...
1086def hr(msg: yuio.string.ToColorable = "", /, *args, weight: int | str = 1, **kwargs):
1087 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
1088 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
1089 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
1091 Print a horizontal ruler.
1093 :param msg:
1094 message to print in the middle of the ruler.
1095 :param args:
1096 arguments for ``%``\\ -formatting the message.
1097 :param weight:
1098 weight or style of the ruler:
1100 - ``0`` prints no ruler (but still prints centered text),
1101 - ``1`` prints normal ruler,
1102 - ``2`` prints bold ruler.
1104 Additional styles can be added through
1105 :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`.
1106 :param kwargs:
1107 any additional keyword arguments will be passed to :func:`raw`.
1109 """
1111 info(yuio.string.Hr(yuio.string._to_colorable(msg, args), weight=weight), **kwargs)
1114@_t.overload
1115def hl(
1116 msg: _t.LiteralString,
1117 /,
1118 *args,
1119 syntax: str | yuio.md.SyntaxHighlighter,
1120 dedent: bool = True,
1121 **kwargs,
1122): ...
1123@_t.overload
1124def hl(
1125 msg: str,
1126 /,
1127 *,
1128 syntax: str | yuio.md.SyntaxHighlighter,
1129 dedent: bool = True,
1130 **kwargs,
1131): ...
1132def hl(
1133 msg: str,
1134 /,
1135 *args,
1136 syntax: str | yuio.md.SyntaxHighlighter,
1137 dedent: bool = True,
1138 **kwargs,
1139):
1140 """hl(msg: typing.LiteralString, /, *args, syntax: str | yuio.md.SyntaxHighlighter, dedent: bool = True, **kwargs)
1141 hl(msg: str, /, *, syntax: str | yuio.md.SyntaxHighlighter, dedent: bool = True, **kwargs) ->
1143 Print highlighted code. See :mod:`yuio.md` for more info.
1145 :param msg:
1146 code to highlight.
1147 :param args:
1148 arguments for ``%``-formatting the highlighted code.
1149 :param syntax:
1150 name of syntax or a :class:`~yuio.md.SyntaxHighlighter` instance.
1151 :param dedent:
1152 whether to remove leading indent from code.
1153 :param kwargs:
1154 any additional keyword arguments will be passed to :func:`raw`.
1156 """
1158 info(yuio.string.Hl(msg, *args, syntax=syntax, dedent=dedent), **kwargs)
1161def raw(
1162 msg: yuio.string.Colorable,
1163 /,
1164 *,
1165 ignore_suspended: bool = False,
1166 tag: str | None = None,
1167 exc_info: ExcInfo | bool | None = None,
1168 add_newline: bool = False,
1169 heading: bool = False,
1170 wrap: bool = False,
1171 ctx: yuio.string.ReprContext | None = None,
1172 term: yuio.term.Term | None = None,
1173 to_stdout: bool = False,
1174 to_stderr: bool = False,
1175 theme: yuio.theme.Theme | None = None,
1176 multiline: bool | None = None,
1177 highlighted: bool | None = None,
1178 max_depth: int | None = None,
1179 width: int | None = None,
1180):
1181 """
1182 Print any :class:`~yuio.string.ToColorable`.
1184 This is a bridge between :mod:`yuio.io` and lower-level
1185 modules like :mod:`yuio.string`.
1187 :param msg:
1188 message to print.
1189 :param ignore_suspended:
1190 whether to ignore :class:`SuspendOutput` context.
1191 :param tag:
1192 tag that will be used to add color and decoration to the message.
1194 Decoration is looked up by path :samp:`{tag}`
1195 (see :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`),
1196 and color is looked up by path :samp:`msg/text:{tag}`
1197 (see :attr:`Theme.colors <yuio.theme.Theme.colors>`).
1198 :param exc_info:
1199 either a boolean indicating that the current exception
1200 should be captured, or a tuple of three elements, as returned
1201 by :func:`sys.exc_info`.
1202 :param add_newline:
1203 adds newline after the message.
1204 :param heading:
1205 whether to separate message by extra newlines.
1207 If :data:`True`, adds extra newline after the message; if this is not the
1208 first message printed so far, adds another newline before the message.
1209 :param wrap:
1210 whether to wrap message before printing it.
1212 :param ctx:
1213 :param term:
1214 if `ctx` is not given, sets terminal where to print this message. Default is
1215 to use :func:`get_term`.
1216 :param to_stdout:
1217 shortcut for setting `term` to ``stdout``.
1218 :param to_stderr:
1219 shortcut for setting `term` to ``stderr``.
1220 :param theme:
1221 if `ctx` is not given, sets theme used to format the message. Default is
1222 to use :func:`get_theme`.
1223 :param multiline:
1224 if `ctx` is not given, sets initial value for
1225 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
1226 Default is :data:`False`.
1227 :param highlighted:
1228 if `ctx` is not given, sets initial value for
1229 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
1230 Default is :data:`False`.
1231 :param max_depth:
1232 if `ctx` is not given, sets initial value for
1233 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
1234 Default is :data:`False`.
1235 :param width:
1236 if `ctx` is not given, sets initial value for
1237 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
1238 If not given, uses current terminal width
1239 or :attr:`Theme.fallback_width <yuio.theme.Theme.fallback_width>`
1240 if terminal width can't be established.
1242 """
1244 if (term is not None) + to_stdout + to_stderr > 1:
1245 raise TypeError("term, to_stdout, to_stderr can't be given together")
1247 manager = _manager()
1249 if ctx is None:
1250 ctx = make_repr_context(
1251 term=term,
1252 to_stdout=to_stdout,
1253 to_stderr=to_stderr,
1254 theme=theme,
1255 multiline=multiline,
1256 highlighted=highlighted,
1257 max_depth=max_depth,
1258 width=width,
1259 )
1261 if tag and (decoration := ctx.get_msg_decoration(tag)):
1262 indent = yuio.string.ColorizedString(
1263 [ctx.get_color(f"msg/decoration:{tag}"), decoration]
1264 )
1265 continuation_indent = " " * indent.width
1266 else:
1267 indent = ""
1268 continuation_indent = ""
1270 if tag:
1271 msg = yuio.string.WithBaseColor(
1272 msg, base_color=ctx.get_color(f"msg/text:{tag}")
1273 )
1275 if wrap:
1276 msg = yuio.string.Wrap(
1277 msg,
1278 indent=indent,
1279 continuation_indent=continuation_indent,
1280 preserve_spaces=True,
1281 )
1282 elif indent or continuation_indent:
1283 msg = yuio.string.Indent(
1284 msg,
1285 indent=indent,
1286 continuation_indent=continuation_indent,
1287 )
1289 msg = ctx.str(msg)
1291 if add_newline:
1292 msg.append_color(yuio.color.Color.NONE)
1293 msg.append_str("\n")
1295 if exc_info is True:
1296 exc_info = sys.exc_info()
1297 elif exc_info is False or exc_info is None:
1298 exc_info = None
1299 elif isinstance(exc_info, BaseException):
1300 exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
1301 elif not isinstance(exc_info, tuple) or len(exc_info) != 3:
1302 raise ValueError(f"invalid exc_info {exc_info!r}")
1303 if exc_info is not None and exc_info != (None, None, None):
1304 tb = "".join(traceback.format_exception(*exc_info))
1305 highlighter = yuio.md.SyntaxHighlighter.get_highlighter("python-traceback")
1306 msg += highlighter.highlight(ctx.theme, tb).indent()
1308 manager.print(
1309 msg.as_code(ctx.term.color_support),
1310 ctx.term,
1311 ignore_suspended=ignore_suspended,
1312 heading=heading,
1313 )
1316class _AskWidget(yuio.widget.Widget[T], _t.Generic[T]):
1317 _layout: yuio.widget.VerticalLayout[T]
1319 def __init__(self, prompt: yuio.string.Colorable, widget: yuio.widget.Widget[T]):
1320 self._prompt = yuio.widget.Text(prompt)
1321 self._error: Exception | None = None
1322 self._inner = widget
1324 def event(self, e: yuio.widget.KeyboardEvent, /) -> yuio.widget.Result[T] | None:
1325 try:
1326 result = self._inner.event(e)
1327 except yuio.parse.ParsingError as err:
1328 self._error = err
1329 else:
1330 self._error = None
1331 return result
1333 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1334 builder = (
1335 yuio.widget.VerticalLayoutBuilder()
1336 .add(self._prompt)
1337 .add(self._inner, receive_events=True)
1338 )
1339 if self._error is not None:
1340 rc.bell()
1341 error_msg = yuio.string.colorize(
1342 "<c msg/decoration:error>▲</c> %s",
1343 yuio.string.Indent(self._error, indent=0, continuation_indent=2),
1344 default_color="msg/text:error",
1345 ctx=rc.make_repr_context(),
1346 )
1347 builder = builder.add(yuio.widget.Text(error_msg))
1349 self._layout = builder.build()
1350 return self._layout.layout(rc)
1352 def draw(self, rc: yuio.widget.RenderContext, /):
1353 self._layout.draw(rc)
1355 @property
1356 def help_data(self) -> yuio.widget.WidgetHelp:
1357 return self._inner.help_data
1360class _AskMeta(type):
1361 __hint = None
1363 @_t.overload
1364 def __call__(
1365 cls: type[ask[S]],
1366 msg: _t.LiteralString,
1367 /,
1368 *args,
1369 default: M | yuio.Missing = yuio.MISSING,
1370 default_non_interactive: _t.Any = yuio.MISSING,
1371 parser: yuio.parse.Parser[S] | None = None,
1372 input_description: str | None = None,
1373 default_description: str | None = None,
1374 ) -> S | M: ...
1375 @_t.overload
1376 def __call__(
1377 cls: type[ask[S]],
1378 msg: str,
1379 /,
1380 *,
1381 default: M | yuio.Missing = yuio.MISSING,
1382 default_non_interactive: _t.Any = yuio.MISSING,
1383 parser: yuio.parse.Parser[S] | None = None,
1384 input_description: str | None = None,
1385 default_description: str | None = None,
1386 ) -> S | M: ...
1387 def __call__(cls, *args, **kwargs):
1388 if "parser" not in kwargs:
1389 hint = cls.__hint
1390 if hint is None:
1391 hint = str
1392 kwargs["parser"] = yuio.parse.from_type_hint(hint)
1393 return _ask(*args, **kwargs)
1395 def __getitem(cls, ty):
1396 return _AskMeta("ask", (), {"_AskMeta__hint": ty})
1398 # A dirty hack to hide `__getitem__` from type checkers. `ask` should look like
1399 # an ordinary class with overloaded `__new__` for the magic to work.
1400 locals()["__getitem__"] = __getitem
1402 def __repr__(cls) -> str:
1403 if cls.__hint is None:
1404 return cls.__name__
1405 else:
1406 return f"{cls.__name__}[{_tx.type_repr(cls.__hint)}]"
1409@_t.final
1410class ask(_t.Generic[S], metaclass=_AskMeta):
1411 """ask[T](msg: typing.LiteralString, /, *args, parser: ~yuio.parse.Parser[T] | None = None, default: U, default_non_interactive: U, input_description: str | None = None, default_description: str | None = None) -> T | U
1412 ask[T](msg: str, /, *, parser: ~yuio.parse.Parser[T] | None = None, default: U, default_non_interactive: U, input_description: str | None = None, default_description: str | None = None) -> T | U
1414 Ask user to provide an input, parse it and return a value.
1416 If current terminal is not interactive, return default if one is present,
1417 or raise a :class:`UserIoError`.
1419 .. vhs:: /_tapes/questions.tape
1420 :alt: Demonstration of the `ask` function.
1421 :scale: 40%
1423 :func:`ask` accepts generic parameters, which determine how input is parsed.
1424 For example, if you're asking for an enum element,
1425 Yuio will show user a choice widget.
1427 You can also supply a custom :class:`~yuio.parse.Parser`,
1428 which will determine the widget that is displayed to the user,
1429 the way autocompletion works, etc.
1431 .. note::
1433 :func:`ask` is designed to interact with users, not to read data. It uses
1434 ``/dev/tty`` on Unix, and console API on Windows, so it will read from
1435 an actual TTY even if ``stdin`` is redirected.
1437 When designing your program, make sure that users have alternative means
1438 to provide values: use configs or CLI arguments, allow passing passwords
1439 via environment variables, etc.
1441 :param msg:
1442 prompt to display to user.
1443 :param args:
1444 arguments for ``%``\\ - formatting the prompt.
1445 :param parser:
1446 parser to use to parse user input. See :mod:`yuio.parse` for more info.
1447 :param default:
1448 default value to return if user input is empty.
1449 :param default_non_interactive:
1450 default value returned if input stream is not readable. If not given,
1451 `default` is used instead. This is handy when you want to ask user if they
1452 want to continue with `default` set to :data:`False`,
1453 but `default_non_interactive` set to :data:`True`.
1454 :param input_description:
1455 description of the expected input, like ``"yes/no"`` for boolean
1456 inputs.
1457 :param default_description:
1458 description of the `default` value.
1459 :returns:
1460 parsed user input.
1461 :raises:
1462 raises :class:`UserIoError` if we're not in interactive environment, and there
1463 is no default to return.
1464 :example:
1465 .. code-block:: python
1467 class Level(enum.Enum):
1468 WARNING = "Warning"
1469 INFO = "Info"
1470 DEBUG = "Debug"
1473 answer = yuio.io.ask[Level]("Choose a logging level", default=Level.INFO)
1475 """
1477 if TYPE_CHECKING:
1479 @_t.overload
1480 def __new__(
1481 cls: type[ask[S]],
1482 msg: _t.LiteralString,
1483 /,
1484 *args,
1485 default: M | yuio.Missing = yuio.MISSING,
1486 default_non_interactive: _t.Any = yuio.MISSING,
1487 parser: yuio.parse.Parser[S] | None = None,
1488 input_description: str | None = None,
1489 default_description: str | None = None,
1490 ) -> S | M: ...
1491 @_t.overload
1492 def __new__(
1493 cls: type[ask[S]],
1494 msg: str,
1495 /,
1496 *,
1497 default: M | yuio.Missing = yuio.MISSING,
1498 default_non_interactive: _t.Any = yuio.MISSING,
1499 parser: yuio.parse.Parser[S] | None = None,
1500 input_description: str | None = None,
1501 default_description: str | None = None,
1502 ) -> S | M: ...
1503 def __new__(cls: _t.Any, *_, **__) -> _t.Any: ...
1506def _ask(
1507 msg: _t.LiteralString,
1508 /,
1509 *args,
1510 parser: yuio.parse.Parser[_t.Any],
1511 default: _t.Any = yuio.MISSING,
1512 default_non_interactive: _t.Any = yuio.MISSING,
1513 input_description: str | None = None,
1514 default_description: str | None = None,
1515) -> _t.Any:
1516 ctx = make_repr_context(term=yuio.term.get_tty())
1518 if not _can_query_user(ctx.term):
1519 # TTY is not available.
1520 if default_non_interactive is yuio.MISSING:
1521 default_non_interactive = default
1522 if default_non_interactive is yuio.MISSING:
1523 raise UserIoError("Can't interact with user in non-interactive environment")
1524 return default_non_interactive
1526 if default is None and not yuio.parse._is_optional_parser(parser):
1527 parser = yuio.parse.Optional(parser)
1529 msg = msg.rstrip()
1530 if msg.endswith(":"):
1531 needs_colon = True
1532 msg = msg[:-1]
1533 else:
1534 needs_colon = msg and msg[-1] not in string.punctuation
1536 base_color = ctx.get_color("msg/text:question")
1537 prompt = yuio.string.colorize(msg, *args, default_color=base_color, ctx=ctx)
1539 if not input_description:
1540 input_description = parser.describe()
1542 if default is not yuio.MISSING and default_description is None:
1543 try:
1544 default_description = parser.describe_value(default)
1545 except TypeError:
1546 default_description = str(default)
1548 if not yuio.term._is_foreground(ctx.term.ostream):
1549 warning(
1550 "User input is requested in background process, use `fg %s` to resume",
1551 os.getpid(),
1552 ctx=ctx,
1553 )
1554 yuio.term._pause()
1556 if ctx.term.can_run_widgets:
1557 # Use widget.
1559 if needs_colon:
1560 prompt.append_color(base_color)
1561 prompt.append_str(":")
1563 if parser.is_secret():
1564 inner_widget = yuio.parse._secret_widget(
1565 parser, default, input_description, default_description
1566 )
1567 else:
1568 inner_widget = parser.widget(
1569 default, input_description, default_description
1570 )
1572 widget = _AskWidget(prompt, inner_widget)
1573 with SuspendOutput() as s:
1574 try:
1575 result = widget.run(ctx.term, ctx.theme)
1576 except (OSError, EOFError) as e: # pragma: no cover
1577 raise UserIoError("Unexpected end of input") from e
1579 if result is yuio.MISSING:
1580 result = default
1582 try:
1583 result_desc = parser.describe_value(result)
1584 except TypeError:
1585 result_desc = str(result)
1587 prompt.append_color(base_color)
1588 prompt.append_str(" ")
1589 prompt.append_color(base_color | ctx.get_color("code"))
1590 prompt.append_str(result_desc)
1592 s.info(prompt, tag="question", ctx=ctx)
1593 return result
1594 else:
1595 # Use raw input.
1597 prompt += base_color
1598 if input_description:
1599 prompt += " ("
1600 prompt += input_description
1601 prompt += ")"
1602 if default_description:
1603 prompt += " ["
1604 prompt += base_color | ctx.get_color("code")
1605 prompt += default_description
1606 prompt += base_color
1607 prompt += "]"
1608 prompt += yuio.string.Esc(": " if needs_colon else " ")
1609 if parser.is_secret():
1610 do_input = _getpass
1611 else:
1612 do_input = _read
1613 with SuspendOutput() as s:
1614 while True:
1615 try:
1616 answer = do_input(ctx.term, prompt)
1617 except (OSError, EOFError) as e: # pragma: no cover
1618 raise UserIoError("Unexpected end of input") from e
1619 if not answer and default is not yuio.MISSING:
1620 return default
1621 elif not answer:
1622 s.error("Input is required.", ctx=ctx)
1623 else:
1624 try:
1625 return parser.parse(answer)
1626 except yuio.parse.ParsingError as e:
1627 s.error(e, ctx=ctx)
1630if os.name == "posix":
1631 # Getpass implementation is based on the standard `getpass` module, with a few
1632 # Yuio-specific modifications.
1634 def _getpass_fallback(
1635 term: yuio.term.Term, prompt: yuio.string.ColorizedString
1636 ) -> str:
1637 warning(
1638 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1639 )
1640 return _read(term, prompt)
1642 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1643 info(
1644 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1645 )
1646 return term.istream.readline().rstrip("\r\n")
1648 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1649 import termios
1651 try:
1652 fd = term.istream.fileno()
1653 except (AttributeError, ValueError):
1654 # We can't control the tty or stdin. Give up and use normal IO.
1655 return _getpass_fallback(term, prompt)
1657 result: str | None = None
1659 try:
1660 prev_mode = termios.tcgetattr(fd)
1661 new_mode = prev_mode.copy()
1662 new_mode[3] &= ~termios.ECHO
1663 tcsetattr_flags = termios.TCSAFLUSH | getattr(termios, "TCSASOFT", 0)
1664 try:
1665 termios.tcsetattr(fd, tcsetattr_flags, new_mode)
1666 info(
1667 prompt,
1668 add_newline=False,
1669 tag="question",
1670 term=term,
1671 ignore_suspended=True,
1672 )
1673 result = term.istream.readline().rstrip("\r\n")
1674 term.ostream.write("\n")
1675 term.ostream.flush()
1676 finally:
1677 termios.tcsetattr(fd, tcsetattr_flags, prev_mode)
1678 except termios.error:
1679 if result is not None:
1680 # `readline` succeeded, the final `tcsetattr` failed. Reraise instead
1681 # of leaving the terminal in an unknown state.
1682 raise
1683 else:
1684 # We can't control the tty or stdin. Give up and use normal IO.
1685 return _getpass_fallback(term, prompt)
1687 assert result is not None
1688 return result
1690elif os.name == "nt":
1692 def _do_read(
1693 term: yuio.term.Term, prompt: yuio.string.ColorizedString, echo: bool
1694 ) -> str:
1695 import msvcrt
1697 if term.ostream_is_tty:
1698 info(
1699 prompt,
1700 add_newline=False,
1701 tag="question",
1702 term=term,
1703 ignore_suspended=True,
1704 )
1705 else:
1706 for c in str(prompt):
1707 msvcrt.putwch(c)
1709 if term.ostream_is_tty and echo:
1710 return term.istream.readline().rstrip("\r\n")
1711 else:
1712 result = ""
1713 while True:
1714 c = msvcrt.getwch()
1715 if c == "\0" or c == "\xe0":
1716 # Read key scan code and ignore it.
1717 msvcrt.getwch()
1718 continue
1719 if c == "\r" or c == "\n":
1720 break
1721 if c == "\x03":
1722 raise KeyboardInterrupt
1723 if c == "\b":
1724 if result:
1725 msvcrt.putwch("\b")
1726 msvcrt.putwch(" ")
1727 msvcrt.putwch("\b")
1728 result = result[:-1]
1729 else:
1730 result = result + c
1731 if echo:
1732 msvcrt.putwch(c)
1733 else:
1734 msvcrt.putwch("*")
1735 msvcrt.putwch("\r")
1736 msvcrt.putwch("\n")
1738 return result
1740 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1741 return _do_read(term, prompt, echo=True)
1743 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1744 return _do_read(term, prompt, echo=False)
1746else:
1748 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1749 warning(
1750 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1751 )
1752 return _read(term, prompt)
1754 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1755 info(
1756 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1757 )
1758 return term.istream.readline().rstrip("\r\n")
1761def _can_query_user(term: yuio.term.Term):
1762 return (
1763 # We're attached to a TTY.
1764 term.is_tty
1765 # On Windows, there is no way to bring a process to foreground (AFAIK?).
1766 # Thus, we need to check if there's a console window.
1767 and (os.name != "nt" or yuio.term._is_foreground(None))
1768 )
1771class _WaitForUserWidget(yuio.widget.Widget[None]):
1772 def __init__(self, prompt: yuio.string.Colorable):
1773 self._prompt = yuio.widget.Text(prompt)
1775 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1776 return self._prompt.layout(rc)
1778 def draw(self, rc: yuio.widget.RenderContext, /):
1779 return self._prompt.draw(rc)
1781 @yuio.widget.bind(yuio.widget.Key.ENTER)
1782 @yuio.widget.bind(yuio.widget.Key.ESCAPE)
1783 @yuio.widget.bind("d", ctrl=True)
1784 @yuio.widget.bind(" ")
1785 def exit(self):
1786 return yuio.widget.Result(None)
1789def wait_for_user(
1790 msg: _t.LiteralString = "Press <c note>enter</c> to continue",
1791 /,
1792 *args,
1793):
1794 """
1795 A simple function to wait for user to press enter.
1797 If current terminal is not interactive, this functions has no effect.
1799 :param msg:
1800 prompt to display to user.
1801 :param args:
1802 arguments for ``%``\\ - formatting the prompt.
1804 """
1806 ctx = make_repr_context(term=yuio.term.get_tty())
1808 if not _can_query_user(ctx.term):
1809 # TTY is not available.
1810 return
1812 if not yuio.term._is_foreground(ctx.term.ostream):
1813 if os.name == "nt":
1814 # AFAIK there's no way to bring job to foreground in Windows.
1815 return
1817 warning(
1818 "User input is requested in background process, use `fg %s` to resume",
1819 os.getpid(),
1820 ctx=ctx,
1821 )
1822 yuio.term._pause()
1824 prompt = yuio.string.colorize(
1825 msg.rstrip(), *args, default_color="msg/text:question", ctx=ctx
1826 )
1827 prompt += yuio.string.Esc(" ")
1829 with SuspendOutput():
1830 try:
1831 if ctx.term.can_run_widgets:
1832 _WaitForUserWidget(prompt).run(ctx.term, ctx.theme)
1833 else:
1834 _read(ctx.term, prompt)
1835 except (OSError, EOFError): # pragma: no cover
1836 return
1839def detect_editor(fallbacks: list[str] | None = None) -> str | None:
1840 """
1841 Detect the user's preferred editor.
1843 This function checks the ``VISUAL`` and ``EDITOR`` environment variables.
1844 If they're not set, it checks if any of the fallback editors are available.
1845 If none can be found, it returns :data:`None`.
1847 :param fallbacks:
1848 list of fallback editors to try. By default, we try "nano", "vim", "vi",
1849 "msedit", "edit", "notepad", "gedit".
1850 :returns:
1851 on Windows, returns an executable name; on Unix, may return a shell command
1852 or an executable name.
1854 """
1856 if os.name != "nt":
1857 if editor := os.environ.get("VISUAL"):
1858 return editor
1859 if editor := os.environ.get("EDITOR"):
1860 return editor
1862 if fallbacks is None:
1863 fallbacks = ["nano", "vim", "vi", "msedit", "edit", "notepad", "gedit"]
1864 for fallback in fallbacks:
1865 if shutil.which(fallback):
1866 return fallback
1867 return None
1870def edit(
1871 text: str,
1872 /,
1873 *,
1874 comment_marker: str | None = None,
1875 editor: str | None = None,
1876 file_ext: str = ".txt",
1877 fallbacks: list[str] | None = None,
1878 dedent: bool = False,
1879) -> str:
1880 """
1881 Ask user to edit some text.
1883 This function creates a temporary file with the given text
1884 and opens it in an editor. After editing is done, it strips away
1885 all lines that start with `comment_marker`, if one is given.
1887 :param text:
1888 text to edit.
1889 :param comment_marker:
1890 lines starting with this marker will be removed from the output after edit.
1891 :param editor:
1892 overrides editor.
1894 On Unix, this should be a shell command, file path will be appended to it;
1895 on Windows, this should be an executable path.
1896 :param file_ext:
1897 extension for the temporary file, can be used to enable syntax highlighting
1898 in editors that support it.
1899 :param fallbacks:
1900 list of fallback editors to try, see :func:`detect_editor` for details.
1901 :param dedent:
1902 remove leading indentation from text before opening an editor.
1903 :returns:
1904 an edited string with comments removed.
1905 :raises:
1906 If editor is not available, returns a non-zero exit code, or launched in
1907 a non-interactive environment, a :class:`UserIoError` is raised.
1909 Also raises :class:`UserIoError` if ``stdin`` or ``stderr`` is piped
1910 or redirected to a file (virtually no editors can work when this happens).
1911 :example:
1912 .. skip: next
1914 .. code-block:: python
1916 message = yuio.io.edit(
1917 \"""
1918 # Please enter the commit message for your changes. Lines starting
1919 # with '#' will be ignored, and an empty message aborts the commit.
1920 \""",
1921 comment_marker="#",
1922 dedent=True,
1923 )
1925 """
1927 term = yuio.term.get_tty()
1929 if not _can_query_user(term):
1930 raise UserIoError("Can't run editor in non-interactive environment")
1932 if editor is None:
1933 editor = detect_editor(fallbacks)
1935 if editor is None:
1936 if os.name == "nt":
1937 raise UserIoError("Can't find a usable editor")
1938 else:
1939 raise UserIoError(
1940 "Can't find a usable editor. Ensure that `$VISUAL` and `$EDITOR` "
1941 "environment variables contain correct path to an editor executable"
1942 )
1944 if dedent:
1945 text = _dedent(text)
1947 if not yuio.term._is_foreground(term.ostream):
1948 warning(
1949 "Background process is waiting for user, use `fg %s` to resume",
1950 os.getpid(),
1951 term=term,
1952 )
1953 yuio.term._pause()
1955 fd, filepath = tempfile.mkstemp(text=True, suffix=file_ext)
1956 try:
1957 with open(fd, "w") as file:
1958 file.write(text)
1960 if os.name == "nt":
1961 # Windows doesn't use $VISUAL/$EDITOR, so shell execution is not needed.
1962 # Plus, quoting arguments for CMD.exe is hard af.
1963 args = [editor, filepath]
1964 shell = False
1965 else:
1966 # $VISUAL/$EDITOR can include flags, so we need to use shell instead.
1967 from shlex import quote
1969 args = f"{editor} {quote(filepath)}"
1970 shell = True
1972 try:
1973 with SuspendOutput():
1974 res = subprocess.run(
1975 args,
1976 shell=shell,
1977 stdin=term.istream.fileno(),
1978 stdout=term.ostream.fileno(),
1979 )
1980 except FileNotFoundError:
1981 raise UserIoError(
1982 "Can't use editor `%r`: no such file or directory",
1983 editor,
1984 )
1986 if res.returncode != 0:
1987 if res.returncode < 0:
1988 import signal
1990 try:
1991 action = "died with"
1992 code = signal.Signals(-res.returncode).name
1993 except ValueError:
1994 action = "died with unknown signal"
1995 code = res.returncode
1996 else:
1997 action = "returned exit code"
1998 code = res.returncode
1999 raise UserIoError(
2000 "Editing failed: editor `%r` %s `%s`",
2001 editor,
2002 action,
2003 code,
2004 )
2006 if not os.path.exists(filepath):
2007 raise UserIoError("Editing failed: can't read back edited file")
2008 else:
2009 with open(filepath) as file:
2010 text = file.read()
2011 finally:
2012 try:
2013 os.remove(filepath)
2014 except OSError:
2015 pass
2017 if comment_marker is not None:
2018 text = re.sub(
2019 r"^\s*" + re.escape(comment_marker) + r".*(\n|$)",
2020 "",
2021 text,
2022 flags=re.MULTILINE,
2023 )
2025 return text
2028class SuspendOutput:
2029 """
2030 A context manager for pausing output.
2032 This is handy for when you need to take control over the output stream.
2033 For example, the :func:`ask` function uses this class internally.
2035 This context manager also suspends all prints that go to :data:`sys.stdout`
2036 and :data:`sys.stderr` if they were wrapped (see :func:`setup`).
2037 To print through them, use :func:`orig_stderr` and :func:`orig_stdout`.
2039 """
2041 def __init__(self):
2042 self._resumed = False
2043 _manager().suspend()
2045 def resume(self):
2046 """
2047 Manually resume the logging process.
2049 """
2051 if not self._resumed:
2052 _manager().resume()
2053 self._resumed = True
2055 @_t.overload
2056 def info(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2057 @_t.overload
2058 def info(self, err: yuio.string.ToColorable, /, **kwargs): ...
2059 def info(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2060 """info(msg: typing.LiteralString, /, *args, **kwargs)
2061 info(msg: ~string.templatelib.Template, /, **kwargs) ->
2062 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2064 Log an :func:`info` message, ignore suspended status.
2066 """
2068 kwargs.setdefault("ignore_suspended", True)
2069 info(msg, *args, **kwargs)
2071 @_t.overload
2072 def warning(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2073 @_t.overload
2074 def warning(self, err: yuio.string.ToColorable, /, **kwargs): ...
2075 def warning(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2076 """warning(msg: typing.LiteralString, /, *args, **kwargs)
2077 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
2078 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2080 Log a :func:`warning` message, ignore suspended status.
2082 """
2084 kwargs.setdefault("ignore_suspended", True)
2085 warning(msg, *args, **kwargs)
2087 @_t.overload
2088 def success(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2089 @_t.overload
2090 def success(self, err: yuio.string.ToColorable, /, **kwargs): ...
2091 def success(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2092 """success(msg: typing.LiteralString, /, *args, **kwargs)
2093 success(msg: ~string.templatelib.Template, /, **kwargs) ->
2094 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2096 Log a :func:`success` message, ignore suspended status.
2098 """
2100 kwargs.setdefault("ignore_suspended", True)
2101 success(msg, *args, **kwargs)
2103 @_t.overload
2104 def error(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2105 @_t.overload
2106 def error(self, err: yuio.string.ToColorable, /, **kwargs): ...
2107 def error(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2108 """error(msg: typing.LiteralString, /, *args, **kwargs)
2109 error(msg: ~string.templatelib.Template, /, **kwargs) ->
2110 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2112 Log an :func:`error` message, ignore suspended status.
2114 """
2116 kwargs.setdefault("ignore_suspended", True)
2117 error(msg, *args, **kwargs)
2119 @_t.overload
2120 def error_with_tb(
2121 self,
2122 msg: _t.LiteralString,
2123 /,
2124 *args,
2125 exc_info: ExcInfo | bool | None = True,
2126 **kwargs,
2127 ): ...
2128 @_t.overload
2129 def error_with_tb(
2130 self,
2131 msg: yuio.string.ToColorable,
2132 /,
2133 *,
2134 exc_info: ExcInfo | bool | None = True,
2135 **kwargs,
2136 ): ...
2137 def error_with_tb(
2138 self,
2139 msg: yuio.string.ToColorable,
2140 /,
2141 *args,
2142 exc_info: ExcInfo | bool | None = True,
2143 **kwargs,
2144 ):
2145 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2146 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2147 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2149 Log an :func:`error_with_tb` message, ignore suspended status.
2151 """
2153 kwargs.setdefault("ignore_suspended", True)
2154 error_with_tb(msg, *args, **kwargs)
2156 @_t.overload
2157 def failure(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2158 @_t.overload
2159 def failure(self, err: yuio.string.ToColorable, /, **kwargs): ...
2160 def failure(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2161 """failure(msg: typing.LiteralString, /, *args, **kwargs)
2162 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
2163 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2165 Log a :func:`failure` message, ignore suspended status.
2167 """
2169 kwargs.setdefault("ignore_suspended", True)
2170 failure(msg, *args, **kwargs)
2172 @_t.overload
2173 def failure_with_tb(
2174 self,
2175 msg: _t.LiteralString,
2176 /,
2177 *args,
2178 exc_info: ExcInfo | bool | None = True,
2179 **kwargs,
2180 ): ...
2181 @_t.overload
2182 def failure_with_tb(
2183 self,
2184 msg: yuio.string.ToColorable,
2185 /,
2186 *,
2187 exc_info: ExcInfo | bool | None = True,
2188 **kwargs,
2189 ): ...
2190 def failure_with_tb(
2191 self,
2192 msg: yuio.string.ToColorable,
2193 /,
2194 *args,
2195 exc_info: ExcInfo | bool | None = True,
2196 **kwargs,
2197 ):
2198 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2199 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2200 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2202 Log a :func:`failure_with_tb` message, ignore suspended status.
2204 """
2206 kwargs.setdefault("ignore_suspended", True)
2207 failure_with_tb(msg, *args, **kwargs)
2209 @_t.overload
2210 def heading(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2211 @_t.overload
2212 def heading(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2213 def heading(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2214 """heading(msg: typing.LiteralString, /, *args, **kwargs)
2215 heading(msg: ~string.templatelib.Template, /, **kwargs)
2216 heading(msg: ~yuio.string.ToColorable, /, **kwargs)
2218 Log a :func:`heading` message, ignore suspended status.
2220 """
2222 kwargs.setdefault("ignore_suspended", True)
2223 heading(msg, *args, **kwargs)
2225 @_t.overload
2226 def md(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2227 @_t.overload
2228 def md(self, msg: str, /, **kwargs): ...
2229 def md(self, msg: str, /, *args, **kwargs):
2230 """md(msg: typing.LiteralString, /, *args, dedent: bool = True, allow_headings: bool = True, **kwargs)
2231 md(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs)
2233 Log an :func:`md` message, ignore suspended status.
2235 """
2237 kwargs.setdefault("ignore_suspended", True)
2238 md(msg, *args, **kwargs)
2240 def br(self, **kwargs):
2241 """br()
2243 Log a :func:`br` message, ignore suspended status.
2245 """
2247 kwargs.setdefault("ignore_suspended", True)
2248 br(**kwargs)
2250 @_t.overload
2251 def hl(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2252 @_t.overload
2253 def hl(self, msg: str, /, **kwargs): ...
2254 def hl(self, msg: str, /, *args, **kwargs):
2255 """hl(msg: typing.LiteralString, /, *args, syntax: str | yuio.md.SyntaxHighlighter, dedent: bool = True, **kwargs)
2256 hl(msg: str, /, *, syntax: str | yuio.md.SyntaxHighlighter, dedent: bool = True, **kwargs)
2258 Log an :func:`hl` message, ignore suspended status.
2260 """
2262 kwargs.setdefault("ignore_suspended", True)
2263 hl(msg, *args, **kwargs)
2265 @_t.overload
2266 def hr(self, msg: _t.LiteralString = "", /, *args, **kwargs): ...
2267 @_t.overload
2268 def hr(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2269 def hr(self, msg: yuio.string.ToColorable = "", /, *args, **kwargs):
2270 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
2271 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
2272 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
2274 Log an :func:`hr` message, ignore suspended status.
2276 """
2278 kwargs.setdefault("ignore_suspended", True)
2279 hr(msg, *args, **kwargs)
2281 def raw(self, msg: yuio.string.Colorable, /, **kwargs):
2282 """
2283 Log a :func:`raw` message, ignore suspended status.
2285 """
2287 kwargs.setdefault("ignore_suspended", True)
2288 raw(msg, **kwargs)
2290 def __enter__(self):
2291 return self
2293 def __exit__(self, exc_type, exc_val, exc_tb):
2294 self.resume()
2297class _IterTask(_t.Generic[T]):
2298 def __init__(
2299 self, collection: _t.Collection[T], task: Task, unit: str, ndigits: int
2300 ):
2301 self._iter = iter(collection)
2302 self._task = task
2303 self._unit = unit
2304 self._ndigits = ndigits
2306 self._i = 0
2307 self._len = len(collection)
2309 def __next__(self) -> T:
2310 self._task.progress(self._i, self._len, unit=self._unit, ndigits=self._ndigits)
2311 if self._i < self._len:
2312 self._i += 1
2313 return self._iter.__next__()
2315 def __iter__(self) -> _IterTask[T]:
2316 return self
2319class Task:
2320 """Task(msg: typing.LiteralString, /, *args, comment: str | None = None)
2321 Task(msg: str, /, *, comment: str | None = None)
2323 A class for indicating progress of some task.
2325 :param msg:
2326 task heading.
2327 :param args:
2328 arguments for ``%``\\ -formatting the task heading.
2329 :param comment:
2330 comment for the task. Can be specified after creation
2331 via the :meth:`~Task.comment` method.
2333 You can have multiple tasks at the same time,
2334 create subtasks, set task's progress or add a comment about
2335 what's currently being done within a task.
2337 .. vhs:: /_tapes/tasks_multithreaded.tape
2338 :alt: Demonstration of the `Task` class.
2339 :scale: 40%
2341 This class can be used as a context manager:
2343 .. code-block:: python
2345 with yuio.io.Task("Processing input") as t:
2346 ...
2347 t.progress(0.3)
2348 ...
2350 """
2352 class _Status(enum.Enum):
2353 DONE = "done"
2354 ERROR = "error"
2355 RUNNING = "running"
2357 @_t.overload
2358 def __init__(
2359 self,
2360 msg: _t.LiteralString,
2361 /,
2362 *args,
2363 comment: str | None = None,
2364 ): ...
2365 @_t.overload
2366 def __init__(
2367 self,
2368 msg: str,
2369 /,
2370 *,
2371 comment: str | None = None,
2372 ): ...
2373 def __init__(
2374 self,
2375 msg: str,
2376 /,
2377 *args,
2378 _parent: Task | None = None,
2379 comment: str | None = None,
2380 ):
2381 # Task properties should not be written to directly.
2382 # Instead, task should be sent to a handler for modification.
2383 # This ensures thread safety, because handler has a lock.
2384 # See handler's implementation details.
2386 self._msg: str = msg
2387 self._args: tuple[object, ...] = args
2388 self._comment: str | None = comment
2389 self._comment_args: tuple[object, ...] | None = None
2390 self._status: Task._Status = Task._Status.RUNNING
2391 self._progress: float | None = None
2392 self._progress_done: str | None = None
2393 self._progress_total: str | None = None
2394 self._subtasks: list[Task] = []
2396 self._cached_msg: yuio.string.ColorizedString | None = None
2397 self._cached_comment: yuio.string.ColorizedString | None = None
2399 if _parent is None:
2400 _manager().start_task(self)
2401 else:
2402 _manager().start_subtask(_parent, self)
2404 @_t.overload
2405 def progress(self, progress: float | None, /, *, ndigits: int = 2): ...
2407 @_t.overload
2408 def progress(
2409 self,
2410 done: float | int,
2411 total: float | int,
2412 /,
2413 *,
2414 unit: str = "",
2415 ndigits: int = 0,
2416 ): ...
2418 def progress(
2419 self,
2420 *args: float | int | None,
2421 unit: str = "",
2422 ndigits: int | None = None,
2423 ):
2424 """progress(progress: float | None, /, *, ndigits: int = 2)
2425 progress(done: float | int, total: float | int, /, *, unit: str = "", ndigits: int = 0) ->
2427 Indicate progress of this task.
2429 If given one argument, it is treated as percentage between ``0`` and ``1``.
2431 If given two arguments, they are treated as amount of finished work,
2432 and a total amount of work. In this case, optional argument `unit`
2433 can be used to indicate units for the progress.
2435 If given a single :data:`None`, reset task progress.
2437 .. note::
2439 Tasks are updated asynchronously once every ~100ms, so calling this method
2440 is relatively cheap. It still requires acquiring a global lock, though:
2441 contention could be an issue in multi-threaded applications.
2443 :param progress:
2444 a percentage between ``0`` and ``1``, or :data:`None`
2445 to reset task progress.
2446 :param done:
2447 amount of finished work, should be less than or equal to `total`.
2448 :param total:
2449 total amount of work.
2450 :param unit:
2451 unit for measuring progress. Only displayed when progress is given
2452 as `done` and `total`.
2453 :param ndigits:
2454 number of digits to display after a decimal point.
2455 :example:
2456 .. code-block:: python
2458 with yuio.io.Task("Loading cargo") as task:
2459 task.progress(110, 150, unit="Kg")
2461 This will print the following:
2463 .. code-block:: text
2465 ■■■■■■■■■■■□□□□ Loading cargo - 110/150Kg
2467 """
2469 progress = None
2471 if len(args) == 1:
2472 progress = done = args[0]
2473 total = None
2474 if ndigits is None:
2475 ndigits = 2
2476 elif len(args) == 2:
2477 done, total = args
2478 if ndigits is None:
2479 ndigits = (
2480 2 if isinstance(done, float) or isinstance(total, float) else 0
2481 )
2482 else:
2483 raise ValueError(
2484 f"Task.progress() takes between one and two arguments "
2485 f"({len(args)} given)"
2486 )
2488 if done is None:
2489 _manager().set_task_progress(self, None, None, None)
2490 return
2492 if len(args) == 1:
2493 done *= 100
2494 unit = "%"
2496 done_str = "%.*f" % (ndigits, done)
2497 if total is None:
2498 _manager().set_task_progress(self, progress, done_str + unit, None)
2499 else:
2500 total_str = "%.*f" % (ndigits, total)
2501 progress = done / total if total else 0
2502 _manager().set_task_progress(self, progress, done_str, total_str + unit)
2504 def progress_size(
2505 self,
2506 done: float | int,
2507 total: float | int,
2508 /,
2509 *,
2510 ndigits: int = 2,
2511 ):
2512 """
2513 Indicate progress of this task using human-readable 1024-based size units.
2515 :param done:
2516 amount of processed data.
2517 :param total:
2518 total amount of data.
2519 :param ndigits:
2520 number of digits to display after a decimal point.
2521 :example:
2522 .. code-block:: python
2524 with yuio.io.Task("Downloading a file") as task:
2525 task.progress_size(31.05 * 2**20, 150 * 2**20)
2527 This will print:
2529 .. code-block:: text
2531 ■■■□□□□□□□□□□□□ Downloading a file - 31.05/150.00M
2533 """
2535 progress = done / total
2536 done, done_unit = self._size(done)
2537 total, total_unit = self._size(total)
2539 if done_unit == total_unit:
2540 done_unit = ""
2542 _manager().set_task_progress(
2543 self,
2544 progress,
2545 "%.*f%s" % (ndigits, done, done_unit),
2546 "%.*f%s" % (ndigits, total, total_unit),
2547 )
2549 @staticmethod
2550 def _size(n):
2551 for unit in "BKMGT":
2552 if n < 1024:
2553 return n, unit
2554 n /= 1024
2555 return n, "P"
2557 def progress_scale(
2558 self,
2559 done: float | int,
2560 total: float | int,
2561 /,
2562 *,
2563 unit: str = "",
2564 ndigits: int = 2,
2565 ):
2566 """
2567 Indicate progress of this task while scaling numbers in accordance
2568 with SI system.
2570 :param done:
2571 amount of finished work, should be less than or equal to `total`.
2572 :param total:
2573 total amount of work.
2574 :param unit:
2575 unit for measuring progress.
2576 :param ndigits:
2577 number of digits to display after a decimal point.
2578 :example:
2579 .. code-block:: python
2581 with yuio.io.Task("Charging a capacitor") as task:
2582 task.progress_scale(889.25e-3, 1, unit="V")
2584 This will print:
2586 .. code-block:: text
2588 ■■■■■■■■■■■■■□□ Charging a capacitor - 889.25mV/1.00V
2590 """
2592 progress = done / total
2593 done, done_unit = self._unit(done)
2594 total, total_unit = self._unit(total)
2596 if unit:
2597 done_unit += unit
2598 total_unit += unit
2600 _manager().set_task_progress(
2601 self,
2602 progress,
2603 "%.*f%s" % (ndigits, done, done_unit),
2604 "%.*f%s" % (ndigits, total, total_unit),
2605 )
2607 @staticmethod
2608 def _unit(n: float) -> tuple[float, str]:
2609 if math.fabs(n) < 1e-33:
2610 return 0, ""
2611 magnitude = max(-8, min(8, int(math.log10(math.fabs(n)) // 3)))
2612 if magnitude < 0:
2613 return n * 10 ** -(3 * magnitude), "munpfazy"[-magnitude - 1]
2614 elif magnitude > 0:
2615 return n / 10 ** (3 * magnitude), "KMGTPEZY"[magnitude - 1]
2616 else:
2617 return n, ""
2619 def iter(
2620 self,
2621 collection: _t.Collection[T],
2622 /,
2623 *,
2624 unit: str = "",
2625 ndigits: int = 0,
2626 ) -> _t.Iterable[T]:
2627 """
2628 Helper for updating progress automatically
2629 while iterating over a collection.
2631 :param collection:
2632 an iterable collection. Should support returning its length.
2633 :param unit:
2634 unit for measuring progress.
2635 :param ndigits:
2636 number of digits to display after a decimal point.
2637 :example:
2638 .. invisible-code-block: python
2640 urls = []
2642 .. code-block:: python
2644 with yuio.io.Task("Fetching data") as t:
2645 for url in t.iter(urls):
2646 ...
2648 This will output the following:
2650 .. code-block:: text
2652 ■■■■■□□□□□□□□□□ Fetching data - 1/3
2654 """
2656 return _IterTask(collection, self, unit, ndigits)
2658 def comment(self, comment: str | None, /, *args):
2659 """
2660 Set a comment for a task.
2662 Comment is displayed after the progress.
2664 :param comment:
2665 comment to display beside task progress.
2666 :param args:
2667 arguments for ``%``\\ -formatting comment.
2668 :example:
2669 .. invisible-code-block: python
2671 urls = []
2673 .. code-block:: python
2675 with yuio.io.Task("Fetching data") as t:
2676 for url in urls:
2677 t.comment("%s", url)
2678 ...
2680 This will output the following:
2682 .. code-block:: text
2684 ⣿ Fetching data - https://google.com
2686 """
2688 _manager().set_task_comment(self, comment, args)
2690 def done(self):
2691 """
2692 Indicate that this task has finished successfully.
2694 """
2696 _manager().finish_task(self, Task._Status.DONE)
2698 def error(self):
2699 """
2700 Indicate that this task has finished with an error.
2702 """
2704 _manager().finish_task(self, Task._Status.ERROR)
2706 @_t.overload
2707 def subtask(
2708 self, msg: _t.LiteralString, /, *args, comment: str | None = None
2709 ) -> Task: ...
2710 @_t.overload
2711 def subtask(self, msg: str, /, *, comment: str | None = None) -> Task: ...
2712 def subtask(self, msg: str, /, *args, comment: str | None = None) -> Task:
2713 """
2714 Create a subtask within this task.
2716 :param msg:
2717 subtask heading.
2718 :param args:
2719 arguments for ``%``\\ -formatting the subtask heading.
2720 :param comment:
2721 comment for the task. Can be specified after creation
2722 via the :meth:`~Task.comment` method.
2723 :returns:
2724 a new :class:`Task` that will be displayed as a sub-task of this task.
2726 """
2728 return Task(msg, *args, comment=comment, **{"_parent": self})
2730 def __enter__(self):
2731 return self
2733 def __exit__(self, exc_type, exc_val, exc_tb):
2734 if exc_type is None:
2735 self.done()
2736 else:
2737 self.error()
2740class Formatter(logging.Formatter):
2741 """
2742 Log formatter that uses ``%`` style with colorized string formatting
2743 and returns a string with ANSI escape characters generated for current
2744 output terminal.
2746 Every part of log message is colored with path :samp:`log/{name}:{level}`.
2747 For example, `asctime` in info log line is colored
2748 with path ``log/asctime:info``.
2750 In addition to the usual `log record attributes`__, this formatter also
2751 adds ``%(colMessage)s``, which is similar to ``%(message)s``, but colorized.
2753 __ https://docs.python.org/3/library/logging.html#logrecord-attributes
2755 """
2757 default_format = "%(asctime)s %(name)s %(levelname)s %(colMessage)s"
2758 default_msec_format = "%s.%03d"
2760 def __init__(
2761 self,
2762 fmt: str | None = None,
2763 datefmt: str | None = None,
2764 validate: bool = True,
2765 *,
2766 defaults: _t.Mapping[str, _t.Any] | None = None,
2767 ):
2768 fmt = fmt or self.default_format
2769 super().__init__(
2770 fmt,
2771 datefmt,
2772 style="%",
2773 validate=validate,
2774 defaults=defaults,
2775 )
2777 def formatMessage(self, record):
2778 level = record.levelname.lower()
2780 ctx = make_repr_context()
2782 if not hasattr(record, "colMessage"):
2783 msg = str(record.msg)
2784 if record.args:
2785 msg = ColorizedString(msg).percent_format(record.args, ctx)
2786 setattr(record, "colMessage", msg)
2788 if defaults := self._style._defaults: # type: ignore
2789 data = defaults | record.__dict__
2790 else:
2791 data = record.__dict__
2793 data = {
2794 k: yuio.string.WithBaseColor(v, base_color=f"log/{k}:{level}")
2795 for k, v in data.items()
2796 }
2798 return "".join(
2799 yuio.string.colorize(
2800 self._fmt or self.default_format, default_color=f"log:{level}", ctx=ctx
2801 )
2802 .percent_format(data, ctx)
2803 .as_code(ctx.term.color_support)
2804 )
2806 def formatException(self, ei):
2807 tb = "".join(traceback.format_exception(*ei)).rstrip()
2808 return self.formatStack(tb)
2810 def formatStack(self, stack_info):
2811 manager = _manager()
2812 theme = manager.theme
2813 term = manager.term
2814 highlighter = yuio.md.SyntaxHighlighter.get_highlighter("python-traceback")
2815 return "".join(
2816 highlighter.highlight(theme, stack_info)
2817 .indent()
2818 .as_code(term.color_support)
2819 )
2822class Handler(logging.Handler):
2823 """
2824 A handler that redirects all log messages to Yuio.
2826 """
2828 def __init__(self, level: int | str = 0):
2829 super().__init__(level)
2830 self.setFormatter(Formatter())
2832 def emit(self, record: LogRecord):
2833 manager = _manager()
2834 manager.print_direct(self.format(record).rstrip() + "\n", manager.term.ostream)
2837class _IoManager(abc.ABC):
2838 # If we see that it took more than this time to render progress bars,
2839 # we assume that the process was suspended, meaning that we might've been moved
2840 # from foreground to background or back. In either way, we should assume that the
2841 # screen was changed, and re-render all tasks accordingly. We have to track time
2842 # because Python might take significant time to call `SIGCONT` handler, so we can't
2843 # rely on it.
2844 TASK_RENDER_TIMEOUT_NS = 250_000_000
2846 def __init__(
2847 self,
2848 term: yuio.term.Term | None = None,
2849 theme: (
2850 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
2851 ) = None,
2852 enable_bg_updates: bool = True,
2853 ):
2854 self._out_term = yuio.term.get_term_from_stream(orig_stdout(), sys.stdin)
2855 self._err_term = yuio.term.get_term_from_stream(orig_stderr(), sys.stdin)
2857 self._term = term or self._err_term
2859 self._theme_ctor = theme
2860 if isinstance(theme, yuio.theme.Theme):
2861 self._theme = theme
2862 else:
2863 self._theme = yuio.theme.load(self._term, theme)
2864 self._rc = yuio.widget.RenderContext(self._term, self._theme)
2865 self._rc.prepare()
2867 self._suspended: int = 0
2868 self._suspended_lines: list[tuple[list[str], _t.TextIO]] = []
2870 self._tasks: list[Task] = []
2871 self._tasks_printed = 0
2872 self._spinner_state = 0
2873 self._needs_update = False
2874 self._last_update_time_us = 0
2875 self._printed_some_lines = False
2877 self._renders = 0
2879 self._stop = False
2880 self._stop_condition = threading.Condition(_IO_LOCK)
2881 self._thread: threading.Thread | None = None
2883 self._enable_bg_updates = enable_bg_updates
2884 self._prev_sigcont_handler: (
2885 None | yuio.Missing | int | _t.Callable[[int, types.FrameType | None], None]
2886 ) = yuio.MISSING
2887 self._seen_sigcont: bool = False
2888 if enable_bg_updates:
2889 self._setup_sigcont()
2890 self._thread = threading.Thread(
2891 target=self._bg_update, name="yuio_io_task_refresh", daemon=True
2892 )
2893 self._thread.start()
2895 atexit.register(self.stop)
2897 @property
2898 def term(self):
2899 return self._term
2901 @property
2902 def out_term(self):
2903 return self._out_term
2905 @property
2906 def err_term(self):
2907 return self._err_term
2909 @property
2910 def theme(self):
2911 return self._theme
2913 @property
2914 def rc(self):
2915 return self._rc
2917 def setup(
2918 self,
2919 term: yuio.term.Term | None = None,
2920 theme: (
2921 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
2922 ) = None,
2923 ):
2924 with _IO_LOCK:
2925 self._clear_tasks()
2927 if term is not None:
2928 self._term = term
2929 if theme is None:
2930 # Refresh theme to reflect changed terminal capabilities.
2931 theme = self._theme_ctor
2932 if theme is not None:
2933 self._theme_ctor = theme
2934 if isinstance(theme, yuio.theme.Theme):
2935 self._theme = theme
2936 else:
2937 self._theme = yuio.theme.load(self._term, theme)
2939 self._rc = yuio.widget.RenderContext(self._term, self._theme)
2940 self._rc.prepare()
2941 self.__dict__.pop("_update_rate_us", None)
2942 self._update_tasks()
2944 def _setup_sigcont(self):
2945 import signal
2947 if not hasattr(signal, "SIGCONT"):
2948 return
2950 self._prev_sigcont_handler = signal.getsignal(signal.SIGCONT)
2951 signal.signal(signal.SIGCONT, self._on_sigcont)
2953 def _reset_sigcont(self):
2954 import signal
2956 if not hasattr(signal, "SIGCONT"):
2957 return
2959 if self._prev_sigcont_handler is not yuio.MISSING:
2960 signal.signal(signal.SIGCONT, self._prev_sigcont_handler)
2962 def _on_sigcont(self, sig: int, frame: types.FrameType | None):
2963 self._seen_sigcont = True
2964 if self._prev_sigcont_handler and not isinstance(
2965 self._prev_sigcont_handler, int
2966 ):
2967 self._prev_sigcont_handler(sig, frame)
2969 def _bg_update(self):
2970 while True:
2971 try:
2972 with _IO_LOCK:
2973 while True:
2974 update_rate_us = self._update_rate_us
2975 start_ns = time.monotonic_ns()
2976 now_us = start_ns // 1_000
2977 sleep_us = update_rate_us - now_us % update_rate_us
2978 deadline_ns = (
2979 start_ns + 2 * sleep_us * 1000 + self.TASK_RENDER_TIMEOUT_NS
2980 )
2982 if self._stop_condition.wait_for(
2983 lambda: self._stop, timeout=sleep_us / 1_000_000
2984 ):
2985 return
2987 self._show_tasks(deadline_ns=deadline_ns)
2988 except Exception:
2989 yuio._logger.critical("exception in bg updater", exc_info=True)
2991 def stop(self):
2992 if self._stop:
2993 return
2995 with _IO_LOCK:
2996 atexit.unregister(self.stop)
2998 self._stop = True
2999 self._stop_condition.notify()
3000 self._show_tasks(immediate_render=True)
3002 if self._thread:
3003 self._thread.join()
3005 if self._prev_sigcont_handler is not yuio.MISSING:
3006 self._reset_sigcont()
3008 def print(
3009 self,
3010 msg: list[str],
3011 term: yuio.term.Term,
3012 *,
3013 ignore_suspended: bool = False,
3014 heading: bool = False,
3015 ):
3016 with _IO_LOCK:
3017 if heading and self.theme.separate_headings:
3018 if self._printed_some_lines:
3019 msg.insert(0, "\n")
3020 msg.append("\n")
3021 self._emit_lines(msg, term.ostream, ignore_suspended)
3022 if heading:
3023 self._printed_some_lines = False
3025 def print_direct(
3026 self,
3027 msg: str,
3028 stream: _t.TextIO | None = None,
3029 ):
3030 with _IO_LOCK:
3031 self._emit_lines([msg], stream, ignore_suspended=False)
3033 def print_direct_lines(
3034 self,
3035 lines: _t.Iterable[str],
3036 stream: _t.TextIO | None = None,
3037 ):
3038 with _IO_LOCK:
3039 self._emit_lines(lines, stream, ignore_suspended=False)
3041 def start_task(self, task: Task):
3042 with _IO_LOCK:
3043 self._start_task(task)
3045 def start_subtask(self, parent: Task, task: Task):
3046 with _IO_LOCK:
3047 self._start_subtask(parent, task)
3049 def finish_task(self, task: Task, status: Task._Status):
3050 with _IO_LOCK:
3051 self._finish_task(task, status)
3053 def set_task_progress(
3054 self,
3055 task: Task,
3056 progress: float | None,
3057 done: str | None,
3058 total: str | None,
3059 ):
3060 with _IO_LOCK:
3061 task._progress = progress
3062 task._progress_done = done
3063 task._progress_total = total
3064 self._update_tasks()
3066 def set_task_comment(self, task: Task, comment: str | None, args):
3067 with _IO_LOCK:
3068 task._comment = comment
3069 task._comment_args = args
3070 task._cached_comment = None
3071 self._update_tasks()
3073 def suspend(self):
3074 with _IO_LOCK:
3075 self._suspend()
3077 def resume(self):
3078 with _IO_LOCK:
3079 self._resume()
3081 # Implementation.
3082 # These functions are always called under a lock.
3084 @functools.cached_property
3085 def _update_rate_us(self) -> int:
3086 update_rate_ms = max(self._theme.spinner_update_rate_ms, 1)
3087 while update_rate_ms < 50:
3088 update_rate_ms *= 2
3089 while update_rate_ms > 250:
3090 update_rate_ms //= 2
3091 return int(update_rate_ms * 1000)
3093 @property
3094 def _spinner_update_rate_us(self) -> int:
3095 return self._theme.spinner_update_rate_ms * 1000
3097 def _emit_lines(
3098 self,
3099 lines: _t.Iterable[str],
3100 stream: _t.TextIO | None = None,
3101 ignore_suspended: bool = False,
3102 ):
3103 stream = stream or self._term.ostream
3104 if self._suspended and not ignore_suspended:
3105 self._suspended_lines.append((list(lines), stream))
3106 else:
3107 self._clear_tasks()
3108 stream.writelines(lines)
3109 self._update_tasks(immediate_render=True)
3110 stream.flush()
3112 self._printed_some_lines = True
3114 def _suspend(self):
3115 self._suspended += 1
3117 if self._suspended == 1:
3118 self._clear_tasks()
3120 def _resume(self):
3121 self._suspended -= 1
3123 if self._suspended == 0:
3124 for lines, stream in self._suspended_lines:
3125 stream.writelines(lines)
3126 if self._suspended_lines:
3127 self._printed_some_lines = True
3128 self._suspended_lines.clear()
3130 self._update_tasks()
3132 if self._suspended < 0:
3133 yuio._logger.warning("unequal number of suspends and resumes")
3134 self._suspended = 0
3136 def _should_draw_interactive_tasks(self):
3137 should_draw_interactive_tasks = (
3138 self._term.color_support >= yuio.term.ColorSupport.ANSI
3139 and self._term.ostream_is_tty
3140 and yuio.term._is_foreground(self._term.ostream)
3141 )
3143 if (
3144 not should_draw_interactive_tasks and self._tasks_printed
3145 ) or self._seen_sigcont:
3146 # We were moved from foreground to background. There's no point in hiding
3147 # tasks now (shell printed something when user sent C-z), but we need
3148 # to make sure that we'll start rendering tasks from scratch whenever
3149 # user brings us to foreground again.
3150 self.rc.prepare(reset_term_pos=True)
3151 self._tasks_printed = 0
3152 self._seen_sigcont = False
3154 return should_draw_interactive_tasks
3156 def _start_task(self, task: Task):
3157 self._tasks.append(task)
3158 if self._should_draw_interactive_tasks():
3159 self._update_tasks()
3160 else:
3161 self._emit_lines(self._format_task(task).as_code(self._term.color_support))
3163 def _start_subtask(self, parent: Task, task: Task):
3164 parent._subtasks.append(task)
3165 if self._should_draw_interactive_tasks():
3166 self._update_tasks()
3167 else:
3168 self._emit_lines(self._format_task(task).as_code(self._term.color_support))
3170 def _finish_task(self, task: Task, status: Task._Status):
3171 if task._status != Task._Status.RUNNING:
3172 yuio._logger.warning("trying to change status of an already stopped task")
3173 return
3175 task._status = status
3176 for subtask in task._subtasks:
3177 if subtask._status == Task._Status.RUNNING:
3178 self._finish_task(subtask, status)
3180 if self._should_draw_interactive_tasks():
3181 if task in self._tasks:
3182 self._tasks.remove(task)
3183 self._emit_lines(
3184 self._format_task(task).as_code(self._term.color_support)
3185 )
3186 else:
3187 self._update_tasks()
3188 else:
3189 if task in self._tasks:
3190 self._tasks.remove(task)
3191 self._emit_lines(self._format_task(task).as_code(self._term.color_support))
3193 def _clear_tasks(self):
3194 if self._should_draw_interactive_tasks() and self._tasks_printed:
3195 self._rc.finalize()
3196 self._tasks_printed = 0
3198 def _update_tasks(self, immediate_render: bool = False):
3199 self._needs_update = True
3200 if immediate_render or not self._enable_bg_updates:
3201 self._show_tasks(immediate_render)
3203 def _show_tasks(
3204 self, immediate_render: bool = False, deadline_ns: int | None = None
3205 ):
3206 if (
3207 self._should_draw_interactive_tasks()
3208 and not self._suspended
3209 and (self._tasks or self._tasks_printed)
3210 ):
3211 start_ns = time.monotonic_ns()
3212 if deadline_ns is None:
3213 deadline_ns = start_ns + self.TASK_RENDER_TIMEOUT_NS
3214 now_us = start_ns // 1000
3215 now_us -= now_us % self._update_rate_us
3217 if not immediate_render and self._enable_bg_updates:
3218 next_update_us = self._last_update_time_us + self._update_rate_us
3219 if now_us < next_update_us:
3220 # Hard-limit update rate by `update_rate_ms`.
3221 return
3222 next_spinner_update_us = (
3223 self._last_update_time_us + self._spinner_update_rate_us
3224 )
3225 if not self._needs_update and now_us < next_spinner_update_us:
3226 # Tasks didn't change, and spinner state didn't change either,
3227 # so we can skip this update.
3228 return
3230 self._last_update_time_us = now_us
3231 self._spinner_state = now_us // self._spinner_update_rate_us
3232 self._tasks_printed = 0
3233 self._needs_update = False
3235 self._prepare_for_rendering_tasks()
3236 for task in self._tasks:
3237 self._draw_task(task, 0)
3238 self._renders += 1
3239 self._rc.set_final_pos(0, self._tasks_printed)
3241 now_ns = time.monotonic_ns()
3242 if not self._seen_sigcont and now_ns < deadline_ns:
3243 self._rc.render()
3244 else:
3245 # We have to skip this render: the process was suspended while we were
3246 # formatting tasks. Because of this, te position of the cursor
3247 # could've changed, so we need to reset rendering context and re-render.
3248 self._seen_sigcont = True
3250 def _prepare_for_rendering_tasks(self):
3251 self._rc.prepare()
3253 self.n_tasks = dict.fromkeys(Task._Status, 0)
3254 self.displayed_tasks = dict.fromkeys(Task._Status, 0)
3256 stack = self._tasks.copy()
3257 while stack:
3258 task = stack.pop()
3259 self.n_tasks[task._status] += 1
3260 stack.extend(task._subtasks)
3262 self.display_tasks = self.n_tasks.copy()
3263 total_tasks = sum(self.display_tasks.values())
3264 height = self._rc.height
3265 if total_tasks > height:
3266 height -= 1 # account for '+x more' message
3267 for status in Task._Status:
3268 to_hide = min(total_tasks - height, self.display_tasks[status])
3269 self.display_tasks[status] -= to_hide
3270 total_tasks -= to_hide
3271 if total_tasks <= height:
3272 break
3274 def _format_task(self, task: Task) -> yuio.string.ColorizedString:
3275 res = yuio.string.ColorizedString()
3277 ctx = task._status.value
3279 if decoration := self._theme.get_msg_decoration(
3280 "task", is_unicode=self._term.is_unicode
3281 ):
3282 res += self._theme.get_color(f"task/decoration:{ctx}")
3283 res += decoration
3285 res += self._format_task_msg(task)
3286 res += self._theme.get_color(f"task:{ctx}")
3287 res += " - "
3288 res += self._theme.get_color(f"task/progress:{ctx}")
3289 res += task._status.value
3290 res += self._theme.get_color(f"task:{ctx}")
3291 res += "\n"
3293 res += yuio.color.Color.NONE
3295 return res
3297 def _format_task_msg(self, task: Task) -> yuio.string.ColorizedString:
3298 if task._cached_msg is None:
3299 msg = yuio.string.colorize(
3300 task._msg,
3301 *task._args,
3302 default_color=f"task/heading:{task._status.value}",
3303 ctx=yuio.string.ReprContext(
3304 term=self._term,
3305 theme=self._theme,
3306 width=self._rc.width,
3307 ),
3308 )
3309 task._cached_msg = msg
3310 return task._cached_msg
3312 def _format_task_comment(self, task: Task) -> yuio.string.ColorizedString | None:
3313 if task._status is not Task._Status.RUNNING:
3314 return None
3315 if task._cached_comment is None and task._comment is not None:
3316 comment = yuio.string.colorize(
3317 task._comment,
3318 *(task._comment_args or ()),
3319 default_color=f"task/comment:{task._status.value}",
3320 ctx=yuio.string.ReprContext(
3321 term=self._term,
3322 theme=self._theme,
3323 width=self._rc.width,
3324 ),
3325 )
3326 task._cached_comment = comment
3327 return task._cached_comment
3329 def _draw_task(self, task: Task, indent: int):
3330 self.displayed_tasks[task._status] += 1
3332 self._tasks_printed += 1
3333 self._rc.move_pos(indent * 2, 0)
3334 self._draw_task_progressbar(task)
3335 self._rc.write(self._format_task_msg(task))
3336 self._draw_task_progress(task)
3337 if comment := self._format_task_comment(task):
3338 self._rc.set_color_path(f"task:{task._status.value}")
3339 self._rc.write(" - ")
3340 self._rc.write(comment)
3341 self._rc.new_line()
3343 for subtask in task._subtasks:
3344 self._draw_task(subtask, indent + 1)
3346 def _draw_task_progress(self, task: Task):
3347 if task._status in (Task._Status.DONE, Task._Status.ERROR):
3348 self._rc.set_color_path(f"task:{task._status.value}")
3349 self._rc.write(" - ")
3350 self._rc.set_color_path(f"task/progress:{task._status.value}")
3351 self._rc.write(task._status.name.lower())
3352 elif task._progress_done is not None:
3353 self._rc.set_color_path(f"task:{task._status.value}")
3354 self._rc.write(" - ")
3355 self._rc.set_color_path(f"task/progress:{task._status.value}")
3356 self._rc.write(task._progress_done)
3357 if task._progress_total is not None:
3358 self._rc.set_color_path(f"task:{task._status.value}")
3359 self._rc.write("/")
3360 self._rc.set_color_path(f"task/progress:{task._status.value}")
3361 self._rc.write(task._progress_total)
3363 def _draw_task_progressbar(self, task: Task):
3364 progress_bar_start_symbol = self._theme.get_msg_decoration(
3365 "progress_bar/start_symbol", is_unicode=self._term.is_unicode
3366 )
3367 progress_bar_end_symbol = self._theme.get_msg_decoration(
3368 "progress_bar/end_symbol", is_unicode=self._term.is_unicode
3369 )
3370 total_width = (
3371 self._theme.progress_bar_width
3372 - yuio.string.line_width(progress_bar_start_symbol)
3373 - yuio.string.line_width(progress_bar_end_symbol)
3374 )
3375 progress_bar_done_symbol = self._theme.get_msg_decoration(
3376 "progress_bar/done_symbol", is_unicode=self._term.is_unicode
3377 )
3378 progress_bar_pending_symbol = self._theme.get_msg_decoration(
3379 "progress_bar/pending_symbol", is_unicode=self._term.is_unicode
3380 )
3381 if task._status != Task._Status.RUNNING:
3382 self._rc.set_color_path(f"task/decoration:{task._status.value}")
3383 self._rc.write(
3384 self._theme.get_msg_decoration(
3385 "spinner/static_symbol", is_unicode=self._term.is_unicode
3386 )
3387 )
3388 elif (
3389 task._progress is None
3390 or total_width <= 1
3391 or not progress_bar_done_symbol
3392 or not progress_bar_pending_symbol
3393 ):
3394 self._rc.set_color_path(f"task/decoration:{task._status.value}")
3395 spinner_pattern = self._theme.get_msg_decoration(
3396 "spinner/pattern", is_unicode=self._term.is_unicode
3397 )
3398 if spinner_pattern:
3399 self._rc.write(
3400 spinner_pattern[self._spinner_state % len(spinner_pattern)]
3401 )
3402 else:
3403 transition_pattern = self._theme.get_msg_decoration(
3404 "progress_bar/transition_pattern", is_unicode=self._term.is_unicode
3405 )
3407 progress = max(0, min(1, task._progress))
3408 if transition_pattern:
3409 done_width = int(total_width * progress)
3410 transition_factor = 1 - (total_width * progress - done_width)
3411 transition_width = 1
3412 else:
3413 done_width = round(total_width * progress)
3414 transition_factor = 0
3415 transition_width = 0
3417 self._rc.set_color_path(f"task/progressbar:{task._status.value}")
3418 self._rc.write(progress_bar_start_symbol)
3420 done_color = yuio.color.Color.lerp(
3421 self._theme.get_color("task/progressbar/done/start"),
3422 self._theme.get_color("task/progressbar/done/end"),
3423 )
3425 for i in range(0, done_width):
3426 self._rc.set_color(done_color(i / (total_width - 1)))
3427 self._rc.write(progress_bar_done_symbol)
3429 if transition_pattern and done_width < total_width:
3430 self._rc.set_color(done_color(done_width / (total_width - 1)))
3431 self._rc.write(
3432 transition_pattern[
3433 int(len(transition_pattern) * transition_factor - 1)
3434 ]
3435 )
3437 pending_color = yuio.color.Color.lerp(
3438 self._theme.get_color("task/progressbar/pending/start"),
3439 self._theme.get_color("task/progressbar/pending/end"),
3440 )
3442 for i in range(done_width + transition_width, total_width):
3443 self._rc.set_color(pending_color(i / (total_width - 1)))
3444 self._rc.write(progress_bar_pending_symbol)
3446 self._rc.set_color_path(f"task/progressbar:{task._status.value}")
3447 self._rc.write(progress_bar_end_symbol)
3449 self._rc.set_color_path(f"task:{task._status.value}")
3450 self._rc.write(" ")
3453class _WrappedOutput(_t.TextIO): # pragma: no cover
3454 def __init__(self, wrapped: _t.TextIO):
3455 self.__wrapped = wrapped
3457 @property
3458 def mode(self) -> str:
3459 return self.__wrapped.mode
3461 @property
3462 def name(self) -> str:
3463 return self.__wrapped.name
3465 def close(self):
3466 self.__wrapped.close()
3468 @property
3469 def closed(self) -> bool:
3470 return self.__wrapped.closed
3472 def fileno(self) -> int:
3473 return self.__wrapped.fileno()
3475 def flush(self):
3476 self.__wrapped.flush()
3478 def isatty(self) -> bool:
3479 return self.__wrapped.isatty()
3481 def writable(self) -> bool:
3482 return self.__wrapped.writable()
3484 def write(self, s: str, /) -> int:
3485 _manager().print_direct(s, self.__wrapped)
3486 return len(s)
3488 def writelines(self, lines: _t.Iterable[str], /):
3489 _manager().print_direct_lines(lines, self.__wrapped)
3491 def readable(self) -> bool:
3492 return self.__wrapped.readable()
3494 def read(self, n: int = -1) -> str:
3495 return self.__wrapped.read(n)
3497 def readline(self, limit: int = -1) -> str:
3498 return self.__wrapped.readline(limit)
3500 def readlines(self, hint: int = -1) -> list[str]:
3501 return self.__wrapped.readlines(hint)
3503 def seek(self, offset: int, whence: int = 0) -> int:
3504 return self.__wrapped.seek(offset, whence)
3506 def seekable(self) -> bool:
3507 return self.__wrapped.seekable()
3509 def tell(self) -> int:
3510 return self.__wrapped.tell()
3512 def truncate(self, size: int | None = None) -> int:
3513 return self.__wrapped.truncate(size)
3515 def __enter__(self) -> _t.TextIO:
3516 return self.__wrapped.__enter__()
3518 def __exit__(self, exc_type, exc_val, exc_tb):
3519 self.__wrapped.__exit__(exc_type, exc_val, exc_tb)
3521 @property
3522 def buffer(self) -> _t.BinaryIO:
3523 return self.__wrapped.buffer
3525 @property
3526 def encoding(self) -> str:
3527 return self.__wrapped.encoding
3529 @property
3530 def errors(self) -> str | None:
3531 return self.__wrapped.errors
3533 @property
3534 def line_buffering(self) -> int:
3535 return self.__wrapped.line_buffering
3537 @property
3538 def newlines(self) -> _t.Any:
3539 return self.__wrapped.newlines