-
Notifications
You must be signed in to change notification settings - Fork 3.3k
/
Copy pathprinter.py
46 lines (37 loc) · 1.44 KB
/
printer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from typing import Any
from rich.console import Console, Group
from rich.live import Live
from rich.spinner import Spinner
class Printer:
"""
Simple wrapper to stream status updates. Used by the financial bot
manager as it orchestrates planning, search and writing.
"""
def __init__(self, console: Console) -> None:
self.live = Live(console=console)
self.items: dict[str, tuple[str, bool]] = {}
self.hide_done_ids: set[str] = set()
self.live.start()
def end(self) -> None:
self.live.stop()
def hide_done_checkmark(self, item_id: str) -> None:
self.hide_done_ids.add(item_id)
def update_item(
self, item_id: str, content: str, is_done: bool = False, hide_checkmark: bool = False
) -> None:
self.items[item_id] = (content, is_done)
if hide_checkmark:
self.hide_done_ids.add(item_id)
self.flush()
def mark_item_done(self, item_id: str) -> None:
self.items[item_id] = (self.items[item_id][0], True)
self.flush()
def flush(self) -> None:
renderables: list[Any] = []
for item_id, (content, is_done) in self.items.items():
if is_done:
prefix = "✅ " if item_id not in self.hide_done_ids else ""
renderables.append(prefix + content)
else:
renderables.append(Spinner("dots", text=content))
self.live.update(Group(*renderables))