将打印和/或记录重定向到 Panel

Question

我制作了一个小应用程序，它使用 Rich 在多个面板中显示实时视图。
有没有办法将标准（或丰富）print 语句放入特定面板？
能够在其自己的专用面板中显示日志输出也可以。

我觉得这将是一个非常常见的用例，但我没有找到任何文档。我认为答案可能是使用Console.capture() 方法，但我无法弄清楚。

Answer 1

我知道你怎么能做到这一点。首先，我们需要拦截Rich logger的stdout进程。我们从一个类开始：

from collections import deque
class Logger():
    _instance = None
    def __init__(self):
        self.messages = deque(["sa"])
        self.size = 10

    def __new__(class_, *args, **kwargs):
        if not isinstance(class_._instance, class_):
            class_._instance = object.__new__(class_, *args, **kwargs)
        return class_._instance

    def write(self, message):
        self.messages.extend(message.splitlines())
        while len(self.messages) > self.size:
            self.messages.popleft()

    def flush(self):
        pass

这是一个单例类。我们需要将这个类传递给控制台 API

from rich.console import Console
c = Console(file=Logger(), width=150)

有一些宽度。然后，我们创建一个日志处理程序

from rich.logging import RichHandler
r = RichHandler(console=c)

这将是我们的日志处理程序

import logging
logging.basicConfig(
    level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[r]
)
logger = logging.getLogger("rich")

稍后，我们需要在您管理布局的地方使用我们的 Logger 类。对我来说，它在 Dashboard 类中。

