我正在使用Python中的“Rich”打印工具。它工作得很好 - 但我有一个问题。当我运行它时,其内容超出了终端窗口的容纳范围,它会切断底部的所有内容(如预期)。但当它完成后,我想让整个表格打印出来。这是一个重现:
import asyncio
import logging
import random
import time
import uuid
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, TextColumn
from rich.table import Table
all_statuses = {}
task_total = 30
console = Console()
table_update_event = asyncio.Event()
table_update_running = False
events_to_progress = []
max_time = 5 # Changed to 5 seconds
logging.basicConfig(level=logging.INFO)
class InstanceStatus:
def __init__(self, region, zone):
# Generate a unique ID for each instance - maximum 6 characters
self.id = f"{region}-{zone}-{uuid.uuid4()}"[:6]
self.region = region
self.zone = zone
self.status = "Initializing"
self.detailed_status = "Initializing"
self.elapsed_time = 0
self.instance_id = None
self.public_ip = None
self.private_ip = None
self.vpc_id = None
def combined_status(self):
return f"{self.status} ({self.detailed_status})"
def make_progress_table():
table = Table(show_header=True, header_style="bold magenta", show_lines=False)
table.add_column("ID", width=8, style="cyan", no_wrap=True)
table.add_column("Region", width=8, style="green", no_wrap=True)
table.add_column("Zone", width=8, style="green", no_wrap=True)
table.add_column("Status", width=15, style="yellow", no_wrap=True)
table.add_column("Elapsed", width=8, justify="right", style="magenta", no_wrap=True)
table.add_column("Instance ID", width=15, style="blue", no_wrap=True)
table.add_column("Public IP", width=15, style="blue", no_wrap=True)
table.add_column("Private IP", width=15, style="blue", no_wrap=True)
sorted_statuses = sorted(all_statuses.values(), key=lambda x: (x.region, x.zone))
for status in sorted_statuses:
table.add_row(
status.id[:8],
status.region[:8],
status.zone[:8],
status.combined_status()[:15],
f"{status.elapsed_time:.1f}s",
(status.instance_id or "")[:15],
(status.public_ip or "")[:15],
(status.private_ip or "")[:15],
)
return table
def create_layout(progress, table):
layout = Layout()
progress_panel = Panel(
progress,
title="Progress",
border_style="green",
padding=(1, 1),
)
layout.split(
Layout(progress_panel, size=5),
Layout(table),
)
return layout
async def update_table(live):
global table_update_running, events_to_progress, all_statuses, console
if table_update_running:
logging.debug("Table update already running. Exiting.")
return
logging.debug("Starting table update.")
try:
table_update_running = True
progress = Progress(
TextColumn("[progress.description]{task.description}"),
BarColumn(bar_width=None),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
TextColumn("[progress.completed]{task.completed} of {task.total}"),
TextColumn("[progress.elapsed]{task.elapsed:>3.0f}s"),
expand=True,
)
task = progress.add_task("Creating Instances", total=task_total)
while not table_update_event.is_set() or events_to_progress:
while events_to_progress:
event = events_to_progress.pop(0)
all_statuses[event.id] = event
progress.update(task, completed=len(all_statuses))
table = make_progress_table()
layout = create_layout(progress, table)
live.update(layout)
await asyncio.sleep(0.05) # Reduce sleep time for more frequent updates
except Exception as e:
logging.error(f"Error in update_table: {str(e)}")
finally:
table_update_running = False
logging.debug("Table update finished.")
async def main():
global events_to_progress, all_statuses
start_time = time.time()
end_time = start_time + 4 # Set to 4 seconds
statuses_to_create = [
InstanceStatus(str(random.randint(1, 100)), str(random.randint(1, 1000)))
for _ in range(task_total)
]
with Live(console=console, refresh_per_second=20) as live:
update_table_task = asyncio.create_task(update_table(live))
# Distribute status creation over 4 seconds
for i in range(task_total):
events_to_progress.append(statuses_to_create[i])
if (i + 1) % 10 == 0:
await asyncio.sleep(0.4)
# Ensure all statuses are processed
all_statuses.update({status.id: status for status in statuses_to_create})
# If we finished early, wait until 4 seconds have passed
time_elapsed = time.time() - start_time
if time_elapsed < 4:
await asyncio.sleep(4 - time_elapsed)
table_update_event.set()
if __name__ == "__main__":
asyncio.run(main())
如果
task_total = 30
小于终端的高度,则它可以正常工作。如果它更大(例如 100),那么它会切断表格并且永远不会打印最终结果。
你有几个问题。首先,您的
Live
未设置为显示溢出。这是一个简单的修复。
with Live(console=console, refresh_per_second=20, vertical_overflow='visible') as live:
其次,
Layout
类将基于终端大小,而不是内容大小。您必须测量表格及其所有边框和标题的大小,然后需要显式设置布局(外部和内部)的大小。然而,由于您只是垂直拆分,因此有一个更简单的解决方案。不要使用布局,而是创建一个 Group
。
将
create_layout
替换为以下内容(需要您将 Group
添加到来自 rich.console
的导入列表中)。
def create_group(progress, table):
progress_panel = Panel(
progress,
title="Progress",
border_style="green",
padding=(1, 1),
height=5,
)
return Group(table, progress_panel)
那么您可能想更改下面的变量名称
layout
(或者只是删除赋值并执行live.update(create_group(progress, table))
)
layout = create_layout(progress, table)
live.update(layout)
如果您不想切换进度面板和表格的顺序,另一种选择是在处理时渲染
Layout
,以便进度条可见,然后在处理完成时显示 Group
。