import json from typing import Optional from PyQt5.QtCore import QObject from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest from UM import i18nCatalog from UM.Logger import Logger from UM.Message import Message from UM.Signal import Signal from plugins.Toolbox.src.UltimakerCloudScope import UltimakerCloudScope from cura.CuraApplication import CuraApplication from plugins.Toolbox.src.CloudApiModel import CloudApiModel from plugins.Toolbox.src.CloudSync.SubscribedPackagesModel import SubscribedPackagesModel class CloudPackageChecker(QObject): def __init__(self, application: CuraApplication) -> None: super().__init__() self.discrepancies = Signal() # Emits SubscribedPackagesModel self._application = application # type: CuraApplication self._scope = UltimakerCloudScope(application) self._model = SubscribedPackagesModel() self._application.initializationFinished.connect(self._onAppInitialized) self._i18n_catalog = i18nCatalog("cura") # 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._fetchUserSubscribedPackages() # check again whenever the login state changes self._application.getCuraAPI().account.loginStateChanged.connect(self._fetchUserSubscribedPackages) def _fetchUserSubscribedPackages(self) -> None: if self._application.getCuraAPI().account.isLoggedIn: self._getUserPackages() def _handleCompatibilityData(self, json_data) -> None: user_subscribed_packages = [plugin["package_id"] for plugin in json_data] user_installed_packages = self._package_manager.getUserInstalledPackages() user_dismissed_packages = self._package_manager.getDismissedPackages() if user_dismissed_packages: user_installed_packages += user_dismissed_packages # We check if there are packages installed in Cloud Marketplace but not in Cura marketplace package_discrepancy = list(set(user_subscribed_packages).difference(user_installed_packages)) self._model.setMetadata(json_data) self._model.addDiscrepancies(package_discrepancy) self._model.initialize() if package_discrepancy: self._handlePackageDiscrepancies() def _handlePackageDiscrepancies(self) -> None: Logger.log("d", "Discrepancy found between Cloud subscribed packages and Cura installed packages") sync_message = Message(self._i18n_catalog.i18nc( "@info:generic", "\nDo you want to sync material and software packages with your account?"), lifetime=0, 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 Cloud subscribed packages to your local environment.", button_align=Message.ActionButtonAlignment.ALIGN_RIGHT) sync_message.actionTriggered.connect(self._onSyncButtonClicked) sync_message.show() def _onSyncButtonClicked(self, sync_message: Message, sync_message_action: str) -> None: sync_message.hide() self.discrepancies.emit(self._model) def _getUserPackages(self) -> None: Logger.log("d", "Requesting subscribed packages metadata from server.") url = CloudApiModel.api_url_user_packages self._application.getHttpRequestManager().get(url, callback = self._onUserPackagesRequestFinished, error_callback = self._onUserPackagesRequestFinished, scope = self._scope) def _onUserPackagesRequestFinished(self, reply: "QNetworkReply", error: Optional["QNetworkReply.NetworkError"] = None) -> None: if error is not None or reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200: Logger.log("w", "Requesting user packages failed, response code %s while trying to connect to %s", reply.attribute(QNetworkRequest.HttpStatusCodeAttribute), reply.url()) 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"]) return self._handleCompatibilityData(json_data["data"]) except json.decoder.JSONDecodeError: Logger.log("w", "Received invalid JSON for user packages")