-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
188 lines (158 loc) · 6.74 KB
/
app.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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
from typing import Callable
from rich.text import TextType
from textual import work
from textual.app import App, ComposeResult
from textual.containers import Vertical, VerticalScroll
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Static, RichLog, TabPane, TabbedContent, Header, LoadingIndicator
from textual_prusa_connect.config import AppSettings
from textual_prusa_connect.connect_api import PrusaConnectAPI
from textual_prusa_connect.app_widgets import PrinterHeader
from textual_prusa_connect.messages import PrinterUpdated
from textual_prusa_connect.widgets.dashboard import DashboardPane
from textual_prusa_connect.widgets.file import PrintJobWidget
SETTINGS = AppSettings()
PRINTING_REFRESH = 5
MAIN_REFRESH = 30
OTHER_REFRESH = 30
dummy = {
'filament': {},
'firmware': 'dummy',
'location': 'dummy',
'printer_model': 'dummy',
'printer_state': 'dummy',
'printer_type': 'dummy',
'printer_type_name': 'dummy',
'name': 'dummy',
'nozzle_diameter': 0.0,
'slots': 0,
'supported_printer_models': [],
'uuid': SETTINGS.printer_uuid,
}
class DataLoaded(Message):
"""Data loaded message"""
def __init__(self, content):
self.content = content
super().__init__()
class LazyTabPane(TabPane):
# reference https://gist.github.com/paulrobello/0a2f807ddd195c42f87cec9ff5825ac8
loaded = False
def __init__(self, title: TextType, init_callable: Callable, *children: Widget):
super().__init__(title, *children)
self.init_callable = init_callable
def compose(self) -> ComposeResult:
yield VerticalScroll()
# LoadingIndicator adds some unwanted behaviour where "navigating"
# to the pane does not work, while "clicking" on the pane works as expected
yield LoadingIndicator()
@work
async def update_data(self):
content = self.init_callable()
self.post_message(DataLoaded(content=content))
async def on_data_loaded(self, msg: DataLoaded) -> None:
msg.stop()
await self.query_one(LoadingIndicator).remove()
vs = self.query_one(VerticalScroll)
for i, element in enumerate(msg.content):
await vs.mount(element)
if i < len(msg.content) - 1:
await vs.mount(Static(" "))
def on_show(self):
if self.loaded:
return
self.update_data()
self.loaded = True
class PrusaConnectApp(App):
DEFAULT_CSS = """
PrusaConnectApp {
TabbedContent {
height: 1fr;
}
Container {
height: auto;
}
}
"""
BINDINGS = [('p', 'toggle_refresh', 'Pause'),
('s', 'screenshot', 'Take screenshot'),
('q', 'quit', 'Quit'),
('d', 'dump', 'Dump tree')]
CSS_PATH = "css.tcss"
do_refresh = True
refresh_rate = PRINTING_REFRESH
def __init__(self, headers: dict[str, str]):
super().__init__()
self.refresh_timer = None
self.client = PrusaConnectAPI(headers)
self.printer = self.client.get_printer(SETTINGS.printer_uuid)
# self.printer = Printer(**dummy)
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with Vertical():
yield PrinterHeader(printer=self.printer)
with TabbedContent():
yield DashboardPane(self.client, self.printer)
yield TabPane("Printer files", disabled=True)
yield TabPane("Print Queue", disabled=True)
def load_print_history():
jobs = self.client.get_jobs(limit=25)
return [PrintJobWidget(job) for job in jobs]
yield LazyTabPane("Print history", load_print_history)
yield TabPane("Control", disabled=True)
yield TabPane("Statistics", disabled=True)
yield TabPane("Telemetry", disabled=True)
yield TabPane("Settings", disabled=True)
with TabPane("App logs", id='logs'):
yield RichLog()
def on_mount(self):
self.screen.set_focus(None)
# self.update_printer(True)
self.query_one(RichLog).write(self.printer)
self.refresh_timer = self.set_interval(self.refresh_rate, self.update_printer)
self.set_interval(MAIN_REFRESH, self.background_loop)
def background_loop(self):
new_rate = OTHER_REFRESH
if self.printer.printer_state == 'PRINTING':
new_rate = PRINTING_REFRESH
if new_rate != self.refresh_rate:
self.refresh_rate = new_rate
self.refresh_timer.stop()
self.query_one(RichLog).write(f'new rate {self.refresh_rate}')
self.refresh_timer = self.set_interval(self.refresh_rate, self.update_printer, pause=not self.do_refresh)
def update_printer(self):
new_printer = self.client.get_printer(self.printer.uuid.get_secret_value())
if self.printer.printer_state != new_printer.printer_state:
self.notify(f"{self.printer.printer_state} -> {new_printer.printer_state}",
title='State change',
timeout=10)
# We are no longer printing, let's remove the currently printing block
# if self.printer.printer_state != 'PRINTING':
# self.query_one(CurrentlyPrinting).remove()
# NOT WORKING Started a new print, let's recompose to add the block back
# if new_printer.printer_state == 'PRINTING':
# self.query_one(DashboardPane).recompose()
for widget in self.query('.--requires-printer'):
widget.post_message(PrinterUpdated(printer=new_printer))
self.printer = new_printer
# self.query_one(RichLog).write(f'updated {self.refresh_rate}')
def action_dump(self):
self.query_one(RichLog).write(self.tree)
def action_toggle_refresh(self):
if self.do_refresh:
self.refresh_timer.pause()
self.query_one(TabbedContent).add_class('--app-paused')
self.query_one(RichLog).write('paused')
else:
self.refresh_timer.resume()
self.query_one(TabbedContent).remove_class('--app-paused')
self.query_one(RichLog).write('resumed')
self.do_refresh = not self.do_refresh
if __name__ == '__main__':
from textual_prusa_connect.version import __version__
my_headers = {
'cookie': f'SESSID="{SETTINGS.session_id}"',
'User-Agent': f"textual-prusa-connect/{__version__}"
}
app = PrusaConnectApp(my_headers)
app.run()