安装

pip install rich

# PDM
pdm add rich

文档 & 源码

基础

from rich import print

print("Hello World")

Rich功能非常丰富,从日志处理到富文本显示甚至动态的终端界面。

要使用Rich用做界面的话,一般是这样的:

from rich.console import Console
from rich.layout import Layout
from rich.panel import 

console = Console()
panel = Panel(name="panel_name")
layout = Layout(panel,name="layout_name")

界面Layout和面板Panel是经常要用到的东西,Layout作为界面布局,Panel内展示元素并风格化,比如经常使用的表格Table;要实现动态变化而不是print需要用到Live。

然后是一些小记录。

进度条 Process

这里展示的是一个模拟文件下载的进度条

from time import sleep

from rich.panel import Panel
from rich.progress import (
    BarColumn,
    FileSizeColumn,
    Progress,
    TextColumn,
    TimeRemainingColumn,
    TransferSpeedColumn,
)

class MyProgress(Progress):
    def get_renderables(self):
        yield Panel(self.make_tasks_table(self.tasks))


text_column = TextColumn("{task.description}")
bar_column = BarColumn(bar_width=None)
progress = MyProgress(
    text_column,
    "•",
    FileSizeColumn(),
    bar_column,
    TransferSpeedColumn(),
    "•",
    TimeRemainingColumn(),
    expand=True,
)
# 描述文本 • 文件大小组件 进度条 速度展示 • 剩余时间 自动扩充
with progress:
    n = 0
    file_size = 1024 * 1024 * 1024
    task_id = progress.add_task(f"Downloading file {n}", total=file_size)
    for n in range(100):
        progress.update(task_id, completed=n / 100 * file_size, description=f"{n}%")
        sleep(0.1)

18bfdba2ce81313f5250c797c1d70f9d.png

可以看到progress提供了非常多的工具来帮助构建进度条

转圈动画 Spinner

一个基本的spinner

from rich.spinner import Spinner

spinner = Spinner("dots")

你可以将它装载到Table上,比如

from rich.console import Console
from rich.spinner import Spinner
from rich.table import Table

table = Table(title="File")
table.add_column("ID", justify="right", style="cyan", no_wrap=True)
table.add_column("Name", style="magenta")
table.add_column("Size", justify="right", style="green")
table.add_column("Status", justify="center")

spinner = Spinner("dots")

table.add_row("1", "1.txt", "0.2MB", spinner)
table.add_row("2", "2.txt", "1.1MB", spinner)
table.add_row("3", "3.txt", "0.3MB", spinner)
table.add_row("4", "4.txt", "0.8MB", spinner)

console = Console()
console.print(table)

注意,此时你是无法看到旋转动画的,因为使用的是print,动画是需要在with console或者with live中操作

import time

from rich.console import Console
from rich.spinner import Spinner

console = Console()

spinner = Spinner("dots")

with console.status(spinner):
    # 模拟进行中的操作
    time.sleep(5)

console.print("Task completed!")

此时动画正常

配合Loguru输出日志

示例:来源Github

from loguru import logger
from rich.console import Console

console = Console()
logger.remove()


def _log_formatter(record: dict) -> str:
    """Log message formatter"""
    color_map = {
        "TRACE": "dim blue",
        "DEBUG": "cyan",
        "INFO": "bold",
        "SUCCESS": "bold green",
        "WARNING": "yellow",
        "ERROR": "bold red",
        "CRITICAL": "bold white on red",
    }
    lvl_color = color_map.get(record["level"].name, "cyan")
    return (
        "[not bold green]{time:YYYY/MM/DD HH:mm:ss}[/not bold green] | {level.icon}"
        + f"  - [{lvl_color}]{{message}}[/{lvl_color}]"
    )


log_handler = logger.add(
    console.print,  # type: ignore
    level="DEBUG",
    format=_log_formatter,  # type: ignore
    colorize=True,
)

logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message")

QQ截图20230728012650.png

用户输入 Prompt

password = Prompt.ask("input your password (hideen)", password=True)

注意,Prompt操作需要停止Live,比如live.stop()

Table行数过多的一种解决方案

来自:StackOverFlow

每次新增行,表格会始终展示最新行,要是超过控制台展示高度,则会截断上面的旧行;是一个很好的参考

import os
import time
from collections import deque

from rich.console import Console
from rich.layout import Layout
from rich.live import Live
from rich.table import Table


def generate_table(rows):
    layout = Layout()
    console = Console()

    table = Table()
    table.add_column("Time")
    table.add_column("Message")

    rows = list(rows)

    # This would also get the height:
    # render_map = layout.render(console, console.options)
    # render_map[layout].region.height
    n_rows = os.get_terminal_size()[1]

    while n_rows >= 0:
        table = Table()
        table.add_column("Time")
        table.add_column("Message")

        for row in rows[-n_rows:]:
            table.add_row(*row)

        layout.update(table)

        render_map = layout.render(console, console.options)
        render_length = len(render_map[layout].render[-1])
        if render_length > 2:
            # The table is overflowing
            n_rows -= 1
        else:
            break

    return table


width, height = os.get_terminal_size()
messages = deque(maxlen=height - 4)  # save space for header and footer

with Live(generate_table(messages), refresh_per_second=5) as live:
    for i in range(100):
        time.sleep(0.2)
        messages.append((time.asctime(), f"Event {i:03d}"))
        live.update(generate_table(messages))

一个复杂界面的案例

来源:Github

我修改了一点,以保证Type Hint正确

import logging
import os
from datetime import datetime
from pathlib import Path
from time import sleep

import numpy as np
from rich.console import Console
from rich.layout import Layout
from rich.live import Live
from rich.panel import Panel
from rich.progress import (
    BarColumn,
    Progress,
    SpinnerColumn,
    TextColumn,
    TimeRemainingColumn,
)
from rich.table import Table


