CreatBot/GeneralKlipperScreen
7
0
Fork 0
forked from CreatBot/CreatBotKlipperScreen
Code Pull Requests Activity

updater: use more descriptive names and reformat

Browse Source
This commit is contained in:
alfrix 2024-05-12 10:44:18 -03:00
parent e2f2023d55
commit 4d72330761
1 changed files with 211 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

311
panels/updater.py
View File

@ -1,4 +1,6 @@
import logging
from gettext import ngettext
import gi
gi.require_version("Gtk", "3.0")
@ -9,59 +11,86 @@ from ks_includes.screen_panel import ScreenPanel
class Panel(ScreenPanel):
def __init__(self, screen, title):
super().__init__(screen, title)
self.labels = {}
self.update_status = None
self.update_all = self._gtk.Button('arrow-up', _('Full Update'), 'color1', self.bts, Gtk.PositionType.LEFT, 1)
self.update_all.connect("clicked", self.show_update_info, "full")
self.update_all.set_vexpand(False)
self.refresh = self._gtk.Button('arrow-down', _('Refresh'), 'color3', self.bts, Gtk.PositionType.LEFT, 1)
self.refresh.connect("clicked", self.refresh_updates)
self.refresh.set_vexpand(False)
self.buttons = {
"update_all": self._gtk.Button(
image_name="arrow-up",
label=_("Full Update"),
style="color1",
scale=self.bts,
position=Gtk.PositionType.LEFT,
lines=1,
),
"refresh": self._gtk.Button(
image_name="arrow-down",
label=_("Refresh"),
style="color3",
scale=self.bts,
position=Gtk.PositionType.LEFT,
lines=1,
),
}
self.buttons["update_all"].connect("clicked", self.show_update_info, "full")
self.buttons["update_all"].set_vexpand(False)
self.buttons["refresh"].connect("clicked", self.refresh_updates)
self.buttons["refresh"].set_vexpand(False)
sbox = Gtk.Box(vexpand=False)
sbox.pack_start(self.update_all, True, True, 0)
sbox.pack_start(self.refresh, True, True, 0)
top_box = Gtk.Box(vexpand=False)
top_box.pack_start(self.buttons["update_all"], True, True, 0)
top_box.pack_start(self.buttons["refresh"], True, True, 0)
self.update_msg = Gtk.Label(label=_("Checking for updates, please wait..."), vexpand=True)
self.update_msg = Gtk.Label(
label=_("Checking for updates, please wait..."), vexpand=True
)
self.scroll = self._gtk.ScrolledWindow()
self.scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
self.scroll.add(self.update_msg)
self.main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, vexpand=True)
self.main_box.pack_start(sbox, False, False, 0)
self.main_box.pack_start(top_box, False, False, 0)
self.main_box.pack_start(self.scroll, True, True, 0)
self.content.add(self.main_box)
def activate(self):
self._screen._ws.send_method('machine.update.status', callback=self.get_updates)
self._screen._ws.send_method("machine.update.status", callback=self.get_updates)
def create_info_grid(self):
infogrid = Gtk.Grid()
infogrid.get_style_context().add_class("system-program-grid")
for i, prog in enumerate(sorted(list(self.update_status['version_info']))):
self.labels[prog] = Gtk.Label(hexpand=True, halign=Gtk.Align.START, ellipsize=Pango.EllipsizeMode.END)
for i, prog in enumerate(sorted(list(self.update_status["version_info"]))):
self.labels[prog] = Gtk.Label(
hexpand=True, halign=Gtk.Align.START, ellipsize=Pango.EllipsizeMode.END
)
self.labels[prog].get_style_context().add_class("updater-item")
self.labels[f"{prog}_status"] = self._gtk.Button()
self.labels[f"{prog}_status"].set_hexpand(False)
self.labels[f"{prog}_status"].connect("clicked", self.show_update_info, prog)
self.buttons[f"{prog}_status"] = self._gtk.Button()
self.buttons[f"{prog}_status"].set_hexpand(False)
self.buttons[f"{prog}_status"].connect(
"clicked", self.show_update_info, prog
)
try:
if prog in self._printer.system_info['available_services']:
self.labels[f"{prog}_restart"] = self._gtk.Button("refresh", _("Restart"),
if prog in self._printer.system_info["available_services"]:
self.buttons[f"{prog}_restart"] = self._gtk.Button(
"refresh",
_("Restart"),
"color2",
position=Gtk.PositionType.LEFT,
scale=self.bts)
self.labels[f"{prog}_restart"].connect("clicked", self.restart, prog)
infogrid.attach(self.labels[f"{prog}_restart"], 0, i, 1, 1)
scale=self.bts,
)
self.buttons[f"{prog}_restart"].connect(
"clicked", self.restart, prog
)
infogrid.attach(self.buttons[f"{prog}_restart"], 0, i, 1, 1)
except Exception as e:
logging.exception(e)
infogrid.attach(self.labels[prog], 1, i, 1, 1)
self.labels[prog].get_style_context().add_class('updater-item')
infogrid.attach(self.labels[f"{prog}_status"], 2, i, 1, 1)
infogrid.attach(self.buttons[f"{prog}_status"], 2, i, 1, 1)
self.update_program_info(prog)
self.clear_scroll()
self.scroll.add(infogrid)
@ -74,35 +103,52 @@ class Panel(ScreenPanel):
self.clear_scroll()
self.scroll.add(self.update_msg)
self._gtk.Button_busy(widget, True)
logging.info('Sending machine.update.refresh')
self._screen._ws.send_method('machine.update.refresh', callback=self.get_updates)
logging.info("Sending machine.update.refresh")
self._screen._ws.send_method(
"machine.update.refresh", callback=self.get_updates
)
def get_updates(self, response, method, params):
self._gtk.Button_busy(self.refresh, False)
self._gtk.Button_busy(self.buttons["refresh"], False)
logging.info(response)
if not response or 'result' not in response:
self.update_all.set_sensitive(False)
if not response or "result" not in response:
self.buttons["update_all"].set_sensitive(False)
self.clear_scroll()
if 'error' in response:
self.scroll.add(Gtk.Label(label=f"Moonraker: {response['error']['message']}", vexpand=True))
if "error" in response:
self.scroll.add(
Gtk.Label(
label=f"Moonraker: {response['error']['message']}", vexpand=True
)
)
else:
self.scroll.add(Gtk.Label(label=_("Not working or not configured"), vexpand=True))
self.scroll.add(
Gtk.Label(label=_("Not working or not configured"), vexpand=True)
)
else:
self.update_status = response['result']
self.update_all.set_sensitive(True)
self.update_status = response["result"]
self.buttons["update_all"].set_sensitive(True)
self.create_info_grid()
self.scroll.show_all()
def restart(self, widget, program):
if self._printer.state in ("printing", "paused"):
self._screen._confirm_send_action(widget, f'{_("Are you sure?")}\n\n'
f'{_("Restart")}: {program}',
"machine.services.restart", {"service": program})
self._screen._confirm_send_action(
widget,
f'{_("Are you sure?")}\n\n' f'{_("Restart")}: {program}',
"machine.services.restart",
{"service": program},
)
else:
self._screen._send_action(widget, "machine.services.restart", {"service": program})
self._screen._send_action(
widget, "machine.services.restart", {"service": program}
)
def show_update_info(self, widget, program):
info = self.update_status['version_info'][program] if program in self.update_status['version_info'] else {}
info = (
self.update_status["version_info"][program]
if program in self.update_status["version_info"]
else {}
)
scroll = self._gtk.ScrolledWindow(steppers=False)
scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
@ -111,61 +157,91 @@ class Panel(ScreenPanel):
label = Gtk.Label(wrap=True, vexpand=True)
if program == "full":
label.set_markup('<b>' + _("Perform a full upgrade?") + '</b>')
label.set_markup("<b>" + _("Perform a full upgrade?") + "</b>")
vbox.add(label)
elif 'configured_type' in info and info['configured_type'] == 'git_repo':
if not info['is_valid'] or info['is_dirty']:
elif "configured_type" in info and info["configured_type"] == "git_repo":
if not info["is_valid"] or info["is_dirty"]:
label.set_markup(_("Do you want to recover %s?") % program)
recoverybuttons = [
{"name": _("Recover Hard"), "response": Gtk.ResponseType.OK, "style": 'dialog-warning'},
{"name": _("Recover Soft"), "response": Gtk.ResponseType.APPLY, "style": 'dialog-info'},
{"name": _("Cancel"), "response": Gtk.ResponseType.CANCEL, "style": 'dialog-error'}
{
"name": _("Recover Hard"),
"response": Gtk.ResponseType.OK,
"style": "dialog-warning",
},
{
"name": _("Recover Soft"),
"response": Gtk.ResponseType.APPLY,
"style": "dialog-info",
},
{
"name": _("Cancel"),
"response": Gtk.ResponseType.CANCEL,
"style": "dialog-error",
},
]
self._gtk.Dialog(_("Recover"), recoverybuttons, label, self.reset_confirm, program)
self._gtk.Dialog(
_("Recover"), recoverybuttons, label, self.reset_confirm, program
)
return
else:
if info['version'] == info['remote_version']:
if info["version"] == info["remote_version"]:
return
ncommits = len(info['commits_behind'])
label.set_markup("<b>" +
_("Outdated by %d") % ncommits +
" " + ngettext("commit", "commits", ncommits) +
":</b>\n")
ncommits = len(info["commits_behind"])
label.set_markup(
"<b>"
+ _("Outdated by %d") % ncommits
+ " "
+ ngettext("commit", "commits", ncommits)
+ ":</b>\n"
)
vbox.add(label)
label.set_vexpand(False)
for c in info['commits_behind']:
for c in info["commits_behind"]:
commit_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
title = Gtk.Label(wrap=True, hexpand=True)
title.set_markup(f"\n<b>{c['subject']}</b>\n<i>{c['author']}</i>\n")
commit_box.add(title)
details = Gtk.Label(label=c['message'], wrap=True, hexpand=True)
details = Gtk.Label(label=c["message"], wrap=True, hexpand=True)
commit_box.add(details)
commit_box.add(Gtk.Separator())
vbox.add(commit_box)
elif "package_count" in info:
label.set_markup((
label.set_markup(
(
f'<b>{info["package_count"]} '
+ ngettext("Package will be updated", "Packages will be updated", info["package_count"])
+ ':</b>\n'
))
+ ngettext(
"Package will be updated",
"Packages will be updated",
info["package_count"],
)
+ ":</b>\n"
)
)
label.set_vexpand(False)
vbox.set_valign(Gtk.Align.CENTER)
vbox.add(label)
grid = Gtk.Grid(column_homogeneous=True, halign=Gtk.Align.CENTER, valign=Gtk.Align.CENTER)
grid = Gtk.Grid(
column_homogeneous=True,
halign=Gtk.Align.CENTER,
valign=Gtk.Align.CENTER,
)
i = 0
for j, c in enumerate(info["package_list"]):
label = Gtk.Label(halign=Gtk.Align.START, ellipsize=Pango.EllipsizeMode.END)
label = Gtk.Label(
halign=Gtk.Align.START, ellipsize=Pango.EllipsizeMode.END
)
label.set_markup(f" {c} ")
pos = (j % 3)
pos = j % 3
grid.attach(label, pos, i, 1, 1)
if pos == 2:
i += 1
vbox.add(grid)
else:
label.set_markup(
"<b>" + _("%s will be updated to version") % program.capitalize()
"<b>"
+ _("%s will be updated to version") % program.capitalize()
+ f": {info['remote_version']}</b>"
)
vbox.add(label)
@ -173,8 +249,16 @@ class Panel(ScreenPanel):
scroll.add(vbox)
buttons = [
{"name": _("Update"), "response": Gtk.ResponseType.OK, "style": 'dialog-info'},
{"name": _("Cancel"), "response": Gtk.ResponseType.CANCEL, "style": 'dialog-error'}
{
"name": _("Update"),
"response": Gtk.ResponseType.OK,
"style": "dialog-info",
},
{
"name": _("Cancel"),
"response": Gtk.ResponseType.CANCEL,
"style": "dialog-error",
},
]
self._gtk.Dialog(_("Update"), buttons, scroll, self.update_confirm, program)
@ -197,28 +281,42 @@ class Panel(ScreenPanel):
if self._screen.updating:
return
self._screen.base_panel.show_update_dialog()
msg = _("Starting recovery for") + f' {program}...'
self._screen._websocket_callback("notify_update_response",
{'application': {program}, 'message': msg, 'complete': False})
msg = _("Starting recovery for") + f" {program}..."
self._screen._websocket_callback(
"notify_update_response",
{"application": {program}, "message": msg, "complete": False},
)
logging.info(f"Sending machine.update.recover name: {program} hard: {hard}")
self._screen._ws.send_method("machine.update.recover", {"name": program, "hard": hard})
self._screen._ws.send_method(
"machine.update.recover", {"name": program, "hard": hard}
)
def update_program(self, widget, program):
if self._screen.updating or not self.update_status:
return
if program in self.update_status['version_info']:
info = self.update_status['version_info'][program]
if program in self.update_status["version_info"]:
info = self.update_status["version_info"][program]
logging.info(f"program: {info}")
if "package_count" in info and info['package_count'] == 0 \
or "version" in info and info['version'] == info['remote_version']:
if (
"package_count" in info
and info["package_count"] == 0
or "version" in info
and info["version"] == info["remote_version"]
):
return
self._screen.base_panel.show_update_dialog()
msg = _("Updating") if program == "full" else _("Starting update for") + f' {program}...'
self._screen._websocket_callback("notify_update_response",
{'application': {program}, 'message': msg, 'complete': False})
msg = (
_("Updating")
if program == "full"
else _("Starting update for") + f" {program}..."
)
self._screen._websocket_callback(
"notify_update_response",
{"application": {program}, "message": msg, "complete": False},
)
if program in ['klipper', 'moonraker', 'system', 'full']:
if program in ["klipper", "moonraker", "system", "full"]:
logging.info(f"Sending machine.update.{program}")
self._screen._ws.send_method(f"machine.update.{program}")
else:
@ -227,53 +325,60 @@ class Panel(ScreenPanel):
def update_program_info(self, p):
if not self.update_status or p not in self.update_status['version_info']:
if not self.update_status or p not in self.update_status["version_info"]:
logging.info(f"Unknown version: {p}")
return
info = self.update_status['version_info'][p]
info = self.update_status["version_info"][p]
if p == "system":
distro = (
self._printer.system_info['distribution']['name']
if 'distribution' in self._printer.system_info
and 'name' in self._printer.system_info['distribution']
else _('System'))
self._printer.system_info["distribution"]["name"]
if "distribution" in self._printer.system_info
and "name" in self._printer.system_info["distribution"]
else _("System")
)
self.labels[p].set_markup(f"<b>{distro}</b>")
if info['package_count'] == 0:
if info["package_count"] == 0:
self._already_updated(p)
else:
self._needs_update(p, local="", remote=info['package_count'])
self._needs_update(p, local="", remote=info["package_count"])
elif 'configured_type' in info and info['configured_type'] == 'git_repo':
if info['is_valid'] and not info['is_dirty']:
if info['version'] == info['remote_version']:
elif "configured_type" in info and info["configured_type"] == "git_repo":
if info["is_valid"] and not info["is_dirty"]:
if info["version"] == info["remote_version"]:
self.labels[p].set_markup(f"<b>{p}</b>\n{info['version']}")
self._already_updated(p)
self.labels[f"{p}_status"].get_style_context().remove_class('invalid')
self.buttons[f"{p}_status"].get_style_context().remove_class(
"invalid"
)
else:
self.labels[p].set_markup(f"<b>{p}</b>\n{info['version']} -> {info['remote_version']}")
self._needs_update(p, info['version'], info['remote_version'])
self.labels[p].set_markup(
f"<b>{p}</b>\n{info['version']} -> {info['remote_version']}"
)
self._needs_update(p, info["version"], info["remote_version"])
else:
logging.info(f"Invalid {p} {info['version']}")
self.labels[p].set_markup(f"<b>{p}</b>\n{info['version']}")
self.labels[f"{p}_status"].set_label(_("Invalid"))
self.labels[f"{p}_status"].get_style_context().add_class('invalid')
self.labels[f"{p}_status"].set_sensitive(True)
elif 'version' in info and info['version'] == info['remote_version']:
self.buttons[f"{p}_status"].set_label(_("Invalid"))
self.buttons[f"{p}_status"].get_style_context().add_class("invalid")
self.buttons[f"{p}_status"].set_sensitive(True)
elif "version" in info and info["version"] == info["remote_version"]:
self.labels[p].set_markup(f"<b>{p}</b>\n{info['version']}")
self._already_updated(p)
else:
self.labels[p].set_markup(f"<b>{p}</b>\n{info['version']} -> {info['remote_version']}")
self._needs_update(p, info['version'], info['remote_version'])
self.labels[p].set_markup(
f"<b>{p}</b>\n{info['version']} -> {info['remote_version']}"
)
self._needs_update(p, info["version"], info["remote_version"])
def _already_updated(self, p):
self.labels[f"{p}_status"].set_label(_("Up To Date"))
self.labels[f"{p}_status"].get_style_context().remove_class('update')
self.labels[f"{p}_status"].set_sensitive(False)
self.buttons[f"{p}_status"].set_label(_("Up To Date"))
self.buttons[f"{p}_status"].get_style_context().remove_class("update")
self.buttons[f"{p}_status"].set_sensitive(False)
def _needs_update(self, p, local="", remote=""):
logging.info(f"{p} {local} -> {remote}")
self.labels[f"{p}_status"].set_label(_("Update"))
self.labels[f"{p}_status"].get_style_context().add_class('update')
self.labels[f"{p}_status"].set_sensitive(True)
self.buttons[f"{p}_status"].set_label(_("Update"))
self.buttons[f"{p}_status"].get_style_context().add_class("update")
self.buttons[f"{p}_status"].set_sensitive(True)