# Copyright (c) 2022 Ultimaker B.V. # Cura is released under the terms of the LGPLv3 or higher. from PyQt6.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, QObject, QUrl from PyQt6.QtGui import QDesktopServices from typing import Dict, Optional, TYPE_CHECKING import zipfile # To export all materials in a .zip archive. import cura.CuraApplication # Imported like this to prevent circular imports. from UM.Resources import Resources from cura.PrinterOutput.UploadMaterialsJob import UploadMaterialsJob, UploadMaterialsError # To export materials to the output printer. from cura.Settings.CuraContainerRegistry import CuraContainerRegistry from UM.i18n import i18nCatalog from UM.Logger import Logger from UM.Message import Message if TYPE_CHECKING: from UM.Signal import Signal catalog = i18nCatalog("cura") class CloudMaterialSync(QObject): """ Handles the synchronisation of material profiles with cloud accounts. """ def __init__(self, parent: QObject = None): super().__init__(parent) self.sync_all_dialog = None # type: Optional[QObject] self._export_upload_status = "idle" self._checkIfNewMaterialsWereInstalled() self._export_progress = 0.0 self._printer_status = {} # type: Dict[str, str] def _checkIfNewMaterialsWereInstalled(self) -> None: """ Checks whether new material packages were installed in the latest startup. If there were, then it shows a message prompting the user to sync the materials with their printers. """ application = cura.CuraApplication.CuraApplication.getInstance() for package_id, package_data in application.getPackageManager().getPackagesInstalledOnStartup().items(): if package_data["package_info"]["package_type"] == "material": # At least one new material was installed self._showSyncNewMaterialsMessage() break def openSyncAllWindow(self): self.reset() if self.sync_all_dialog is None: qml_path = Resources.getPath(cura.CuraApplication.CuraApplication.ResourceTypes.QmlFiles, "Preferences", "Materials", "MaterialsSyncDialog.qml") self.sync_all_dialog = cura.CuraApplication.CuraApplication.getInstance().createQmlComponent( qml_path, {}) if self.sync_all_dialog is None: # Failed to load QML file. return self.sync_all_dialog.setProperty("syncModel", self) self.sync_all_dialog.setProperty("pageIndex", 0) # Return to first page. self.sync_all_dialog.setProperty("hasExportedUsb", False) # If the user exported USB before, reset that page. self.sync_all_dialog.setProperty("syncStatusText", "") # Reset any previous error messages. self.sync_all_dialog.show() def _showSyncNewMaterialsMessage(self) -> None: sync_materials_message = Message( text = catalog.i18nc("@action:button", "Please sync the material profiles with your printers before starting to print."), title = catalog.i18nc("@action:button", "New materials installed"), message_type = Message.MessageType.WARNING, lifetime = 0 ) sync_materials_message.addAction( "sync", name = catalog.i18nc("@action:button", "Sync materials"), icon = "", description = "Sync your newly installed materials with your printers.", button_align = Message.ActionButtonAlignment.ALIGN_RIGHT ) sync_materials_message.addAction( "learn_more", name = catalog.i18nc("@action:button", "Learn more"), icon = "", description = "Learn more about syncing your newly installed materials with your printers.", button_align = Message.ActionButtonAlignment.ALIGN_LEFT, button_style = Message.ActionButtonStyle.LINK ) sync_materials_message.actionTriggered.connect(self._onSyncMaterialsMessageActionTriggered) # Show the message only if there are printers that support material export container_registry = cura.CuraApplication.CuraApplication.getInstance().getContainerRegistry() global_stacks = container_registry.findContainerStacks(type = "machine") if any([stack.supportsMaterialExport for stack in global_stacks]): sync_materials_message.show() def _onSyncMaterialsMessageActionTriggered(self, sync_message: Message, sync_message_action: str): if sync_message_action == "sync": self.openSyncAllWindow() sync_message.hide() elif sync_message_action == "learn_more": QDesktopServices.openUrl(QUrl("https://support.ultimaker.com/hc/en-us/articles/360013137919?utm_source=cura&utm_medium=software&utm_campaign=sync-material-printer-message")) @pyqtSlot(result = QUrl) def getPreferredExportAllPath(self) -> QUrl: """ Get the preferred path to export materials to. If there is a removable drive, that should be the preferred path. Otherwise it should be the most recent local file path. :return: The preferred path to export all materials to. """ cura_application = cura.CuraApplication.CuraApplication.getInstance() device_manager = cura_application.getOutputDeviceManager() devices = device_manager.getOutputDevices() for device in devices: if device.__class__.__name__ == "RemovableDriveOutputDevice": return QUrl.fromLocalFile(device.getId()) else: # No removable drives? Use local path. return cura_application.getDefaultPath("dialog_material_path") @pyqtSlot(QUrl) def exportAll(self, file_path: QUrl, notify_progress: Optional["Signal"] = None) -> None: """ Export all materials to a certain file path. :param file_path: The path to export the materials to. """ registry = CuraContainerRegistry.getInstance() # Create empty archive. try: archive = zipfile.ZipFile(file_path.toLocalFile(), "w", compression = zipfile.ZIP_DEFLATED) except OSError as e: Logger.log("e", f"Can't write to destination {file_path.toLocalFile()}: {type(e)} - {str(e)}") error_message = Message( text = catalog.i18nc("@message:text", "Could not save material archive to {}:").format(file_path.toLocalFile()) + " " + str(e), title = catalog.i18nc("@message:title", "Failed to save material archive"), message_type = Message.MessageType.ERROR ) error_message.show() return materials_metadata = registry.findInstanceContainersMetadata(type = "material") for index, metadata in enumerate(materials_metadata): if notify_progress is not None: progress = index / len(materials_metadata) notify_progress.emit(progress) if metadata["base_file"] != metadata["id"]: # Only process base files. continue if metadata["id"] == "empty_material": # Don't export the empty material. continue # Ignore materials that are marked as not visible for whatever reason if not bool(metadata.get("visible", True)): continue material = registry.findContainers(id = metadata["id"])[0] suffix = registry.getMimeTypeForContainer(type(material)).preferredSuffix filename = metadata["id"] + "." + suffix try: archive.writestr(filename, material.serialize()) except OSError as e: Logger.log("e", f"An error has occurred while writing the material \'{metadata['id']}\' in the file \'{filename}\': {e}.") exportUploadStatusChanged = pyqtSignal() @pyqtProperty(str, notify = exportUploadStatusChanged) def exportUploadStatus(self) -> str: return self._export_upload_status @pyqtSlot() def exportUpload(self) -> None: """ Export all materials and upload them to the user's account. """ self._export_upload_status = "uploading" self.exportUploadStatusChanged.emit() job = UploadMaterialsJob(self) job.uploadProgressChanged.connect(self._onUploadProgressChanged) job.uploadCompleted.connect(self.exportUploadCompleted) job.start() def _onUploadProgressChanged(self, progress: float, printers_status: Dict[str, str]): self.setExportProgress(progress) self.setPrinterStatus(printers_status) def exportUploadCompleted(self, job_result: UploadMaterialsJob.Result, job_error: Optional[Exception]): if not self.sync_all_dialog: # Shouldn't get triggered before the dialog is open, but better to check anyway. return if job_result == UploadMaterialsJob.Result.FAILED: if isinstance(job_error, UploadMaterialsError): self.sync_all_dialog.setProperty("syncStatusText", str(job_error)) else: # Could be "None" self.sync_all_dialog.setProperty("syncStatusText", catalog.i18nc("@text", "Unknown error.")) self._export_upload_status = "error" else: self._export_upload_status = "success" self.exportUploadStatusChanged.emit() exportProgressChanged = pyqtSignal(float) def setExportProgress(self, progress: float) -> None: self._export_progress = progress self.exportProgressChanged.emit(self._export_progress) @pyqtProperty(float, fset = setExportProgress, notify = exportProgressChanged) def exportProgress(self) -> float: return self._export_progress printerStatusChanged = pyqtSignal() def setPrinterStatus(self, new_status: Dict[str, str]) -> None: self._printer_status = new_status self.printerStatusChanged.emit() @pyqtProperty("QVariantMap", fset = setPrinterStatus, notify = printerStatusChanged) def printerStatus(self) -> Dict[str, str]: return self._printer_status def reset(self) -> None: self.setPrinterStatus({}) self.setExportProgress(0.0) self._export_upload_status = "idle" self.exportUploadStatusChanged.emit()