def get_file_handler(log_dir: Path) -> logging.FileHandler:
    log_format = (
        "[%(asctime)s]-[%(funcName)s(%(lineno)d)]-[%(levelname)s]-[%(message)s]"
    )
    log_file = log_dir / "test.log"
    file_handler = logging.FileHandler(log_file)
    file_handler.setFormatter(logging.Formatter(log_format))
    return file_handler


def get_logger(log_dir: Path, console: Console) -> logging.Logger:
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.INFO)
    logger.addHandler(get_file_handler(log_dir))
    return logger


def make_layout() -> Layout:
    layout = Layout(name="root")
    layout.split(Layout(name="header", size=3), Layout(name="main"))
    layout["main"].split_row(Layout(name="leftside"), Layout(name="termoutput"))
    layout["leftside"].split_column(
        Layout(name="stats", ratio=2), Layout(name="progbar")
    )
    return layout


def get_stats() -> Table:
    rand_arr = np.random.randint(low=2, high=10, size=(10, 3)) + np.random.random(
        (10, 3)
    )
    stats_table = Table(
        expand=True,
        show_header=True,
        header_style="bold",
        title="[magenta][b]Hot Stats![/b]",
        highlight=True,
    )
    stats_table.add_column("Column", justify="right")
    stats_table.add_column("Mean", justify="center")
    stats_table.add_column("Std", justify="center")
    stats_table.add_column("Max", justify="center")
    stats_table.add_column("Min", justify="center")

    for col in range(rand_arr.shape[1]):
        stats_table.add_row(
            f"Col {col}",
            f"{rand_arr[:, col].mean():.2f}",
            f"{rand_arr[:, col].std():.2f}",
            f"{rand_arr[:, col].max():.2f}",
            f"{rand_arr[:, col].min():.2f}",
        )

    return stats_table


# Old code for reading a the last line of a log file.
# def get_last_log(logger) -> str:
# 	log_file_path = logger.handlers[0].baseFilename
# 	with open(log_file_path, "r") as file:
# 		log_entries = file.readlines()
# 		last_entry = log_entries[-1].strip() if log_entries else ""

# 	return last_entry


class make_header:
    """Display header with clock."""

    def __rich__(self) -> Panel:
        grid = Table.grid(expand=True)
        grid.add_column(justify="center", ratio=1)
        grid.add_column(justify="right")
        grid.add_row(
            "[b]Super[/b] cool application",
            datetime.now().ctime().replace(":", "[blink]:[/]"),
        )
        return Panel(grid, style="red on black")


class MainTableHandler(logging.Handler):
    def __init__(self, main_table: Table, layout: Layout, log_level: str):
        super().__init__()
        self.main_table = main_table
        self.log_list = []
        self.layout = layout
        self.log_format = (
            "%(asctime)s-(%(funcName)s)-%(lineno)d-%(levelname)s-[%(message)s]"
        )
        self.setLevel(log_level)
        # Could set colors for levels here.

    def emit(self, record):
        record.asctime = record.asctime.split(",")[0]
        # msg = self.format(record) #if you want just the message info switch comment lines
        msg = self.log_format % record.__dict__
        tsize = os.get_terminal_size().lines // 2
        if len(self.log_list) > tsize:
            self.log_list.append(msg)
            self.log_list.pop(0)
            self.main_table = redraw_main_table(self.log_list)
            self.layout["termoutput"].update(Panel(self.main_table, border_style="red"))
        else:
            self.main_table.add_row(msg)
            self.log_list.append(msg)


def redraw_main_table(temp_list: list) -> Table:
    main_table = Table(
        expand=True,
        show_header=False,
        header_style="bold",
        title="[blue][b]Log Entries[/b]",
        highlight=True,
    )
    main_table.add_column("Log Entries")
    for row in temp_list:
        main_table.add_row(row)

    return main_table


def main():
    console = Console(color_system="truecolor")
    logger = get_logger(Path.cwd(), console=console)

    main_table = Table(
        expand=True,
        show_header=False,
        header_style="bold",
        title="[blue][b]Log Messages[/b]",
        highlight=True,
    )
    main_table.add_column("Log Output")

    my_progress_bar = Progress(
        SpinnerColumn(),
        TextColumn("{task.description}"),
        BarColumn(),
        "time remain:",
        TimeRemainingColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
    )
    my_counter = range(30)
    my_task = my_progress_bar.add_task("The jobs", total=len(my_counter))

    progress_table = Table.grid(expand=True)
    progress_table.add_row(
        Panel(
            my_progress_bar,
            title="Super awesome progress bar",
            border_style="green",
            padding=(1, 1),
        )
    )

    stats_table = get_stats()
    layout = make_layout()
    layout["header"].update(make_header())
    layout["progbar"].update(Panel(progress_table, border_style="green"))
    layout["termoutput"].update(Panel(main_table, border_style="blue"))
    layout["stats"].update(Panel(stats_table, border_style="magenta"))

    with Live(layout, refresh_per_second=10, screen=True) as live:
        # Add MainTableHandler to logger
        logger.addHandler(MainTableHandler(main_table, layout, logger.level))  # type: ignore
        for count in my_counter:
            sleep(1)
            logger.info(f"I made it to count {count}")
            my_progress_bar.update(my_task, completed=count)

            if count % 5 == 0:
                logger.warning("Gettin some STATS")
                stats_table = get_stats()
                layout["stats"].update(Panel(stats_table, border_style="magenta"))
                logger.critical("Stranger Danger")
            live.refresh()

    logger.info("Done logging.")


if __name__ == "__main__":
    main()

QQ截图20230728020916.png

这个示例可以说好,常用的都有,是个非常好的参考。