Coverage for yuio / io.py: 90%
1142 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-03 15:42 +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- :cli:env:`FORCE_NO_COLOR`: disable colored output,
17- :cli:env:`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:: rst
70.. autofunction:: hl
72.. autofunction:: br
74.. autofunction:: hr
76.. autofunction:: raw
79.. _percent-format:
81Formatting the output
82---------------------
84Yuio supports `printf-style formatting`__, similar to :mod:`logging`. If you're using
85Python 3.14 or later, you can also use `template strings`__.
87__ https://docs.python.org/3/library/stdtypes.html#printf-style-string-formatting
88__ https://docs.python.org/3/library/string.html#template-strings
90.. invisible-code-block: python
92 config = ...
94.. tab-set::
95 :sync-group: formatting-method
97 .. tab-item:: Printf-style formatting
98 :sync: printf
100 ``%s`` and ``%r`` specifiers are handled to respect colors and `rich repr protocol`__.
101 Additionally, they allow specifying flags to control whether rendered values should
102 be highlighted, and should they be rendered in multiple lines:
104 __ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
106 - ``#`` enables colors in repr (i.e. ``%#r``);
107 - ``+`` splits repr into multiple lines (i.e. ``%+r``, ``%#+r``).
109 .. code-block:: python
111 yuio.io.info("Loaded config: %#+r", config)
113 .. tab-item:: Template strings
114 :sync: template
116 When formatting template strings, default format specification is extended
117 to respect colors and `rich repr protocol`__. Additionally, it allows
118 specifying flags to control whether rendered values should be highlighted,
119 and should they be rendered in multiple lines:
121 __ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
123 - ``#`` enables colors in repr (i.e. ``{var:#}``);
124 - ``+`` splits repr into multiple lines (i.e. ``{var:+}``, ``{var:#+}``);
125 - unless explicit conversion is given (i.e. ``!s``, ``!r``, or ``!a``),
126 this format specification applies to objects that don't define
127 custom ``__format__`` method;
128 - full format specification is available :ref:`here <t-string-spec>`.
130 .. code-block:: python
132 yuio.io.info(t"Loaded config: {config!r:#+}")
134 .. note::
136 The formatting algorithm is as follows:
138 1. if formatting conversion is specified (i.e. ``!s``, ``!r``, or ``!a``),
139 the object is passed to
140 :meth:`ReprContext.convert() <yuio.string.ReprContext.convert>`;
141 2. otherwise, if object defines custom ``__format__`` method,
142 this method is used;
143 3. otherwise, we fall back to
144 :meth:`ReprContext.convert() <yuio.string.ReprContext.convert>`
145 with assumed conversion flag ``"s"``.
147To support highlighted formatting, define ``__colorized_str__``
148or ``__colorized_repr__`` on your class. See :ref:`pretty-protocol` for implementation
149details.
151To support rich repr protocol, define function ``__rich_repr__`` on your class.
152This method should return an iterable of tuples, as described in Rich__ documentation.
154__ https://rich.readthedocs.io/en/stable/pretty.html#rich-repr-protocol
157.. _color-tags:
159Coloring the output
160-------------------
162By default, all messages are colored according to their level (i.e. which function
163you use to print them).
165If you need inline colors, you can use special tags in your log messages:
167.. code-block:: python
169 yuio.io.info("Using the <c code>code</c> tag.")
171You can combine multiple colors in the same tag:
173.. code-block:: python
175 yuio.io.info("<c bold green>Success!</c>")
177Only tags that appear in the message itself are processed:
179.. tab-set::
180 :sync-group: formatting-method
182 .. tab-item:: Printf-style formatting
183 :sync: printf
185 .. code-block:: python
187 yuio.io.info("Tags in this message --> %s are printed as-is", "<c color>")
189 .. tab-item:: Template strings
190 :sync: template
192 .. code-block:: python
194 value = "<c color>"
195 yuio.io.info(t"Tags in this message --> {value} are printed as-is")
197For highlighting inline code, Yuio supports parsing CommonMark's backticks:
199.. code-block:: python
201 yuio.io.info("Using the `backticks`.")
202 yuio.io.info("Using the `` nested `backticks` ``")
204Any punctuation symbol can be escaped with backslash:
206.. code-block:: python
208 yuio.io.info(r"\\`\\<c red> this is normal text \\</c>\\`.")
210See full list of tags in :ref:`yuio.theme <common-tags>`.
213Message channels
214----------------
216.. autoclass:: MessageChannel
217 :members:
220Formatting utilities
221--------------------
223There are several :ref:`formatting utilities <formatting-utilities>` defined
224in :mod:`yuio.string` and re-exported in :mod:`yuio.io`. These utilities
225perform various formatting tasks when converted to strings, allowing you to lazily
226build more complex messages.
229Indicating progress
230-------------------
232You can use the :class:`Task` class to indicate status and progress
233of some task.
235.. autoclass:: TaskBase
236 :members:
238.. autoclass:: Task
239 :members:
242Querying user input
243-------------------
245If you need to get something from the user, :func:`ask` is the way to do it.
247.. autofunction:: ask
249.. autofunction:: wait_for_user
251You can also prompt the user to edit something with the :func:`edit` function:
253.. autofunction:: edit
255.. autofunction:: detect_editor
257If you need to spawn a sub-shell for user to interact with, you can use :func:`shell`:
259.. autofunction:: shell
261.. autofunction:: detect_shell
263All of these functions throw an error if something goes wrong:
265.. autoclass:: UserIoError
268Suspending the output
269---------------------
271You can temporarily disable printing of tasks and messages
272using the :class:`SuspendOutput` context manager.
274.. autoclass:: SuspendOutput
275 :members:
278Python's `logging` and yuio
279---------------------------
281If you want to direct messages from the :mod:`logging` to Yuio,
282you can add a :class:`Handler`:
284.. autoclass:: Handler
286.. autoclass:: Formatter
289Helpers
290-------
292.. autofunction:: make_repr_context
294.. type:: ExcInfo
295 :canonical: tuple[type[BaseException] | None, BaseException | None, types.TracebackType | None]
297 Exception information as returned by :func:`sys.exc_info`.
300Re-imports
301----------
303.. type:: And
304 :no-index:
306 Alias of :obj:`yuio.string.And`.
308.. type:: ColorizedString
309 :no-index:
311 Alias of :obj:`yuio.string.ColorizedString`.
313.. type:: Format
314 :no-index:
316 Alias of :obj:`yuio.string.Format`.
318.. type:: Hl
319 :no-index:
321 Alias of :obj:`yuio.string.Hl`.
323.. type:: Hr
324 :no-index:
326 Alias of :obj:`yuio.string.Hr`.
328.. type:: Indent
329 :no-index:
331 Alias of :obj:`yuio.string.Indent`.
333.. type:: JoinRepr
334 :no-index:
336 Alias of :obj:`yuio.string.JoinRepr`.
338.. type:: JoinStr
339 :no-index:
341 Alias of :obj:`yuio.string.JoinStr`.
343.. type:: Link
344 :no-index:
346 Alias of :obj:`yuio.string.Link`.
348.. type:: Md
349 :no-index:
351 Alias of :obj:`yuio.string.Md`.
353.. type:: Rst
354 :no-index:
356 Alias of :obj:`yuio.string.Rst`.
358.. type:: Or
359 :no-index:
361 Alias of :obj:`yuio.string.Or`.
363.. type:: Repr
364 :no-index:
366 Alias of :obj:`yuio.string.Repr`.
368.. type:: Stack
369 :no-index:
371 Alias of :obj:`yuio.string.Stack`.
373.. type:: TypeRepr
374 :no-index:
376 Alias of :obj:`yuio.string.TypeRepr`.
378.. type:: WithBaseColor
379 :no-index:
381 Alias of :obj:`yuio.string.WithBaseColor`.
383.. type:: Wrap
384 :no-index:
386 Alias of :obj:`yuio.string.Wrap`.
389"""
391from __future__ import annotations
393import abc
394import atexit
395import functools
396import logging
397import os
398import re
399import shutil
400import string
401import subprocess
402import sys
403import tempfile
404import textwrap
405import threading
406import time
407import traceback
408import types
409from logging import LogRecord
411import yuio.color
412import yuio.hl
413import yuio.parse
414import yuio.string
415import yuio.term
416import yuio.theme
417import yuio.widget
418from yuio._dist.dsu import DisjointSet as _DisjointSet
419from yuio.string import (
420 And,
421 ColorizedString,
422 Format,
423 Hl,
424 Hr,
425 Indent,
426 JoinRepr,
427 JoinStr,
428 Link,
429 Md,
430 Or,
431 Repr,
432 Rst,
433 Stack,
434 TypeRepr,
435 WithBaseColor,
436 Wrap,
437)
438from yuio.util import dedent as _dedent
440import yuio._typing_ext as _tx
441from typing import TYPE_CHECKING
442from typing import ClassVar as _ClassVar
444if TYPE_CHECKING:
445 import typing_extensions as _t
446else:
447 from yuio import _typing as _t
449__all__ = [
450 "And",
451 "ColorizedString",
452 "ExcInfo",
453 "Format",
454 "Formatter",
455 "Handler",
456 "Hl",
457 "Hr",
458 "Indent",
459 "JoinRepr",
460 "JoinStr",
461 "Link",
462 "Md",
463 "MessageChannel",
464 "Or",
465 "Repr",
466 "Rst",
467 "Stack",
468 "SuspendOutput",
469 "Task",
470 "TaskBase",
471 "TypeRepr",
472 "UserIoError",
473 "WithBaseColor",
474 "Wrap",
475 "ask",
476 "br",
477 "detect_editor",
478 "detect_shell",
479 "edit",
480 "error",
481 "error_with_tb",
482 "failure",
483 "failure_with_tb",
484 "get_term",
485 "get_theme",
486 "heading",
487 "hl",
488 "hr",
489 "info",
490 "make_repr_context",
491 "md",
492 "orig_stderr",
493 "orig_stdout",
494 "raw",
495 "restore_streams",
496 "rst",
497 "setup",
498 "shell",
499 "streams_wrapped",
500 "success",
501 "wait_for_user",
502 "warning",
503 "wrap_streams",
504]
506T = _t.TypeVar("T")
507M = _t.TypeVar("M", default=_t.Never)
508S = _t.TypeVar("S", default=str)
510ExcInfo: _t.TypeAlias = tuple[
511 type[BaseException] | None,
512 BaseException | None,
513 types.TracebackType | None,
514]
515"""
516Exception information as returned by :func:`sys.exc_info`.
518"""
521_IO_LOCK = threading.RLock()
522_IO_MANAGER: _IoManager | None = None
523_STREAMS_WRAPPED: bool = False
524_ORIG_STDERR: _t.TextIO | None = None
525_ORIG_STDOUT: _t.TextIO | None = None
528def _manager() -> _IoManager:
529 global _IO_MANAGER
531 if _IO_MANAGER is None:
532 with _IO_LOCK:
533 if _IO_MANAGER is None:
534 _IO_MANAGER = _IoManager()
535 return _IO_MANAGER
538class UserIoError(yuio.PrettyException, IOError):
539 """
540 Raised when interaction with user fails.
542 """
545def setup(
546 *,
547 term: yuio.term.Term | None = None,
548 theme: (
549 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
550 ) = None,
551 wrap_stdio: bool = True,
552):
553 """
554 Initial setup of the logging facilities.
556 :param term:
557 terminal that will be used for output.
559 If not passed, the global terminal is not re-configured;
560 the default is to use a term attached to :data:`sys.stderr`.
561 :param theme:
562 either a theme that will be used for output, or a theme constructor that takes
563 a :class:`~yuio.term.Term` and returns a theme.
565 If not passed, the global theme is not re-configured; the default is to use
566 :class:`yuio.theme.DefaultTheme` then.
567 :param wrap_stdio:
568 if set to :data:`True`, wraps :data:`sys.stdout` and :data:`sys.stderr`
569 in a special wrapper that ensures better interaction
570 with Yuio's progress bars and widgets.
572 .. note::
574 If you're working with some other library that wraps :data:`sys.stdout`
575 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
577 .. _colorama: https://github.com/tartley/colorama
579 .. warning::
581 This operation is not atomic. Call this function before creating new threads
582 that use :mod:`yuio.io` or output streams to avoid race conditions.
584 """
586 global _IO_MANAGER
588 if not (manager := _IO_MANAGER):
589 with _IO_LOCK:
590 if not (manager := _IO_MANAGER):
591 _IO_MANAGER = _IoManager(term, theme)
592 if manager is not None:
593 manager.setup(term, theme)
595 if wrap_stdio:
596 wrap_streams()
599def get_term() -> yuio.term.Term:
600 """
601 Get the global instance of :class:`~yuio.term.Term` that is used
602 with :mod:`yuio.io`.
604 If global setup wasn't performed, this function implicitly performs it.
606 :returns:
607 Instance of :class:`~yuio.term.Term` that's used to print messages and tasks.
609 """
611 return _manager().term
614def get_theme() -> yuio.theme.Theme:
615 """
616 Get the global instance of :class:`~yuio.theme.Theme`
617 that is used with :mod:`yuio.io`.
619 If global setup wasn't performed, this function implicitly performs it.
621 :returns:
622 Instance of :class:`~yuio.theme.Theme` that's used to format messages and tasks.
624 """
626 return _manager().theme
629def make_repr_context(
630 *,
631 term: yuio.term.Term | None = None,
632 to_stdout: bool = False,
633 to_stderr: bool = False,
634 theme: yuio.theme.Theme | None = None,
635 multiline: bool | None = None,
636 highlighted: bool | None = None,
637 max_depth: int | None = None,
638 width: int | None = None,
639) -> yuio.string.ReprContext:
640 """
641 Create new :class:`~yuio.string.ReprContext` for the given term and theme.
643 .. warning::
645 :class:`~yuio.string.ReprContext`\\ s are not thread safe. As such,
646 you shouldn't create them for long term use.
648 :param term:
649 terminal where to print this message. If not given, terminal from
650 :func:`get_term` is used.
651 :param to_stdout:
652 shortcut for setting `term` to ``stdout``.
653 :param to_stderr:
654 shortcut for setting `term` to ``stderr``.
655 :param theme:
656 theme used to format the message. If not given, theme from
657 :func:`get_theme` is used.
658 :param multiline:
659 sets initial value for
660 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
661 Default is :data:`False`.
662 :param highlighted:
663 sets initial value for
664 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
665 Default is :data:`False`.
666 :param max_depth:
667 sets initial value for
668 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
669 Default is :data:`False`.
670 :param width:
671 sets initial value for
672 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
673 If not given, uses current terminal width or
674 :attr:`Theme.fallback_width <yuio.theme.Theme.fallback_width>`
675 depending on whether `term` is attached to a TTY device and whether colors
676 are supported by the target terminal.
678 """
680 if (term is not None) + to_stdout + to_stderr > 1:
681 names = []
682 if term is not None:
683 names.append("term")
684 if to_stdout:
685 names.append("to_stdout")
686 if to_stderr:
687 names.append("to_stderr")
688 raise TypeError(f"{And(names)} can't be given together")
690 manager = _manager()
692 theme = manager.theme
693 if term is None:
694 if to_stdout:
695 term = manager.out_term
696 elif to_stderr:
697 term = manager.err_term
698 else:
699 term = manager.term
700 if width is None and (term.ostream_is_tty or term.supports_colors):
701 width = manager.rc.canvas_width
703 return yuio.string.ReprContext(
704 term=term,
705 theme=theme,
706 multiline=multiline,
707 highlighted=highlighted,
708 max_depth=max_depth,
709 width=width,
710 )
713def wrap_streams():
714 """
715 Wrap :data:`sys.stdout` and :data:`sys.stderr` so that they honor
716 Yuio tasks and widgets. If strings are already wrapped, this function
717 has no effect.
719 .. note::
721 If you're working with some other library that wraps :data:`sys.stdout`
722 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
724 .. seealso::
726 :func:`setup`.
728 .. _colorama: https://github.com/tartley/colorama
730 """
732 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
734 if _STREAMS_WRAPPED:
735 return
737 with _IO_LOCK:
738 if _STREAMS_WRAPPED: # pragma: no cover
739 return
741 if yuio.term._output_is_tty(sys.stdout):
742 _ORIG_STDOUT, sys.stdout = sys.stdout, _YuioOutputWrapper(sys.stdout)
743 if yuio.term._output_is_tty(sys.stderr):
744 _ORIG_STDERR, sys.stderr = sys.stderr, _YuioOutputWrapper(sys.stderr)
745 _STREAMS_WRAPPED = True
747 atexit.register(restore_streams)
750def restore_streams():
751 """
752 Restore wrapped streams. If streams weren't wrapped, this function
753 has no effect.
755 .. seealso::
757 :func:`wrap_streams`, :func:`setup`
759 """
761 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
763 if not _STREAMS_WRAPPED:
764 return
766 with _IO_LOCK:
767 if not _STREAMS_WRAPPED: # pragma: no cover
768 return
770 if _ORIG_STDOUT is not None:
771 sys.stdout = _ORIG_STDOUT
772 _ORIG_STDOUT = None
773 if _ORIG_STDERR is not None:
774 sys.stderr = _ORIG_STDERR
775 _ORIG_STDERR = None
776 _STREAMS_WRAPPED = False
779def streams_wrapped() -> bool:
780 """
781 Check if :data:`sys.stdout` and :data:`sys.stderr` are wrapped.
782 See :func:`setup`.
784 :returns:
785 :data:`True` is streams are currently wrapped, :data:`False` otherwise.
787 """
789 return _STREAMS_WRAPPED
792def orig_stderr() -> _t.TextIO:
793 """
794 Return the original :data:`sys.stderr` before wrapping.
796 """
798 return _ORIG_STDERR or sys.stderr
801def orig_stdout() -> _t.TextIO:
802 """
803 Return the original :data:`sys.stdout` before wrapping.
805 """
807 return _ORIG_STDOUT or sys.stdout
810@_t.overload
811def info(msg: _t.LiteralString, /, *args, **kwargs): ...
812@_t.overload
813def info(msg: yuio.string.ToColorable, /, **kwargs): ...
814def info(msg: yuio.string.ToColorable, /, *args, **kwargs):
815 """info(msg: typing.LiteralString, /, *args, **kwargs)
816 info(msg: ~string.templatelib.Template, /, **kwargs) ->
817 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
819 Print an info message.
821 :param msg:
822 message to print.
823 :param args:
824 arguments for ``%``\\ -formatting the message.
825 :param kwargs:
826 any additional keyword arguments will be passed to :func:`raw`.
828 """
830 msg_colorable = yuio.string._to_colorable(msg, args)
831 kwargs.setdefault("tag", "info")
832 kwargs.setdefault("wrap", True)
833 kwargs.setdefault("add_newline", True)
834 raw(msg_colorable, **kwargs)
837@_t.overload
838def warning(msg: _t.LiteralString, /, *args, **kwargs): ...
839@_t.overload
840def warning(msg: yuio.string.ToColorable, /, **kwargs): ...
841def warning(msg: yuio.string.ToColorable, /, *args, **kwargs):
842 """warning(msg: typing.LiteralString, /, *args, **kwargs)
843 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
844 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
846 Print a warning message.
848 :param msg:
849 message to print.
850 :param args:
851 arguments for ``%``\\ -formatting the message.
852 :param kwargs:
853 any additional keyword arguments will be passed to :func:`raw`.
855 """
857 msg_colorable = yuio.string._to_colorable(msg, args)
858 kwargs.setdefault("tag", "warning")
859 kwargs.setdefault("wrap", True)
860 kwargs.setdefault("add_newline", True)
861 raw(msg_colorable, **kwargs)
864@_t.overload
865def success(msg: _t.LiteralString, /, *args, **kwargs): ...
866@_t.overload
867def success(msg: yuio.string.ToColorable, /, **kwargs): ...
868def success(msg: yuio.string.ToColorable, /, *args, **kwargs):
869 """success(msg: typing.LiteralString, /, *args, **kwargs)
870 success(msg: ~string.templatelib.Template, /, **kwargs) ->
871 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
873 Print a success message.
875 :param msg:
876 message to print.
877 :param args:
878 arguments for ``%``\\ -formatting the message.
879 :param kwargs:
880 any additional keyword arguments will be passed to :func:`raw`.
882 """
884 msg_colorable = yuio.string._to_colorable(msg, args)
885 kwargs.setdefault("tag", "success")
886 kwargs.setdefault("wrap", True)
887 kwargs.setdefault("add_newline", True)
888 raw(msg_colorable, **kwargs)
891@_t.overload
892def error(msg: _t.LiteralString, /, *args, **kwargs): ...
893@_t.overload
894def error(msg: yuio.string.ToColorable, /, **kwargs): ...
895def error(msg: yuio.string.ToColorable, /, *args, **kwargs):
896 """error(msg: typing.LiteralString, /, *args, **kwargs)
897 error(msg: ~string.templatelib.Template, /, **kwargs) ->
898 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
900 Print an error message.
902 :param msg:
903 message to print.
904 :param args:
905 arguments for ``%``\\ -formatting the message.
906 :param kwargs:
907 any additional keyword arguments will be passed to :func:`raw`.
909 """
911 msg_colorable = yuio.string._to_colorable(msg, args)
912 kwargs.setdefault("tag", "error")
913 kwargs.setdefault("wrap", True)
914 kwargs.setdefault("add_newline", True)
915 raw(msg_colorable, **kwargs)
918@_t.overload
919def error_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
920@_t.overload
921def error_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
922def error_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
923 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
924 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
925 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
927 Print an error message and capture the current exception.
929 Call this function in the ``except`` clause of a ``try`` block
930 or in an ``__exit__`` function of a context manager to attach
931 current exception details to the log message.
933 :param msg:
934 message to print.
935 :param args:
936 arguments for ``%``\\ -formatting the message.
937 :param kwargs:
938 any additional keyword arguments will be passed to :func:`raw`.
940 """
942 msg_colorable = yuio.string._to_colorable(msg, args)
943 kwargs.setdefault("tag", "error")
944 kwargs.setdefault("wrap", True)
945 kwargs.setdefault("add_newline", True)
946 kwargs.setdefault("exc_info", True)
947 raw(msg_colorable, **kwargs)
950@_t.overload
951def failure(msg: _t.LiteralString, /, *args, **kwargs): ...
952@_t.overload
953def failure(msg: yuio.string.ToColorable, /, **kwargs): ...
954def failure(msg: yuio.string.ToColorable, /, *args, **kwargs):
955 """failure(msg: typing.LiteralString, /, *args, **kwargs)
956 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
957 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
959 Print a failure message.
961 :param msg:
962 message to print.
963 :param args:
964 arguments for ``%``\\ -formatting the message.
965 :param kwargs:
966 any additional keyword arguments will be passed to :func:`raw`.
968 """
970 msg_colorable = yuio.string._to_colorable(msg, args)
971 kwargs.setdefault("tag", "failure")
972 kwargs.setdefault("wrap", True)
973 kwargs.setdefault("add_newline", True)
974 raw(msg_colorable, **kwargs)
977@_t.overload
978def failure_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
979@_t.overload
980def failure_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
981def failure_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
982 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
983 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
984 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
986 Print a failure message and capture the current exception.
988 Call this function in the ``except`` clause of a ``try`` block
989 or in an ``__exit__`` function of a context manager to attach
990 current exception details to the log message.
992 :param msg:
993 message to print.
994 :param args:
995 arguments for ``%``\\ -formatting the message.
996 :param kwargs:
997 any additional keyword arguments will be passed to :func:`raw`.
999 """
1001 msg_colorable = yuio.string._to_colorable(msg, args)
1002 kwargs.setdefault("tag", "failure")
1003 kwargs.setdefault("wrap", True)
1004 kwargs.setdefault("add_newline", True)
1005 kwargs.setdefault("exc_info", True)
1006 raw(msg_colorable, **kwargs)
1009@_t.overload
1010def heading(msg: _t.LiteralString, /, *args, level: int = 1, **kwargs): ...
1011@_t.overload
1012def heading(msg: yuio.string.ToColorable, /, *, level: int = 1, **kwargs): ...
1013def heading(msg: yuio.string.ToColorable, /, *args, level: int = 1, **kwargs):
1014 """heading(msg: typing.LiteralString, /, *args, level: int = 1, **kwargs)
1015 heading(msg: ~string.templatelib.Template, /, *, level: int = 1, **kwargs) ->
1016 heading(msg: ~yuio.string.ToColorable, /, *, level: int = 1, **kwargs) ->
1018 Print a heading message.
1020 :param msg:
1021 message to print.
1022 :param args:
1023 arguments for ``%``\\ -formatting the message.
1024 :param level:
1025 level of the heading.
1026 :param kwargs:
1027 any additional keyword arguments will be passed to :func:`raw`.
1029 """
1031 msg_colorable = yuio.string._to_colorable(msg, args)
1032 level = kwargs.pop("level", 1)
1033 kwargs.setdefault("heading", True)
1034 kwargs.setdefault("tag", f"heading/{level}")
1035 kwargs.setdefault("wrap", True)
1036 kwargs.setdefault("add_newline", True)
1037 raw(msg_colorable, **kwargs)
1040def md(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs):
1041 """
1042 Print a markdown-formatted text.
1044 Yuio supports all CommonMark block markup except tables.
1045 See :mod:`yuio.md` for more info.
1047 :param msg:
1048 message to print.
1049 :param dedent:
1050 whether to remove leading indent from `msg`.
1051 :param allow_headings:
1052 whether to render headings as actual headings or as paragraphs.
1053 :param kwargs:
1054 any additional keyword arguments will be passed to :func:`raw`.
1056 """
1058 info(
1059 yuio.string.Md(msg, dedent=dedent, allow_headings=allow_headings),
1060 **kwargs,
1061 )
1064def rst(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs):
1065 """
1066 Print a RST-formatted text.
1068 Yuio supports all RST block markup except tables and field lists.
1069 See :mod:`yuio.rst` for more info.
1071 :param msg:
1072 message to print.
1073 :param dedent:
1074 whether to remove leading indent from `msg`.
1075 :param allow_headings:
1076 whether to render headings as actual headings or as paragraphs.
1077 :param kwargs:
1078 any additional keyword arguments will be passed to :func:`raw`.
1080 """
1082 info(
1083 yuio.string.Rst(msg, dedent=dedent, allow_headings=allow_headings),
1084 **kwargs,
1085 )
1088def br(**kwargs):
1089 """
1090 Print an empty string.
1092 :param kwargs:
1093 any additional keyword arguments will be passed to :func:`raw`.
1095 """
1097 raw("\n", **kwargs)
1100@_t.overload
1101def hr(msg: _t.LiteralString = "", /, *args, weight: int | str = 1, **kwargs): ...
1102@_t.overload
1103def hr(msg: yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs): ...
1104def hr(msg: yuio.string.ToColorable = "", /, *args, weight: int | str = 1, **kwargs):
1105 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
1106 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
1107 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
1109 Print a horizontal ruler.
1111 :param msg:
1112 message to print in the middle of the ruler.
1113 :param args:
1114 arguments for ``%``\\ -formatting the message.
1115 :param weight:
1116 weight or style of the ruler:
1118 - ``0`` prints no ruler (but still prints centered text),
1119 - ``1`` prints normal ruler,
1120 - ``2`` prints bold ruler.
1122 Additional styles can be added through
1123 :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`.
1124 :param kwargs:
1125 any additional keyword arguments will be passed to :func:`raw`.
1127 """
1129 info(yuio.string.Hr(yuio.string._to_colorable(msg, args), weight=weight), **kwargs)
1132@_t.overload
1133def hl(
1134 msg: _t.LiteralString,
1135 /,
1136 *args,
1137 syntax: str,
1138 dedent: bool = True,
1139 **kwargs,
1140): ...
1141@_t.overload
1142def hl(
1143 msg: str,
1144 /,
1145 *,
1146 syntax: str,
1147 dedent: bool = True,
1148 **kwargs,
1149): ...
1150def hl(
1151 msg: str,
1152 /,
1153 *args,
1154 syntax: str,
1155 dedent: bool = True,
1156 **kwargs,
1157):
1158 """hl(msg: typing.LiteralString, /, *args, syntax: str, dedent: bool = True, **kwargs)
1159 hl(msg: str, /, *, syntax: str, dedent: bool = True, **kwargs) ->
1161 Print highlighted code. See :mod:`yuio.hl` for more info.
1163 :param msg:
1164 code to highlight.
1165 :param args:
1166 arguments for ``%``-formatting the highlighted code.
1167 :param syntax:
1168 name of syntax or a :class:`~yuio.hl.SyntaxHighlighter` instance.
1169 :param dedent:
1170 whether to remove leading indent from `msg`.
1171 :param kwargs:
1172 any additional keyword arguments will be passed to :func:`raw`.
1174 """
1176 info(yuio.string.Hl(msg, *args, syntax=syntax, dedent=dedent), **kwargs)
1179def raw(
1180 msg: yuio.string.Colorable,
1181 /,
1182 *,
1183 ignore_suspended: bool = False,
1184 tag: str | None = None,
1185 exc_info: ExcInfo | bool | None = None,
1186 add_newline: bool = False,
1187 heading: bool = False,
1188 wrap: bool = False,
1189 ctx: yuio.string.ReprContext | None = None,
1190 term: yuio.term.Term | None = None,
1191 to_stdout: bool = False,
1192 to_stderr: bool = False,
1193 theme: yuio.theme.Theme | None = None,
1194 multiline: bool | None = None,
1195 highlighted: bool | None = None,
1196 max_depth: int | None = None,
1197 width: int | None = None,
1198):
1199 """
1200 Print any :class:`~yuio.string.ToColorable`.
1202 This is a bridge between :mod:`yuio.io` and lower-level
1203 modules like :mod:`yuio.string`.
1205 :param msg:
1206 message to print.
1207 :param ignore_suspended:
1208 whether to ignore :class:`SuspendOutput` context.
1209 :param tag:
1210 tag that will be used to add color and decoration to the message.
1212 Decoration is looked up by path :samp:`{tag}`
1213 (see :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`),
1214 and color is looked up by path :samp:`msg/text:{tag}`
1215 (see :attr:`Theme.colors <yuio.theme.Theme.colors>`).
1216 :param exc_info:
1217 either a boolean indicating that the current exception
1218 should be captured, or a tuple of three elements, as returned
1219 by :func:`sys.exc_info`.
1220 :param add_newline:
1221 adds newline after the message.
1222 :param heading:
1223 whether to separate message by extra newlines.
1225 If :data:`True`, adds extra newline after the message; if this is not the
1226 first message printed so far, adds another newline before the message.
1227 :param wrap:
1228 whether to wrap message before printing it.
1229 :param ctx:
1230 :class:`~yuio.string.ReprContext` that should be used for formatting
1231 and printing the message.
1232 :param term:
1233 if `ctx` is not given, sets terminal where to print this message. Default is
1234 to use :func:`get_term`.
1235 :param to_stdout:
1236 shortcut for setting `term` to ``stdout``.
1237 :param to_stderr:
1238 shortcut for setting `term` to ``stderr``.
1239 :param theme:
1240 if `ctx` is not given, sets theme used to format the message. Default is
1241 to use :func:`get_theme`.
1242 :param multiline:
1243 if `ctx` is not given, sets initial value for
1244 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
1245 Default is :data:`False`.
1246 :param highlighted:
1247 if `ctx` is not given, sets initial value for
1248 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
1249 Default is :data:`False`.
1250 :param max_depth:
1251 if `ctx` is not given, sets initial value for
1252 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
1253 Default is :data:`False`.
1254 :param width:
1255 if `ctx` is not given, sets initial value for
1256 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
1257 If not given, uses current terminal width
1258 or :attr:`Theme.fallback_width <yuio.theme.Theme.fallback_width>`
1259 if terminal width can't be established.
1261 """
1263 if (ctx is not None) + (term is not None) + to_stdout + to_stderr > 1:
1264 names = []
1265 if ctx is not None:
1266 names.append("ctx")
1267 if term is not None:
1268 names.append("term")
1269 if to_stdout:
1270 names.append("to_stdout")
1271 if to_stderr:
1272 names.append("to_stderr")
1273 raise TypeError(f"{And(names)} can't be given together")
1275 manager = _manager()
1277 if ctx is None:
1278 ctx = make_repr_context(
1279 term=term,
1280 to_stdout=to_stdout,
1281 to_stderr=to_stderr,
1282 theme=theme,
1283 multiline=multiline,
1284 highlighted=highlighted,
1285 max_depth=max_depth,
1286 width=width,
1287 )
1289 if tag and (decoration := ctx.get_msg_decoration(tag)):
1290 indent = yuio.string.ColorizedString(
1291 [ctx.get_color(f"msg/decoration:{tag}"), decoration]
1292 )
1293 continuation_indent = " " * indent.width
1294 else:
1295 indent = ""
1296 continuation_indent = ""
1298 if tag:
1299 msg = yuio.string.WithBaseColor(
1300 msg, base_color=ctx.get_color(f"msg/text:{tag}")
1301 )
1303 if wrap:
1304 msg = yuio.string.Wrap(
1305 msg,
1306 indent=indent,
1307 continuation_indent=continuation_indent,
1308 )
1309 elif indent or continuation_indent:
1310 msg = yuio.string.Indent(
1311 msg,
1312 indent=indent,
1313 continuation_indent=continuation_indent,
1314 )
1316 msg = ctx.str(msg)
1318 if add_newline:
1319 msg.append_color(yuio.color.Color.NONE)
1320 msg.append_str("\n")
1322 if exc_info is True:
1323 exc_info = sys.exc_info()
1324 elif exc_info is False or exc_info is None:
1325 exc_info = None
1326 elif isinstance(exc_info, BaseException):
1327 exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
1328 elif not isinstance(exc_info, tuple) or len(exc_info) != 3:
1329 raise ValueError(f"invalid exc_info {exc_info!r}")
1330 if exc_info is not None and exc_info != (None, None, None):
1331 tb = "".join(traceback.format_exception(*exc_info))
1332 highlighter, syntax_name = yuio.hl.get_highlighter("python-traceback")
1333 msg += highlighter.highlight(tb, theme=ctx.theme, syntax=syntax_name).indent()
1335 manager.print(
1336 msg.as_code(ctx.term.color_support),
1337 ctx.term,
1338 ignore_suspended=ignore_suspended,
1339 heading=heading,
1340 )
1343class _AskWidget(yuio.widget.Widget[T], _t.Generic[T]):
1344 _layout: yuio.widget.VerticalLayout[T]
1346 def __init__(self, prompt: yuio.string.Colorable, widget: yuio.widget.Widget[T]):
1347 self._prompt = yuio.widget.Text(prompt)
1348 self._error: Exception | None = None
1349 self._inner = widget
1351 def event(self, e: yuio.widget.KeyboardEvent, /) -> yuio.widget.Result[T] | None:
1352 try:
1353 result = self._inner.event(e)
1354 except yuio.parse.ParsingError as err:
1355 self._error = err
1356 else:
1357 self._error = None
1358 return result
1360 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1361 builder = (
1362 yuio.widget.VerticalLayoutBuilder()
1363 .add(self._prompt)
1364 .add(self._inner, receive_events=True)
1365 )
1366 if self._error is not None:
1367 rc.bell()
1368 error_msg = yuio.string.colorize(
1369 "<c msg/decoration:error>▲</c> %s",
1370 yuio.string.Indent(self._error, indent=0, continuation_indent=2),
1371 default_color="msg/text:error",
1372 ctx=rc.make_repr_context(),
1373 )
1374 builder = builder.add(yuio.widget.Text(error_msg))
1376 self._layout = builder.build()
1377 return self._layout.layout(rc)
1379 def draw(self, rc: yuio.widget.RenderContext, /):
1380 self._layout.draw(rc)
1382 @property
1383 def help_data(self) -> yuio.widget.WidgetHelp:
1384 return self._inner.help_data
1387class _AskMeta(type):
1388 __hint = None
1390 @_t.overload
1391 def __call__(
1392 cls: type[ask[S]],
1393 msg: _t.LiteralString,
1394 /,
1395 *args,
1396 default: M | yuio.Missing = yuio.MISSING,
1397 default_non_interactive: _t.Any = yuio.MISSING,
1398 parser: yuio.parse.Parser[S] | None = None,
1399 input_description: str | None = None,
1400 default_description: str | None = None,
1401 ) -> S | M: ...
1402 @_t.overload
1403 def __call__(
1404 cls: type[ask[S]],
1405 msg: str,
1406 /,
1407 *,
1408 default: M | yuio.Missing = yuio.MISSING,
1409 default_non_interactive: _t.Any = yuio.MISSING,
1410 parser: yuio.parse.Parser[S] | None = None,
1411 input_description: str | None = None,
1412 default_description: str | None = None,
1413 ) -> S | M: ...
1414 def __call__(cls, *args, **kwargs):
1415 if "parser" not in kwargs:
1416 hint = cls.__hint
1417 if hint is None:
1418 hint = str
1419 kwargs["parser"] = yuio.parse.from_type_hint(hint)
1420 return _ask(*args, **kwargs)
1422 def __getitem(cls, ty):
1423 return _AskMeta("ask", (), {"_AskMeta__hint": ty})
1425 # A dirty hack to hide `__getitem__` from type checkers. `ask` should look like
1426 # an ordinary class with overloaded `__new__` for the magic to work.
1427 locals()["__getitem__"] = __getitem
1429 def __repr__(cls) -> str:
1430 if cls.__hint is None:
1431 return cls.__name__
1432 else:
1433 return f"{cls.__name__}[{_tx.type_repr(cls.__hint)}]"
1436@_t.final
1437class ask(_t.Generic[S], metaclass=_AskMeta):
1438 """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
1439 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
1441 Ask user to provide an input, parse it and return a value.
1443 If current terminal is not interactive, return default if one is present,
1444 or raise a :class:`UserIoError`.
1446 .. vhs:: /_tapes/questions.tape
1447 :alt: Demonstration of the `ask` function.
1448 :width: 480
1449 :height: 240
1451 :func:`ask` accepts generic parameters, which determine how input is parsed.
1452 For example, if you're asking for an enum element,
1453 Yuio will show user a choice widget.
1455 You can also supply a custom :class:`~yuio.parse.Parser`,
1456 which will determine the widget that is displayed to the user,
1457 the way autocompletion works, etc.
1459 .. note::
1461 :func:`ask` is designed to interact with users, not to read data. It uses
1462 ``/dev/tty`` on Unix, and console API on Windows, so it will read from
1463 an actual TTY even if ``stdin`` is redirected.
1465 When designing your program, make sure that users have alternative means
1466 to provide values: use configs or CLI arguments, allow passing passwords
1467 via environment variables, etc.
1469 :param msg:
1470 prompt to display to user.
1471 :param args:
1472 arguments for ``%``\\ - formatting the prompt.
1473 :param parser:
1474 parser to use to parse user input. See :mod:`yuio.parse` for more info.
1475 :param default:
1476 default value to return if user input is empty.
1477 :param default_non_interactive:
1478 default value returned if input stream is not readable. If not given,
1479 `default` is used instead. This is handy when you want to ask user if they
1480 want to continue with `default` set to :data:`False`,
1481 but `default_non_interactive` set to :data:`True`.
1482 :param input_description:
1483 description of the expected input, like ``"yes/no"`` for boolean
1484 inputs.
1485 :param default_description:
1486 description of the `default` value.
1487 :returns:
1488 parsed user input.
1489 :raises:
1490 raises :class:`UserIoError` if we're not in interactive environment, and there
1491 is no default to return.
1492 :example:
1493 .. invisible-code-block: python
1495 import enum
1497 .. code-block:: python
1499 class Level(enum.Enum):
1500 WARNING = "Warning"
1501 INFO = "Info"
1502 DEBUG = "Debug"
1505 answer = yuio.io.ask[Level]("Choose a logging level", default=Level.INFO)
1507 """
1509 if TYPE_CHECKING:
1511 @_t.overload
1512 def __new__(
1513 cls: type[ask[S]],
1514 msg: _t.LiteralString,
1515 /,
1516 *args,
1517 default: M | yuio.Missing = yuio.MISSING,
1518 default_non_interactive: _t.Any = yuio.MISSING,
1519 parser: yuio.parse.Parser[S] | None = None,
1520 input_description: str | None = None,
1521 default_description: str | None = None,
1522 ) -> S | M: ...
1523 @_t.overload
1524 def __new__(
1525 cls: type[ask[S]],
1526 msg: str,
1527 /,
1528 *,
1529 default: M | yuio.Missing = yuio.MISSING,
1530 default_non_interactive: _t.Any = yuio.MISSING,
1531 parser: yuio.parse.Parser[S] | None = None,
1532 input_description: str | None = None,
1533 default_description: str | None = None,
1534 ) -> S | M: ...
1535 def __new__(cls: _t.Any, *_, **__) -> _t.Any: ...
1538def _ask(
1539 msg: _t.LiteralString,
1540 /,
1541 *args,
1542 parser: yuio.parse.Parser[_t.Any],
1543 default: _t.Any = yuio.MISSING,
1544 default_non_interactive: _t.Any = yuio.MISSING,
1545 input_description: str | None = None,
1546 default_description: str | None = None,
1547) -> _t.Any:
1548 ctx = make_repr_context(term=yuio.term.get_tty())
1550 if not _can_query_user(ctx.term):
1551 # TTY is not available.
1552 if default_non_interactive is yuio.MISSING:
1553 default_non_interactive = default
1554 if default_non_interactive is yuio.MISSING:
1555 raise UserIoError("Can't interact with user in non-interactive environment")
1556 return default_non_interactive
1558 if default is None and not yuio.parse._is_optional_parser(parser):
1559 parser = yuio.parse.Optional(parser)
1561 msg = msg.rstrip()
1562 if msg.endswith(":"):
1563 needs_colon = True
1564 msg = msg[:-1]
1565 else:
1566 needs_colon = msg and msg[-1] not in string.punctuation
1568 base_color = ctx.get_color("msg/text:question")
1569 prompt = yuio.string.colorize(msg, *args, default_color=base_color, ctx=ctx)
1571 if not input_description:
1572 input_description = parser.describe()
1574 if default is not yuio.MISSING and default_description is None:
1575 try:
1576 default_description = parser.describe_value(default)
1577 except TypeError:
1578 default_description = str(default)
1580 if not yuio.term._is_foreground(ctx.term.ostream):
1581 warning(
1582 "User input is requested in background process, use `fg %s` to resume",
1583 os.getpid(),
1584 ctx=ctx,
1585 )
1586 yuio.term._pause()
1588 if ctx.term.can_run_widgets:
1589 # Use widget.
1591 if needs_colon:
1592 prompt.append_color(base_color)
1593 prompt.append_str(":")
1595 if parser.is_secret():
1596 inner_widget = yuio.parse._secret_widget(
1597 parser, default, input_description, default_description
1598 )
1599 else:
1600 inner_widget = parser.widget(
1601 default, input_description, default_description
1602 )
1604 widget = _AskWidget(prompt, inner_widget)
1605 with SuspendOutput() as s:
1606 try:
1607 result = widget.run(ctx.term, ctx.theme)
1608 except (OSError, EOFError) as e: # pragma: no cover
1609 raise UserIoError("Unexpected end of input") from e
1611 if result is yuio.MISSING:
1612 result = default
1614 try:
1615 result_desc = parser.describe_value(result)
1616 except TypeError:
1617 result_desc = str(result)
1619 prompt.append_color(base_color)
1620 prompt.append_str(" ")
1621 prompt.append_color(base_color | ctx.get_color("code"))
1622 prompt.append_str(result_desc)
1624 # note: print to default terminal, not to tty
1625 s.info(prompt, tag="question")
1627 return result
1628 else:
1629 # Use raw input.
1631 prompt += base_color
1632 if input_description:
1633 prompt += " ("
1634 prompt += input_description
1635 prompt += ")"
1636 if default_description:
1637 prompt += " ["
1638 prompt += base_color | ctx.get_color("code")
1639 prompt += default_description
1640 prompt += base_color
1641 prompt += "]"
1642 prompt += yuio.string.Esc(": " if needs_colon else " ")
1643 if parser.is_secret():
1644 do_input = _getpass
1645 else:
1646 do_input = _read
1647 with SuspendOutput() as s:
1648 while True:
1649 try:
1650 answer = do_input(ctx.term, prompt)
1651 except (OSError, EOFError) as e: # pragma: no cover
1652 raise UserIoError("Unexpected end of input") from e
1653 if not answer and default is not yuio.MISSING:
1654 return default
1655 elif not answer:
1656 s.error("Input is required.", ctx=ctx)
1657 else:
1658 try:
1659 return parser.parse(answer)
1660 except yuio.parse.ParsingError as e:
1661 s.error(e, ctx=ctx)
1664if os.name == "posix":
1665 # Getpass implementation is based on the standard `getpass` module, with a few
1666 # Yuio-specific modifications.
1668 def _getpass_fallback(
1669 term: yuio.term.Term, prompt: yuio.string.ColorizedString
1670 ) -> str:
1671 warning(
1672 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1673 )
1674 return _read(term, prompt)
1676 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1677 info(
1678 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1679 )
1680 return term.istream.readline().rstrip("\r\n")
1682 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1683 import termios
1685 try:
1686 fd = term.istream.fileno()
1687 except (AttributeError, ValueError):
1688 # We can't control the tty or stdin. Give up and use normal IO.
1689 return _getpass_fallback(term, prompt)
1691 result: str | None = None
1693 try:
1694 prev_mode = termios.tcgetattr(fd)
1695 new_mode = prev_mode.copy()
1696 new_mode[3] &= ~termios.ECHO
1697 tcsetattr_flags = termios.TCSAFLUSH | getattr(termios, "TCSASOFT", 0)
1698 try:
1699 termios.tcsetattr(fd, tcsetattr_flags, new_mode)
1700 info(
1701 prompt,
1702 add_newline=False,
1703 tag="question",
1704 term=term,
1705 ignore_suspended=True,
1706 )
1707 result = term.istream.readline().rstrip("\r\n")
1708 term.ostream.write("\n")
1709 term.ostream.flush()
1710 finally:
1711 termios.tcsetattr(fd, tcsetattr_flags, prev_mode)
1712 except termios.error:
1713 if result is not None:
1714 # `readline` succeeded, the final `tcsetattr` failed. Reraise instead
1715 # of leaving the terminal in an unknown state.
1716 raise
1717 else:
1718 # We can't control the tty or stdin. Give up and use normal IO.
1719 return _getpass_fallback(term, prompt)
1721 assert result is not None
1722 return result
1724elif os.name == "nt":
1726 def _do_read(
1727 term: yuio.term.Term, prompt: yuio.string.ColorizedString, echo: bool
1728 ) -> str:
1729 import msvcrt
1731 if term.ostream_is_tty:
1732 info(
1733 prompt,
1734 add_newline=False,
1735 tag="question",
1736 term=term,
1737 ignore_suspended=True,
1738 )
1739 else:
1740 for c in str(prompt):
1741 msvcrt.putwch(c)
1743 if term.ostream_is_tty and echo:
1744 return term.istream.readline().rstrip("\r\n")
1745 else:
1746 result = ""
1747 while True:
1748 c = msvcrt.getwch()
1749 if c == "\0" or c == "\xe0":
1750 # Read key scan code and ignore it.
1751 msvcrt.getwch()
1752 continue
1753 if c == "\r" or c == "\n":
1754 break
1755 if c == "\x03":
1756 raise KeyboardInterrupt
1757 if c == "\b":
1758 if result:
1759 msvcrt.putwch("\b")
1760 msvcrt.putwch(" ")
1761 msvcrt.putwch("\b")
1762 result = result[:-1]
1763 else:
1764 result = result + c
1765 if echo:
1766 msvcrt.putwch(c)
1767 else:
1768 msvcrt.putwch("*")
1769 msvcrt.putwch("\r")
1770 msvcrt.putwch("\n")
1772 return result
1774 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1775 return _do_read(term, prompt, echo=True)
1777 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1778 return _do_read(term, prompt, echo=False)
1780else:
1782 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1783 warning(
1784 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1785 )
1786 return _read(term, prompt)
1788 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1789 info(
1790 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1791 )
1792 return term.istream.readline().rstrip("\r\n")
1795def _can_query_user(term: yuio.term.Term):
1796 return (
1797 # We're attached to a TTY.
1798 term.is_tty
1799 # On Windows, there is no way to bring a process to foreground (AFAIK?).
1800 # Thus, we need to check if there's a console window.
1801 and (os.name != "nt" or yuio.term._is_foreground(None))
1802 )
1805class _WaitForUserWidget(yuio.widget.Widget[None]):
1806 def __init__(self, prompt: yuio.string.Colorable):
1807 self._prompt = yuio.widget.Text(prompt)
1809 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1810 return self._prompt.layout(rc)
1812 def draw(self, rc: yuio.widget.RenderContext, /):
1813 return self._prompt.draw(rc)
1815 @yuio.widget.bind(yuio.widget.Key.ENTER)
1816 @yuio.widget.bind(yuio.widget.Key.ESCAPE)
1817 @yuio.widget.bind("d", ctrl=True)
1818 @yuio.widget.bind(" ")
1819 def exit(self):
1820 return yuio.widget.Result(None)
1823def wait_for_user(
1824 msg: _t.LiteralString = "Press <c note>enter</c> to continue",
1825 /,
1826 *args,
1827):
1828 """
1829 A simple function to wait for user to press enter.
1831 If current terminal is not interactive, this functions has no effect.
1833 :param msg:
1834 prompt to display to user.
1835 :param args:
1836 arguments for ``%``\\ - formatting the prompt.
1838 """
1840 ctx = make_repr_context(term=yuio.term.get_tty())
1842 if not _can_query_user(ctx.term):
1843 # TTY is not available.
1844 return
1846 if not yuio.term._is_foreground(ctx.term.ostream):
1847 if os.name == "nt":
1848 # AFAIK there's no way to bring job to foreground in Windows.
1849 return
1851 warning(
1852 "User input is requested in background process, use `fg %s` to resume",
1853 os.getpid(),
1854 ctx=ctx,
1855 )
1856 yuio.term._pause()
1858 prompt = yuio.string.colorize(
1859 msg.rstrip(), *args, default_color="msg/text:question", ctx=ctx
1860 )
1861 prompt += yuio.string.Esc(" ")
1863 with SuspendOutput():
1864 try:
1865 if ctx.term.can_run_widgets:
1866 _WaitForUserWidget(prompt).run(ctx.term, ctx.theme)
1867 else:
1868 _read(ctx.term, prompt)
1869 except (OSError, EOFError): # pragma: no cover
1870 return
1873def detect_editor(fallbacks: list[str] | None = None) -> str | None:
1874 """
1875 Detect the user's preferred editor.
1877 This function checks the ``VISUAL`` and ``EDITOR`` environment variables.
1878 If they're not set, it checks if any of the fallback editors are available.
1879 If none can be found, it returns :data:`None`.
1881 :param fallbacks:
1882 list of fallback editors to try. By default, we try "nano", "vim", "vi",
1883 "msedit", "edit", "notepad", "gedit".
1884 :returns:
1885 on Windows, returns an executable name; on Unix, may return a shell command
1886 or an executable name.
1888 """
1890 if os.name != "nt":
1891 if editor := os.environ.get("VISUAL"):
1892 return editor
1893 if editor := os.environ.get("EDITOR"):
1894 return editor
1896 if fallbacks is None:
1897 fallbacks = ["nano", "vim", "vi", "msedit", "edit", "notepad", "gedit"]
1898 for fallback in fallbacks:
1899 if shutil.which(fallback):
1900 return fallback
1901 return None
1904def detect_shell(fallbacks: list[str] | None = None) -> str | None:
1905 """
1906 Detect the user's preferred shell.
1908 This function checks the ``SHELL`` environment variable.
1909 If it's not set, it checks if any of the fallback shells are available.
1910 If none can be found, it returns :data:`None`.
1912 :param fallbacks:
1913 list of fallback shells to try. By default, we try "pwsh" and "powershell"
1914 on Windows, and "bash", "sh", "/bin/sh" on Linux/MacOS.
1915 :returns:
1916 returns an executable name.
1918 """
1920 if os.name != "nt" and (shell := os.environ.get("SHELL")):
1921 return shell
1923 if fallbacks is None:
1924 if os.name != "nt":
1925 fallbacks = ["bash", "sh", "/bin/sh"]
1926 else:
1927 fallbacks = ["pwsh", "powershell"]
1928 for fallback in fallbacks:
1929 if shutil.which(fallback):
1930 return fallback
1931 return None
1934def edit(
1935 text: str,
1936 /,
1937 *,
1938 comment_marker: str | None = None,
1939 editor: str | None = None,
1940 file_ext: str = ".txt",
1941 fallbacks: list[str] | None = None,
1942 dedent: bool = False,
1943) -> str:
1944 """
1945 Ask user to edit some text.
1947 This function creates a temporary file with the given text
1948 and opens it in an editor. After editing is done, it strips away
1949 all lines that start with `comment_marker`, if one is given.
1951 :param text:
1952 text to edit.
1953 :param comment_marker:
1954 lines starting with this marker will be removed from the output after edit.
1955 :param editor:
1956 overrides editor.
1958 On Unix, this should be a shell command, file path will be appended to it;
1959 on Windows, this should be an executable path.
1960 :param file_ext:
1961 extension for the temporary file, can be used to enable syntax highlighting
1962 in editors that support it.
1963 :param fallbacks:
1964 list of fallback editors to try, see :func:`detect_editor` for details.
1965 :param dedent:
1966 remove leading indentation from text before opening an editor.
1967 :returns:
1968 an edited string with comments removed.
1969 :raises:
1970 If editor is not available, returns a non-zero exit code, or launched in
1971 a non-interactive environment, a :class:`UserIoError` is raised.
1973 Also raises :class:`UserIoError` if ``stdin`` or ``stderr`` is piped
1974 or redirected to a file (virtually no editors can work when this happens).
1975 :example:
1976 .. skip: next
1978 .. code-block:: python
1980 message = yuio.io.edit(
1981 \"""
1982 # Please enter the commit message for your changes. Lines starting
1983 # with '#' will be ignored, and an empty message aborts the commit.
1984 \""",
1985 comment_marker="#",
1986 dedent=True,
1987 )
1989 """
1991 term = yuio.term.get_tty()
1993 if not _can_query_user(term):
1994 raise UserIoError("Can't run editor in non-interactive environment")
1996 if editor is None:
1997 editor = detect_editor(fallbacks)
1999 if editor is None:
2000 if os.name == "nt":
2001 raise UserIoError("Can't find a usable editor")
2002 else:
2003 raise UserIoError(
2004 "Can't find a usable editor. Ensure that `$VISUAL` and `$EDITOR` "
2005 "environment variables contain correct path to an editor executable"
2006 )
2008 if dedent:
2009 text = _dedent(text)
2011 if not yuio.term._is_foreground(term.ostream):
2012 warning(
2013 "Background process is waiting for user, use `fg %s` to resume",
2014 os.getpid(),
2015 term=term,
2016 )
2017 yuio.term._pause()
2019 fd, filepath = tempfile.mkstemp(text=True, suffix=file_ext)
2020 try:
2021 with open(fd, "w") as file:
2022 file.write(text)
2024 if os.name == "nt":
2025 # Windows doesn't use $VISUAL/$EDITOR, so shell execution is not needed.
2026 # Plus, quoting arguments for CMD.exe is hard af.
2027 args = [editor, filepath]
2028 shell = False
2029 else:
2030 # $VISUAL/$EDITOR can include flags, so we need to use shell instead.
2031 from shlex import quote
2033 args = f"{editor} {quote(filepath)}"
2034 shell = True
2036 try:
2037 with SuspendOutput():
2038 res = subprocess.run(
2039 args,
2040 shell=shell,
2041 stdin=term.istream.fileno(),
2042 stdout=term.ostream.fileno(),
2043 )
2044 except FileNotFoundError:
2045 raise UserIoError(
2046 "Can't use editor `%r`: no such file or directory",
2047 editor,
2048 )
2050 if res.returncode != 0:
2051 if res.returncode < 0:
2052 import signal
2054 try:
2055 action = "died with"
2056 code = signal.Signals(-res.returncode).name
2057 except ValueError:
2058 action = "died with unknown signal"
2059 code = res.returncode
2060 else:
2061 action = "returned exit code"
2062 code = res.returncode
2063 raise UserIoError(
2064 "Editing failed: editor `%r` %s `%s`",
2065 editor,
2066 action,
2067 code,
2068 )
2070 if not os.path.exists(filepath):
2071 raise UserIoError("Editing failed: can't read back edited file")
2072 else:
2073 with open(filepath) as file:
2074 text = file.read()
2075 finally:
2076 try:
2077 os.remove(filepath)
2078 except OSError:
2079 pass
2081 if comment_marker is not None:
2082 text = re.sub(
2083 r"^\s*" + re.escape(comment_marker) + r".*(\n|$)",
2084 "",
2085 text,
2086 flags=re.MULTILINE,
2087 )
2089 return text
2092def shell(
2093 *,
2094 shell: str | None = None,
2095 fallbacks: list[str] | None = None,
2096 prompt_marker: str = "",
2097):
2098 """
2099 Launch an interactive shell and give user control over it.
2101 This function is useful in interactive scripts. For example, if the script is
2102 creating a release commit, it might be desired to give user a chance to inspect
2103 repository status before proceeding.
2105 :param shell:
2106 overrides shell executable.
2107 :param fallbacks:
2108 list of fallback shells to try, see :func:`detect_shell` for details.
2109 :param prompt_marker:
2110 if given, Yuio will try to add this marker to the shell's prompt
2111 to remind users that this shell is a sub-process of some script.
2113 This only works with Bash, Zsh, Fish, and PowerShell.
2115 """
2117 term = yuio.term.get_tty()
2119 if not _can_query_user(term):
2120 raise UserIoError("Can't run editor in non-interactive environment")
2122 if shell is None:
2123 shell = detect_shell(fallbacks=fallbacks)
2125 if shell is None:
2126 if os.name == "nt":
2127 raise UserIoError("Can't find a usable shell")
2128 else:
2129 raise UserIoError(
2130 "Can't find a usable shell. Ensure that `$SHELL`"
2131 "environment variable contain correct path to a shell executable"
2132 )
2134 args = [shell]
2135 env = os.environ.copy()
2137 rcpath = None
2138 rcpath_is_dir = False
2139 if prompt_marker:
2140 env["__YUIO_PROMPT_MARKER"] = prompt_marker
2142 if shell == "bash" or shell.endswith(os.path.sep + "bash"):
2143 fd, rcpath = tempfile.mkstemp(text=True, suffix=".bash")
2145 rc = textwrap.dedent(
2146 """
2147 [ -f ~/.bashrc ] && source ~/.bashrc;
2148 PS1='\\e[33m$__YUIO_PROMPT_MARKER\\e[m'\" $PS1\"
2149 """
2150 )
2152 with open(fd, "w") as file:
2153 file.write(rc)
2155 args += ["--rcfile", rcpath, "-i"]
2156 elif shell == "zsh" or shell.endswith(os.path.sep + "zsh"):
2157 rcpath = tempfile.mkdtemp()
2158 rcpath_is_dir = True
2160 rc = textwrap.dedent(
2161 """
2162 ZDOTDIR=$ZDOTDIR_ORIG
2163 [ -f $ZDOTDIR/.zprofile ] && source $ZDOTDIR/.zprofile
2164 [ -f $ZDOTDIR/.zshrc ] && source $ZDOTDIR/.zshrc
2165 autoload -U colors && colors
2166 PS1='%F{yellow}$__YUIO_PROMPT_MARKER%f'" $PS1"
2167 """
2168 )
2170 with open(os.path.join(rcpath, ".zshrc"), "w") as file:
2171 file.write(rc)
2173 if "ZDOTDIR" in env:
2174 zdotdir = env["ZDOTDIR"]
2175 else:
2176 zdotdir = os.path.expanduser("~")
2178 env["ZDOTDIR"] = rcpath
2179 env["ZDOTDIR_ORIG"] = zdotdir
2181 args += ["-i"]
2182 elif shell == "fish" or shell.endswith(os.path.sep + "fish"):
2183 rc = textwrap.dedent(
2184 """
2185 functions -c fish_prompt _yuio_old_fish_prompt
2186 function fish_prompt
2187 set -l old_status $status
2188 printf "%s%s%s " (set_color yellow) $__YUIO_PROMPT_MARKER (set_color normal)
2189 echo "exit $old_status" | .
2190 _yuio_old_fish_prompt
2191 end
2192 """
2193 )
2195 args += ["--init-command", rc, "-i"]
2196 elif shell in ["powershell", "pwsh"] or shell.endswith(
2197 (os.path.sep + "powershell", os.path.sep + "pwsh")
2198 ):
2199 fd, rcpath = tempfile.mkstemp(text=True, suffix=".ps1")
2201 rc = textwrap.dedent(
2202 """
2203 function global:_yuio_old_pwsh_prompt { "" }
2204 Copy-Item -Path function:prompt -Destination function:_yuio_old_pwsh_prompt
2206 function global:prompt {
2207 Write-Host -NoNewline -ForegroundColor Yellow "$env:__YUIO_PROMPT_MARKER "
2208 _yuio_old_pwsh_prompt
2209 }
2210 """
2211 )
2213 with open(fd, "w") as file:
2214 file.write(rc)
2216 args += ["-NoExit", "-File", rcpath]
2218 try:
2219 with SuspendOutput():
2220 subprocess.run(
2221 args,
2222 env=env,
2223 stdin=term.istream.fileno(),
2224 stdout=term.ostream.fileno(),
2225 )
2226 except FileNotFoundError:
2227 raise UserIoError(
2228 "Can't use shell `%r`: no such file or directory",
2229 shell,
2230 )
2231 finally:
2232 if rcpath:
2233 try:
2234 if rcpath_is_dir:
2235 shutil.rmtree(rcpath)
2236 else:
2237 os.remove(rcpath)
2238 except OSError:
2239 pass
2242class MessageChannel:
2243 """
2244 Message channels are similar to logging adapters: they allow adding additional
2245 arguments for calls to :func:`raw` and other message functions.
2247 This is useful when you need to control destination for messages, but don't want
2248 to override global settings via :func:`setup`. One example for them is described
2249 in :ref:`cookbook-print-to-file`.
2251 .. dropdown:: Protected members
2253 .. autoattribute:: _msg_kwargs
2255 .. automethod:: _update_kwargs
2257 .. automethod:: _is_enabled
2259 """
2261 enabled: bool
2262 """
2263 Message channel can be disabled, in which case messages are not printed.
2265 """
2267 _msg_kwargs: dict[str, _t.Any]
2268 """
2269 Keyword arguments that will be added to every message.
2271 """
2273 if _t.TYPE_CHECKING:
2275 def __init__(
2276 self,
2277 *,
2278 ignore_suspended: bool = False,
2279 term: yuio.term.Term | None = None,
2280 to_stdout: bool = False,
2281 to_stderr: bool = False,
2282 theme: yuio.theme.Theme | None = None,
2283 multiline: bool | None = None,
2284 highlighted: bool | None = None,
2285 max_depth: int | None = None,
2286 width: int | None = None,
2287 ): ...
2288 else:
2290 def __init__(self, **kwargs):
2291 self._msg_kwargs: dict[str, _t.Any] = kwargs
2292 self.enabled: bool = True
2294 def _update_kwargs(self, kwargs: dict[str, _t.Any]):
2295 """
2296 A hook that updates method's `kwargs` before calling its implementation.
2298 """
2300 for name, option in self._msg_kwargs.items():
2301 kwargs.setdefault(name, option)
2303 def _is_enabled(self):
2304 """
2305 A hook that check if the message should be printed. By default, returns value
2306 of :attr:`~MessageChannel.enabled`.
2308 """
2310 return self.enabled
2312 @_t.overload
2313 def info(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2314 @_t.overload
2315 def info(self, err: yuio.string.ToColorable, /, **kwargs): ...
2316 def info(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2317 """info(msg: typing.LiteralString, /, *args, **kwargs)
2318 info(msg: ~string.templatelib.Template, /, **kwargs) ->
2319 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2321 Print an :func:`info` message.
2323 """
2325 if not self._is_enabled():
2326 return
2328 self._update_kwargs(kwargs)
2329 info(msg, *args, **kwargs)
2331 @_t.overload
2332 def warning(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2333 @_t.overload
2334 def warning(self, err: yuio.string.ToColorable, /, **kwargs): ...
2335 def warning(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2336 """warning(msg: typing.LiteralString, /, *args, **kwargs)
2337 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
2338 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2340 Print a :func:`warning` message.
2342 """
2344 if not self._is_enabled():
2345 return
2347 self._update_kwargs(kwargs)
2348 warning(msg, *args, **kwargs)
2350 @_t.overload
2351 def success(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2352 @_t.overload
2353 def success(self, err: yuio.string.ToColorable, /, **kwargs): ...
2354 def success(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2355 """success(msg: typing.LiteralString, /, *args, **kwargs)
2356 success(msg: ~string.templatelib.Template, /, **kwargs) ->
2357 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2359 Print a :func:`success` message.
2361 """
2363 if not self._is_enabled():
2364 return
2366 self._update_kwargs(kwargs)
2367 success(msg, *args, **kwargs)
2369 @_t.overload
2370 def error(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2371 @_t.overload
2372 def error(self, err: yuio.string.ToColorable, /, **kwargs): ...
2373 def error(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2374 """error(msg: typing.LiteralString, /, *args, **kwargs)
2375 error(msg: ~string.templatelib.Template, /, **kwargs) ->
2376 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2378 Print an :func:`error` message.
2380 """
2382 if not self._is_enabled():
2383 return
2385 self._update_kwargs(kwargs)
2386 error(msg, *args, **kwargs)
2388 @_t.overload
2389 def error_with_tb(
2390 self,
2391 msg: _t.LiteralString,
2392 /,
2393 *args,
2394 exc_info: ExcInfo | bool | None = True,
2395 **kwargs,
2396 ): ...
2397 @_t.overload
2398 def error_with_tb(
2399 self,
2400 msg: yuio.string.ToColorable,
2401 /,
2402 *,
2403 exc_info: ExcInfo | bool | None = True,
2404 **kwargs,
2405 ): ...
2406 def error_with_tb(
2407 self,
2408 msg: yuio.string.ToColorable,
2409 /,
2410 *args,
2411 exc_info: ExcInfo | bool | None = True,
2412 **kwargs,
2413 ):
2414 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2415 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2416 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2418 Print an :func:`error_with_tb` message.
2420 """
2422 if not self._is_enabled():
2423 return
2425 self._update_kwargs(kwargs)
2426 error_with_tb(msg, *args, **kwargs)
2428 @_t.overload
2429 def failure(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2430 @_t.overload
2431 def failure(self, err: yuio.string.ToColorable, /, **kwargs): ...
2432 def failure(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2433 """failure(msg: typing.LiteralString, /, *args, **kwargs)
2434 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
2435 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2437 Print a :func:`failure` message.
2439 """
2441 if not self._is_enabled():
2442 return
2444 self._update_kwargs(kwargs)
2445 failure(msg, *args, **kwargs)
2447 @_t.overload
2448 def failure_with_tb(
2449 self,
2450 msg: _t.LiteralString,
2451 /,
2452 *args,
2453 exc_info: ExcInfo | bool | None = True,
2454 **kwargs,
2455 ): ...
2456 @_t.overload
2457 def failure_with_tb(
2458 self,
2459 msg: yuio.string.ToColorable,
2460 /,
2461 *,
2462 exc_info: ExcInfo | bool | None = True,
2463 **kwargs,
2464 ): ...
2465 def failure_with_tb(
2466 self,
2467 msg: yuio.string.ToColorable,
2468 /,
2469 *args,
2470 exc_info: ExcInfo | bool | None = True,
2471 **kwargs,
2472 ):
2473 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2474 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2475 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2477 Print a :func:`failure_with_tb` message.
2479 """
2481 if not self._is_enabled():
2482 return
2484 self._update_kwargs(kwargs)
2485 failure_with_tb(msg, *args, **kwargs)
2487 @_t.overload
2488 def heading(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2489 @_t.overload
2490 def heading(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2491 def heading(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2492 """heading(msg: typing.LiteralString, /, *args, **kwargs)
2493 heading(msg: ~string.templatelib.Template, /, **kwargs)
2494 heading(msg: ~yuio.string.ToColorable, /, **kwargs)
2496 Print a :func:`heading` message.
2498 """
2500 if not self._is_enabled():
2501 return
2503 self._update_kwargs(kwargs)
2504 heading(msg, *args, **kwargs)
2506 def md(self, msg: str, /, **kwargs):
2507 """
2508 Print an :func:`md` message.
2510 """
2512 if not self._is_enabled():
2513 return
2515 self._update_kwargs(kwargs)
2516 md(msg, **kwargs)
2518 def rst(self, msg: str, /, **kwargs):
2519 """
2520 Print an :func:`rst` message.
2522 """
2524 if not self._is_enabled():
2525 return
2527 self._update_kwargs(kwargs)
2528 rst(msg, **kwargs)
2530 def br(self, **kwargs):
2531 """br()
2533 Print a :func:`br` message.
2535 """
2537 if not self._is_enabled():
2538 return
2540 self._update_kwargs(kwargs)
2541 br(**kwargs)
2543 @_t.overload
2544 def hl(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2545 @_t.overload
2546 def hl(self, msg: str, /, **kwargs): ...
2547 def hl(self, msg: str, /, *args, **kwargs):
2548 """hl(msg: typing.LiteralString, /, *args, syntax: str, dedent: bool = True, **kwargs)
2549 hl(msg: str, /, *, syntax: str, dedent: bool = True, **kwargs)
2551 Print an :func:`hl` message.
2553 """
2555 if not self._is_enabled():
2556 return
2558 self._update_kwargs(kwargs)
2559 hl(msg, *args, **kwargs)
2561 @_t.overload
2562 def hr(self, msg: _t.LiteralString = "", /, *args, **kwargs): ...
2563 @_t.overload
2564 def hr(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2565 def hr(self, msg: yuio.string.ToColorable = "", /, *args, **kwargs):
2566 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
2567 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
2568 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
2570 Print an :func:`hr` message.
2572 """
2574 if not self._is_enabled():
2575 return
2577 self._update_kwargs(kwargs)
2578 hr(msg, *args, **kwargs)
2580 def raw(self, msg: yuio.string.Colorable, /, **kwargs):
2581 """
2582 Print a :func:`raw` message.
2584 """
2586 if not self._is_enabled():
2587 return
2589 self._update_kwargs(kwargs)
2590 raw(msg, **kwargs)
2592 def make_repr_context(self) -> yuio.string.ReprContext:
2593 """
2594 Make a :class:`~yuio.string.ReprContext` using settings
2595 from :attr:`~MessageChannel._msg_kwargs`.
2597 """
2599 return make_repr_context(
2600 term=self._msg_kwargs.get("term"),
2601 to_stdout=self._msg_kwargs.get("to_stdout", False),
2602 to_stderr=self._msg_kwargs.get("to_stderr", False),
2603 theme=self._msg_kwargs.get("theme"),
2604 multiline=self._msg_kwargs.get("multiline"),
2605 highlighted=self._msg_kwargs.get("highlighted"),
2606 max_depth=self._msg_kwargs.get("max_depth"),
2607 width=self._msg_kwargs.get("width"),
2608 )
2611class SuspendOutput(MessageChannel):
2612 """
2613 A context manager for pausing output.
2615 This is handy for when you need to take control over the output stream.
2616 For example, the :func:`ask` function uses this class internally.
2618 This context manager also suspends all prints that go to :data:`sys.stdout`
2619 and :data:`sys.stderr` if they were wrapped (see :func:`setup`).
2620 To print through them, use :func:`orig_stderr` and :func:`orig_stdout`.
2622 Each instance of this class is a :class:`MessageChannel`; calls to its printing
2623 methods bypass output suppression:
2625 .. code-block:: python
2627 with SuspendOutput() as out:
2628 print("Suspended") # [1]_
2629 out.info("Not suspended") # [2]_
2631 .. code-annotations::
2633 1. This message is suspended; it will be printed when output is resumed.
2634 2. This message bypasses suspension; it will be printed immediately.
2636 """
2638 def __init__(self, initial_channel: MessageChannel | None = None, /):
2639 super().__init__()
2641 if initial_channel is not None:
2642 self._msg_kwargs.update(initial_channel._msg_kwargs)
2643 self._msg_kwargs["ignore_suspended"] = True
2645 self._resumed = False
2646 _manager().suspend()
2648 def resume(self):
2649 """
2650 Manually resume the logging process.
2652 """
2654 if not self._resumed:
2655 _manager().resume()
2656 self._resumed = True
2658 def __enter__(self):
2659 return self
2661 def __exit__(self, exc_type, exc_val, exc_tb):
2662 self.resume()
2665class _IterTask(_t.Generic[T]):
2666 def __init__(
2667 self, collection: _t.Collection[T], task: Task, unit: str, ndigits: int
2668 ):
2669 self._iter = iter(collection)
2670 self._task = task
2671 self._unit = unit
2672 self._ndigits = ndigits
2674 self._i = 0
2675 self._len = len(collection)
2677 def __next__(self) -> T:
2678 self._task.progress(self._i, self._len, unit=self._unit, ndigits=self._ndigits)
2679 if self._i < self._len:
2680 self._i += 1
2681 return self._iter.__next__()
2683 def __iter__(self) -> _IterTask[T]:
2684 return self
2687class TaskBase:
2688 """
2689 Base class for tasks and other objects that you might show to the user.
2691 Example of a custom task can be found in :ref:`cookbook <cookbook-custom-tasks>`.
2693 .. dropdown:: Protected members
2695 .. autoproperty:: _lock
2697 .. automethod:: _get_widget
2699 .. automethod:: _get_priority
2701 .. automethod:: _request_update
2703 .. automethod:: _widgets_are_displayed
2705 .. automethod:: _get_parent
2707 .. automethod:: _is_toplevel
2709 .. automethod:: _get_children
2711 """
2713 def __init__(self):
2714 self.__parent: TaskBase | None = None
2715 self.__children: list[TaskBase] = []
2717 def attach(self, parent: TaskBase | None):
2718 """
2719 Attach this task and all of its children to the task tree.
2721 :param parent:
2722 parent task in the tree. Pass :data:`None` to attach to root.
2724 """
2726 with self._lock:
2727 if parent is None:
2728 parent = _manager().tasks_root
2729 if self.__parent is not None:
2730 self.__parent.__children.remove(self)
2731 self.__parent = parent
2732 parent.__children.append(self)
2733 self._request_update()
2735 def detach(self):
2736 """
2737 Remove this task and all of its children from the task tree.
2739 """
2741 with self._lock:
2742 if self.__parent is not None:
2743 self.__parent.__children.remove(self)
2744 self.__parent = None
2745 self._request_update()
2747 @property
2748 def _lock(self):
2749 """
2750 Global IO lock.
2752 All protected methods, as well as state mutations, should happen
2753 under this lock.
2755 """
2757 return _IO_LOCK
2759 @abc.abstractmethod
2760 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
2761 """
2762 This method should return widget that renders the task.
2764 .. warning::
2766 This method should be called under :attr:`~TaskBase._lock`.
2768 """
2770 raise NotImplementedError()
2772 @abc.abstractmethod
2773 def _get_priority(self) -> int:
2774 """
2775 This method should return priority that will be used to hide non-important
2776 tasks when there is not enough space to show all of them.
2778 Default priority is ``1``, priority for finished tasks is ``0``.
2780 .. warning::
2782 This method should be called under :attr:`~TaskBase._lock`.
2784 """
2786 raise NotImplementedError()
2788 def _request_update(self, *, immediate_render: bool = False):
2789 """
2790 Indicate that task's state has changed, and update is necessary.
2792 .. warning::
2794 This method should be called under :attr:`~TaskBase._lock`.
2796 :param immediate_render:
2797 by default, tasks are updated lazily from a background thread; set this
2798 parameter to :data:`True` to redraw them immediately from this thread.
2800 """
2802 _manager()._update_tasks(immediate_render)
2804 def _widgets_are_displayed(self) -> bool:
2805 """
2806 Return :data:`True` if we're in an interactive foreground process which
2807 renders tasks.
2809 If this function returns :data:`False`, you should print log messages about
2810 task status instead of relying on task's widget being presented to the user.
2812 .. warning::
2814 This method should be called under :attr:`~TaskBase._lock`.
2816 """
2818 return _manager()._should_draw_interactive_tasks()
2820 def _get_parent(self) -> TaskBase | None:
2821 """
2822 Get parent task.
2824 .. warning::
2826 This method should be called under :attr:`~TaskBase._lock`.
2828 """
2830 return self.__parent
2832 def _is_toplevel(self) -> bool:
2833 """
2834 Check if this task is attached to the first level of the tree.
2836 .. warning::
2838 This method should be called under :attr:`~TaskBase._lock`.
2840 """
2842 return self._get_parent() is _manager().tasks_root
2844 def _get_children(self) -> _t.Sequence[TaskBase]:
2845 """
2846 Get child tasks.
2848 .. warning::
2850 This method should be called under :attr:`~TaskBase._lock`.
2852 """
2854 return self.__children
2857class _TasksRoot(TaskBase):
2858 _widget = yuio.widget.Empty()
2860 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
2861 return self._widget
2863 def _get_priority(self) -> int:
2864 return 0
2867class Task(TaskBase):
2868 """Task(msg: typing.LiteralString, /, *args, comment: str | None = None, parent: Task | None = None)
2869 Task(msg: str, /, *, comment: str | None = None, parent: Task | None = None)
2871 A class for indicating progress of some task.
2873 :param msg:
2874 task heading.
2875 :param args:
2876 arguments for ``%``\\ -formatting the task heading.
2877 :param comment:
2878 comment for the task. Can be specified after creation
2879 via the :meth:`~Task.comment` method.
2880 :param persistent:
2881 whether to keep showing this task after it finishes.
2882 Default is :data:`False`.
2884 To manually hide the task, call :meth:`~TaskBase.detach`.
2885 :param initial_status:
2886 initial status of the task.
2887 :param parent:
2888 parent task.
2890 You can have multiple tasks at the same time,
2891 create subtasks, set task's progress or add a comment about
2892 what's currently being done within a task.
2894 .. vhs:: /_tapes/tasks_multithreaded.tape
2895 :alt: Demonstration of the `Task` class.
2896 :width: 480
2897 :height: 240
2899 This class can be used as a context manager:
2901 .. code-block:: python
2903 with yuio.io.Task("Processing input") as t:
2904 ...
2905 t.progress(0.3)
2906 ...
2908 .. dropdown:: Protected members
2910 .. autoattribute:: _widget_class
2912 """
2914 Status = yuio.widget.Task.Status
2916 _widget_class: _ClassVar[type[yuio.widget.Task]] = yuio.widget.Task
2917 """
2918 Class of the widget that will be used to draw this task, can be overridden
2919 in subclasses.
2921 """
2923 @_t.overload
2924 def __init__(
2925 self,
2926 msg: _t.LiteralString,
2927 /,
2928 *args,
2929 comment: str | None = None,
2930 persistent: bool = False,
2931 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2932 parent: TaskBase | None = None,
2933 ): ...
2934 @_t.overload
2935 def __init__(
2936 self,
2937 msg: str,
2938 /,
2939 *,
2940 comment: str | None = None,
2941 persistent: bool = False,
2942 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2943 parent: TaskBase | None = None,
2944 ): ...
2945 def __init__(
2946 self,
2947 msg: str,
2948 /,
2949 *args,
2950 comment: str | None = None,
2951 persistent: bool = False,
2952 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2953 parent: TaskBase | None = None,
2954 ):
2955 super().__init__()
2957 self._widget = self._widget_class(msg, *args, comment=comment)
2958 self._persistent = persistent
2959 with self._lock:
2960 self.set_status(initial_status)
2961 self.attach(parent)
2963 @_t.overload
2964 def progress(self, progress: float | None, /, *, ndigits: int = 2): ...
2966 @_t.overload
2967 def progress(
2968 self,
2969 done: float | int,
2970 total: float | int,
2971 /,
2972 *,
2973 unit: str = "",
2974 ndigits: int = 0,
2975 ): ...
2977 def progress(
2978 self,
2979 *args: float | int | None,
2980 unit: str = "",
2981 ndigits: int | None = None,
2982 ):
2983 """progress(progress: float | None, /, *, ndigits: int = 2)
2984 progress(done: float | int, total: float | int, /, *, unit: str = "", ndigits: int = 0) ->
2986 Indicate progress of this task.
2988 If given one argument, it is treated as percentage between ``0`` and ``1``.
2990 If given two arguments, they are treated as amount of finished work,
2991 and a total amount of work. In this case, optional argument `unit`
2992 can be used to indicate units for the progress.
2994 If given a single :data:`None`, reset task progress.
2996 .. note::
2998 Tasks are updated asynchronously once every ~100ms, so calling this method
2999 is relatively cheap. It still requires acquiring a global lock, though:
3000 contention could be an issue in multi-threaded applications.
3002 :param progress:
3003 a percentage between ``0`` and ``1``, or :data:`None`
3004 to reset task progress.
3005 :param done:
3006 amount of finished work, should be less than or equal to `total`.
3007 :param total:
3008 total amount of work.
3009 :param unit:
3010 unit for measuring progress. Only displayed when progress is given
3011 as `done` and `total`.
3012 :param ndigits:
3013 number of digits to display after a decimal point.
3014 :example:
3015 .. code-block:: python
3017 with yuio.io.Task("Loading cargo") as task:
3018 task.progress(110, 150, unit="Kg")
3020 This will print the following:
3022 .. code-block:: text
3024 ■■■■■■■■■■■□□□□ Loading cargo - 110/150Kg
3026 """
3028 with self._lock:
3029 self._widget.progress(*args, unit=unit, ndigits=ndigits) # type: ignore
3030 self._request_update()
3032 def progress_size(
3033 self,
3034 done: float | int,
3035 total: float | int,
3036 /,
3037 *,
3038 ndigits: int = 2,
3039 ):
3040 """
3041 Indicate progress of this task using human-readable 1024-based size units.
3043 :param done:
3044 amount of processed data.
3045 :param total:
3046 total amount of data.
3047 :param ndigits:
3048 number of digits to display after a decimal point.
3049 :example:
3050 .. code-block:: python
3052 with yuio.io.Task("Downloading a file") as task:
3053 task.progress_size(31.05 * 2**20, 150 * 2**20)
3055 This will print:
3057 .. code-block:: text
3059 ■■■□□□□□□□□□□□□ Downloading a file - 31.05/150.00M
3061 """
3063 with self._lock:
3064 self._widget.progress_size(done, total, ndigits=ndigits)
3065 self._request_update()
3067 def progress_scale(
3068 self,
3069 done: float | int,
3070 total: float | int,
3071 /,
3072 *,
3073 unit: str = "",
3074 ndigits: int = 2,
3075 ):
3076 """
3077 Indicate progress of this task while scaling numbers in accordance
3078 with SI system.
3080 :param done:
3081 amount of finished work, should be less than or equal to `total`.
3082 :param total:
3083 total amount of work.
3084 :param unit:
3085 unit for measuring progress.
3086 :param ndigits:
3087 number of digits to display after a decimal point.
3088 :example:
3089 .. code-block:: python
3091 with yuio.io.Task("Charging a capacitor") as task:
3092 task.progress_scale(889.25e-3, 1, unit="V")
3094 This will print:
3096 .. code-block:: text
3098 ■■■■■■■■■■■■■□□ Charging a capacitor - 889.25mV/1.00V
3100 """
3102 with self._lock:
3103 self._widget.progress_scale(done, total, unit=unit, ndigits=ndigits)
3104 self._request_update()
3106 def iter(
3107 self,
3108 collection: _t.Collection[T],
3109 /,
3110 *,
3111 unit: str = "",
3112 ndigits: int = 0,
3113 ) -> _t.Iterable[T]:
3114 """
3115 Helper for updating progress automatically
3116 while iterating over a collection.
3118 :param collection:
3119 an iterable collection. Should support returning its length.
3120 :param unit:
3121 unit for measuring progress.
3122 :param ndigits:
3123 number of digits to display after a decimal point.
3124 :example:
3125 .. invisible-code-block: python
3127 urls = []
3129 .. code-block:: python
3131 with yuio.io.Task("Fetching data") as t:
3132 for url in t.iter(urls):
3133 ...
3135 This will output the following:
3137 .. code-block:: text
3139 ■■■■■□□□□□□□□□□ Fetching data - 1/3
3141 """
3143 return _IterTask(collection, self, unit, ndigits)
3145 def comment(self, comment: str | None, /, *args):
3146 """
3147 Set a comment for a task.
3149 Comment is displayed after the progress.
3151 :param comment:
3152 comment to display beside task progress.
3153 :param args:
3154 arguments for ``%``\\ -formatting comment.
3155 :example:
3156 .. invisible-code-block: python
3158 urls = []
3160 .. code-block:: python
3162 with yuio.io.Task("Fetching data") as t:
3163 for url in urls:
3164 t.comment("%s", url)
3165 ...
3167 This will output the following:
3169 .. code-block:: text
3171 ⣿ Fetching data - https://google.com
3173 """
3175 with self._lock:
3176 self._widget.comment(comment, *args)
3177 self._request_update()
3179 def set_status(self, status: Task.Status):
3180 """
3181 Set task status.
3183 :param status:
3184 New status.
3186 """
3188 with self._lock:
3189 if self._widget.status == status:
3190 return
3192 self._widget.status = status
3193 if status in [Task.Status.DONE, Task.Status.ERROR] and not self._persistent:
3194 self.detach()
3195 if self._widgets_are_displayed():
3196 self._request_update()
3197 else:
3198 raw(self._widget, add_newline=True)
3200 def running(self):
3201 """
3202 Indicate that this task is running.
3204 """
3206 self.set_status(Task.Status.RUNNING)
3208 def pending(self):
3209 """
3210 Indicate that this task is pending.
3212 """
3214 self.set_status(Task.Status.PENDING)
3216 def done(self):
3217 """
3218 Indicate that this task has finished successfully.
3220 """
3222 self.set_status(Task.Status.DONE)
3224 def error(self):
3225 """
3226 Indicate that this task has finished with an error.
3228 """
3230 self.set_status(Task.Status.ERROR)
3232 @_t.overload
3233 def subtask(
3234 self,
3235 msg: _t.LiteralString,
3236 /,
3237 *args,
3238 comment: str | None = None,
3239 persistent: bool = True,
3240 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3241 ) -> Task: ...
3242 @_t.overload
3243 def subtask(
3244 self,
3245 msg: str,
3246 /,
3247 *,
3248 comment: str | None = None,
3249 persistent: bool = True,
3250 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3251 ) -> Task: ...
3252 def subtask(
3253 self,
3254 msg: str,
3255 /,
3256 *args,
3257 comment: str | None = None,
3258 persistent: bool = True,
3259 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3260 ) -> Task:
3261 """
3262 Create a subtask within this task.
3264 :param msg:
3265 subtask heading.
3266 :param args:
3267 arguments for ``%``\\ -formatting the subtask heading.
3268 :param comment:
3269 comment for the task. Can be specified after creation
3270 via the :meth:`~Task.comment` method.
3271 :param persistent:
3272 whether to keep showing this subtask after it finishes. Default
3273 is :data:`True`.
3274 :param initial_status:
3275 initial status of the task.
3276 :returns:
3277 a new :class:`Task` that will be displayed as a sub-task of this task.
3279 """
3281 return Task(
3282 msg,
3283 *args,
3284 comment=comment,
3285 persistent=persistent,
3286 initial_status=initial_status,
3287 parent=self,
3288 )
3290 def __enter__(self):
3291 self.running()
3292 return self
3294 def __exit__(self, exc_type, exc_val, exc_tb):
3295 if exc_type is None:
3296 self.done()
3297 else:
3298 self.error()
3300 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
3301 return self._widget
3303 def _get_priority(self) -> int:
3304 return 1 if self._widget.status is yuio.widget.Task.Status.RUNNING else 0
3307class _TaskTree(yuio.widget.Widget[_t.Never]):
3308 def __init__(self, root: TaskBase):
3309 super().__init__()
3311 self._root = root
3313 def layout(self, rc: yuio.widget.RenderContext) -> tuple[int, int]:
3314 widgets: list[yuio.widget.Widget[_t.Never]] = [] # widget
3315 tree: dict[
3316 int, tuple[int | None, int, int]
3317 ] = {} # index -> parent, level, priority
3319 # Build widgets tree.
3320 to_visit: list[tuple[TaskBase, int, int | None]] = [(self._root, 0, None)]
3321 while to_visit:
3322 node, level, parent = to_visit.pop()
3323 widget = node._get_widget()
3324 tree[len(widgets)] = parent, level, node._get_priority()
3325 to_visit.extend(
3326 (child, level + 1, len(widgets))
3327 for child in reversed(node._get_children())
3328 )
3329 widgets.append(widget)
3331 # Prepare layouts.
3332 layouts: dict[yuio.widget.Widget[_t.Never], tuple[int, int, int]] = {}
3333 self.__layouts = layouts
3334 total_min_h = 0
3335 total_max_h = 0
3336 for index, widget in enumerate(widgets):
3337 min_h, max_h = widget.layout(rc)
3338 assert min_h <= max_h, "incorrect layout"
3339 _, level, _ = tree[index]
3340 layouts[widget] = min_h, max_h, level
3341 total_min_h += min_h
3342 total_max_h += max_h
3344 if total_min_h <= rc.height:
3345 # All widgets fit.
3346 self.__min_h = total_min_h
3347 self.__max_h = total_max_h
3348 self.__widgets = widgets
3349 return total_min_h, total_max_h
3351 # Propagate priority upwards, ensure that parents are at least as important
3352 # as children.
3353 for index, widget in enumerate(widgets):
3354 parent, _, priority = tree[index]
3355 while parent is not None:
3356 grandparent, parent_level, parent_priority = tree[parent]
3357 if parent_priority >= priority:
3358 break
3359 tree[parent] = grandparent, parent_level, priority
3360 widget = parent
3361 parent = grandparent
3363 # Sort by (-priority, level, -index). Since we've propagated priorities, we can
3364 # be sure that parents are always included first. Hence in the loop below,
3365 # we will visit children before parents.
3366 widgets_sorted = list(enumerate(widgets))
3367 widgets_sorted.sort(key=lambda w: (-tree[w[0]][2], tree[w[0]][1], -w[0]))
3369 # Decide which widgets to hide by introducing "holes" to widgets sequence.
3370 total_h = total_min_h
3371 holes = _DisjointSet[int]()
3372 for index, widget in reversed(widgets_sorted):
3373 if total_h <= rc.height:
3374 break
3376 min_h, max_h = widget.layout(rc)
3378 # We need to hide this widget.
3379 _, level, _ = tree[index]
3380 holes.add(index)
3381 total_h -= min_h
3382 total_h += 1 # Size of a message.
3384 # Join this hole with the next one.
3385 if index + 1 < len(widgets) and index + 1 in holes:
3386 _, next_level, _ = tree[index + 1]
3387 if next_level >= level:
3388 holes.union(index, index + 1)
3389 total_h -= 1
3390 # Join this hole with the previous one.
3391 if index - 1 >= 0 and index - 1 in holes:
3392 _, prev_level, _ = tree[index - 1]
3393 if prev_level <= level:
3394 holes.union(index, index - 1)
3395 total_h -= 1
3397 # Assemble the final sequence of widgets.
3398 hole_color = rc.theme.get_color("task/hole")
3399 hole_num_color = rc.theme.get_color("task/hole/num")
3400 prev_hole_id: int | None = None
3401 prev_hole_size = 0
3402 prev_hole_level: int | None = None
3403 displayed_widgets: list[yuio.widget.Widget[_t.Never]] = []
3404 for index, widget in enumerate(widgets):
3405 if index in holes:
3406 hole_id = holes.find(index)
3407 if hole_id == prev_hole_id:
3408 prev_hole_size += 1
3409 if prev_hole_level is None:
3410 prev_hole_level = tree[index][1]
3411 else:
3412 prev_hole_level = min(prev_hole_level, tree[index][1])
3413 else:
3414 if prev_hole_id is not None:
3415 hole_widget = yuio.widget.Line(
3416 yuio.string.ColorizedString(
3417 hole_num_color,
3418 "+",
3419 str(prev_hole_size),
3420 hole_color,
3421 " more",
3422 )
3423 )
3424 displayed_widgets.append(hole_widget)
3425 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3426 prev_hole_id = hole_id
3427 prev_hole_size = 1
3428 prev_hole_level = tree[index][1]
3429 else:
3430 if prev_hole_id is not None:
3431 hole_widget = yuio.widget.Line(
3432 yuio.string.ColorizedString(
3433 hole_num_color,
3434 "+",
3435 str(prev_hole_size),
3436 hole_color,
3437 " more",
3438 )
3439 )
3440 displayed_widgets.append(hole_widget)
3441 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3442 prev_hole_id = None
3443 prev_hole_size = 0
3444 prev_hole_level = None
3445 displayed_widgets.append(widget)
3447 if prev_hole_id is not None:
3448 hole_widget = yuio.widget.Line(
3449 yuio.string.ColorizedString(
3450 hole_num_color,
3451 "+",
3452 str(prev_hole_size),
3453 hole_color,
3454 " more",
3455 )
3456 )
3457 displayed_widgets.append(hole_widget)
3458 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3460 total_min_h = 0
3461 total_max_h = 0
3462 for widget in displayed_widgets:
3463 min_h, max_h, _ = layouts[widget]
3464 total_min_h += min_h
3465 total_max_h += max_h
3467 self.__min_h = total_min_h
3468 self.__max_h = total_max_h
3469 self.__widgets = displayed_widgets
3470 return total_min_h, total_max_h
3472 def draw(self, rc: yuio.widget.RenderContext):
3473 if rc.height <= self.__min_h:
3474 scale = 0.0
3475 elif rc.height >= self.__max_h:
3476 scale = 1.0
3477 else:
3478 scale = (rc.height - self.__min_h) / (self.__max_h - self.__min_h)
3480 y1 = 0.0
3481 for widget in self.__widgets:
3482 min_h, max_h, level = self.__layouts[widget]
3483 y2 = y1 + min_h + scale * (max_h - min_h)
3485 iy1 = round(y1)
3486 iy2 = round(y2)
3488 with rc.frame(max((level - 1) * 2, 0), iy1, height=iy2 - iy1):
3489 widget.draw(rc)
3491 y1 = y2
3493 rc.set_final_pos(0, round(y1))
3496class Formatter(logging.Formatter):
3497 """
3498 Log formatter that uses ``%`` style with colorized string formatting
3499 and returns a string with ANSI escape characters generated for current
3500 output terminal.
3502 Every part of log message is colored with path :samp:`log/{name}:{level}`.
3503 For example, `asctime` in info log line is colored
3504 with path ``log/asctime:info``.
3506 In addition to the usual `log record attributes`__, this formatter also
3507 adds ``%(colMessage)s``, which is similar to ``%(message)s``, but colorized.
3509 __ https://docs.python.org/3/library/logging.html#logrecord-attributes
3511 """
3513 default_format = "%(asctime)s %(name)s %(levelname)s %(colMessage)s"
3514 default_msec_format = "%s.%03d"
3516 def __init__(
3517 self,
3518 fmt: str | None = None,
3519 datefmt: str | None = None,
3520 validate: bool = True,
3521 *,
3522 defaults: _t.Mapping[str, _t.Any] | None = None,
3523 ):
3524 fmt = fmt or self.default_format
3525 super().__init__(
3526 fmt,
3527 datefmt,
3528 style="%",
3529 validate=validate,
3530 defaults=defaults,
3531 )
3533 def formatMessage(self, record):
3534 level = record.levelname.lower()
3536 ctx = make_repr_context()
3538 if not hasattr(record, "colMessage"):
3539 msg = str(record.msg)
3540 if record.args:
3541 msg = ColorizedString(msg).percent_format(record.args, ctx)
3542 setattr(record, "colMessage", msg)
3544 if defaults := self._style._defaults: # type: ignore
3545 data = defaults | record.__dict__
3546 else:
3547 data = record.__dict__
3549 data = {
3550 k: yuio.string.WithBaseColor(v, base_color=f"log/{k}:{level}")
3551 for k, v in data.items()
3552 }
3554 return "".join(
3555 yuio.string.colorize(
3556 self._fmt or self.default_format, default_color=f"log:{level}", ctx=ctx
3557 )
3558 .percent_format(data, ctx)
3559 .as_code(ctx.term.color_support)
3560 )
3562 def formatException(self, ei):
3563 tb = "".join(traceback.format_exception(*ei)).rstrip()
3564 return self.formatStack(tb)
3566 def formatStack(self, stack_info):
3567 manager = _manager()
3568 theme = manager.theme
3569 term = manager.term
3570 highlighter, syntax_name = yuio.hl.get_highlighter("python-traceback")
3571 return "".join(
3572 highlighter.highlight(stack_info, theme=theme, syntax=syntax_name)
3573 .indent()
3574 .as_code(term.color_support)
3575 )
3578class Handler(logging.Handler):
3579 """
3580 A handler that redirects all log messages to Yuio.
3582 """
3584 def __init__(self, level: int | str = 0):
3585 super().__init__(level)
3586 self.setFormatter(Formatter())
3588 def emit(self, record: LogRecord):
3589 manager = _manager()
3590 manager.print_direct(self.format(record).rstrip() + "\n", manager.term.ostream)
3593class _IoManager(abc.ABC):
3594 # If we see that it took more than this time to render progress bars,
3595 # we assume that the process was suspended, meaning that we might've been moved
3596 # from foreground to background or back. In either way, we should assume that the
3597 # screen was changed, and re-render all tasks accordingly. We have to track time
3598 # because Python might take significant time to call `SIGCONT` handler, so we can't
3599 # rely on it.
3600 TASK_RENDER_TIMEOUT_NS = 250_000_000
3602 def __init__(
3603 self,
3604 term: yuio.term.Term | None = None,
3605 theme: (
3606 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
3607 ) = None,
3608 enable_bg_updates: bool = True,
3609 ):
3610 self._out_term = yuio.term.get_term_from_stream(
3611 orig_stdout(), sys.stdin, allow_env_overrides=True
3612 )
3613 self._err_term = yuio.term.get_term_from_stream(
3614 orig_stderr(), sys.stdin, allow_env_overrides=True
3615 )
3617 self._term = term or self._err_term
3619 self._theme_ctor = theme
3620 if isinstance(theme, yuio.theme.Theme):
3621 self._theme = theme
3622 else:
3623 self._theme = yuio.theme.load(self._term, theme)
3624 self._rc = yuio.widget.RenderContext(self._term, self._theme)
3625 self._rc.prepare()
3627 self._suspended: int = 0
3628 self._suspended_lines: list[tuple[list[str], _t.TextIO]] = []
3630 self._tasks_root = _TasksRoot()
3631 self._tasks_widet = _TaskTree(self._tasks_root)
3632 self._printed_tasks: bool = False
3633 self._needs_update = False
3634 self._last_update_time_us = 0
3635 self._printed_some_lines = False
3637 self._stop = False
3638 self._stop_condition = threading.Condition(_IO_LOCK)
3639 self._thread: threading.Thread | None = None
3641 self._enable_bg_updates = enable_bg_updates
3642 self._prev_sigcont_handler: (
3643 None | yuio.Missing | int | _t.Callable[[int, types.FrameType | None], None]
3644 ) = yuio.MISSING
3645 self._seen_sigcont: bool = False
3646 if enable_bg_updates:
3647 self._setup_sigcont()
3648 self._thread = threading.Thread(
3649 target=self._bg_update, name="yuio_io_task_refresh", daemon=True
3650 )
3651 self._thread.start()
3653 atexit.register(self.stop)
3655 @property
3656 def term(self):
3657 return self._term
3659 @property
3660 def out_term(self):
3661 return self._out_term
3663 @property
3664 def err_term(self):
3665 return self._err_term
3667 @property
3668 def theme(self):
3669 return self._theme
3671 @property
3672 def rc(self):
3673 return self._rc
3675 @property
3676 def tasks_root(self):
3677 return self._tasks_root
3679 def setup(
3680 self,
3681 term: yuio.term.Term | None = None,
3682 theme: (
3683 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
3684 ) = None,
3685 ):
3686 with _IO_LOCK:
3687 self._clear_tasks()
3689 if term is not None:
3690 self._term = term
3691 if theme is None:
3692 # Refresh theme to reflect changed terminal capabilities.
3693 theme = self._theme_ctor
3694 if theme is not None:
3695 self._theme_ctor = theme
3696 if isinstance(theme, yuio.theme.Theme):
3697 self._theme = theme
3698 else:
3699 self._theme = yuio.theme.load(self._term, theme)
3701 self._rc = yuio.widget.RenderContext(self._term, self._theme)
3702 self._rc.prepare()
3703 self.__dict__.pop("_update_rate_us", None) # type: ignore
3704 self._update_tasks()
3706 def _setup_sigcont(self):
3707 import signal
3709 if not hasattr(signal, "SIGCONT"):
3710 return
3712 self._prev_sigcont_handler = signal.getsignal(signal.SIGCONT)
3713 signal.signal(signal.SIGCONT, self._on_sigcont)
3715 def _reset_sigcont(self):
3716 import signal
3718 if not hasattr(signal, "SIGCONT"):
3719 return
3721 if self._prev_sigcont_handler is not yuio.MISSING:
3722 signal.signal(signal.SIGCONT, self._prev_sigcont_handler)
3724 def _on_sigcont(self, sig: int, frame: types.FrameType | None):
3725 self._seen_sigcont = True
3726 if self._prev_sigcont_handler and not isinstance(
3727 self._prev_sigcont_handler, int
3728 ):
3729 self._prev_sigcont_handler(sig, frame)
3731 def _bg_update(self):
3732 while True:
3733 try:
3734 with _IO_LOCK:
3735 while True:
3736 update_rate_us = self._update_rate_us
3737 start_ns = time.monotonic_ns()
3738 now_us = start_ns // 1_000
3739 sleep_us = update_rate_us - now_us % update_rate_us
3740 deadline_ns = (
3741 start_ns + 2 * sleep_us * 1000 + self.TASK_RENDER_TIMEOUT_NS
3742 )
3744 if self._stop_condition.wait_for(
3745 lambda: self._stop, timeout=sleep_us / 1_000_000
3746 ):
3747 return
3749 self._show_tasks(deadline_ns=deadline_ns)
3750 except Exception:
3751 yuio._logger.critical("exception in bg updater", exc_info=True)
3753 def stop(self):
3754 if self._stop:
3755 return
3757 with _IO_LOCK:
3758 atexit.unregister(self.stop)
3760 self._stop = True
3761 self._stop_condition.notify()
3762 self._show_tasks(immediate_render=True)
3764 if self._thread:
3765 self._thread.join()
3767 if self._prev_sigcont_handler is not yuio.MISSING:
3768 self._reset_sigcont()
3770 def print(
3771 self,
3772 msg: list[str],
3773 term: yuio.term.Term,
3774 *,
3775 ignore_suspended: bool = False,
3776 heading: bool = False,
3777 ):
3778 with _IO_LOCK:
3779 if heading and self.theme.separate_headings:
3780 if self._printed_some_lines:
3781 msg.insert(0, "\n")
3782 msg.append("\n")
3783 self._emit_lines(msg, term.ostream, ignore_suspended)
3784 if heading:
3785 self._printed_some_lines = False
3787 def print_direct(
3788 self,
3789 msg: str,
3790 stream: _t.TextIO | None = None,
3791 ):
3792 with _IO_LOCK:
3793 self._emit_lines([msg], stream, ignore_suspended=False)
3795 def print_direct_lines(
3796 self,
3797 lines: _t.Iterable[str],
3798 stream: _t.TextIO | None = None,
3799 ):
3800 with _IO_LOCK:
3801 self._emit_lines(lines, stream, ignore_suspended=False)
3803 def suspend(self):
3804 with _IO_LOCK:
3805 self._suspend()
3807 def resume(self):
3808 with _IO_LOCK:
3809 self._resume()
3811 # Implementation.
3812 # These functions are always called under a lock.
3814 @functools.cached_property
3815 def _update_rate_us(self) -> int:
3816 update_rate_ms = max(self._theme.spinner_update_rate_ms, 1)
3817 while update_rate_ms < 50:
3818 update_rate_ms *= 2
3819 while update_rate_ms > 250:
3820 update_rate_ms //= 2
3821 return int(update_rate_ms * 1000)
3823 @property
3824 def _spinner_update_rate_us(self) -> int:
3825 return self._theme.spinner_update_rate_ms * 1000
3827 def _emit_lines(
3828 self,
3829 lines: _t.Iterable[str],
3830 stream: _t.TextIO | None = None,
3831 ignore_suspended: bool = False,
3832 ):
3833 stream = stream or self._term.ostream
3834 if self._suspended and not ignore_suspended:
3835 self._suspended_lines.append((list(lines), stream))
3836 else:
3837 self._clear_tasks()
3838 stream.writelines(lines)
3839 self._update_tasks(immediate_render=True)
3840 stream.flush()
3842 self._printed_some_lines = True
3844 def _suspend(self):
3845 self._suspended += 1
3847 if self._suspended == 1:
3848 self._clear_tasks()
3850 def _resume(self):
3851 self._suspended -= 1
3853 if self._suspended == 0:
3854 for lines, stream in self._suspended_lines:
3855 stream.writelines(lines)
3856 if self._suspended_lines:
3857 self._printed_some_lines = True
3858 self._suspended_lines.clear()
3860 self._update_tasks()
3862 if self._suspended < 0:
3863 yuio._logger.warning("unequal number of suspends and resumes")
3864 self._suspended = 0
3866 def _should_draw_interactive_tasks(self):
3867 should_draw_interactive_tasks = (
3868 self._term.color_support >= yuio.term.ColorSupport.ANSI
3869 and self._term.ostream_is_tty
3870 and yuio.term._is_foreground(self._term.ostream)
3871 )
3873 if (
3874 not should_draw_interactive_tasks and self._printed_tasks
3875 ) or self._seen_sigcont:
3876 # We were moved from foreground to background. There's no point in hiding
3877 # tasks now (shell printed something when user sent C-z), but we need
3878 # to make sure that we'll start rendering tasks from scratch whenever
3879 # user brings us to foreground again.
3880 self.rc.prepare(reset_term_pos=True)
3881 self._printed_tasks = False
3882 self._seen_sigcont = False
3884 return should_draw_interactive_tasks
3886 def _clear_tasks(self):
3887 if self._should_draw_interactive_tasks() and self._printed_tasks:
3888 self._rc.finalize()
3889 self._printed_tasks = False
3891 def _update_tasks(self, immediate_render: bool = False):
3892 self._needs_update = True
3893 if immediate_render or not self._enable_bg_updates:
3894 self._show_tasks(immediate_render)
3896 def _show_tasks(
3897 self, immediate_render: bool = False, deadline_ns: int | None = None
3898 ):
3899 if (
3900 self._should_draw_interactive_tasks()
3901 and not self._suspended
3902 and (self._tasks_root._get_children() or self._printed_tasks)
3903 ):
3904 start_ns = time.monotonic_ns()
3905 if deadline_ns is None:
3906 deadline_ns = start_ns + self.TASK_RENDER_TIMEOUT_NS
3907 now_us = start_ns // 1000
3908 now_us -= now_us % self._update_rate_us
3910 if not immediate_render and self._enable_bg_updates:
3911 next_update_us = self._last_update_time_us + self._update_rate_us
3912 if now_us < next_update_us:
3913 # Hard-limit update rate by `update_rate_ms`.
3914 return
3915 next_spinner_update_us = (
3916 self._last_update_time_us + self._spinner_update_rate_us
3917 )
3918 if not self._needs_update and now_us < next_spinner_update_us:
3919 # Tasks didn't change, and spinner state didn't change either,
3920 # so we can skip this update.
3921 return
3923 self._last_update_time_us = now_us
3924 self._printed_tasks = bool(self._tasks_root._get_children())
3925 self._needs_update = False
3927 self._rc.prepare()
3928 self._tasks_widet.layout(self._rc)
3929 self._tasks_widet.draw(self._rc)
3931 now_ns = time.monotonic_ns()
3932 if not self._seen_sigcont and now_ns < deadline_ns:
3933 self._rc.render()
3934 else:
3935 # We have to skip this render: the process was suspended while we were
3936 # formatting tasks. Because of this, te position of the cursor
3937 # could've changed, so we need to reset rendering context and re-render.
3938 self._seen_sigcont = True
3941class _YuioOutputWrapper(_t.TextIO): # pragma: no cover
3942 def __init__(self, wrapped: _t.TextIO):
3943 self.__wrapped = wrapped
3945 @property
3946 def mode(self) -> str:
3947 return self.__wrapped.mode
3949 @property
3950 def name(self) -> str:
3951 return self.__wrapped.name
3953 def close(self):
3954 self.__wrapped.close()
3956 @property
3957 def closed(self) -> bool:
3958 return self.__wrapped.closed
3960 def fileno(self) -> int:
3961 return self.__wrapped.fileno()
3963 def flush(self):
3964 self.__wrapped.flush()
3966 def isatty(self) -> bool:
3967 return self.__wrapped.isatty()
3969 def writable(self) -> bool:
3970 return self.__wrapped.writable()
3972 def write(self, s: str, /) -> int:
3973 _manager().print_direct(s, self.__wrapped)
3974 return len(s)
3976 def writelines(self, lines: _t.Iterable[str], /):
3977 _manager().print_direct_lines(lines, self.__wrapped)
3979 def readable(self) -> bool:
3980 return self.__wrapped.readable()
3982 def read(self, n: int = -1) -> str:
3983 return self.__wrapped.read(n)
3985 def readline(self, limit: int = -1) -> str:
3986 return self.__wrapped.readline(limit)
3988 def readlines(self, hint: int = -1) -> list[str]:
3989 return self.__wrapped.readlines(hint)
3991 def seek(self, offset: int, whence: int = 0) -> int:
3992 return self.__wrapped.seek(offset, whence)
3994 def seekable(self) -> bool:
3995 return self.__wrapped.seekable()
3997 def tell(self) -> int:
3998 return self.__wrapped.tell()
4000 def truncate(self, size: int | None = None) -> int:
4001 return self.__wrapped.truncate(size)
4003 def __enter__(self) -> _t.TextIO:
4004 return self.__wrapped.__enter__()
4006 def __exit__(self, exc_type, exc_val, exc_tb):
4007 self.__wrapped.__exit__(exc_type, exc_val, exc_tb)
4009 @property
4010 def buffer(self) -> _t.BinaryIO:
4011 return self.__wrapped.buffer
4013 @property
4014 def encoding(self) -> str:
4015 return self.__wrapped.encoding
4017 @property
4018 def errors(self) -> str | None:
4019 return self.__wrapped.errors
4021 @property
4022 def line_buffering(self) -> int:
4023 return self.__wrapped.line_buffering
4025 @property
4026 def newlines(self) -> _t.Any:
4027 return self.__wrapped.newlines
4029 def __repr__(self) -> str:
4030 return f"{self.__class__.__name__}({self.__wrapped!r})"