Coverage for yuio / io.py: 89%
1147 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-29 19:55 +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:: Ordinal
309 :no-index:
311 Alias of :obj:`yuio.string.Ordinal`.
313.. type:: ColorizedString
314 :no-index:
316 Alias of :obj:`yuio.string.ColorizedString`.
318.. type:: Format
319 :no-index:
321 Alias of :obj:`yuio.string.Format`.
323.. type:: Hl
324 :no-index:
326 Alias of :obj:`yuio.string.Hl`.
328.. type:: Hr
329 :no-index:
331 Alias of :obj:`yuio.string.Hr`.
333.. type:: Indent
334 :no-index:
336 Alias of :obj:`yuio.string.Indent`.
338.. type:: JoinRepr
339 :no-index:
341 Alias of :obj:`yuio.string.JoinRepr`.
343.. type:: JoinStr
344 :no-index:
346 Alias of :obj:`yuio.string.JoinStr`.
348.. type:: Link
349 :no-index:
351 Alias of :obj:`yuio.string.Link`.
353.. type:: Md
354 :no-index:
356 Alias of :obj:`yuio.string.Md`.
358.. type:: Rst
359 :no-index:
361 Alias of :obj:`yuio.string.Rst`.
363.. type:: Or
364 :no-index:
366 Alias of :obj:`yuio.string.Or`.
368.. type:: Plural
369 :no-index:
371 Alias of :obj:`yuio.string.Plural`.
373.. type:: Repr
374 :no-index:
376 Alias of :obj:`yuio.string.Repr`.
378.. type:: Stack
379 :no-index:
381 Alias of :obj:`yuio.string.Stack`.
383.. type:: TypeRepr
384 :no-index:
386 Alias of :obj:`yuio.string.TypeRepr`.
388.. type:: WithBaseColor
389 :no-index:
391 Alias of :obj:`yuio.string.WithBaseColor`.
393.. type:: Wrap
394 :no-index:
396 Alias of :obj:`yuio.string.Wrap`.
399"""
401from __future__ import annotations
403import abc
404import atexit
405import functools
406import logging
407import os
408import re
409import shutil
410import string
411import subprocess
412import sys
413import tempfile
414import textwrap
415import threading
416import time
417import traceback
418import types
419from logging import LogRecord
421import yuio.color
422import yuio.hl
423import yuio.parse
424import yuio.string
425import yuio.term
426import yuio.theme
427import yuio.widget
428from yuio._dist.dsu import DisjointSet as _DisjointSet
429from yuio.string import (
430 And,
431 ColorizedString,
432 Format,
433 Hl,
434 Hr,
435 Indent,
436 JoinRepr,
437 JoinStr,
438 Link,
439 Md,
440 Or,
441 Ordinal,
442 Plural,
443 Repr,
444 Rst,
445 Stack,
446 TypeRepr,
447 WithBaseColor,
448 Wrap,
449)
450from yuio.util import dedent as _dedent
452import yuio._typing_ext as _tx
453from typing import TYPE_CHECKING
454from typing import ClassVar as _ClassVar
456if TYPE_CHECKING:
457 import typing_extensions as _t
458else:
459 from yuio import _typing as _t
461__all__ = [
462 "And",
463 "ColorizedString",
464 "ExcInfo",
465 "Format",
466 "Formatter",
467 "Handler",
468 "Hl",
469 "Hr",
470 "Indent",
471 "JoinRepr",
472 "JoinStr",
473 "Link",
474 "Md",
475 "MessageChannel",
476 "Or",
477 "Ordinal",
478 "Plural",
479 "Repr",
480 "Rst",
481 "Stack",
482 "SuspendOutput",
483 "Task",
484 "TaskBase",
485 "TypeRepr",
486 "UserIoError",
487 "WithBaseColor",
488 "Wrap",
489 "ask",
490 "br",
491 "detect_editor",
492 "detect_shell",
493 "edit",
494 "error",
495 "error_with_tb",
496 "failure",
497 "failure_with_tb",
498 "get_term",
499 "get_theme",
500 "heading",
501 "hl",
502 "hr",
503 "info",
504 "make_repr_context",
505 "md",
506 "orig_stderr",
507 "orig_stdout",
508 "raw",
509 "restore_streams",
510 "rst",
511 "setup",
512 "shell",
513 "streams_wrapped",
514 "success",
515 "wait_for_user",
516 "warning",
517 "wrap_streams",
518]
520T = _t.TypeVar("T")
521M = _t.TypeVar("M", default=_t.Never)
522S = _t.TypeVar("S", default=str)
524ExcInfo: _t.TypeAlias = tuple[
525 type[BaseException] | None,
526 BaseException | None,
527 types.TracebackType | None,
528]
529"""
530Exception information as returned by :func:`sys.exc_info`.
532"""
535_IO_LOCK = threading.RLock()
536_IO_MANAGER: _IoManager | None = None
537_STREAMS_WRAPPED: bool = False
538_ORIG_STDERR: _t.TextIO | None = None
539_ORIG_STDOUT: _t.TextIO | None = None
542def _manager() -> _IoManager:
543 global _IO_MANAGER
545 if _IO_MANAGER is None:
546 with _IO_LOCK:
547 if _IO_MANAGER is None:
548 _IO_MANAGER = _IoManager()
549 return _IO_MANAGER
552class UserIoError(yuio.PrettyException, IOError):
553 """
554 Raised when interaction with user fails.
556 """
559def setup(
560 *,
561 term: yuio.term.Term | None = None,
562 theme: (
563 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
564 ) = None,
565 wrap_stdio: bool = True,
566):
567 """
568 Initial setup of the logging facilities.
570 :param term:
571 terminal that will be used for output.
573 If not passed, the global terminal is not re-configured;
574 the default is to use a term attached to :data:`sys.stderr`.
575 :param theme:
576 either a theme that will be used for output, or a theme constructor that takes
577 a :class:`~yuio.term.Term` and returns a theme.
579 If not passed, the global theme is not re-configured; the default is to use
580 :class:`yuio.theme.DefaultTheme` then.
581 :param wrap_stdio:
582 if set to :data:`True`, wraps :data:`sys.stdout` and :data:`sys.stderr`
583 in a special wrapper that ensures better interaction
584 with Yuio's progress bars and widgets.
586 .. note::
588 If you're working with some other library that wraps :data:`sys.stdout`
589 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
591 .. _colorama: https://github.com/tartley/colorama
593 .. warning::
595 This operation is not atomic. Call this function before creating new threads
596 that use :mod:`yuio.io` or output streams to avoid race conditions.
598 """
600 global _IO_MANAGER
602 if not (manager := _IO_MANAGER):
603 with _IO_LOCK:
604 if not (manager := _IO_MANAGER):
605 _IO_MANAGER = _IoManager(term, theme)
606 if manager is not None:
607 manager.setup(term, theme)
609 if wrap_stdio:
610 wrap_streams()
613def get_term() -> yuio.term.Term:
614 """
615 Get the global instance of :class:`~yuio.term.Term` that is used
616 with :mod:`yuio.io`.
618 If global setup wasn't performed, this function implicitly performs it.
620 :returns:
621 Instance of :class:`~yuio.term.Term` that's used to print messages and tasks.
623 """
625 return _manager().term
628def get_theme() -> yuio.theme.Theme:
629 """
630 Get the global instance of :class:`~yuio.theme.Theme`
631 that is used with :mod:`yuio.io`.
633 If global setup wasn't performed, this function implicitly performs it.
635 :returns:
636 Instance of :class:`~yuio.theme.Theme` that's used to format messages and tasks.
638 """
640 return _manager().theme
643def make_repr_context(
644 *,
645 term: yuio.term.Term | None = None,
646 to_stdout: bool = False,
647 to_stderr: bool = False,
648 theme: yuio.theme.Theme | None = None,
649 multiline: bool | None = None,
650 highlighted: bool | None = None,
651 max_depth: int | None = None,
652 width: int | None = None,
653) -> yuio.string.ReprContext:
654 """
655 Create new :class:`~yuio.string.ReprContext` for the given term and theme.
657 .. warning::
659 :class:`~yuio.string.ReprContext`\\ s are not thread safe. As such,
660 you shouldn't create them for long term use.
662 :param term:
663 terminal where to print this message. If not given, terminal from
664 :func:`get_term` is used.
665 :param to_stdout:
666 shortcut for setting `term` to ``stdout``.
667 :param to_stderr:
668 shortcut for setting `term` to ``stderr``.
669 :param theme:
670 theme used to format the message. If not given, theme from
671 :func:`get_theme` is used.
672 :param multiline:
673 sets initial value for
674 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
675 Default is :data:`False`.
676 :param highlighted:
677 sets initial value for
678 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
679 Default is :data:`False`.
680 :param max_depth:
681 sets initial value for
682 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
683 Default is :data:`False`.
684 :param width:
685 sets initial value for
686 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
687 If not given, uses current terminal width or
688 :attr:`Theme.fallback_width <yuio.theme.Theme.fallback_width>`
689 depending on whether `term` is attached to a TTY device and whether colors
690 are supported by the target terminal.
692 """
694 if (term is not None) + to_stdout + to_stderr > 1:
695 names = []
696 if term is not None:
697 names.append("term")
698 if to_stdout:
699 names.append("to_stdout")
700 if to_stderr:
701 names.append("to_stderr")
702 raise TypeError(f"{And(names)} can't be given together")
704 manager = _manager()
706 theme = manager.theme
707 if term is None:
708 if to_stdout:
709 term = manager.out_term
710 elif to_stderr:
711 term = manager.err_term
712 else:
713 term = manager.term
714 if width is None and (term.ostream_is_tty or term.supports_colors):
715 width = manager.rc.canvas_width
717 return yuio.string.ReprContext(
718 term=term,
719 theme=theme,
720 multiline=multiline,
721 highlighted=highlighted,
722 max_depth=max_depth,
723 width=width,
724 )
727def wrap_streams():
728 """
729 Wrap :data:`sys.stdout` and :data:`sys.stderr` so that they honor
730 Yuio tasks and widgets. If strings are already wrapped, this function
731 has no effect.
733 .. note::
735 If you're working with some other library that wraps :data:`sys.stdout`
736 and :data:`sys.stderr`, such as colorama_, initialize it before Yuio.
738 .. seealso::
740 :func:`setup`.
742 .. _colorama: https://github.com/tartley/colorama
744 """
746 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
748 if _STREAMS_WRAPPED:
749 return
751 with _IO_LOCK:
752 if _STREAMS_WRAPPED: # pragma: no cover
753 return
755 if yuio.term._output_is_tty(sys.stdout):
756 _ORIG_STDOUT, sys.stdout = sys.stdout, _YuioOutputWrapper(sys.stdout)
757 if yuio.term._output_is_tty(sys.stderr):
758 _ORIG_STDERR, sys.stderr = sys.stderr, _YuioOutputWrapper(sys.stderr)
759 _STREAMS_WRAPPED = True
761 atexit.register(restore_streams)
764def restore_streams():
765 """
766 Restore wrapped streams. If streams weren't wrapped, this function
767 has no effect.
769 .. seealso::
771 :func:`wrap_streams`, :func:`setup`
773 """
775 global _STREAMS_WRAPPED, _ORIG_STDOUT, _ORIG_STDERR
777 if not _STREAMS_WRAPPED:
778 return
780 with _IO_LOCK:
781 if not _STREAMS_WRAPPED: # pragma: no cover
782 return
784 if _ORIG_STDOUT is not None:
785 sys.stdout = _ORIG_STDOUT
786 _ORIG_STDOUT = None
787 if _ORIG_STDERR is not None:
788 sys.stderr = _ORIG_STDERR
789 _ORIG_STDERR = None
790 _STREAMS_WRAPPED = False
793def streams_wrapped() -> bool:
794 """
795 Check if :data:`sys.stdout` and :data:`sys.stderr` are wrapped.
796 See :func:`setup`.
798 :returns:
799 :data:`True` is streams are currently wrapped, :data:`False` otherwise.
801 """
803 return _STREAMS_WRAPPED
806def orig_stderr() -> _t.TextIO:
807 """
808 Return the original :data:`sys.stderr` before wrapping.
810 """
812 return _ORIG_STDERR or sys.stderr
815def orig_stdout() -> _t.TextIO:
816 """
817 Return the original :data:`sys.stdout` before wrapping.
819 """
821 return _ORIG_STDOUT or sys.stdout
824@_t.overload
825def info(msg: _t.LiteralString, /, *args, **kwargs): ...
826@_t.overload
827def info(msg: yuio.string.ToColorable, /, **kwargs): ...
828def info(msg: yuio.string.ToColorable, /, *args, **kwargs):
829 """info(msg: typing.LiteralString, /, *args, **kwargs)
830 info(msg: ~string.templatelib.Template, /, **kwargs) ->
831 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
833 Print an info message.
835 :param msg:
836 message to print.
837 :param args:
838 arguments for ``%``\\ -formatting the message.
839 :param kwargs:
840 any additional keyword arguments will be passed to :func:`raw`.
842 """
844 msg_colorable = yuio.string._to_colorable(msg, args)
845 kwargs.setdefault("tag", "info")
846 kwargs.setdefault("wrap", True)
847 kwargs.setdefault("add_newline", True)
848 raw(msg_colorable, **kwargs)
851@_t.overload
852def warning(msg: _t.LiteralString, /, *args, **kwargs): ...
853@_t.overload
854def warning(msg: yuio.string.ToColorable, /, **kwargs): ...
855def warning(msg: yuio.string.ToColorable, /, *args, **kwargs):
856 """warning(msg: typing.LiteralString, /, *args, **kwargs)
857 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
858 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
860 Print a warning message.
862 :param msg:
863 message to print.
864 :param args:
865 arguments for ``%``\\ -formatting the message.
866 :param kwargs:
867 any additional keyword arguments will be passed to :func:`raw`.
869 """
871 msg_colorable = yuio.string._to_colorable(msg, args)
872 kwargs.setdefault("tag", "warning")
873 kwargs.setdefault("wrap", True)
874 kwargs.setdefault("add_newline", True)
875 raw(msg_colorable, **kwargs)
878@_t.overload
879def success(msg: _t.LiteralString, /, *args, **kwargs): ...
880@_t.overload
881def success(msg: yuio.string.ToColorable, /, **kwargs): ...
882def success(msg: yuio.string.ToColorable, /, *args, **kwargs):
883 """success(msg: typing.LiteralString, /, *args, **kwargs)
884 success(msg: ~string.templatelib.Template, /, **kwargs) ->
885 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
887 Print a success message.
889 :param msg:
890 message to print.
891 :param args:
892 arguments for ``%``\\ -formatting the message.
893 :param kwargs:
894 any additional keyword arguments will be passed to :func:`raw`.
896 """
898 msg_colorable = yuio.string._to_colorable(msg, args)
899 kwargs.setdefault("tag", "success")
900 kwargs.setdefault("wrap", True)
901 kwargs.setdefault("add_newline", True)
902 raw(msg_colorable, **kwargs)
905@_t.overload
906def error(msg: _t.LiteralString, /, *args, **kwargs): ...
907@_t.overload
908def error(msg: yuio.string.ToColorable, /, **kwargs): ...
909def error(msg: yuio.string.ToColorable, /, *args, **kwargs):
910 """error(msg: typing.LiteralString, /, *args, **kwargs)
911 error(msg: ~string.templatelib.Template, /, **kwargs) ->
912 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
914 Print an error message.
916 :param msg:
917 message to print.
918 :param args:
919 arguments for ``%``\\ -formatting the message.
920 :param kwargs:
921 any additional keyword arguments will be passed to :func:`raw`.
923 """
925 msg_colorable = yuio.string._to_colorable(msg, args)
926 kwargs.setdefault("tag", "error")
927 kwargs.setdefault("wrap", True)
928 kwargs.setdefault("add_newline", True)
929 raw(msg_colorable, **kwargs)
932@_t.overload
933def error_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
934@_t.overload
935def error_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
936def error_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
937 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
938 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
939 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
941 Print an error message and capture the current exception.
943 Call this function in the ``except`` clause of a ``try`` block
944 or in an ``__exit__`` function of a context manager to attach
945 current exception details to the log message.
947 :param msg:
948 message to print.
949 :param args:
950 arguments for ``%``\\ -formatting the message.
951 :param kwargs:
952 any additional keyword arguments will be passed to :func:`raw`.
954 """
956 msg_colorable = yuio.string._to_colorable(msg, args)
957 kwargs.setdefault("tag", "error")
958 kwargs.setdefault("wrap", True)
959 kwargs.setdefault("add_newline", True)
960 kwargs.setdefault("exc_info", True)
961 raw(msg_colorable, **kwargs)
964@_t.overload
965def failure(msg: _t.LiteralString, /, *args, **kwargs): ...
966@_t.overload
967def failure(msg: yuio.string.ToColorable, /, **kwargs): ...
968def failure(msg: yuio.string.ToColorable, /, *args, **kwargs):
969 """failure(msg: typing.LiteralString, /, *args, **kwargs)
970 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
971 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
973 Print a failure message.
975 :param msg:
976 message to print.
977 :param args:
978 arguments for ``%``\\ -formatting the message.
979 :param kwargs:
980 any additional keyword arguments will be passed to :func:`raw`.
982 """
984 msg_colorable = yuio.string._to_colorable(msg, args)
985 kwargs.setdefault("tag", "failure")
986 kwargs.setdefault("wrap", True)
987 kwargs.setdefault("add_newline", True)
988 raw(msg_colorable, **kwargs)
991@_t.overload
992def failure_with_tb(msg: _t.LiteralString, /, *args, **kwargs): ...
993@_t.overload
994def failure_with_tb(msg: yuio.string.ToColorable, /, **kwargs): ...
995def failure_with_tb(msg: yuio.string.ToColorable, /, *args, **kwargs):
996 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
997 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
998 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
1000 Print a failure message and capture the current exception.
1002 Call this function in the ``except`` clause of a ``try`` block
1003 or in an ``__exit__`` function of a context manager to attach
1004 current exception details to the log message.
1006 :param msg:
1007 message to print.
1008 :param args:
1009 arguments for ``%``\\ -formatting the message.
1010 :param kwargs:
1011 any additional keyword arguments will be passed to :func:`raw`.
1013 """
1015 msg_colorable = yuio.string._to_colorable(msg, args)
1016 kwargs.setdefault("tag", "failure")
1017 kwargs.setdefault("wrap", True)
1018 kwargs.setdefault("add_newline", True)
1019 kwargs.setdefault("exc_info", True)
1020 raw(msg_colorable, **kwargs)
1023@_t.overload
1024def heading(msg: _t.LiteralString, /, *args, level: int = 1, **kwargs): ...
1025@_t.overload
1026def heading(msg: yuio.string.ToColorable, /, *, level: int = 1, **kwargs): ...
1027def heading(msg: yuio.string.ToColorable, /, *args, level: int = 1, **kwargs):
1028 """heading(msg: typing.LiteralString, /, *args, level: int = 1, **kwargs)
1029 heading(msg: ~string.templatelib.Template, /, *, level: int = 1, **kwargs) ->
1030 heading(msg: ~yuio.string.ToColorable, /, *, level: int = 1, **kwargs) ->
1032 Print a heading message.
1034 :param msg:
1035 message to print.
1036 :param args:
1037 arguments for ``%``\\ -formatting the message.
1038 :param level:
1039 level of the heading.
1040 :param kwargs:
1041 any additional keyword arguments will be passed to :func:`raw`.
1043 """
1045 msg_colorable = yuio.string._to_colorable(msg, args)
1046 level = kwargs.pop("level", 1)
1047 kwargs.setdefault("heading", True)
1048 kwargs.setdefault("tag", f"heading/{level}")
1049 kwargs.setdefault("wrap", True)
1050 kwargs.setdefault("add_newline", True)
1051 raw(msg_colorable, **kwargs)
1054def md(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs):
1055 """
1056 Print a markdown-formatted text.
1058 Yuio supports all CommonMark block markup except tables.
1059 See :mod:`yuio.md` for more info.
1061 :param msg:
1062 message to print.
1063 :param dedent:
1064 whether to remove leading indent from `msg`.
1065 :param allow_headings:
1066 whether to render headings as actual headings or as paragraphs.
1067 :param kwargs:
1068 any additional keyword arguments will be passed to :func:`raw`.
1070 """
1072 info(
1073 yuio.string.Md(msg, dedent=dedent, allow_headings=allow_headings),
1074 **kwargs,
1075 )
1078def rst(msg: str, /, *, dedent: bool = True, allow_headings: bool = True, **kwargs):
1079 """
1080 Print a RST-formatted text.
1082 Yuio supports all RST block markup except tables and field lists.
1083 See :mod:`yuio.rst` for more info.
1085 :param msg:
1086 message to print.
1087 :param dedent:
1088 whether to remove leading indent from `msg`.
1089 :param allow_headings:
1090 whether to render headings as actual headings or as paragraphs.
1091 :param kwargs:
1092 any additional keyword arguments will be passed to :func:`raw`.
1094 """
1096 info(
1097 yuio.string.Rst(msg, dedent=dedent, allow_headings=allow_headings),
1098 **kwargs,
1099 )
1102def br(**kwargs):
1103 """
1104 Print an empty string.
1106 :param kwargs:
1107 any additional keyword arguments will be passed to :func:`raw`.
1109 """
1111 raw("\n", **kwargs)
1114@_t.overload
1115def hr(msg: _t.LiteralString = "", /, *args, weight: int | str = 1, **kwargs): ...
1116@_t.overload
1117def hr(msg: yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs): ...
1118def hr(msg: yuio.string.ToColorable = "", /, *args, weight: int | str = 1, **kwargs):
1119 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
1120 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
1121 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
1123 Print a horizontal ruler.
1125 :param msg:
1126 message to print in the middle of the ruler.
1127 :param args:
1128 arguments for ``%``\\ -formatting the message.
1129 :param weight:
1130 weight or style of the ruler:
1132 - ``0`` prints no ruler (but still prints centered text),
1133 - ``1`` prints normal ruler,
1134 - ``2`` prints bold ruler.
1136 Additional styles can be added through
1137 :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`.
1138 :param kwargs:
1139 any additional keyword arguments will be passed to :func:`raw`.
1141 """
1143 info(yuio.string.Hr(yuio.string._to_colorable(msg, args), weight=weight), **kwargs)
1146@_t.overload
1147def hl(
1148 msg: _t.LiteralString,
1149 /,
1150 *args,
1151 syntax: str,
1152 dedent: bool = True,
1153 **kwargs,
1154): ...
1155@_t.overload
1156def hl(
1157 msg: str,
1158 /,
1159 *,
1160 syntax: str,
1161 dedent: bool = True,
1162 **kwargs,
1163): ...
1164def hl(
1165 msg: str,
1166 /,
1167 *args,
1168 syntax: str,
1169 dedent: bool = True,
1170 **kwargs,
1171):
1172 """hl(msg: typing.LiteralString, /, *args, syntax: str, dedent: bool = True, **kwargs)
1173 hl(msg: str, /, *, syntax: str, dedent: bool = True, **kwargs) ->
1175 Print highlighted code. See :mod:`yuio.hl` for more info.
1177 :param msg:
1178 code to highlight.
1179 :param args:
1180 arguments for ``%``-formatting the highlighted code.
1181 :param syntax:
1182 name of syntax or a :class:`~yuio.hl.SyntaxHighlighter` instance.
1183 :param dedent:
1184 whether to remove leading indent from `msg`.
1185 :param kwargs:
1186 any additional keyword arguments will be passed to :func:`raw`.
1188 """
1190 info(yuio.string.Hl(msg, *args, syntax=syntax, dedent=dedent), **kwargs)
1193def raw(
1194 msg: yuio.string.Colorable,
1195 /,
1196 *,
1197 ignore_suspended: bool = False,
1198 tag: str | None = None,
1199 exc_info: ExcInfo | bool | None = None,
1200 add_newline: bool = False,
1201 heading: bool = False,
1202 wrap: bool = False,
1203 ctx: yuio.string.ReprContext | None = None,
1204 term: yuio.term.Term | None = None,
1205 to_stdout: bool = False,
1206 to_stderr: bool = False,
1207 theme: yuio.theme.Theme | None = None,
1208 multiline: bool | None = None,
1209 highlighted: bool | None = None,
1210 max_depth: int | None = None,
1211 width: int | None = None,
1212):
1213 """
1214 Print any :class:`~yuio.string.ToColorable`.
1216 This is a bridge between :mod:`yuio.io` and lower-level
1217 modules like :mod:`yuio.string`.
1219 :param msg:
1220 message to print.
1221 :param ignore_suspended:
1222 whether to ignore :class:`SuspendOutput` context.
1223 :param tag:
1224 tag that will be used to add color and decoration to the message.
1226 Decoration is looked up by path :samp:`{tag}`
1227 (see :attr:`Theme.msg_decorations <yuio.theme.Theme.msg_decorations_unicode>`),
1228 and color is looked up by path :samp:`msg/text:{tag}`
1229 (see :attr:`Theme.colors <yuio.theme.Theme.colors>`).
1230 :param exc_info:
1231 either a boolean indicating that the current exception
1232 should be captured, or a tuple of three elements, as returned
1233 by :func:`sys.exc_info`.
1234 :param add_newline:
1235 adds newline after the message.
1236 :param heading:
1237 whether to separate message by extra newlines.
1239 If :data:`True`, adds extra newline after the message; if this is not the
1240 first message printed so far, adds another newline before the message.
1241 :param wrap:
1242 whether to wrap message before printing it.
1243 :param ctx:
1244 :class:`~yuio.string.ReprContext` that should be used for formatting
1245 and printing the message.
1246 :param term:
1247 if `ctx` is not given, sets terminal where to print this message. Default is
1248 to use :func:`get_term`.
1249 :param to_stdout:
1250 shortcut for setting `term` to ``stdout``.
1251 :param to_stderr:
1252 shortcut for setting `term` to ``stderr``.
1253 :param theme:
1254 if `ctx` is not given, sets theme used to format the message. Default is
1255 to use :func:`get_theme`.
1256 :param multiline:
1257 if `ctx` is not given, sets initial value for
1258 :attr:`ReprContext.multiline <yuio.string.ReprContext.multiline>`.
1259 Default is :data:`False`.
1260 :param highlighted:
1261 if `ctx` is not given, sets initial value for
1262 :attr:`ReprContext.highlighted <yuio.string.ReprContext.highlighted>`.
1263 Default is :data:`False`.
1264 :param max_depth:
1265 if `ctx` is not given, sets initial value for
1266 :attr:`ReprContext.max_depth <yuio.string.ReprContext.max_depth>`.
1267 Default is :data:`False`.
1268 :param width:
1269 if `ctx` is not given, sets initial value for
1270 :attr:`ReprContext.width <yuio.string.ReprContext.width>`.
1271 If not given, uses current terminal width
1272 or :attr:`Theme.fallback_width <yuio.theme.Theme.fallback_width>`
1273 if terminal width can't be established.
1275 """
1277 if (ctx is not None) + (term is not None) + to_stdout + to_stderr > 1:
1278 names = []
1279 if ctx is not None:
1280 names.append("ctx")
1281 if term is not None:
1282 names.append("term")
1283 if to_stdout:
1284 names.append("to_stdout")
1285 if to_stderr:
1286 names.append("to_stderr")
1287 raise TypeError(f"{And(names)} can't be given together")
1289 manager = _manager()
1291 if ctx is None:
1292 ctx = make_repr_context(
1293 term=term,
1294 to_stdout=to_stdout,
1295 to_stderr=to_stderr,
1296 theme=theme,
1297 multiline=multiline,
1298 highlighted=highlighted,
1299 max_depth=max_depth,
1300 width=width,
1301 )
1303 if tag and (decoration := ctx.get_msg_decoration(tag)):
1304 indent = yuio.string.ColorizedString(
1305 [ctx.get_color(f"msg/decoration:{tag}"), decoration]
1306 )
1307 continuation_indent = " " * indent.width
1308 else:
1309 indent = ""
1310 continuation_indent = ""
1312 if tag:
1313 msg = yuio.string.WithBaseColor(
1314 msg, base_color=ctx.get_color(f"msg/text:{tag}")
1315 )
1317 if wrap:
1318 msg = yuio.string.Wrap(
1319 msg,
1320 indent=indent,
1321 continuation_indent=continuation_indent,
1322 )
1323 elif indent or continuation_indent:
1324 msg = yuio.string.Indent(
1325 msg,
1326 indent=indent,
1327 continuation_indent=continuation_indent,
1328 )
1330 msg = ctx.str(msg)
1332 if add_newline:
1333 msg.append_color(yuio.color.Color.NONE)
1334 msg.append_str("\n")
1336 if exc_info is True:
1337 exc_info = sys.exc_info()
1338 elif exc_info is False or exc_info is None:
1339 exc_info = None
1340 elif isinstance(exc_info, BaseException):
1341 exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
1342 elif not isinstance(exc_info, tuple) or len(exc_info) != 3:
1343 raise ValueError(f"invalid exc_info {exc_info!r}")
1344 if exc_info is not None and exc_info != (None, None, None):
1345 tb = "".join(traceback.format_exception(*exc_info))
1346 highlighter, syntax_name = yuio.hl.get_highlighter("python-traceback")
1347 msg += highlighter.highlight(tb, theme=ctx.theme, syntax=syntax_name).indent()
1349 manager.print(
1350 msg.as_code(ctx.term.color_support),
1351 ctx.term,
1352 ignore_suspended=ignore_suspended,
1353 heading=heading,
1354 )
1357class _AskWidget(yuio.widget.Widget[T], _t.Generic[T]):
1358 _layout: yuio.widget.VerticalLayout[T]
1360 def __init__(self, prompt: yuio.string.Colorable, widget: yuio.widget.Widget[T]):
1361 self._prompt = yuio.widget.Text(prompt)
1362 self._error: Exception | None = None
1363 self._inner = widget
1365 def event(self, e: yuio.widget.KeyboardEvent, /) -> yuio.widget.Result[T] | None:
1366 try:
1367 result = self._inner.event(e)
1368 except yuio.parse.ParsingError as err:
1369 self._error = err
1370 else:
1371 self._error = None
1372 return result
1374 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1375 builder = (
1376 yuio.widget.VerticalLayoutBuilder()
1377 .add(self._prompt)
1378 .add(self._inner, receive_events=True)
1379 )
1380 if self._error is not None:
1381 rc.bell()
1382 error_msg = yuio.string.colorize(
1383 "<c msg/decoration:error>▲</c> %s",
1384 yuio.string.Indent(self._error, indent=0, continuation_indent=2),
1385 default_color="msg/text:error",
1386 ctx=rc.make_repr_context(),
1387 )
1388 builder = builder.add(yuio.widget.Text(error_msg))
1390 self._layout = builder.build()
1391 return self._layout.layout(rc)
1393 def draw(self, rc: yuio.widget.RenderContext, /):
1394 self._layout.draw(rc)
1396 @property
1397 def help_data(self) -> yuio.widget.WidgetHelp:
1398 return self._inner.help_data
1401class _AskMeta(type):
1402 __hint = None
1404 @_t.overload
1405 def __call__(
1406 cls: type[ask[S]],
1407 msg: _t.LiteralString,
1408 /,
1409 *args,
1410 default: M | yuio.Missing = yuio.MISSING,
1411 default_non_interactive: _t.Any = yuio.MISSING,
1412 parser: yuio.parse.Parser[S] | None = None,
1413 input_description: str | None = None,
1414 default_description: str | None = None,
1415 ) -> S | M: ...
1416 @_t.overload
1417 def __call__(
1418 cls: type[ask[S]],
1419 msg: str,
1420 /,
1421 *,
1422 default: M | yuio.Missing = yuio.MISSING,
1423 default_non_interactive: _t.Any = yuio.MISSING,
1424 parser: yuio.parse.Parser[S] | None = None,
1425 input_description: str | None = None,
1426 default_description: str | None = None,
1427 ) -> S | M: ...
1428 def __call__(cls, *args, **kwargs):
1429 if "parser" not in kwargs:
1430 hint = cls.__hint
1431 if hint is None:
1432 hint = str
1433 kwargs["parser"] = yuio.parse.from_type_hint(hint)
1434 return _ask(*args, **kwargs)
1436 def __getitem(cls, ty):
1437 return _AskMeta("ask", (), {"_AskMeta__hint": ty})
1439 # A dirty hack to hide `__getitem__` from type checkers. `ask` should look like
1440 # an ordinary class with overloaded `__new__` for the magic to work.
1441 locals()["__getitem__"] = __getitem
1443 def __repr__(cls) -> str:
1444 if cls.__hint is None:
1445 return cls.__name__
1446 else:
1447 return f"{cls.__name__}[{_tx.type_repr(cls.__hint)}]"
1450@_t.final
1451class ask(_t.Generic[S], metaclass=_AskMeta):
1452 """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
1453 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
1455 Ask user to provide an input, parse it and return a value.
1457 If current terminal is not interactive, return default if one is present,
1458 or raise a :class:`UserIoError`.
1460 .. vhs:: /_tapes/questions.tape
1461 :alt: Demonstration of the `ask` function.
1462 :width: 480
1463 :height: 240
1465 :func:`ask` accepts generic parameters, which determine how input is parsed.
1466 For example, if you're asking for an enum element,
1467 Yuio will show user a choice widget.
1469 You can also supply a custom :class:`~yuio.parse.Parser`,
1470 which will determine the widget that is displayed to the user,
1471 the way autocompletion works, etc.
1473 .. note::
1475 :func:`ask` is designed to interact with users, not to read data. It uses
1476 ``/dev/tty`` on Unix, and console API on Windows, so it will read from
1477 an actual TTY even if ``stdin`` is redirected.
1479 When designing your program, make sure that users have alternative means
1480 to provide values: use configs or CLI arguments, allow passing passwords
1481 via environment variables, etc.
1483 :param msg:
1484 prompt to display to user.
1485 :param args:
1486 arguments for ``%``\\ - formatting the prompt.
1487 :param parser:
1488 parser to use to parse user input. See :mod:`yuio.parse` for more info.
1489 :param default:
1490 default value to return if user input is empty.
1491 :param default_non_interactive:
1492 default value returned if input stream is not readable. If not given,
1493 `default` is used instead. This is handy when you want to ask user if they
1494 want to continue with `default` set to :data:`False`,
1495 but `default_non_interactive` set to :data:`True`.
1496 :param input_description:
1497 description of the expected input, like ``"yes/no"`` for boolean
1498 inputs.
1499 :param default_description:
1500 description of the `default` value.
1501 :returns:
1502 parsed user input.
1503 :raises:
1504 raises :class:`UserIoError` if we're not in interactive environment, and there
1505 is no default to return.
1506 :example:
1507 .. invisible-code-block: python
1509 import enum
1511 .. code-block:: python
1513 class Level(enum.Enum):
1514 WARNING = "Warning"
1515 INFO = "Info"
1516 DEBUG = "Debug"
1519 answer = yuio.io.ask[Level]("Choose a logging level", default=Level.INFO)
1521 """
1523 if TYPE_CHECKING:
1525 @_t.overload
1526 def __new__(
1527 cls: type[ask[S]],
1528 msg: _t.LiteralString,
1529 /,
1530 *args,
1531 default: M | yuio.Missing = yuio.MISSING,
1532 default_non_interactive: _t.Any = yuio.MISSING,
1533 parser: yuio.parse.Parser[S] | None = None,
1534 input_description: str | None = None,
1535 default_description: str | None = None,
1536 ) -> S | M: ...
1537 @_t.overload
1538 def __new__(
1539 cls: type[ask[S]],
1540 msg: str,
1541 /,
1542 *,
1543 default: M | yuio.Missing = yuio.MISSING,
1544 default_non_interactive: _t.Any = yuio.MISSING,
1545 parser: yuio.parse.Parser[S] | None = None,
1546 input_description: str | None = None,
1547 default_description: str | None = None,
1548 ) -> S | M: ...
1549 def __new__(cls: _t.Any, *_, **__) -> _t.Any: ...
1552def _ask(
1553 msg: _t.LiteralString,
1554 /,
1555 *args,
1556 parser: yuio.parse.Parser[_t.Any],
1557 default: _t.Any = yuio.MISSING,
1558 default_non_interactive: _t.Any = yuio.MISSING,
1559 input_description: str | None = None,
1560 default_description: str | None = None,
1561) -> _t.Any:
1562 ctx = make_repr_context(term=yuio.term.get_tty())
1564 if not _can_query_user(ctx.term):
1565 # TTY is not available.
1566 if default_non_interactive is yuio.MISSING:
1567 default_non_interactive = default
1568 if default_non_interactive is yuio.MISSING:
1569 raise UserIoError("Can't interact with user in non-interactive environment")
1570 return default_non_interactive
1572 if default is None and not yuio.parse._is_optional_parser(parser):
1573 parser = yuio.parse.Optional(parser)
1575 msg = msg.rstrip()
1576 if msg.endswith(":"):
1577 needs_colon = True
1578 msg = msg[:-1]
1579 else:
1580 needs_colon = msg and msg[-1] not in string.punctuation
1582 base_color = ctx.get_color("msg/text:question")
1583 prompt = yuio.string.colorize(msg, *args, default_color=base_color, ctx=ctx)
1585 if not input_description:
1586 input_description = parser.describe()
1588 if default is not yuio.MISSING and default_description is None:
1589 try:
1590 default_description = parser.describe_value(default)
1591 except TypeError:
1592 default_description = str(default)
1594 if not yuio.term._is_foreground(ctx.term.ostream):
1595 warning(
1596 "User input is requested in background process, use `fg %s` to resume",
1597 os.getpid(),
1598 ctx=ctx,
1599 )
1600 yuio.term._pause()
1602 if ctx.term.can_run_widgets:
1603 # Use widget.
1605 if needs_colon:
1606 prompt.append_color(base_color)
1607 prompt.append_str(":")
1609 if parser.is_secret():
1610 inner_widget = yuio.parse._secret_widget(
1611 parser, default, input_description, default_description
1612 )
1613 else:
1614 inner_widget = parser.widget(
1615 default, input_description, default_description
1616 )
1618 widget = _AskWidget(prompt, inner_widget)
1619 with SuspendOutput() as s:
1620 try:
1621 result = widget.run(ctx.term, ctx.theme)
1622 except (OSError, EOFError) as e: # pragma: no cover
1623 raise UserIoError("Unexpected end of input") from e
1625 if result is yuio.MISSING:
1626 result = default
1628 try:
1629 result_desc = parser.describe_value(result)
1630 except TypeError:
1631 result_desc = str(result)
1633 prompt.append_color(base_color)
1634 prompt.append_str(" ")
1635 prompt.append_color(base_color | ctx.get_color("code"))
1636 prompt.append_str(result_desc)
1638 # note: print to default terminal, not to tty
1639 s.info(prompt, tag="question")
1641 return result
1642 else:
1643 # Use raw input.
1645 prompt += base_color
1646 if input_description:
1647 prompt += " ("
1648 prompt += input_description
1649 prompt += ")"
1650 if default_description:
1651 prompt += " ["
1652 prompt += base_color | ctx.get_color("code")
1653 prompt += default_description
1654 prompt += base_color
1655 prompt += "]"
1656 prompt += yuio.string.Esc(": " if needs_colon else " ")
1657 if parser.is_secret():
1658 do_input = _getpass
1659 else:
1660 do_input = _read
1661 with SuspendOutput() as s:
1662 while True:
1663 try:
1664 answer = do_input(ctx.term, prompt)
1665 except (OSError, EOFError) as e: # pragma: no cover
1666 raise UserIoError("Unexpected end of input") from e
1667 if not answer and default is not yuio.MISSING:
1668 return default
1669 elif not answer:
1670 s.error("Input is required.", ctx=ctx)
1671 else:
1672 try:
1673 return parser.parse(answer)
1674 except yuio.parse.ParsingError as e:
1675 s.error(e, ctx=ctx)
1678if os.name == "posix":
1679 # Getpass implementation is based on the standard `getpass` module, with a few
1680 # Yuio-specific modifications.
1682 def _getpass_fallback(
1683 term: yuio.term.Term, prompt: yuio.string.ColorizedString
1684 ) -> str:
1685 warning(
1686 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1687 )
1688 return _read(term, prompt)
1690 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1691 info(
1692 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1693 )
1694 return term.istream.readline().rstrip("\r\n")
1696 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1697 import termios
1699 try:
1700 fd = term.istream.fileno()
1701 except (AttributeError, ValueError):
1702 # We can't control the tty or stdin. Give up and use normal IO.
1703 return _getpass_fallback(term, prompt)
1705 result: str | None = None
1707 try:
1708 prev_mode = termios.tcgetattr(fd)
1709 new_mode = prev_mode.copy()
1710 new_mode[3] &= ~termios.ECHO
1711 tcsetattr_flags = termios.TCSAFLUSH | getattr(termios, "TCSASOFT", 0)
1712 try:
1713 termios.tcsetattr(fd, tcsetattr_flags, new_mode)
1714 info(
1715 prompt,
1716 add_newline=False,
1717 tag="question",
1718 term=term,
1719 ignore_suspended=True,
1720 )
1721 result = term.istream.readline().rstrip("\r\n")
1722 term.ostream.write("\n")
1723 term.ostream.flush()
1724 finally:
1725 termios.tcsetattr(fd, tcsetattr_flags, prev_mode)
1726 except termios.error:
1727 if result is not None:
1728 # `readline` succeeded, the final `tcsetattr` failed. Reraise instead
1729 # of leaving the terminal in an unknown state.
1730 raise
1731 else:
1732 # We can't control the tty or stdin. Give up and use normal IO.
1733 return _getpass_fallback(term, prompt)
1735 assert result is not None
1736 return result
1738elif os.name == "nt":
1740 def _do_read(
1741 term: yuio.term.Term, prompt: yuio.string.ColorizedString, echo: bool
1742 ) -> str:
1743 import msvcrt
1745 if term.ostream_is_tty:
1746 info(
1747 prompt,
1748 add_newline=False,
1749 tag="question",
1750 term=term,
1751 ignore_suspended=True,
1752 )
1753 else:
1754 for c in str(prompt):
1755 msvcrt.putwch(c)
1757 if term.ostream_is_tty and echo:
1758 return term.istream.readline().rstrip("\r\n")
1759 else:
1760 result = ""
1761 while True:
1762 c = msvcrt.getwch()
1763 if c == "\0" or c == "\xe0":
1764 # Read key scan code and ignore it.
1765 msvcrt.getwch()
1766 continue
1767 if c == "\r" or c == "\n":
1768 break
1769 if c == "\x03":
1770 raise KeyboardInterrupt
1771 if c == "\b":
1772 if result:
1773 msvcrt.putwch("\b")
1774 msvcrt.putwch(" ")
1775 msvcrt.putwch("\b")
1776 result = result[:-1]
1777 else:
1778 result = result + c
1779 if echo:
1780 msvcrt.putwch(c)
1781 else:
1782 msvcrt.putwch("*")
1783 msvcrt.putwch("\r")
1784 msvcrt.putwch("\n")
1786 return result
1788 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1789 return _do_read(term, prompt, echo=True)
1791 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString):
1792 return _do_read(term, prompt, echo=False)
1794else:
1796 def _getpass(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1797 warning(
1798 "Warning: Password input may be echoed.", term=term, ignore_suspended=True
1799 )
1800 return _read(term, prompt)
1802 def _read(term: yuio.term.Term, prompt: yuio.string.ColorizedString) -> str:
1803 info(
1804 prompt, add_newline=False, tag="question", term=term, ignore_suspended=True
1805 )
1806 return term.istream.readline().rstrip("\r\n")
1809def _can_query_user(term: yuio.term.Term):
1810 return (
1811 # We're attached to a TTY.
1812 term.is_tty
1813 # On Windows, there is no way to bring a process to foreground (AFAIK?).
1814 # Thus, we need to check if there's a console window.
1815 and (os.name != "nt" or yuio.term._is_foreground(None))
1816 )
1819class _WaitForUserWidget(yuio.widget.Widget[None]):
1820 def __init__(self, prompt: yuio.string.Colorable):
1821 self._prompt = yuio.widget.Text(prompt)
1823 def layout(self, rc: yuio.widget.RenderContext, /) -> tuple[int, int]:
1824 return self._prompt.layout(rc)
1826 def draw(self, rc: yuio.widget.RenderContext, /):
1827 return self._prompt.draw(rc)
1829 @yuio.widget.bind(yuio.widget.Key.ENTER)
1830 @yuio.widget.bind(yuio.widget.Key.ESCAPE)
1831 @yuio.widget.bind("d", ctrl=True)
1832 @yuio.widget.bind(" ")
1833 def exit(self):
1834 return yuio.widget.Result(None)
1837def wait_for_user(
1838 msg: _t.LiteralString = "Press <c note>enter</c> to continue",
1839 /,
1840 *args,
1841):
1842 """
1843 A simple function to wait for user to press enter.
1845 If current terminal is not interactive, this functions has no effect.
1847 :param msg:
1848 prompt to display to user.
1849 :param args:
1850 arguments for ``%``\\ - formatting the prompt.
1852 """
1854 ctx = make_repr_context(term=yuio.term.get_tty())
1856 if not _can_query_user(ctx.term):
1857 # TTY is not available.
1858 return
1860 if not yuio.term._is_foreground(ctx.term.ostream):
1861 if os.name == "nt":
1862 # AFAIK there's no way to bring job to foreground in Windows.
1863 return
1865 warning(
1866 "User input is requested in background process, use `fg %s` to resume",
1867 os.getpid(),
1868 ctx=ctx,
1869 )
1870 yuio.term._pause()
1872 prompt = yuio.string.colorize(
1873 msg.rstrip(), *args, default_color="msg/text:question", ctx=ctx
1874 )
1875 prompt += yuio.string.Esc(" ")
1877 with SuspendOutput():
1878 try:
1879 if ctx.term.can_run_widgets:
1880 _WaitForUserWidget(prompt).run(ctx.term, ctx.theme)
1881 else:
1882 _read(ctx.term, prompt)
1883 except (OSError, EOFError): # pragma: no cover
1884 return
1887def detect_editor(fallbacks: list[str] | None = None) -> str | None:
1888 """
1889 Detect the user's preferred editor.
1891 This function checks the ``VISUAL`` and ``EDITOR`` environment variables.
1892 If they're not set, it checks if any of the fallback editors are available.
1893 If none can be found, it returns :data:`None`.
1895 :param fallbacks:
1896 list of fallback editors to try. By default, we try "nano", "vim", "vi",
1897 "msedit", "edit", "notepad", "gedit".
1898 :returns:
1899 on Windows, returns an executable name; on Unix, may return a shell command
1900 or an executable name.
1902 """
1904 if os.name != "nt":
1905 if editor := os.environ.get("VISUAL"):
1906 return editor
1907 if editor := os.environ.get("EDITOR"):
1908 return editor
1910 if fallbacks is None:
1911 fallbacks = ["nano", "vim", "vi", "msedit", "edit", "notepad", "gedit"]
1912 for fallback in fallbacks:
1913 if shutil.which(fallback):
1914 return fallback
1915 return None
1918def detect_shell(fallbacks: list[str] | None = None) -> str | None:
1919 """
1920 Detect the user's preferred shell.
1922 This function checks the ``SHELL`` environment variable.
1923 If it's not set, it checks if any of the fallback shells are available.
1924 If none can be found, it returns :data:`None`.
1926 :param fallbacks:
1927 list of fallback shells to try. By default, we try "pwsh" and "powershell"
1928 on Windows, and "bash", "sh", "/bin/sh" on Linux/MacOS.
1929 :returns:
1930 returns an executable name.
1932 """
1934 if os.name != "nt" and (shell := os.environ.get("SHELL")):
1935 return shell
1937 if fallbacks is None:
1938 if os.name != "nt":
1939 fallbacks = ["bash", "sh", "/bin/sh"]
1940 else:
1941 fallbacks = ["pwsh", "powershell"]
1942 for fallback in fallbacks:
1943 if shutil.which(fallback):
1944 return fallback
1945 return None
1948def edit(
1949 text: str,
1950 /,
1951 *,
1952 comment_marker: str | None = None,
1953 editor: str | None = None,
1954 file_ext: str = ".txt",
1955 fallbacks: list[str] | None = None,
1956 dedent: bool = False,
1957) -> str:
1958 """
1959 Ask user to edit some text.
1961 This function creates a temporary file with the given text
1962 and opens it in an editor. After editing is done, it strips away
1963 all lines that start with `comment_marker`, if one is given.
1965 :param text:
1966 text to edit.
1967 :param comment_marker:
1968 lines starting with this marker will be removed from the output after edit.
1969 :param editor:
1970 overrides editor.
1972 On Unix, this should be a shell command, file path will be appended to it;
1973 on Windows, this should be an executable path.
1974 :param file_ext:
1975 extension for the temporary file, can be used to enable syntax highlighting
1976 in editors that support it.
1977 :param fallbacks:
1978 list of fallback editors to try, see :func:`detect_editor` for details.
1979 :param dedent:
1980 remove leading indentation from text before opening an editor.
1981 :returns:
1982 an edited string with comments removed.
1983 :raises:
1984 If editor is not available, returns a non-zero exit code, or launched in
1985 a non-interactive environment, a :class:`UserIoError` is raised.
1987 Also raises :class:`UserIoError` if ``stdin`` or ``stderr`` is piped
1988 or redirected to a file (virtually no editors can work when this happens).
1989 :example:
1990 .. skip: next
1992 .. code-block:: python
1994 message = yuio.io.edit(
1995 \"""
1996 # Please enter the commit message for your changes. Lines starting
1997 # with '#' will be ignored, and an empty message aborts the commit.
1998 \""",
1999 comment_marker="#",
2000 dedent=True,
2001 )
2003 """
2005 term = yuio.term.get_tty()
2007 if not _can_query_user(term):
2008 raise UserIoError("Can't run editor in non-interactive environment")
2010 if editor is None:
2011 editor = detect_editor(fallbacks)
2013 if editor is None:
2014 if os.name == "nt":
2015 raise UserIoError("Can't find a usable editor")
2016 else:
2017 raise UserIoError(
2018 "Can't find a usable editor. Ensure that `$VISUAL` and `$EDITOR` "
2019 "environment variables contain correct path to an editor executable"
2020 )
2022 if dedent:
2023 text = _dedent(text)
2025 if not yuio.term._is_foreground(term.ostream):
2026 warning(
2027 "Background process is waiting for user, use `fg %s` to resume",
2028 os.getpid(),
2029 term=term,
2030 )
2031 yuio.term._pause()
2033 fd, filepath = tempfile.mkstemp(text=True, suffix=file_ext)
2034 try:
2035 with open(fd, "w") as file:
2036 file.write(text)
2038 if os.name == "nt":
2039 # Windows doesn't use $VISUAL/$EDITOR, so shell execution is not needed.
2040 # Plus, quoting arguments for CMD.exe is hard af.
2041 args = [editor, filepath]
2042 shell = False
2043 else:
2044 # $VISUAL/$EDITOR can include flags, so we need to use shell instead.
2045 from shlex import quote
2047 args = f"{editor} {quote(filepath)}"
2048 shell = True
2050 try:
2051 with SuspendOutput():
2052 res = subprocess.run(
2053 args,
2054 shell=shell,
2055 stdin=term.istream.fileno(),
2056 stdout=term.ostream.fileno(),
2057 )
2058 except FileNotFoundError:
2059 raise UserIoError(
2060 "Can't use editor `%r`: no such file or directory",
2061 editor,
2062 )
2064 if res.returncode != 0:
2065 if res.returncode < 0:
2066 import signal
2068 try:
2069 action = "died with"
2070 code = signal.Signals(-res.returncode).name
2071 except ValueError:
2072 action = "died with unknown signal"
2073 code = res.returncode
2074 else:
2075 action = "returned exit code"
2076 code = res.returncode
2077 raise UserIoError(
2078 "Editing failed: editor `%r` %s `%s`",
2079 editor,
2080 action,
2081 code,
2082 )
2084 if not os.path.exists(filepath):
2085 raise UserIoError("Editing failed: can't read back edited file")
2086 else:
2087 with open(filepath) as file:
2088 text = file.read()
2089 finally:
2090 try:
2091 os.remove(filepath)
2092 except OSError:
2093 pass
2095 if comment_marker is not None:
2096 text = re.sub(
2097 r"^\s*" + re.escape(comment_marker) + r".*(\n|$)",
2098 "",
2099 text,
2100 flags=re.MULTILINE,
2101 )
2103 return text
2106def shell(
2107 *,
2108 shell: str | None = None,
2109 fallbacks: list[str] | None = None,
2110 prompt_marker: str = "",
2111):
2112 """
2113 Launch an interactive shell and give user control over it.
2115 This function is useful in interactive scripts. For example, if the script is
2116 creating a release commit, it might be desired to give user a chance to inspect
2117 repository status before proceeding.
2119 :param shell:
2120 overrides shell executable.
2121 :param fallbacks:
2122 list of fallback shells to try, see :func:`detect_shell` for details.
2123 :param prompt_marker:
2124 if given, Yuio will try to add this marker to the shell's prompt
2125 to remind users that this shell is a sub-process of some script.
2127 This only works with Bash, Zsh, Fish, and PowerShell.
2129 """
2131 term = yuio.term.get_tty()
2133 if not _can_query_user(term):
2134 raise UserIoError("Can't run editor in non-interactive environment")
2136 if shell is None:
2137 shell = detect_shell(fallbacks=fallbacks)
2139 if shell is None:
2140 if os.name == "nt":
2141 raise UserIoError("Can't find a usable shell")
2142 else:
2143 raise UserIoError(
2144 "Can't find a usable shell. Ensure that `$SHELL`"
2145 "environment variable contain correct path to a shell executable"
2146 )
2148 args = [shell]
2149 env = os.environ.copy()
2151 rcpath = None
2152 rcpath_is_dir = False
2153 if prompt_marker:
2154 env["__YUIO_PROMPT_MARKER"] = prompt_marker
2156 if shell == "bash" or shell.endswith(os.path.sep + "bash"):
2157 fd, rcpath = tempfile.mkstemp(text=True, suffix=".bash")
2159 rc = textwrap.dedent(
2160 """
2161 [ -f ~/.bashrc ] && source ~/.bashrc;
2162 PS1='\\e[33m$__YUIO_PROMPT_MARKER\\e[m'\" $PS1\"
2163 """
2164 )
2166 with open(fd, "w") as file:
2167 file.write(rc)
2169 args += ["--rcfile", rcpath, "-i"]
2170 elif shell == "zsh" or shell.endswith(os.path.sep + "zsh"):
2171 rcpath = tempfile.mkdtemp()
2172 rcpath_is_dir = True
2174 rc = textwrap.dedent(
2175 """
2176 ZDOTDIR=$ZDOTDIR_ORIG
2177 [ -f $ZDOTDIR/.zprofile ] && source $ZDOTDIR/.zprofile
2178 [ -f $ZDOTDIR/.zshrc ] && source $ZDOTDIR/.zshrc
2179 autoload -U colors && colors
2180 PS1='%F{yellow}$__YUIO_PROMPT_MARKER%f'" $PS1"
2181 """
2182 )
2184 with open(os.path.join(rcpath, ".zshrc"), "w") as file:
2185 file.write(rc)
2187 if "ZDOTDIR" in env:
2188 zdotdir = env["ZDOTDIR"]
2189 else:
2190 zdotdir = os.path.expanduser("~")
2192 env["ZDOTDIR"] = rcpath
2193 env["ZDOTDIR_ORIG"] = zdotdir
2195 args += ["-i"]
2196 elif shell == "fish" or shell.endswith(os.path.sep + "fish"):
2197 rc = textwrap.dedent(
2198 """
2199 functions -c fish_prompt _yuio_old_fish_prompt
2200 function fish_prompt
2201 set -l old_status $status
2202 printf "%s%s%s " (set_color yellow) $__YUIO_PROMPT_MARKER (set_color normal)
2203 echo "exit $old_status" | .
2204 _yuio_old_fish_prompt
2205 end
2206 """
2207 )
2209 args += ["--init-command", rc, "-i"]
2210 elif shell in ["powershell", "pwsh"] or shell.endswith(
2211 (os.path.sep + "powershell", os.path.sep + "pwsh")
2212 ):
2213 fd, rcpath = tempfile.mkstemp(text=True, suffix=".ps1")
2215 rc = textwrap.dedent(
2216 """
2217 function global:_yuio_old_pwsh_prompt { "" }
2218 Copy-Item -Path function:prompt -Destination function:_yuio_old_pwsh_prompt
2220 function global:prompt {
2221 Write-Host -NoNewline -ForegroundColor Yellow "$env:__YUIO_PROMPT_MARKER "
2222 _yuio_old_pwsh_prompt
2223 }
2224 """
2225 )
2227 with open(fd, "w") as file:
2228 file.write(rc)
2230 args += ["-NoExit", "-File", rcpath]
2232 try:
2233 with SuspendOutput():
2234 subprocess.run(
2235 args,
2236 env=env,
2237 stdin=term.istream.fileno(),
2238 stdout=term.ostream.fileno(),
2239 )
2240 except FileNotFoundError:
2241 raise UserIoError(
2242 "Can't use shell `%r`: no such file or directory",
2243 shell,
2244 )
2245 finally:
2246 if rcpath:
2247 try:
2248 if rcpath_is_dir:
2249 shutil.rmtree(rcpath)
2250 else:
2251 os.remove(rcpath)
2252 except OSError:
2253 pass
2256class MessageChannel:
2257 """
2258 Message channels are similar to logging adapters: they allow adding additional
2259 arguments for calls to :func:`raw` and other message functions.
2261 This is useful when you need to control destination for messages, but don't want
2262 to override global settings via :func:`setup`. One example for them is described
2263 in :ref:`cookbook-print-to-file`.
2265 .. dropdown:: Protected members
2267 .. autoattribute:: _msg_kwargs
2269 .. automethod:: _update_kwargs
2271 .. automethod:: _is_enabled
2273 """
2275 enabled: bool
2276 """
2277 Message channel can be disabled, in which case messages are not printed.
2279 """
2281 _msg_kwargs: dict[str, _t.Any]
2282 """
2283 Keyword arguments that will be added to every message.
2285 """
2287 if _t.TYPE_CHECKING:
2289 def __init__(
2290 self,
2291 *,
2292 ignore_suspended: bool = False,
2293 term: yuio.term.Term | None = None,
2294 to_stdout: bool = False,
2295 to_stderr: bool = False,
2296 theme: yuio.theme.Theme | None = None,
2297 multiline: bool | None = None,
2298 highlighted: bool | None = None,
2299 max_depth: int | None = None,
2300 width: int | None = None,
2301 ): ...
2302 else:
2304 def __init__(self, **kwargs):
2305 self._msg_kwargs: dict[str, _t.Any] = kwargs
2306 self.enabled: bool = True
2308 def _update_kwargs(self, kwargs: dict[str, _t.Any]):
2309 """
2310 A hook that updates method's `kwargs` before calling its implementation.
2312 """
2314 for name, option in self._msg_kwargs.items():
2315 kwargs.setdefault(name, option)
2317 def _is_enabled(self):
2318 """
2319 A hook that check if the message should be printed. By default, returns value
2320 of :attr:`~MessageChannel.enabled`.
2322 """
2324 return self.enabled
2326 @_t.overload
2327 def info(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2328 @_t.overload
2329 def info(self, err: yuio.string.ToColorable, /, **kwargs): ...
2330 def info(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2331 """info(msg: typing.LiteralString, /, *args, **kwargs)
2332 info(msg: ~string.templatelib.Template, /, **kwargs) ->
2333 info(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2335 Print an :func:`info` message.
2337 """
2339 if not self._is_enabled():
2340 return
2342 self._update_kwargs(kwargs)
2343 info(msg, *args, **kwargs)
2345 @_t.overload
2346 def warning(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2347 @_t.overload
2348 def warning(self, err: yuio.string.ToColorable, /, **kwargs): ...
2349 def warning(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2350 """warning(msg: typing.LiteralString, /, *args, **kwargs)
2351 warning(msg: ~string.templatelib.Template, /, **kwargs) ->
2352 warning(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2354 Print a :func:`warning` message.
2356 """
2358 if not self._is_enabled():
2359 return
2361 self._update_kwargs(kwargs)
2362 warning(msg, *args, **kwargs)
2364 @_t.overload
2365 def success(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2366 @_t.overload
2367 def success(self, err: yuio.string.ToColorable, /, **kwargs): ...
2368 def success(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2369 """success(msg: typing.LiteralString, /, *args, **kwargs)
2370 success(msg: ~string.templatelib.Template, /, **kwargs) ->
2371 success(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2373 Print a :func:`success` message.
2375 """
2377 if not self._is_enabled():
2378 return
2380 self._update_kwargs(kwargs)
2381 success(msg, *args, **kwargs)
2383 @_t.overload
2384 def error(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2385 @_t.overload
2386 def error(self, err: yuio.string.ToColorable, /, **kwargs): ...
2387 def error(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2388 """error(msg: typing.LiteralString, /, *args, **kwargs)
2389 error(msg: ~string.templatelib.Template, /, **kwargs) ->
2390 error(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2392 Print an :func:`error` message.
2394 """
2396 if not self._is_enabled():
2397 return
2399 self._update_kwargs(kwargs)
2400 error(msg, *args, **kwargs)
2402 @_t.overload
2403 def error_with_tb(
2404 self,
2405 msg: _t.LiteralString,
2406 /,
2407 *args,
2408 exc_info: ExcInfo | bool | None = True,
2409 **kwargs,
2410 ): ...
2411 @_t.overload
2412 def error_with_tb(
2413 self,
2414 msg: yuio.string.ToColorable,
2415 /,
2416 *,
2417 exc_info: ExcInfo | bool | None = True,
2418 **kwargs,
2419 ): ...
2420 def error_with_tb(
2421 self,
2422 msg: yuio.string.ToColorable,
2423 /,
2424 *args,
2425 exc_info: ExcInfo | bool | None = True,
2426 **kwargs,
2427 ):
2428 """error_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2429 error_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2430 error_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2432 Print an :func:`error_with_tb` message.
2434 """
2436 if not self._is_enabled():
2437 return
2439 self._update_kwargs(kwargs)
2440 error_with_tb(msg, *args, **kwargs)
2442 @_t.overload
2443 def failure(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2444 @_t.overload
2445 def failure(self, err: yuio.string.ToColorable, /, **kwargs): ...
2446 def failure(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2447 """failure(msg: typing.LiteralString, /, *args, **kwargs)
2448 failure(msg: ~string.templatelib.Template, /, **kwargs) ->
2449 failure(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2451 Print a :func:`failure` message.
2453 """
2455 if not self._is_enabled():
2456 return
2458 self._update_kwargs(kwargs)
2459 failure(msg, *args, **kwargs)
2461 @_t.overload
2462 def failure_with_tb(
2463 self,
2464 msg: _t.LiteralString,
2465 /,
2466 *args,
2467 exc_info: ExcInfo | bool | None = True,
2468 **kwargs,
2469 ): ...
2470 @_t.overload
2471 def failure_with_tb(
2472 self,
2473 msg: yuio.string.ToColorable,
2474 /,
2475 *,
2476 exc_info: ExcInfo | bool | None = True,
2477 **kwargs,
2478 ): ...
2479 def failure_with_tb(
2480 self,
2481 msg: yuio.string.ToColorable,
2482 /,
2483 *args,
2484 exc_info: ExcInfo | bool | None = True,
2485 **kwargs,
2486 ):
2487 """failure_with_tb(msg: typing.LiteralString, /, *args, **kwargs)
2488 failure_with_tb(msg: ~string.templatelib.Template, /, **kwargs) ->
2489 failure_with_tb(msg: ~yuio.string.ToColorable, /, **kwargs) ->
2491 Print a :func:`failure_with_tb` message.
2493 """
2495 if not self._is_enabled():
2496 return
2498 self._update_kwargs(kwargs)
2499 failure_with_tb(msg, *args, **kwargs)
2501 @_t.overload
2502 def heading(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2503 @_t.overload
2504 def heading(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2505 def heading(self, msg: yuio.string.ToColorable, /, *args, **kwargs):
2506 """heading(msg: typing.LiteralString, /, *args, **kwargs)
2507 heading(msg: ~string.templatelib.Template, /, **kwargs)
2508 heading(msg: ~yuio.string.ToColorable, /, **kwargs)
2510 Print a :func:`heading` message.
2512 """
2514 if not self._is_enabled():
2515 return
2517 self._update_kwargs(kwargs)
2518 heading(msg, *args, **kwargs)
2520 def md(self, msg: str, /, **kwargs):
2521 """
2522 Print an :func:`md` message.
2524 """
2526 if not self._is_enabled():
2527 return
2529 self._update_kwargs(kwargs)
2530 md(msg, **kwargs)
2532 def rst(self, msg: str, /, **kwargs):
2533 """
2534 Print an :func:`rst` message.
2536 """
2538 if not self._is_enabled():
2539 return
2541 self._update_kwargs(kwargs)
2542 rst(msg, **kwargs)
2544 def br(self, **kwargs):
2545 """br()
2547 Print a :func:`br` message.
2549 """
2551 if not self._is_enabled():
2552 return
2554 self._update_kwargs(kwargs)
2555 br(**kwargs)
2557 @_t.overload
2558 def hl(self, msg: _t.LiteralString, /, *args, **kwargs): ...
2559 @_t.overload
2560 def hl(self, msg: str, /, **kwargs): ...
2561 def hl(self, msg: str, /, *args, **kwargs):
2562 """hl(msg: typing.LiteralString, /, *args, syntax: str, dedent: bool = True, **kwargs)
2563 hl(msg: str, /, *, syntax: str, dedent: bool = True, **kwargs)
2565 Print an :func:`hl` message.
2567 """
2569 if not self._is_enabled():
2570 return
2572 self._update_kwargs(kwargs)
2573 hl(msg, *args, **kwargs)
2575 @_t.overload
2576 def hr(self, msg: _t.LiteralString = "", /, *args, **kwargs): ...
2577 @_t.overload
2578 def hr(self, msg: yuio.string.ToColorable, /, **kwargs): ...
2579 def hr(self, msg: yuio.string.ToColorable = "", /, *args, **kwargs):
2580 """hr(msg: typing.LiteralString = "", /, *args, weight: int | str = 1, **kwargs)
2581 hr(msg: ~string.templatelib.Template, /, *, weight: int | str = 1, **kwargs) ->
2582 hr(msg: ~yuio.string.ToColorable, /, *, weight: int | str = 1, **kwargs) ->
2584 Print an :func:`hr` message.
2586 """
2588 if not self._is_enabled():
2589 return
2591 self._update_kwargs(kwargs)
2592 hr(msg, *args, **kwargs)
2594 def raw(self, msg: yuio.string.Colorable, /, **kwargs):
2595 """
2596 Print a :func:`raw` message.
2598 """
2600 if not self._is_enabled():
2601 return
2603 self._update_kwargs(kwargs)
2604 raw(msg, **kwargs)
2606 def make_repr_context(self) -> yuio.string.ReprContext:
2607 """
2608 Make a :class:`~yuio.string.ReprContext` using settings
2609 from :attr:`~MessageChannel._msg_kwargs`.
2611 """
2613 return make_repr_context(
2614 term=self._msg_kwargs.get("term"),
2615 to_stdout=self._msg_kwargs.get("to_stdout", False),
2616 to_stderr=self._msg_kwargs.get("to_stderr", False),
2617 theme=self._msg_kwargs.get("theme"),
2618 multiline=self._msg_kwargs.get("multiline"),
2619 highlighted=self._msg_kwargs.get("highlighted"),
2620 max_depth=self._msg_kwargs.get("max_depth"),
2621 width=self._msg_kwargs.get("width"),
2622 )
2625class SuspendOutput(MessageChannel):
2626 """
2627 A context manager for pausing output.
2629 This is handy for when you need to take control over the output stream.
2630 For example, the :func:`ask` function uses this class internally.
2632 This context manager also suspends all prints that go to :data:`sys.stdout`
2633 and :data:`sys.stderr` if they were wrapped (see :func:`setup`).
2634 To print through them, use :func:`orig_stderr` and :func:`orig_stdout`.
2636 Each instance of this class is a :class:`MessageChannel`; calls to its printing
2637 methods bypass output suppression:
2639 .. code-block:: python
2641 with SuspendOutput() as out:
2642 print("Suspended") # [1]_
2643 out.info("Not suspended") # [2]_
2645 .. code-annotations::
2647 1. This message is suspended; it will be printed when output is resumed.
2648 2. This message bypasses suspension; it will be printed immediately.
2650 """
2652 def __init__(self, initial_channel: MessageChannel | None = None, /):
2653 super().__init__()
2655 if initial_channel is not None:
2656 self._msg_kwargs.update(initial_channel._msg_kwargs)
2657 self._msg_kwargs["ignore_suspended"] = True
2659 self._resumed = False
2660 _manager().suspend()
2662 def resume(self):
2663 """
2664 Manually resume the logging process.
2666 """
2668 if not self._resumed:
2669 _manager().resume()
2670 self._resumed = True
2672 def __enter__(self):
2673 return self
2675 def __exit__(self, exc_type, exc_val, exc_tb):
2676 self.resume()
2679class _IterTask(_t.Generic[T]):
2680 def __init__(
2681 self, collection: _t.Collection[T], task: Task, unit: str, ndigits: int
2682 ):
2683 self._iter = iter(collection)
2684 self._task = task
2685 self._unit = unit
2686 self._ndigits = ndigits
2688 self._i = 0
2689 self._len = len(collection)
2691 def __next__(self) -> T:
2692 self._task.progress(self._i, self._len, unit=self._unit, ndigits=self._ndigits)
2693 if self._i < self._len:
2694 self._i += 1
2695 return self._iter.__next__()
2697 def __iter__(self) -> _IterTask[T]:
2698 return self
2701class TaskBase:
2702 """
2703 Base class for tasks and other objects that you might show to the user.
2705 Example of a custom task can be found in :ref:`cookbook <cookbook-custom-tasks>`.
2707 .. dropdown:: Protected members
2709 .. autoproperty:: _lock
2711 .. automethod:: _get_widget
2713 .. automethod:: _get_priority
2715 .. automethod:: _request_update
2717 .. automethod:: _widgets_are_displayed
2719 .. automethod:: _get_parent
2721 .. automethod:: _is_toplevel
2723 .. automethod:: _get_children
2725 """
2727 def __init__(self):
2728 self.__parent: TaskBase | None = None
2729 self.__children: list[TaskBase] = []
2731 def attach(self, parent: TaskBase | None):
2732 """
2733 Attach this task and all of its children to the task tree.
2735 :param parent:
2736 parent task in the tree. Pass :data:`None` to attach to root.
2738 """
2740 with self._lock:
2741 if parent is None:
2742 parent = _manager().tasks_root
2743 if self.__parent is not None:
2744 self.__parent.__children.remove(self)
2745 self.__parent = parent
2746 parent.__children.append(self)
2747 self._request_update()
2749 def detach(self):
2750 """
2751 Remove this task and all of its children from the task tree.
2753 """
2755 with self._lock:
2756 if self.__parent is not None:
2757 self.__parent.__children.remove(self)
2758 self.__parent = None
2759 self._request_update()
2761 @property
2762 def _lock(self):
2763 """
2764 Global IO lock.
2766 All protected methods, as well as state mutations, should happen
2767 under this lock.
2769 """
2771 return _IO_LOCK
2773 @abc.abstractmethod
2774 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
2775 """
2776 This method should return widget that renders the task.
2778 .. warning::
2780 This method should be called under :attr:`~TaskBase._lock`.
2782 """
2784 raise NotImplementedError()
2786 @abc.abstractmethod
2787 def _get_priority(self) -> int:
2788 """
2789 This method should return priority that will be used to hide non-important
2790 tasks when there is not enough space to show all of them.
2792 Default priority is ``1``, priority for finished tasks is ``0``.
2794 .. warning::
2796 This method should be called under :attr:`~TaskBase._lock`.
2798 """
2800 raise NotImplementedError()
2802 def _request_update(self, *, immediate_render: bool = False):
2803 """
2804 Indicate that task's state has changed, and update is necessary.
2806 .. warning::
2808 This method should be called under :attr:`~TaskBase._lock`.
2810 :param immediate_render:
2811 by default, tasks are updated lazily from a background thread; set this
2812 parameter to :data:`True` to redraw them immediately from this thread.
2814 """
2816 _manager()._update_tasks(immediate_render or not streams_wrapped())
2818 def _widgets_are_displayed(self) -> bool:
2819 """
2820 Return :data:`True` if we're in an interactive foreground process which
2821 renders tasks.
2823 If this function returns :data:`False`, you should print log messages about
2824 task status instead of relying on task's widget being presented to the user.
2826 .. warning::
2828 This method should be called under :attr:`~TaskBase._lock`.
2830 """
2832 return _manager()._should_draw_interactive_tasks()
2834 def _get_parent(self) -> TaskBase | None:
2835 """
2836 Get parent task.
2838 .. warning::
2840 This method should be called under :attr:`~TaskBase._lock`.
2842 """
2844 return self.__parent
2846 def _is_toplevel(self) -> bool:
2847 """
2848 Check if this task is attached to the first level of the tree.
2850 .. warning::
2852 This method should be called under :attr:`~TaskBase._lock`.
2854 """
2856 return self._get_parent() is _manager().tasks_root
2858 def _get_children(self) -> _t.Sequence[TaskBase]:
2859 """
2860 Get child tasks.
2862 .. warning::
2864 This method should be called under :attr:`~TaskBase._lock`.
2866 """
2868 return self.__children
2871class _TasksRoot(TaskBase):
2872 _widget = yuio.widget.Empty()
2874 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
2875 return self._widget
2877 def _get_priority(self) -> int:
2878 return 0
2881class Task(TaskBase):
2882 """Task(msg: typing.LiteralString, /, *args, comment: str | None = None, parent: Task | None = None)
2883 Task(msg: str, /, *, comment: str | None = None, parent: Task | None = None)
2885 A class for indicating progress of some task.
2887 :param msg:
2888 task heading.
2889 :param args:
2890 arguments for ``%``\\ -formatting the task heading.
2891 :param comment:
2892 comment for the task. Can be specified after creation
2893 via the :meth:`~Task.comment` method.
2894 :param persistent:
2895 whether to keep showing this task after it finishes.
2896 Default is :data:`False`.
2898 To manually hide the task, call :meth:`~TaskBase.detach`.
2899 :param initial_status:
2900 initial status of the task.
2901 :param parent:
2902 parent task.
2904 You can have multiple tasks at the same time,
2905 create subtasks, set task's progress or add a comment about
2906 what's currently being done within a task.
2908 .. vhs:: /_tapes/tasks_multithreaded.tape
2909 :alt: Demonstration of the `Task` class.
2910 :width: 480
2911 :height: 240
2913 This class can be used as a context manager:
2915 .. code-block:: python
2917 with yuio.io.Task("Processing input") as t:
2918 ...
2919 t.progress(0.3)
2920 ...
2922 .. dropdown:: Protected members
2924 .. autoattribute:: _widget_class
2926 """
2928 Status = yuio.widget.Task.Status
2930 _widget_class: _ClassVar[type[yuio.widget.Task]] = yuio.widget.Task
2931 """
2932 Class of the widget that will be used to draw this task, can be overridden
2933 in subclasses.
2935 """
2937 @_t.overload
2938 def __init__(
2939 self,
2940 msg: _t.LiteralString,
2941 /,
2942 *args,
2943 comment: str | None = None,
2944 persistent: bool = False,
2945 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2946 parent: TaskBase | None = None,
2947 ): ...
2948 @_t.overload
2949 def __init__(
2950 self,
2951 msg: str,
2952 /,
2953 *,
2954 comment: str | None = None,
2955 persistent: bool = False,
2956 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2957 parent: TaskBase | None = None,
2958 ): ...
2959 def __init__(
2960 self,
2961 msg: str,
2962 /,
2963 *args,
2964 comment: str | None = None,
2965 persistent: bool = False,
2966 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
2967 parent: TaskBase | None = None,
2968 ):
2969 super().__init__()
2971 self._widget = self._widget_class(msg, *args, comment=comment)
2972 self._persistent = persistent
2973 with self._lock:
2974 self.set_status(initial_status)
2975 self.attach(parent)
2977 @_t.overload
2978 def progress(self, progress: float | None, /, *, ndigits: int = 2): ...
2980 @_t.overload
2981 def progress(
2982 self,
2983 done: float | int,
2984 total: float | int,
2985 /,
2986 *,
2987 unit: str = "",
2988 ndigits: int = 0,
2989 ): ...
2991 def progress(
2992 self,
2993 *args: float | int | None,
2994 unit: str = "",
2995 ndigits: int | None = None,
2996 ):
2997 """progress(progress: float | None, /, *, ndigits: int = 2)
2998 progress(done: float | int, total: float | int, /, *, unit: str = "", ndigits: int = 0) ->
3000 Indicate progress of this task.
3002 If given one argument, it is treated as percentage between ``0`` and ``1``.
3004 If given two arguments, they are treated as amount of finished work,
3005 and a total amount of work. In this case, optional argument `unit`
3006 can be used to indicate units for the progress.
3008 If given a single :data:`None`, reset task progress.
3010 .. note::
3012 Tasks are updated asynchronously once every ~100ms, so calling this method
3013 is relatively cheap. It still requires acquiring a global lock, though:
3014 contention could be an issue in multi-threaded applications.
3016 :param progress:
3017 a percentage between ``0`` and ``1``, or :data:`None`
3018 to reset task progress.
3019 :param done:
3020 amount of finished work, should be less than or equal to `total`.
3021 :param total:
3022 total amount of work.
3023 :param unit:
3024 unit for measuring progress. Only displayed when progress is given
3025 as `done` and `total`.
3026 :param ndigits:
3027 number of digits to display after a decimal point.
3028 :example:
3029 .. code-block:: python
3031 with yuio.io.Task("Loading cargo") as task:
3032 task.progress(110, 150, unit="Kg")
3034 This will print the following:
3036 .. code-block:: text
3038 ■■■■■■■■■■■□□□□ Loading cargo - 110/150Kg
3040 """
3042 with self._lock:
3043 self._widget.progress(*args, unit=unit, ndigits=ndigits) # type: ignore
3044 self._request_update()
3046 def progress_size(
3047 self,
3048 done: float | int,
3049 total: float | int,
3050 /,
3051 *,
3052 ndigits: int = 2,
3053 ):
3054 """
3055 Indicate progress of this task using human-readable 1024-based size units.
3057 :param done:
3058 amount of processed data.
3059 :param total:
3060 total amount of data.
3061 :param ndigits:
3062 number of digits to display after a decimal point.
3063 :example:
3064 .. code-block:: python
3066 with yuio.io.Task("Downloading a file") as task:
3067 task.progress_size(31.05 * 2**20, 150 * 2**20)
3069 This will print:
3071 .. code-block:: text
3073 ■■■□□□□□□□□□□□□ Downloading a file - 31.05/150.00M
3075 """
3077 with self._lock:
3078 self._widget.progress_size(done, total, ndigits=ndigits)
3079 self._request_update()
3081 def progress_scale(
3082 self,
3083 done: float | int,
3084 total: float | int,
3085 /,
3086 *,
3087 unit: str = "",
3088 ndigits: int = 2,
3089 ):
3090 """
3091 Indicate progress of this task while scaling numbers in accordance
3092 with SI system.
3094 :param done:
3095 amount of finished work, should be less than or equal to `total`.
3096 :param total:
3097 total amount of work.
3098 :param unit:
3099 unit for measuring progress.
3100 :param ndigits:
3101 number of digits to display after a decimal point.
3102 :example:
3103 .. code-block:: python
3105 with yuio.io.Task("Charging a capacitor") as task:
3106 task.progress_scale(889.25e-3, 1, unit="V")
3108 This will print:
3110 .. code-block:: text
3112 ■■■■■■■■■■■■■□□ Charging a capacitor - 889.25mV/1.00V
3114 """
3116 with self._lock:
3117 self._widget.progress_scale(done, total, unit=unit, ndigits=ndigits)
3118 self._request_update()
3120 def iter(
3121 self,
3122 collection: _t.Collection[T],
3123 /,
3124 *,
3125 unit: str = "",
3126 ndigits: int = 0,
3127 ) -> _t.Iterable[T]:
3128 """
3129 Helper for updating progress automatically
3130 while iterating over a collection.
3132 :param collection:
3133 an iterable collection. Should support returning its length.
3134 :param unit:
3135 unit for measuring progress.
3136 :param ndigits:
3137 number of digits to display after a decimal point.
3138 :example:
3139 .. invisible-code-block: python
3141 urls = []
3143 .. code-block:: python
3145 with yuio.io.Task("Fetching data") as t:
3146 for url in t.iter(urls):
3147 ...
3149 This will output the following:
3151 .. code-block:: text
3153 ■■■■■□□□□□□□□□□ Fetching data - 1/3
3155 """
3157 return _IterTask(collection, self, unit, ndigits)
3159 def comment(self, comment: str | None, /, *args):
3160 """
3161 Set a comment for a task.
3163 Comment is displayed after the progress.
3165 :param comment:
3166 comment to display beside task progress.
3167 :param args:
3168 arguments for ``%``\\ -formatting comment.
3169 :example:
3170 .. invisible-code-block: python
3172 urls = []
3174 .. code-block:: python
3176 with yuio.io.Task("Fetching data") as t:
3177 for url in urls:
3178 t.comment("%s", url)
3179 ...
3181 This will output the following:
3183 .. code-block:: text
3185 ⣿ Fetching data - https://google.com
3187 """
3189 with self._lock:
3190 self._widget.comment(comment, *args)
3191 self._request_update()
3193 def set_status(self, status: Task.Status):
3194 """
3195 Set task status.
3197 :param status:
3198 New status.
3200 """
3202 with self._lock:
3203 if self._widget.status == status:
3204 return
3206 self._widget.status = status
3207 if status in [Task.Status.DONE, Task.Status.ERROR] and not self._persistent:
3208 self.detach()
3209 if self._widgets_are_displayed():
3210 self._request_update()
3211 else:
3212 raw(self._widget, add_newline=True)
3214 def running(self):
3215 """
3216 Indicate that this task is running.
3218 """
3220 self.set_status(Task.Status.RUNNING)
3222 def pending(self):
3223 """
3224 Indicate that this task is pending.
3226 """
3228 self.set_status(Task.Status.PENDING)
3230 def done(self):
3231 """
3232 Indicate that this task has finished successfully.
3234 """
3236 self.set_status(Task.Status.DONE)
3238 def error(self):
3239 """
3240 Indicate that this task has finished with an error.
3242 """
3244 self.set_status(Task.Status.ERROR)
3246 @_t.overload
3247 def subtask(
3248 self,
3249 msg: _t.LiteralString,
3250 /,
3251 *args,
3252 comment: str | None = None,
3253 persistent: bool = True,
3254 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3255 ) -> Task: ...
3256 @_t.overload
3257 def subtask(
3258 self,
3259 msg: str,
3260 /,
3261 *,
3262 comment: str | None = None,
3263 persistent: bool = True,
3264 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3265 ) -> Task: ...
3266 def subtask(
3267 self,
3268 msg: str,
3269 /,
3270 *args,
3271 comment: str | None = None,
3272 persistent: bool = True,
3273 initial_status: Task.Status = yuio.widget.Task.Status.RUNNING,
3274 ) -> Task:
3275 """
3276 Create a subtask within this task.
3278 :param msg:
3279 subtask heading.
3280 :param args:
3281 arguments for ``%``\\ -formatting the subtask heading.
3282 :param comment:
3283 comment for the task. Can be specified after creation
3284 via the :meth:`~Task.comment` method.
3285 :param persistent:
3286 whether to keep showing this subtask after it finishes. Default
3287 is :data:`True`.
3288 :param initial_status:
3289 initial status of the task.
3290 :returns:
3291 a new :class:`Task` that will be displayed as a sub-task of this task.
3293 """
3295 return Task(
3296 msg,
3297 *args,
3298 comment=comment,
3299 persistent=persistent,
3300 initial_status=initial_status,
3301 parent=self,
3302 )
3304 def __enter__(self):
3305 self.running()
3306 return self
3308 def __exit__(self, exc_type, exc_val, exc_tb):
3309 if exc_type is None:
3310 self.done()
3311 else:
3312 self.error()
3314 def _get_widget(self) -> yuio.widget.Widget[_t.Never]:
3315 return self._widget
3317 def _get_priority(self) -> int:
3318 return 1 if self._widget.status is yuio.widget.Task.Status.RUNNING else 0
3321class _TaskTree(yuio.widget.Widget[_t.Never]):
3322 def __init__(self, root: TaskBase):
3323 super().__init__()
3325 self._root = root
3327 def layout(self, rc: yuio.widget.RenderContext) -> tuple[int, int]:
3328 widgets: list[yuio.widget.Widget[_t.Never]] = [] # widget
3329 tree: dict[
3330 int, tuple[int | None, int, int]
3331 ] = {} # index -> parent, level, priority
3333 # Build widgets tree.
3334 to_visit: list[tuple[TaskBase, int, int | None]] = [(self._root, 0, None)]
3335 while to_visit:
3336 node, level, parent = to_visit.pop()
3337 widget = node._get_widget()
3338 tree[len(widgets)] = parent, level, node._get_priority()
3339 to_visit.extend(
3340 (child, level + 1, len(widgets))
3341 for child in reversed(node._get_children())
3342 )
3343 widgets.append(widget)
3345 # Prepare layouts.
3346 layouts: dict[yuio.widget.Widget[_t.Never], tuple[int, int, int]] = {}
3347 self.__layouts = layouts
3348 total_min_h = 0
3349 total_max_h = 0
3350 for index, widget in enumerate(widgets):
3351 min_h, max_h = widget.layout(rc)
3352 assert min_h <= max_h, "incorrect layout"
3353 _, level, _ = tree[index]
3354 layouts[widget] = min_h, max_h, level
3355 total_min_h += min_h
3356 total_max_h += max_h
3358 if total_min_h <= rc.height:
3359 # All widgets fit.
3360 self.__min_h = total_min_h
3361 self.__max_h = total_max_h
3362 self.__widgets = widgets
3363 return total_min_h, total_max_h
3365 # Propagate priority upwards, ensure that parents are at least as important
3366 # as children.
3367 for index, widget in enumerate(widgets):
3368 parent, _, priority = tree[index]
3369 while parent is not None:
3370 grandparent, parent_level, parent_priority = tree[parent]
3371 if parent_priority >= priority:
3372 break
3373 tree[parent] = grandparent, parent_level, priority
3374 widget = parent
3375 parent = grandparent
3377 # Sort by (-priority, level, -index). Since we've propagated priorities, we can
3378 # be sure that parents are always included first. Hence in the loop below,
3379 # we will visit children before parents.
3380 widgets_sorted = list(enumerate(widgets))
3381 widgets_sorted.sort(key=lambda w: (-tree[w[0]][2], tree[w[0]][1], -w[0]))
3383 # Decide which widgets to hide by introducing "holes" to widgets sequence.
3384 total_h = total_min_h
3385 holes = _DisjointSet[int]()
3386 for index, widget in reversed(widgets_sorted):
3387 if total_h <= rc.height:
3388 break
3390 min_h, max_h = widget.layout(rc)
3392 # We need to hide this widget.
3393 _, level, _ = tree[index]
3394 holes.add(index)
3395 total_h -= min_h
3396 total_h += 1 # Size of a message.
3398 # Join this hole with the next one.
3399 if index + 1 < len(widgets) and index + 1 in holes:
3400 _, next_level, _ = tree[index + 1]
3401 if next_level >= level:
3402 holes.union(index, index + 1)
3403 total_h -= 1
3404 # Join this hole with the previous one.
3405 if index - 1 >= 0 and index - 1 in holes:
3406 _, prev_level, _ = tree[index - 1]
3407 if prev_level <= level:
3408 holes.union(index, index - 1)
3409 total_h -= 1
3411 # Assemble the final sequence of widgets.
3412 hole_color = rc.theme.get_color("task/hole")
3413 hole_num_color = rc.theme.get_color("task/hole/num")
3414 prev_hole_id: int | None = None
3415 prev_hole_size = 0
3416 prev_hole_level: int | None = None
3417 displayed_widgets: list[yuio.widget.Widget[_t.Never]] = []
3418 for index, widget in enumerate(widgets):
3419 if index in holes:
3420 hole_id = holes.find(index)
3421 if hole_id == prev_hole_id:
3422 prev_hole_size += 1
3423 if prev_hole_level is None:
3424 prev_hole_level = tree[index][1]
3425 else:
3426 prev_hole_level = min(prev_hole_level, tree[index][1])
3427 else:
3428 if prev_hole_id is not None:
3429 hole_widget = yuio.widget.Line(
3430 yuio.string.ColorizedString(
3431 hole_num_color,
3432 "+",
3433 str(prev_hole_size),
3434 hole_color,
3435 " more",
3436 )
3437 )
3438 displayed_widgets.append(hole_widget)
3439 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3440 prev_hole_id = hole_id
3441 prev_hole_size = 1
3442 prev_hole_level = tree[index][1]
3443 else:
3444 if prev_hole_id is not None:
3445 hole_widget = yuio.widget.Line(
3446 yuio.string.ColorizedString(
3447 hole_num_color,
3448 "+",
3449 str(prev_hole_size),
3450 hole_color,
3451 " more",
3452 )
3453 )
3454 displayed_widgets.append(hole_widget)
3455 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3456 prev_hole_id = None
3457 prev_hole_size = 0
3458 prev_hole_level = None
3459 displayed_widgets.append(widget)
3461 if prev_hole_id is not None:
3462 hole_widget = yuio.widget.Line(
3463 yuio.string.ColorizedString(
3464 hole_num_color,
3465 "+",
3466 str(prev_hole_size),
3467 hole_color,
3468 " more",
3469 )
3470 )
3471 displayed_widgets.append(hole_widget)
3472 layouts[hole_widget] = 1, 1, prev_hole_level or 1
3474 total_min_h = 0
3475 total_max_h = 0
3476 for widget in displayed_widgets:
3477 min_h, max_h, _ = layouts[widget]
3478 total_min_h += min_h
3479 total_max_h += max_h
3481 self.__min_h = total_min_h
3482 self.__max_h = total_max_h
3483 self.__widgets = displayed_widgets
3484 return total_min_h, total_max_h
3486 def draw(self, rc: yuio.widget.RenderContext):
3487 if rc.height <= self.__min_h:
3488 scale = 0.0
3489 elif rc.height >= self.__max_h:
3490 scale = 1.0
3491 else:
3492 scale = (rc.height - self.__min_h) / (self.__max_h - self.__min_h)
3494 y1 = 0.0
3495 for widget in self.__widgets:
3496 min_h, max_h, level = self.__layouts[widget]
3497 y2 = y1 + min_h + scale * (max_h - min_h)
3499 iy1 = round(y1)
3500 iy2 = round(y2)
3502 with rc.frame(max((level - 1) * 2, 0), iy1, height=iy2 - iy1):
3503 widget.draw(rc)
3505 y1 = y2
3507 rc.set_final_pos(0, round(y1))
3510class Formatter(logging.Formatter):
3511 """
3512 Log formatter that uses ``%`` style with colorized string formatting
3513 and returns a string with ANSI escape characters generated for current
3514 output terminal.
3516 Every part of log message is colored with path :samp:`log/{name}:{level}`.
3517 For example, `asctime` in info log line is colored
3518 with path ``log/asctime:info``.
3520 In addition to the usual `log record attributes`__, this formatter also
3521 adds ``%(colMessage)s``, which is similar to ``%(message)s``, but colorized.
3523 __ https://docs.python.org/3/library/logging.html#logrecord-attributes
3525 """
3527 default_format = "%(asctime)s %(name)s %(levelname)s %(colMessage)s"
3528 default_msec_format = "%s.%03d"
3530 def __init__(
3531 self,
3532 fmt: str | None = None,
3533 datefmt: str | None = None,
3534 validate: bool = True,
3535 *,
3536 defaults: _t.Mapping[str, _t.Any] | None = None,
3537 ):
3538 fmt = fmt or self.default_format
3539 super().__init__(
3540 fmt,
3541 datefmt,
3542 style="%",
3543 validate=validate,
3544 defaults=defaults,
3545 )
3547 def formatMessage(self, record):
3548 level = record.levelname.lower()
3550 ctx = make_repr_context()
3552 if not hasattr(record, "colMessage"):
3553 msg = str(record.msg)
3554 if record.args:
3555 msg = ColorizedString(msg).percent_format(record.args, ctx)
3556 setattr(record, "colMessage", msg)
3558 if defaults := self._style._defaults: # type: ignore
3559 data = defaults | record.__dict__
3560 else:
3561 data = record.__dict__
3563 data = {
3564 k: yuio.string.WithBaseColor(v, base_color=f"log/{k}:{level}")
3565 for k, v in data.items()
3566 }
3568 return "".join(
3569 yuio.string.colorize(
3570 self._fmt or self.default_format, default_color=f"log:{level}", ctx=ctx
3571 )
3572 .percent_format(data, ctx)
3573 .as_code(ctx.term.color_support)
3574 )
3576 def formatException(self, ei):
3577 tb = "".join(traceback.format_exception(*ei)).rstrip()
3578 return self.formatStack(tb)
3580 def formatStack(self, stack_info):
3581 manager = _manager()
3582 theme = manager.theme
3583 term = manager.term
3584 highlighter, syntax_name = yuio.hl.get_highlighter("python-traceback")
3585 return "".join(
3586 highlighter.highlight(stack_info, theme=theme, syntax=syntax_name)
3587 .indent()
3588 .as_code(term.color_support)
3589 )
3592class Handler(logging.Handler):
3593 """
3594 A handler that redirects all log messages to Yuio.
3596 """
3598 def __init__(self, level: int | str = 0):
3599 super().__init__(level)
3600 self.setFormatter(Formatter())
3602 def emit(self, record: LogRecord):
3603 manager = _manager()
3604 manager.print_direct(self.format(record).rstrip() + "\n", manager.term.ostream)
3607class _IoManager(abc.ABC):
3608 # If we see that it took more than this time to render progress bars,
3609 # we assume that the process was suspended, meaning that we might've been moved
3610 # from foreground to background or back. In either way, we should assume that the
3611 # screen was changed, and re-render all tasks accordingly. We have to track time
3612 # because Python might take significant time to call `SIGCONT` handler, so we can't
3613 # rely on it.
3614 TASK_RENDER_TIMEOUT_NS = 250_000_000
3616 def __init__(
3617 self,
3618 term: yuio.term.Term | None = None,
3619 theme: (
3620 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
3621 ) = None,
3622 enable_bg_updates: bool = True,
3623 ):
3624 self._out_term = yuio.term.get_term_from_stream(
3625 orig_stdout(), sys.stdin, allow_env_overrides=True
3626 )
3627 self._err_term = yuio.term.get_term_from_stream(
3628 orig_stderr(), sys.stdin, allow_env_overrides=True
3629 )
3631 self._term = term or self._err_term
3633 self._theme_ctor = theme
3634 if isinstance(theme, yuio.theme.Theme):
3635 self._theme = theme
3636 else:
3637 self._theme = yuio.theme.load(self._term, theme)
3638 self._rc = yuio.widget.RenderContext(self._term, self._theme)
3639 self._rc.prepare()
3641 self._suspended: int = 0
3642 self._suspended_lines: list[tuple[list[str], _t.TextIO]] = []
3644 self._tasks_root = _TasksRoot()
3645 self._tasks_widet = _TaskTree(self._tasks_root)
3646 self._printed_tasks: bool = False
3647 self._needs_update = False
3648 self._last_update_time_us = 0
3649 self._printed_some_lines = False
3651 self._stop = False
3652 self._stop_condition = threading.Condition(_IO_LOCK)
3653 self._thread: threading.Thread | None = None
3655 self._enable_bg_updates = enable_bg_updates
3656 self._prev_sigcont_handler: (
3657 None | yuio.Missing | int | _t.Callable[[int, types.FrameType | None], None]
3658 ) = yuio.MISSING
3659 self._seen_sigcont: bool = False
3660 if enable_bg_updates:
3661 self._setup_sigcont()
3662 self._thread = threading.Thread(
3663 target=self._bg_update, name="yuio_io_task_refresh", daemon=True
3664 )
3665 self._thread.start()
3667 atexit.register(self.stop)
3669 @property
3670 def term(self):
3671 return self._term
3673 @property
3674 def out_term(self):
3675 return self._out_term
3677 @property
3678 def err_term(self):
3679 return self._err_term
3681 @property
3682 def theme(self):
3683 return self._theme
3685 @property
3686 def rc(self):
3687 return self._rc
3689 @property
3690 def tasks_root(self):
3691 return self._tasks_root
3693 def setup(
3694 self,
3695 term: yuio.term.Term | None = None,
3696 theme: (
3697 yuio.theme.Theme | _t.Callable[[yuio.term.Term], yuio.theme.Theme] | None
3698 ) = None,
3699 ):
3700 with _IO_LOCK:
3701 self._clear_tasks()
3703 if term is not None:
3704 self._term = term
3705 if theme is None:
3706 # Refresh theme to reflect changed terminal capabilities.
3707 theme = self._theme_ctor
3708 if theme is not None:
3709 self._theme_ctor = theme
3710 if isinstance(theme, yuio.theme.Theme):
3711 self._theme = theme
3712 else:
3713 self._theme = yuio.theme.load(self._term, theme)
3715 self._rc = yuio.widget.RenderContext(self._term, self._theme)
3716 self._rc.prepare()
3717 self.__dict__.pop("_update_rate_us", None) # type: ignore
3718 self._update_tasks()
3720 def _setup_sigcont(self):
3721 import signal
3723 if not hasattr(signal, "SIGCONT"):
3724 return
3726 self._prev_sigcont_handler = signal.getsignal(signal.SIGCONT)
3727 signal.signal(signal.SIGCONT, self._on_sigcont)
3729 def _reset_sigcont(self):
3730 import signal
3732 if not hasattr(signal, "SIGCONT"):
3733 return
3735 if self._prev_sigcont_handler is not yuio.MISSING:
3736 signal.signal(signal.SIGCONT, self._prev_sigcont_handler)
3738 def _on_sigcont(self, sig: int, frame: types.FrameType | None):
3739 self._seen_sigcont = True
3740 if self._prev_sigcont_handler and not isinstance(
3741 self._prev_sigcont_handler, int
3742 ):
3743 self._prev_sigcont_handler(sig, frame)
3745 def _bg_update(self):
3746 while True:
3747 try:
3748 with _IO_LOCK:
3749 while True:
3750 update_rate_us = self._update_rate_us
3751 start_ns = time.monotonic_ns()
3752 now_us = start_ns // 1_000
3753 sleep_us = update_rate_us - now_us % update_rate_us
3754 deadline_ns = (
3755 start_ns + 2 * sleep_us * 1000 + self.TASK_RENDER_TIMEOUT_NS
3756 )
3758 if self._stop_condition.wait_for(
3759 lambda: self._stop, timeout=sleep_us / 1_000_000
3760 ):
3761 return
3763 self._show_tasks(deadline_ns=deadline_ns)
3764 except Exception:
3765 yuio._logger.critical("exception in bg updater", exc_info=True)
3767 def stop(self):
3768 if self._stop:
3769 return
3771 with _IO_LOCK:
3772 atexit.unregister(self.stop)
3774 self._stop = True
3775 self._stop_condition.notify()
3776 self._show_tasks(immediate_render=True)
3778 if self._thread:
3779 self._thread.join()
3781 if self._prev_sigcont_handler is not yuio.MISSING:
3782 self._reset_sigcont()
3784 def print(
3785 self,
3786 msg: list[str],
3787 term: yuio.term.Term,
3788 *,
3789 ignore_suspended: bool = False,
3790 heading: bool = False,
3791 ):
3792 with _IO_LOCK:
3793 if heading and self.theme.separate_headings:
3794 if self._printed_some_lines:
3795 msg.insert(0, "\n")
3796 msg.append("\n")
3797 self._emit_lines(msg, term.ostream, ignore_suspended)
3798 if heading:
3799 self._printed_some_lines = False
3801 def print_direct(
3802 self,
3803 msg: str,
3804 stream: _t.TextIO | None = None,
3805 ):
3806 if not msg:
3807 return
3808 with _IO_LOCK:
3809 self._emit_lines([msg], stream, ignore_suspended=False)
3811 def print_direct_lines(
3812 self,
3813 lines: list[str],
3814 stream: _t.TextIO | None = None,
3815 ):
3816 with _IO_LOCK:
3817 self._emit_lines(lines, stream, ignore_suspended=False)
3819 def suspend(self):
3820 with _IO_LOCK:
3821 self._suspend()
3823 def resume(self):
3824 with _IO_LOCK:
3825 self._resume()
3827 # Implementation.
3828 # These functions are always called under a lock.
3830 @functools.cached_property
3831 def _update_rate_us(self) -> int:
3832 update_rate_ms = max(self._theme.spinner_update_rate_ms, 1)
3833 while update_rate_ms < 50:
3834 update_rate_ms *= 2
3835 while update_rate_ms > 250:
3836 update_rate_ms //= 2
3837 return int(update_rate_ms * 1000)
3839 @property
3840 def _spinner_update_rate_us(self) -> int:
3841 return self._theme.spinner_update_rate_ms * 1000
3843 def _emit_lines(
3844 self,
3845 lines: list[str],
3846 stream: _t.TextIO | None = None,
3847 ignore_suspended: bool = False,
3848 ):
3849 if not lines or not any(lines):
3850 return
3851 stream = stream or self._term.ostream
3852 if self._suspended and not ignore_suspended:
3853 self._suspended_lines.append((list(lines), stream))
3854 else:
3855 self._clear_tasks()
3856 stream.writelines(lines)
3857 if lines and lines[-1].endswith("\n"):
3858 self._update_tasks(immediate_render=True)
3859 stream.flush()
3861 self._printed_some_lines = True
3863 def _suspend(self):
3864 self._suspended += 1
3866 if self._suspended == 1:
3867 self._clear_tasks()
3869 def _resume(self):
3870 self._suspended -= 1
3872 if self._suspended == 0:
3873 for lines, stream in self._suspended_lines:
3874 stream.writelines(lines)
3875 if self._suspended_lines:
3876 self._printed_some_lines = True
3877 self._suspended_lines.clear()
3879 self._update_tasks()
3881 if self._suspended < 0:
3882 yuio._logger.warning("unequal number of suspends and resumes")
3883 self._suspended = 0
3885 def _should_draw_interactive_tasks(self):
3886 should_draw_interactive_tasks = (
3887 self._term.color_support >= yuio.term.ColorSupport.ANSI
3888 and self._term.ostream_is_tty
3889 and yuio.term._is_foreground(self._term.ostream)
3890 )
3892 if (
3893 not should_draw_interactive_tasks and self._printed_tasks
3894 ) or self._seen_sigcont:
3895 # We were moved from foreground to background. There's no point in hiding
3896 # tasks now (shell printed something when user sent C-z), but we need
3897 # to make sure that we'll start rendering tasks from scratch whenever
3898 # user brings us to foreground again.
3899 self.rc.prepare(reset_term_pos=True)
3900 self._printed_tasks = False
3901 self._seen_sigcont = False
3903 return should_draw_interactive_tasks
3905 def _clear_tasks(self):
3906 if self._should_draw_interactive_tasks() and self._printed_tasks:
3907 self._rc.finalize()
3908 self._printed_tasks = False
3910 def _update_tasks(self, immediate_render: bool = False):
3911 self._needs_update = True
3912 if immediate_render or not self._enable_bg_updates:
3913 self._show_tasks(immediate_render)
3915 def _show_tasks(
3916 self, immediate_render: bool = False, deadline_ns: int | None = None
3917 ):
3918 if (
3919 self._should_draw_interactive_tasks()
3920 and not self._suspended
3921 and (self._tasks_root._get_children() or self._printed_tasks)
3922 ):
3923 start_ns = time.monotonic_ns()
3924 if deadline_ns is None:
3925 deadline_ns = start_ns + self.TASK_RENDER_TIMEOUT_NS
3926 now_us = start_ns // 1000
3927 now_us -= now_us % self._update_rate_us
3929 if not immediate_render and self._enable_bg_updates:
3930 next_update_us = self._last_update_time_us + self._update_rate_us
3931 if now_us < next_update_us:
3932 # Hard-limit update rate by `update_rate_ms`.
3933 return
3934 next_spinner_update_us = (
3935 self._last_update_time_us + self._spinner_update_rate_us
3936 )
3937 if not self._needs_update and now_us < next_spinner_update_us:
3938 # Tasks didn't change, and spinner state didn't change either,
3939 # so we can skip this update.
3940 return
3942 self._last_update_time_us = now_us
3943 self._printed_tasks = bool(self._tasks_root._get_children())
3944 self._needs_update = False
3946 self._rc.prepare()
3947 self._tasks_widet.layout(self._rc)
3948 self._tasks_widet.draw(self._rc)
3950 now_ns = time.monotonic_ns()
3951 if not self._seen_sigcont and now_ns < deadline_ns:
3952 self._rc.render()
3953 else:
3954 # We have to skip this render: the process was suspended while we were
3955 # formatting tasks. Because of this, te position of the cursor
3956 # could've changed, so we need to reset rendering context and re-render.
3957 self._seen_sigcont = True
3960class _YuioOutputWrapper(_t.TextIO): # pragma: no cover
3961 def __init__(self, wrapped: _t.TextIO):
3962 self.__wrapped = wrapped
3964 @property
3965 def mode(self) -> str:
3966 return self.__wrapped.mode
3968 @property
3969 def name(self) -> str:
3970 return self.__wrapped.name
3972 def close(self):
3973 self.__wrapped.close()
3975 @property
3976 def closed(self) -> bool:
3977 return self.__wrapped.closed
3979 def fileno(self) -> int:
3980 return self.__wrapped.fileno()
3982 def flush(self):
3983 self.__wrapped.flush()
3985 def isatty(self) -> bool:
3986 return self.__wrapped.isatty()
3988 def writable(self) -> bool:
3989 return self.__wrapped.writable()
3991 def write(self, s: str, /) -> int:
3992 _manager().print_direct(s, self.__wrapped)
3993 return len(s)
3995 def writelines(self, lines: _t.Iterable[str], /):
3996 _manager().print_direct_lines(list(lines), self.__wrapped)
3998 def readable(self) -> bool:
3999 return self.__wrapped.readable()
4001 def read(self, n: int = -1) -> str:
4002 return self.__wrapped.read(n)
4004 def readline(self, limit: int = -1) -> str:
4005 return self.__wrapped.readline(limit)
4007 def readlines(self, hint: int = -1) -> list[str]:
4008 return self.__wrapped.readlines(hint)
4010 def seek(self, offset: int, whence: int = 0) -> int:
4011 return self.__wrapped.seek(offset, whence)
4013 def seekable(self) -> bool:
4014 return self.__wrapped.seekable()
4016 def tell(self) -> int:
4017 return self.__wrapped.tell()
4019 def truncate(self, size: int | None = None) -> int:
4020 return self.__wrapped.truncate(size)
4022 def __enter__(self) -> _t.TextIO:
4023 return self.__wrapped.__enter__()
4025 def __exit__(self, exc_type, exc_val, exc_tb):
4026 self.__wrapped.__exit__(exc_type, exc_val, exc_tb)
4028 @property
4029 def buffer(self) -> _t.BinaryIO:
4030 return self.__wrapped.buffer
4032 @property
4033 def encoding(self) -> str:
4034 return self.__wrapped.encoding
4036 @property
4037 def errors(self) -> str | None:
4038 return self.__wrapped.errors
4040 @property
4041 def line_buffering(self) -> int:
4042 return self.__wrapped.line_buffering
4044 @property
4045 def newlines(self) -> _t.Any:
4046 return self.__wrapped.newlines
4048 def __repr__(self) -> str:
4049 return f"{self.__class__.__name__}({self.__wrapped!r})"