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
|
# Copyright (c) 2022 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json
from typing import List, Dict, Any, Set
from typing import Optional
from PyQt6.QtCore import QObject
from PyQt6.QtNetwork import QNetworkReply
from UM import i18nCatalog
from UM.Logger import Logger
from UM.Message import Message
from UM.Signal import Signal
from UM.TaskManagement.HttpRequestManager import HttpRequestManager
from UM.TaskManagement.HttpRequestScope import JsonDecoratorScope
from cura.API.Account import SyncState
from cura.CuraApplication import CuraApplication, ApplicationMetadata
from cura.UltimakerCloud.UltimakerCloudScope import UltimakerCloudScope
from .SubscribedPackagesModel import SubscribedPackagesModel
from ..CloudApiModel import CloudApiModel
class CloudPackageChecker(QObject):
SYNC_SERVICE_NAME = "CloudPackageChecker"
def __init__(self, application: CuraApplication) -> None:
super().__init__()
self.discrepancies = Signal() # Emits SubscribedPackagesModel
self._application: CuraApplication = application
self._scope = JsonDecoratorScope(UltimakerCloudScope(application))
self._model = SubscribedPackagesModel()
self._message: Optional[Message] = None
self._application.initializationFinished.connect(self._onAppInitialized)
self._i18n_catalog = i18nCatalog("cura")
self._sdk_version = ApplicationMetadata.CuraSDKVersion
self._last_notified_packages = set() # type: Set[str]
"""Packages for which a notification has been shown. No need to bother the user twice for equal content"""
# This is a plugin, so most of the components required are not ready when
# this is initialized. Therefore, we wait until the application is ready.
def _onAppInitialized(self) -> None:
self._package_manager = self._application.getPackageManager()
# initial check
self._getPackagesIfLoggedIn()
self._application.getCuraAPI().account.loginStateChanged.connect(self._onLoginStateChanged)
self._application.getCuraAPI().account.syncRequested.connect(self._getPackagesIfLoggedIn)
def _onLoginStateChanged(self) -> None:
# reset session
self._last_notified_packages = set()
self._getPackagesIfLoggedIn()
def _getPackagesIfLoggedIn(self) -> None:
if self._application.getCuraAPI().account.isLoggedIn:
self._getUserSubscribedPackages()
else:
self._hideSyncMessage()
def _getUserSubscribedPackages(self) -> None:
self._application.getCuraAPI().account.setSyncState(self.SYNC_SERVICE_NAME, SyncState.SYNCING)
url = CloudApiModel.api_url_user_packages
self._application.getHttpRequestManager().get(url,
callback = self._onUserPackagesRequestFinished,
error_callback = self._onUserPackagesRequestFinished,
timeout = 10,
scope = self._scope)
def _onUserPackagesRequestFinished(self, reply: "QNetworkReply", error: Optional["QNetworkReply.NetworkError"] = None) -> None:
if error is not None or HttpRequestManager.safeHttpStatus(reply) != 200:
Logger.log("w",
"Requesting user packages failed, response code %s while trying to connect to %s",
HttpRequestManager.safeHttpStatus(reply), reply.url())
self._application.getCuraAPI().account.setSyncState(self.SYNC_SERVICE_NAME, SyncState.ERROR)
return
try:
json_data = json.loads(bytes(reply.readAll()).decode("utf-8"))
# Check for errors:
if "errors" in json_data:
for error in json_data["errors"]:
Logger.log("e", "%s", error["title"])
self._application.getCuraAPI().account.setSyncState(self.SYNC_SERVICE_NAME, SyncState.ERROR)
return
self._handleCompatibilityData(json_data["data"])
except json.decoder.JSONDecodeError:
Logger.log("w", "Received invalid JSON for user subscribed packages from the Web Marketplace")
self._application.getCuraAPI().account.setSyncState(self.SYNC_SERVICE_NAME, SyncState.SUCCESS)
def _handleCompatibilityData(self, subscribed_packages_payload: List[Dict[str, Any]]) -> None:
user_subscribed_packages = {plugin["package_id"] for plugin in subscribed_packages_payload}
user_installed_packages = self._package_manager.getAllInstalledPackageIDs()
# We need to re-evaluate the dismissed packages
# (i.e. some package might got updated to the correct SDK version in the meantime,
# hence remove them from the Dismissed Incompatible list)
self._package_manager.reEvaluateDismissedPackages(subscribed_packages_payload, self._sdk_version)
user_dismissed_packages = self._package_manager.getDismissedPackages()
if user_dismissed_packages:
user_installed_packages.update(user_dismissed_packages)
# We check if there are packages installed in Web Marketplace but not in Cura marketplace
package_discrepancy = list(user_subscribed_packages.difference(user_installed_packages))
if user_subscribed_packages != self._last_notified_packages:
# scenario:
# 1. user subscribes to a package
# 2. dismisses the license/unsubscribes
# 3. subscribes to the same package again
# in this scenario we want to notify the user again. To capture that there was a change during
# step 2, we clear the last_notified after step 2. This way, the user will be notified after
# step 3 even though the list of packages for step 1 and 3 are equal
self._last_notified_packages = set()
if package_discrepancy:
account = self._application.getCuraAPI().account
account.setUpdatePackagesAction(lambda: self._onSyncButtonClicked(None, None))
if user_subscribed_packages == self._last_notified_packages:
# already notified user about these
return
Logger.log("d", "Discrepancy found between Cloud subscribed packages and Cura installed packages")
self._model.addDiscrepancies(package_discrepancy)
self._model.initialize(self._package_manager, subscribed_packages_payload)
self._showSyncMessage()
self._last_notified_packages = user_subscribed_packages
def _showSyncMessage(self) -> None:
"""Show the message if it is not already shown"""
if self._message is not None:
self._message.show()
return
sync_message = Message(self._i18n_catalog.i18nc(
"@info:generic",
"Do you want to sync material and software packages with your account?"),
title = self._i18n_catalog.i18nc("@info:title", "Changes detected from your Ultimaker account", ))
sync_message.addAction("sync",
name = self._i18n_catalog.i18nc("@action:button", "Sync"),
icon = "",
description = "Sync your plugins and print profiles to Ultimaker Cura.",
button_align = Message.ActionButtonAlignment.ALIGN_RIGHT)
sync_message.actionTriggered.connect(self._onSyncButtonClicked)
sync_message.show()
self._message = sync_message
def _hideSyncMessage(self) -> None:
"""Hide the message if it is showing"""
if self._message is not None:
self._message.hide()
self._message = None
def _onSyncButtonClicked(self, sync_message: Optional[Message], sync_message_action: Optional[str]) -> None:
if sync_message is not None:
sync_message.hide()
self._hideSyncMessage() # Should be the same message, but also sets _message to None
self.discrepancies.emit(self._model)
|
