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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
|
# Copyright (c) 2021 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from datetime import datetime
from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot, pyqtProperty, QTimer, Q_ENUMS
from typing import Any, Optional, Dict, TYPE_CHECKING, Callable
from UM.Logger import Logger
from UM.Message import Message
from UM.i18n import i18nCatalog
from cura.OAuth2.AuthorizationService import AuthorizationService
from cura.OAuth2.Models import OAuth2Settings, UserProfile
from cura.UltimakerCloud import UltimakerCloudConstants
if TYPE_CHECKING:
from cura.CuraApplication import CuraApplication
i18n_catalog = i18nCatalog("cura")
class SyncState:
"""QML: Cura.AccountSyncState"""
SYNCING = 0
SUCCESS = 1
ERROR = 2
IDLE = 3
class Account(QObject):
"""The account API provides a version-proof bridge to use Ultimaker Accounts
Usage:
.. code-block:: python
from cura.API import CuraAPI
api = CuraAPI()
api.account.login()
api.account.logout()
api.account.userProfile # Who is logged in
"""
# The interval in which sync services are automatically triggered
SYNC_INTERVAL = 60.0 # seconds
Q_ENUMS(SyncState)
loginStateChanged = pyqtSignal(bool)
"""Signal emitted when user logged in or out"""
userProfileChanged = pyqtSignal()
"""Signal emitted when new account information is available."""
additionalRightsChanged = pyqtSignal("QVariantMap")
"""Signal emitted when a users additional rights change"""
accessTokenChanged = pyqtSignal()
syncRequested = pyqtSignal()
"""Sync services may connect to this signal to receive sync triggers.
Services should be resilient to receiving a signal while they are still syncing,
either by ignoring subsequent signals or restarting a sync.
See setSyncState() for providing user feedback on the state of your service.
"""
lastSyncDateTimeChanged = pyqtSignal()
syncStateChanged = pyqtSignal(int) # because SyncState is an int Enum
manualSyncEnabledChanged = pyqtSignal(bool)
updatePackagesEnabledChanged = pyqtSignal(bool)
CLIENT_SCOPES = "account.user.read drive.backup.read drive.backup.write packages.download " \
"packages.rating.read packages.rating.write connect.cluster.read connect.cluster.write connect.material.write " \
"library.project.read library.project.write cura.printjob.read cura.printjob.write " \
"cura.mesh.read cura.mesh.write"
def __init__(self, application: "CuraApplication", parent = None) -> None:
super().__init__(parent)
self._application = application
self._new_cloud_printers_detected = False
self._error_message: Optional[Message] = None
self._logged_in = False
self._user_profile: Optional[UserProfile] = None
self._additional_rights: Dict[str, Any] = {}
self._sync_state = SyncState.IDLE
self._manual_sync_enabled = False
self._update_packages_enabled = False
self._update_packages_action: Optional[Callable] = None
self._last_sync_str = "-"
self._callback_port = 32118
self._oauth_root = UltimakerCloudConstants.CuraCloudAccountAPIRoot
self._oauth_settings = OAuth2Settings(
OAUTH_SERVER_URL= self._oauth_root,
CALLBACK_PORT=self._callback_port,
CALLBACK_URL="http://localhost:{}/callback".format(self._callback_port),
CLIENT_ID="um----------------------------ultimaker_cura",
CLIENT_SCOPES=self.CLIENT_SCOPES,
AUTH_DATA_PREFERENCE_KEY="general/ultimaker_auth_data",
AUTH_SUCCESS_REDIRECT="{}/app/auth-success".format(self._oauth_root),
AUTH_FAILED_REDIRECT="{}/app/auth-error".format(self._oauth_root)
)
self._authorization_service = AuthorizationService(self._oauth_settings)
# Create a timer for automatic account sync
self._update_timer = QTimer()
self._update_timer.setInterval(int(self.SYNC_INTERVAL * 1000))
# The timer is restarted explicitly after an update was processed. This prevents 2 concurrent updates
self._update_timer.setSingleShot(True)
self._update_timer.timeout.connect(self.sync)
self._sync_services: Dict[str, int] = {}
"""contains entries "service_name" : SyncState"""
def initialize(self) -> None:
self._authorization_service.initialize(self._application.getPreferences())
self._authorization_service.onAuthStateChanged.connect(self._onLoginStateChanged)
self._authorization_service.onAuthenticationError.connect(self._onLoginStateChanged)
self._authorization_service.accessTokenChanged.connect(self._onAccessTokenChanged)
self._authorization_service.loadAuthDataFromPreferences()
@pyqtProperty(int, notify=syncStateChanged)
def syncState(self):
return self._sync_state
def setSyncState(self, service_name: str, state: int) -> None:
""" Can be used to register sync services and update account sync states
Contract: A sync service is expected exit syncing state in all cases, within reasonable time
Example: `setSyncState("PluginSyncService", SyncState.SYNCING)`
:param service_name: A unique name for your service, such as `plugins` or `backups`
:param state: One of SyncState
"""
prev_state = self._sync_state
self._sync_services[service_name] = state
if any(val == SyncState.SYNCING for val in self._sync_services.values()):
self._sync_state = SyncState.SYNCING
self._setManualSyncEnabled(False)
elif any(val == SyncState.ERROR for val in self._sync_services.values()):
self._sync_state = SyncState.ERROR
self._setManualSyncEnabled(True)
else:
self._sync_state = SyncState.SUCCESS
self._setManualSyncEnabled(False)
if self._sync_state != prev_state:
self.syncStateChanged.emit(self._sync_state)
if self._sync_state == SyncState.SUCCESS:
self._last_sync_str = datetime.now().strftime("%d/%m/%Y %H:%M")
self.lastSyncDateTimeChanged.emit()
if self._sync_state != SyncState.SYNCING:
# schedule new auto update after syncing completed (for whatever reason)
if not self._update_timer.isActive():
self._update_timer.start()
def setUpdatePackagesAction(self, action: Callable) -> None:
""" Set the callback which will be invoked when the user clicks the update packages button
Should be invoked after your service sets the sync state to SYNCING and before setting the
sync state to SUCCESS.
Action will be reset to None when the next sync starts
"""
self._update_packages_action = action
self._update_packages_enabled = True
self.updatePackagesEnabledChanged.emit(self._update_packages_enabled)
def _onAccessTokenChanged(self):
self.accessTokenChanged.emit()
@property
def is_staging(self) -> bool:
"""Indication whether the given authentication is applied against staging or not."""
return "staging" in self._oauth_root
@pyqtProperty(bool, notify=loginStateChanged)
def isLoggedIn(self) -> bool:
return self._logged_in
def _onLoginStateChanged(self, logged_in: bool = False, error_message: Optional[str] = None) -> None:
if error_message:
if self._error_message:
self._error_message.hide()
Logger.log("w", "Failed to login: %s", error_message)
self._error_message = Message(error_message,
title = i18n_catalog.i18nc("@info:title", "Login failed"),
message_type = Message.MessageType.ERROR)
self._error_message.show()
self._logged_in = False
self.loginStateChanged.emit(False)
if self._update_timer.isActive():
self._update_timer.stop()
return
if self._logged_in != logged_in:
self._logged_in = logged_in
self.loginStateChanged.emit(logged_in)
if logged_in:
self._authorization_service.getUserProfile(self._onProfileChanged)
self._setManualSyncEnabled(False)
self._sync()
else:
if self._update_timer.isActive():
self._update_timer.stop()
def _onProfileChanged(self, profile: Optional[UserProfile]) -> None:
self._user_profile = profile
self.userProfileChanged.emit()
def _sync(self) -> None:
"""Signals all sync services to start syncing
This can be considered a forced sync: even when a
sync is currently running, a sync will be requested.
"""
self._update_packages_action = None
self._update_packages_enabled = False
self.updatePackagesEnabledChanged.emit(self._update_packages_enabled)
if self._update_timer.isActive():
self._update_timer.stop()
elif self._sync_state == SyncState.SYNCING:
Logger.debug("Starting a new sync while previous sync was not completed")
self.syncRequested.emit()
def _setManualSyncEnabled(self, enabled: bool) -> None:
if self._manual_sync_enabled != enabled:
self._manual_sync_enabled = enabled
self.manualSyncEnabledChanged.emit(enabled)
@pyqtSlot()
@pyqtSlot(bool)
def login(self, force_logout_before_login: bool = False) -> None:
"""
Initializes the login process. If the user is logged in already and force_logout_before_login is true, Cura will
logout from the account before initiating the authorization flow. If the user is logged in and
force_logout_before_login is false, the function will return, as there is nothing to do.
:param force_logout_before_login: Optional boolean parameter
:return: None
"""
if self._logged_in:
if force_logout_before_login:
self.logout()
else:
# Nothing to do, user already logged in.
return
self._authorization_service.startAuthorizationFlow(force_logout_before_login)
@pyqtProperty(str, notify = userProfileChanged)
def userName(self):
if not self._user_profile:
return ""
return self._user_profile.username
@pyqtProperty(str, notify = userProfileChanged)
def profileImageUrl(self):
if not self._user_profile:
return ""
return self._user_profile.profile_image_url
@pyqtProperty(str, notify=accessTokenChanged)
def accessToken(self) -> Optional[str]:
return self._authorization_service.getAccessToken()
@pyqtProperty("QVariantMap", notify = userProfileChanged)
def userProfile(self) -> Optional[Dict[str, Optional[str]]]:
"""None if no user is logged in otherwise the logged in user as a dict containing containing user_id, username and profile_image_url """
if not self._user_profile:
return None
return self._user_profile.__dict__
@pyqtProperty(str, notify=lastSyncDateTimeChanged)
def lastSyncDateTime(self) -> str:
return self._last_sync_str
@pyqtProperty(bool, notify=manualSyncEnabledChanged)
def manualSyncEnabled(self) -> bool:
return self._manual_sync_enabled
@pyqtProperty(bool, notify=updatePackagesEnabledChanged)
def updatePackagesEnabled(self) -> bool:
return self._update_packages_enabled
@pyqtSlot()
@pyqtSlot(bool)
def sync(self, user_initiated: bool = False) -> None:
if user_initiated:
self._setManualSyncEnabled(False)
self._sync()
@pyqtSlot()
def onUpdatePackagesClicked(self) -> None:
if self._update_packages_action is not None:
self._update_packages_action()
@pyqtSlot()
def popupOpened(self) -> None:
self._setManualSyncEnabled(True)
@pyqtSlot()
def logout(self) -> None:
if not self._logged_in:
return # Nothing to do, user isn't logged in.
self._authorization_service.deleteAuthData()
def updateAdditionalRight(self, **kwargs) -> None:
"""Update the additional rights of the account.
The argument(s) are the rights that need to be set"""
self._additional_rights.update(kwargs)
self.additionalRightsChanged.emit(self._additional_rights)
@pyqtProperty("QVariantMap", notify = additionalRightsChanged)
def additionalRights(self) -> Dict[str, Any]:
"""A dictionary which can be queried for additional account rights."""
return self._additional_rights
|