-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconsole_richlog.py
More file actions
62 lines (49 loc) · 1.79 KB
/
Copy pathconsole_richlog.py
File metadata and controls
62 lines (49 loc) · 1.79 KB
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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""A test with RichLog.
This example shows capturing output from a console script which runs using the
subprocess module, in a threaded worker. The problem it hasn't solved is that in
a terminal, three progressbars are displayed at the bottom of the screen while
text is occassionally printed above it the progress bars, while in RichLog, each
screen update writes three new lines to the log. The result is that the log is
flooded with hundreds of lines of progress bars slowly building up to
completion.
"""
import subprocess
from rich.text import Text
from textual import work
from textual.app import App, ComposeResult
from textual.widgets import RichLog
class MyLog(RichLog):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.border_title = "Console"
self.auto_scroll = True
def mywrite(self, text: Text):
self.write(text)
class ConsoleApp(App):
CSS = """
#console {
border: double lightgreen;
}
"""
def compose(self) -> ComposeResult:
yield MyLog(id="console", auto_scroll=False)
async def on_mount(self) -> None:
self.run_process()
@work(thread=True)
def run_process(self) -> None:
console: MyLog = self.query_one("#console")
with subprocess.Popen(
"FORCE_COLOR=yes python -m rich.progress",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
encoding="utf-8",
) as proc:
while True:
output = proc.stdout.readline()
if output == "" and proc.poll() is not None:
break
self.call_from_thread(console.mywrite, Text.from_ansi(output))
if __name__ == "__main__":
app = ConsoleApp()
app.run()