FirmwareUpdateCheckerJob.py 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. # Copyright (c) 2020 Ultimaker B.V.
  2. # Cura is released under the terms of the LGPLv3 or higher.
  3. from UM.Application import Application
  4. from UM.Message import Message
  5. from UM.Logger import Logger
  6. from UM.Job import Job
  7. from UM.Version import Version
  8. import urllib.request
  9. from urllib.error import URLError
  10. from typing import Dict
  11. import ssl
  12. import certifi
  13. from .FirmwareUpdateCheckerLookup import FirmwareUpdateCheckerLookup, getSettingsKeyForMachine
  14. from .FirmwareUpdateCheckerMessage import FirmwareUpdateCheckerMessage
  15. from UM.i18n import i18nCatalog
  16. i18n_catalog = i18nCatalog("cura")
  17. class FirmwareUpdateCheckerJob(Job):
  18. """This job checks if there is an update available on the provided URL."""
  19. STRING_ZERO_VERSION = "0.0.0"
  20. STRING_EPSILON_VERSION = "0.0.1"
  21. ZERO_VERSION = Version(STRING_ZERO_VERSION)
  22. EPSILON_VERSION = Version(STRING_EPSILON_VERSION)
  23. def __init__(self, silent, machine_name, metadata, callback) -> None:
  24. super().__init__()
  25. self.silent = silent
  26. self._callback = callback
  27. self._machine_name = machine_name
  28. self._metadata = metadata
  29. self._lookups = FirmwareUpdateCheckerLookup(self._machine_name, self._metadata)
  30. self._headers = {} # type:Dict[str, str] # Don't set headers yet.
  31. def getUrlResponse(self, url: str) -> str:
  32. result = self.STRING_ZERO_VERSION
  33. try:
  34. # CURA-6698 Create an SSL context and use certifi CA certificates for verification.
  35. context = ssl.SSLContext(protocol = ssl.PROTOCOL_TLSv1_2)
  36. context.verify_mode = ssl.CERT_REQUIRED
  37. context.load_verify_locations(cafile = certifi.where())
  38. request = urllib.request.Request(url, headers = self._headers)
  39. response = urllib.request.urlopen(request, context = context)
  40. result = response.read().decode("utf-8")
  41. except URLError:
  42. Logger.log("w", "Could not reach '{0}', if this URL is old, consider removal.".format(url))
  43. return result
  44. def parseVersionResponse(self, response: str) -> Version:
  45. raw_str = response.split("\n", 1)[0].rstrip()
  46. return Version(raw_str)
  47. def getCurrentVersion(self) -> Version:
  48. max_version = self.ZERO_VERSION
  49. if self._lookups is None:
  50. return max_version
  51. machine_urls = self._lookups.getCheckUrls()
  52. if machine_urls is not None:
  53. for url in machine_urls:
  54. version = self.parseVersionResponse(self.getUrlResponse(url))
  55. if version > max_version:
  56. max_version = version
  57. if max_version < self.EPSILON_VERSION:
  58. Logger.log("w", "MachineID {0} not handled!".format(self._lookups.getMachineName()))
  59. return max_version
  60. def run(self):
  61. try:
  62. # Initialize a Preference that stores the last version checked for this printer.
  63. Application.getInstance().getPreferences().addPreference(
  64. getSettingsKeyForMachine(self._lookups.getMachineId()), "")
  65. # Get headers
  66. application_name = Application.getInstance().getApplicationName()
  67. application_version = Application.getInstance().getVersion()
  68. self._headers = {"User-Agent": "%s - %s" % (application_name, application_version)}
  69. # If it is not None, then we compare between the checked_version and the current_version
  70. machine_id = self._lookups.getMachineId()
  71. if machine_id is not None:
  72. Logger.log("i", "You have a(n) {0} in the printer list. Do firmware-check.".format(self._machine_name))
  73. current_version = self.getCurrentVersion()
  74. # This case indicates that was an error checking the version.
  75. # It happens for instance when not connected to internet.
  76. if current_version == self.ZERO_VERSION:
  77. return
  78. # If it is the first time the version is checked, the checked_version is ""
  79. setting_key_str = getSettingsKeyForMachine(machine_id)
  80. checked_version = Version(Application.getInstance().getPreferences().getValue(setting_key_str))
  81. # If the checked_version is "", it's because is the first time we check firmware and in this case
  82. # we will not show the notification, but we will store it for the next time
  83. Application.getInstance().getPreferences().setValue(setting_key_str, current_version)
  84. Logger.log("i", "Reading firmware version of %s: checked = %s - latest = %s",
  85. self._machine_name, checked_version, current_version)
  86. # The first time we want to store the current version, the notification will not be shown,
  87. # because the new version of Cura will be release before the firmware and we don't want to
  88. # notify the user when no new firmware version is available.
  89. if (checked_version != "") and (checked_version != current_version):
  90. Logger.log("i", "Showing firmware update message for new version: {version}".format(version = current_version))
  91. message = FirmwareUpdateCheckerMessage(machine_id, self._machine_name, current_version,
  92. self._lookups.getRedirectUserUrl())
  93. message.actionTriggered.connect(self._callback)
  94. message.show()
  95. else:
  96. Logger.log("i", "No machine with name {0} in list of firmware to check.".format(self._machine_name))
  97. except Exception as e:
  98. Logger.logException("w", "Failed to check for new version: %s", e)
  99. if not self.silent:
  100. Message(i18n_catalog.i18nc("@info", "Could not access update information.")).show()
  101. return