class Dashboard:
    def __init__(self):
        self.log_std = Logger()
    def update(self, new_parameters):
        self.layout["logs"].update(Panel(Text(
            "
".join(self.log_std.messages), justify="right"), padding=(1, 2),
        title="[b red]Logs",
        ))

每次我调用update 方法时，它都会更新我的布局。 我的布局更复杂，self.layout["logs"] 我在其中显示日志。

Answer 2

因此，我以@Mete Yildirim 的回答为灵感，提出了一个使用现有logging 处理程序而不是创建新记录器的细微变化。

logging.handlers 模块有一个 BufferingHandler() 用于对我的解决方案进行原型设计。通过将处理程序传递给rich，我可以窥探其内容并将它们打印到我的rich 面板中，而无需对logging 进行修改。

LOG_BUFFER_MAX_MSGS = 20

# Set up the main/root logger
main_logger = logging.getLogger()
main_logger.setLevel(logging.DEBUG)

# Instantiate the buffering handler we will watch from within Rich
buffering_handler = BufferingHandler(capacity=LOG_BUFFER_MAX_MSGS)
main_logger.addHandler(buffering_handler)

# Local logger
log = logging.getLogger("rich")


# Create a basic Rich layout
layout = Layout(name="root")

def get_log():
    """
    We call this method from within Rich to snoop the messages
    within the BufferingHandler and put them in a form Rich 
    can display.
    Check the BufferingHandler() source to see how we can access
    its data.
    """
    log_messages = []
    for li in buffering_handler.buffer:
        log_messages.append(li.msg)
    return Panel("
".join(log_messages))

# Run Rich, displaying each log message to the screen
with Live(layout, refresh_per_second=4) as live:
    while True:
        layout["root"].update(get_log())
        time.sleep(0.25)

为了测试上述内容，我们可以在后台生成一些日志消息：

def create_log_messages():
    msgs = (
        "Test message 1",
        "Test message 2",
        "Test message 3",
    )
    for li in msgs:
        log.info(li)
        time.sleep(2)

threading.Thread(target=create_log_messages).start()

缺点： 当行数超过传递给其capacity 参数的值时，BufferingHandler() 将清除它的缓冲区。 我更希望它删除旧消息，但这需要重载现有的 BufferHandler() 实现或编写新的处理程序。 BufferHandler() 代码很短，所以简单地编写一个新代码应该不会太费力。

Answer 3

我有一个类似的问题，我可以使用 print 来显示和对象具有丰富的样式，但是一旦我将对象放在面板中，它就不再是样式了。

只需打印：

from rich import print
print(self.model)

为了解决这个问题，我使用了漂亮的打印：

from rich import print
from rich.panel import Panel
from rich.pretty import Pretty
prettyModel = Pretty(self.model)
print(Panel(prettyModel))

我希望它可以帮助。

Answer 4

我的解决方案违反了一些规则，但我让日志记录异步工作。我将RichHandler 子类化以更新布局而不是打印到标准输出。我无法弄清楚如何调整垂直裁剪或消息数量以适应可用的屏幕空间，因此我将消息数量限制为五个。希望更聪明的人能把它捡起来，让它变得更好。

class RichHandlerPanel(RichHandler):
    """Send logs to a layout."""

    def __init__(
        self,
        layout: Layout,
        level: Union[int, str] = logging.NOTSET,
        console: Optional[Console] = None,
        panel_title: str = "Log",
        max_display: int = 5,
        **kwargs
    ) -> None:
        super().__init__(level=level, console=console, **kwargs)
        self.layout = layout
        self.full_log = None
        self.panel_title = panel_title
        self.max_display = max_display

    def emit(self, record: LogRecord) -> None:
        """Invoke by logging. This is a copy of the original with a change on how emit is done."""
        message = self.format(record)
        traceback = None
        if (
            self.rich_tracebacks
            and record.exc_info
            and record.exc_info != (None, None, None)
        ):
            exc_type, exc_value, exc_traceback = record.exc_info
            assert exc_type is not None
            assert exc_value is not None
            traceback = Traceback.from_exception(
                exc_type,
                exc_value,
                exc_traceback,
                width=self.tracebacks_width,
                extra_lines=self.tracebacks_extra_lines,
                theme=self.tracebacks_theme,
                word_wrap=self.tracebacks_word_wrap,
                show_locals=self.tracebacks_show_locals,
                locals_max_length=self.locals_max_length,
                locals_max_string=self.locals_max_string,
                suppress=self.tracebacks_suppress,
            )
            message = record.getMessage()
            if self.formatter:
                record.message = record.getMessage()
                formatter = self.formatter
                if hasattr(formatter, "usesTime") and formatter.usesTime():
                    record.asctime = formatter.formatTime(record, formatter.datefmt)
                message = formatter.formatMessage(record)

        message_renderable = self.render_message(record, message)
        log_renderable = self.render(
            record=record, traceback=traceback, message_renderable=message_renderable
        )
        if not self.full_log:
            self.full_log = log_renderable
        else:
            for r in range(log_renderable.row_count):
                self.full_log.add_row(
                    *[
                        log_renderable.columns[c]._cells[r]
                        for c in range(len(log_renderable.columns))
                    ]
                )

        while len(self.full_log.rows) > self.max_display:
            for c in range(len(log_renderable.columns)):
                self.full_log.columns[c]._cells.pop(0)
            self.full_log.rows.pop(0)

        try:
            p = Panel(self.full_log, title=self.panel_title, title_align="left")
            self.layout.update(p)
        except Exception:
            self.handleError(record)

在实例化处理程序时，我提供了布局单元：

    def _init_logging(self) -> None:
        # Handler
        self._log_handler = RichHandlerPanel(
            level=logging.INFO, layout=self.layout["log"]
        )
        # Logger
        self._logger = logging.getLogger()
        self._logger.addHandler(self._log_handler)

Answer 5

我发现的最简单的解决方案是利用console protocal 使控制台本身可渲染。这是一个例子：

from rich.console import Console
from rich.layout import Layout
from rich.live import Live
import os
import time
from datetime import datetime

class ConsolePanel(Console):
    def __init__(self,*args,**kwargs):
        console_file = open(os.devnull,'w')
        super().__init__(record=True,file=console_file,*args,**kwargs)

    def __rich_console__(self,console,options):
        texts = self.export_text(clear=False).split('
')
        for line in texts[-options.height:]:
            yield line

class Interface():
    def __init__(self) -> None:
        self.console:list[ConsolePanel] = [ConsolePanel() for _ in range(2)]
    
    def get_renderable(self):
        layout = Layout()
        layout.split_column(
            Layout(self.console[0],name='top'),
            Layout(self.console[1],name='bottom',size=6)
        )
        return layout

db = Interface()
with Live(get_renderable=db.get_renderable):
    while True:
        time.sleep(1)
        db.console[0].print(datetime.now().ctime()+'='*100)
        db.console[1].print(datetime.now().ctime())

通过在Console 类中添加__rich_console__ 方法，可以直接将其视为其他可渲染对象，您可以将其放在布局中的任何位置。