FirmwareUpdateChecker: Small refactors due to code review.

This commit is contained in:
Remco Burema 2018-10-13 19:21:22 +02:00
parent 1b7055f0f3
commit 60408c14bc
4 changed files with 27 additions and 24 deletions

View file

@ -5,19 +5,20 @@ import os
from PyQt5.QtCore import QUrl from PyQt5.QtCore import QUrl
from PyQt5.QtGui import QDesktopServices from PyQt5.QtGui import QDesktopServices
from typing import List from typing import Set
from UM.Extension import Extension from UM.Extension import Extension
from UM.Application import Application from UM.Application import Application
from UM.Logger import Logger from UM.Logger import Logger
from UM.PluginRegistry import PluginRegistry from UM.PluginRegistry import PluginRegistry
from UM.Qt.QtApplication import QtApplication
from UM.i18n import i18nCatalog from UM.i18n import i18nCatalog
from UM.Settings.ContainerRegistry import ContainerRegistry from UM.Settings.ContainerRegistry import ContainerRegistry
from cura.Settings.GlobalStack import GlobalStack from cura.Settings.GlobalStack import GlobalStack
from .FirmwareUpdateCheckerJob import FirmwareUpdateCheckerJob from .FirmwareUpdateCheckerJob import FirmwareUpdateCheckerJob
from .FirmwareUpdateCheckerLookup import FirmwareUpdateCheckerLookup, get_settings_key_for_machine from .FirmwareUpdateCheckerLookup import FirmwareUpdateCheckerLookup, getSettingsKeyForMachine
i18n_catalog = i18nCatalog("cura") i18n_catalog = i18nCatalog("cura")
@ -36,22 +37,23 @@ class FirmwareUpdateChecker(Extension):
if Application.getInstance().getPreferences().getValue("info/automatic_update_check"): if Application.getInstance().getPreferences().getValue("info/automatic_update_check"):
ContainerRegistry.getInstance().containerAdded.connect(self._onContainerAdded) ContainerRegistry.getInstance().containerAdded.connect(self._onContainerAdded)
self._late_init = True # Init some things after creation, since we need the path from the plugin-mgr. # Partly initialize after creation, since we need our own path from the plugin-manager.
self._download_url = None self._download_url = None
self._check_job = None self._check_job = None
self._name_cache = [] # type: List[str] self._checked_printer_names = [] # type: Set[str]
self._lookups = None self._lookups = None
QtApplication.pluginsLoaded.connect(self._onPluginsLoaded)
## Callback for the message that is spawned when there is a new version. ## Callback for the message that is spawned when there is a new version.
def _onActionTriggered(self, message, action): def _onActionTriggered(self, message, action):
try:
download_url = self._lookups.getRedirectUserFor(int(action)) download_url = self._lookups.getRedirectUserFor(int(action))
if download_url is not None: if download_url is not None:
QDesktopServices.openUrl(QUrl(download_url)) if QDesktopServices.openUrl(QUrl(download_url)):
Logger.log("i", "Redirected browser to {0} to show newly available firmware.".format(download_url))
else:
Logger.log("e", "Can't reach URL: {0}".format(download_url))
else: else:
Logger.log("e", "Can't find URL for {0}".format(action)) Logger.log("e", "Can't find URL for {0}".format(action))
except Exception as ex:
Logger.log("e", "Don't know what to do with '{0}' because {1}".format(action, ex))
def _onContainerAdded(self, container): def _onContainerAdded(self, container):
# Only take care when a new GlobalStack was added # Only take care when a new GlobalStack was added
@ -61,8 +63,9 @@ class FirmwareUpdateChecker(Extension):
def _onJobFinished(self, *args, **kwargs): def _onJobFinished(self, *args, **kwargs):
self._check_job = None self._check_job = None
def doLateInit(self): def _onPluginsLoaded(self):
self._late_init = False if self._lookups is not None:
return
self._lookups = FirmwareUpdateCheckerLookup(os.path.join(PluginRegistry.getInstance().getPluginPath( self._lookups = FirmwareUpdateCheckerLookup(os.path.join(PluginRegistry.getInstance().getPluginPath(
"FirmwareUpdateChecker"), "resources/machines.json")) "FirmwareUpdateChecker"), "resources/machines.json"))
@ -70,7 +73,7 @@ class FirmwareUpdateChecker(Extension):
# Initialize the Preference called `latest_checked_firmware` that stores the last version # Initialize the Preference called `latest_checked_firmware` that stores the last version
# checked for each printer. # checked for each printer.
for machine_id in self._lookups.getMachineIds(): for machine_id in self._lookups.getMachineIds():
Application.getInstance().getPreferences().addPreference(get_settings_key_for_machine(machine_id), "") Application.getInstance().getPreferences().addPreference(getSettingsKeyForMachine(machine_id), "")
## Connect with software.ultimaker.com, load latest.version and check version info. ## Connect with software.ultimaker.com, load latest.version and check version info.
# If the version info is different from the current version, spawn a message to # If the version info is different from the current version, spawn a message to
@ -79,13 +82,13 @@ class FirmwareUpdateChecker(Extension):
# \param silent type(boolean) Suppresses messages other than "new version found" messages. # \param silent type(boolean) Suppresses messages other than "new version found" messages.
# This is used when checking for a new firmware version at startup. # This is used when checking for a new firmware version at startup.
def checkFirmwareVersion(self, container = None, silent = False): def checkFirmwareVersion(self, container = None, silent = False):
if self._late_init: if self._lookups is None:
self.doLateInit() self._onPluginsLoaded()
container_name = container.definition.getName() container_name = container.definition.getName()
if container_name in self._name_cache: if container_name in self._checked_printer_names:
return return
self._name_cache.append(container_name) self._checked_printer_names.append(container_name)
self._check_job = FirmwareUpdateCheckerJob(container = container, silent = silent, self._check_job = FirmwareUpdateCheckerJob(container = container, silent = silent,
lookups = self._lookups, lookups = self._lookups,

View file

@ -12,7 +12,7 @@ from urllib.error import URLError
from typing import Dict from typing import Dict
import codecs import codecs
from .FirmwareUpdateCheckerLookup import FirmwareUpdateCheckerLookup, get_settings_key_for_machine from .FirmwareUpdateCheckerLookup import FirmwareUpdateCheckerLookup, getSettingsKeyForMachine
from UM.i18n import i18nCatalog from UM.i18n import i18nCatalog
i18n_catalog = i18nCatalog("cura") i18n_catalog = i18nCatalog("cura")
@ -78,12 +78,12 @@ class FirmwareUpdateCheckerJob(Job):
# If it is not None, then we compare between the checked_version and the current_version # If it is not None, then we compare between the checked_version and the current_version
machine_id = self._lookups.getMachineByName(machine_name.lower()) machine_id = self._lookups.getMachineByName(machine_name.lower())
if machine_id is not None: if machine_id is not None:
Logger.log("i", "You have a {0} in the printer list. Let's check the firmware!".format(machine_name)) Logger.log("i", "You have a(n) {0} in the printer list. Let's check the firmware!".format(machine_name))
current_version = self.getCurrentVersionForMachine(machine_id) current_version = self.getCurrentVersionForMachine(machine_id)
# If it is the first time the version is checked, the checked_version is "" # If it is the first time the version is checked, the checked_version is ""
setting_key_str = get_settings_key_for_machine(machine_id) setting_key_str = getSettingsKeyForMachine(machine_id)
checked_version = Version(Application.getInstance().getPreferences().getValue(setting_key_str)) checked_version = Version(Application.getInstance().getPreferences().getValue(setting_key_str))
# If the checked_version is "", it's because is the first time we check firmware and in this case # If the checked_version is "", it's because is the first time we check firmware and in this case

View file

@ -12,17 +12,17 @@ from UM.i18n import i18nCatalog
i18n_catalog = i18nCatalog("cura") i18n_catalog = i18nCatalog("cura")
def get_settings_key_for_machine(machine_id: int) -> str: def getSettingsKeyForMachine(machine_id: int) -> str:
return "info/latest_checked_firmware_for_{0}".format(machine_id) return "info/latest_checked_firmware_for_{0}".format(machine_id)
def default_parse_version_response(response: str) -> Version: def defaultParseVersionResponse(response: str) -> Version:
raw_str = response.split("\n", 1)[0].rstrip() raw_str = response.split("\n", 1)[0].rstrip()
return Version(raw_str.split(".")) # Split it into a list; the default parsing of "single string" is different. return Version(raw_str)
class FirmwareUpdateCheckerLookup: class FirmwareUpdateCheckerLookup:
JSON_NAME_TO_VERSION_PARSE_FUNCTION = {"default": default_parse_version_response} JSON_NAME_TO_VERSION_PARSE_FUNCTION = {"default": defaultParseVersionResponse}
def __init__(self, json_path) -> None: def __init__(self, json_path) -> None:
# Open the .json file with the needed lookup-lists for each machine(/model) and retrieve "raw" json. # Open the .json file with the needed lookup-lists for each machine(/model) and retrieve "raw" json.
@ -48,7 +48,7 @@ class FirmwareUpdateCheckerLookup:
self.JSON_NAME_TO_VERSION_PARSE_FUNCTION.get(machine_json.get("version_parser")) self.JSON_NAME_TO_VERSION_PARSE_FUNCTION.get(machine_json.get("version_parser"))
if version_parse_function is None: if version_parse_function is None:
Logger.log("w", "No version-parse-function specified for machine {0}.".format(machine_name)) Logger.log("w", "No version-parse-function specified for machine {0}.".format(machine_name))
version_parse_function = default_parse_version_response # Use default instead if nothing is found. version_parse_function = defaultParseVersionResponse # Use default instead if nothing is found.
self._parse_version_url_per_machine[machine_id] = version_parse_function self._parse_version_url_per_machine[machine_id] = version_parse_function
self._check_urls_per_machine[machine_id] = [] # Multiple check-urls: see "_comment" in the .json file. self._check_urls_per_machine[machine_id] = [] # Multiple check-urls: see "_comment" in the .json file.
for check_url in machine_json.get("check_urls"): for check_url in machine_json.get("check_urls"):

View file

@ -1,5 +1,5 @@
{ {
"_comment": "Multiple 'update_urls': The 'new'-style URL is the only way to check for S5 firmware, and in the future, the UM3 line will also switch over, but for now the old 'JEDI'-style URL is still needed.", "_comment": "There are multiple 'check_urls', because sometimes an URL is about to be phased out, and it's useful to have a new 'future-proof' one at the ready.",
"machines": "machines":
[ [