-
Notifications
You must be signed in to change notification settings - Fork 0
/
issue_await_dismiss_with_None.py
88 lines (69 loc) · 2.3 KB
/
issue_await_dismiss_with_None.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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
import time
from textual import on, work
from textual.app import App, ComposeResult
from textual.events import ScreenResume
from textual.message import Message
from textual.screen import ModalScreen
from textual.widgets import Button, LoadingIndicator, Static
from textual.worker import Worker, WorkerState
class MyTask(ModalScreen):
def compose(self) -> ComposeResult:
yield Button("Close")
yield LoadingIndicator(id="loading_indicator")
@on(Button.Pressed)
def close(self, event: Button.Pressed) -> None:
# bug: if you do not return a value, await push_screen will hang and
# subsequent tasks will not be performed in speedrun
self.dismiss()
@on(ScreenResume)
def resume(self) -> None:
self.run_task()
@work(thread=True)
def run_task(self):
time.sleep(1)
@on(Worker.StateChanged)
def exit_task(self, event: Worker.StateChanged):
if event.state == WorkerState.SUCCESS:
self.query_one("#loading_indicator").remove()
class TaskButton(Button):
def __init__(
self, task_class: type[MyTask], label: str | None = None, *args, **kwargs
) -> None:
super().__init__(label, *args, **kwargs)
self.task_class = task_class
@on(Button.Pressed)
def execute(self):
task = self.task_class()
self.app.push_screen(task)
class SpeedRunApp(App[None]):
# CSS = """
# Screen, ModalScreen {
# height: 100%;
# align: center middle;
# }
# Static {
# text-align: center;
# }
# LoadingIndicator {
# height: auto;
# }
# """
def compose(self) -> ComposeResult:
for i in range(1, 4):
yield TaskButton(MyTask, f"Task {i}", id=f"task{i}")
yield Button("Run all", id="speedrun")
@on(Button.Pressed, "#speedrun")
def execute_speedrun(self, event: Button.Pressed):
self.speedrun()
@work()
async def speedrun(self):
print("Starting tasks...")
for i in range(3):
print(f"Running task {i}")
task = MyTask()
await self.push_screen(task, wait_for_dismiss=True)
print(f"Finished task {i}")
print("All tasks done!")
app = SpeedRunApp()
if __name__ == "__main__":
app.run()