diff --git a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py index e319614bc7..017fd0f0ed 100644 --- a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py +++ b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py @@ -30,7 +30,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): def __init__(self, device_id, address: str, properties: Dict[bytes, bytes], parent: QObject = None) -> None: super().__init__(device_id = device_id, parent = parent) - self._manager = None # type: QNetworkAccessManager + self._manager = None # type: Optional[QNetworkAccessManager] self._last_manager_create_time = None # type: Optional[float] self._recreate_network_manager_time = 30 self._timeout_time = 10 # After how many seconds of no response should a timeout occur? @@ -162,7 +162,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): request.setHeader(QNetworkRequest.UserAgentHeader, self._user_agent) return request - def _createFormPart(self, content_header: str, data: str, content_type: Optional[str] = None) -> QHttpPart: + def _createFormPart(self, content_header: str, data: bytes, content_type: Optional[str] = None) -> QHttpPart: part = QHttpPart() if not content_header.startswith("form-data;"): @@ -191,7 +191,6 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): def put(self, target: str, data: str, on_finished: Optional[Callable[[QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() - assert(self._manager is not None) request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.put(request, data.encode()) @@ -200,7 +199,6 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): def get(self, target: str, on_finished: Optional[Callable[[QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() - assert(self._manager is not None) request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.get(request) @@ -209,7 +207,6 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): def post(self, target: str, data: str, on_finished: Optional[Callable[[QNetworkReply], None]], on_progress: Callable = None) -> None: if self._manager is None: self._createNetworkManager() - assert(self._manager is not None) request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.post(request, data) @@ -217,10 +214,9 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): reply.uploadProgress.connect(on_progress) self._registerOnFinishedCallback(reply, on_finished) - def postFormWithParts(self, target:str, parts: List[QHttpPart], on_finished: Optional[Callable[[QNetworkReply], None]], on_progress: Callable = None) -> None: + def postFormWithParts(self, target:str, parts: List[QHttpPart], on_finished: Optional[Callable[[QNetworkReply], None]], on_progress: Callable = None) -> QNetworkReply: if self._manager is None: self._createNetworkManager() - assert(self._manager is not None) request = self._createEmptyRequest(target, content_type=None) multi_post_part = QHttpMultiPart(QHttpMultiPart.FormDataType) for part in parts: diff --git a/plugins/UM3NetworkPrinting/ClusterUM3OutputDevice.py b/plugins/UM3NetworkPrinting/ClusterUM3OutputDevice.py index 197debb58b..495bbe1315 100644 --- a/plugins/UM3NetworkPrinting/ClusterUM3OutputDevice.py +++ b/plugins/UM3NetworkPrinting/ClusterUM3OutputDevice.py @@ -1,15 +1,12 @@ # Copyright (c) 2018 Ultimaker B.V. # Cura is released under the terms of the LGPLv3 or higher. -from typing import Generator -from typing import Set -from typing import Tuple -from typing import Union + +from typing import Any, cast, Set, Tuple, Union from UM.FileHandler.FileHandler import FileHandler from UM.FileHandler.FileWriter import FileWriter #To choose based on the output file mode (text vs. binary). from UM.FileHandler.WriteFileJob import WriteFileJob #To call the file writer asynchronously. from UM.Logger import Logger -from UM.Application import Application from UM.Settings.ContainerRegistry import ContainerRegistry from UM.i18n import i18nCatalog from UM.Message import Message @@ -18,6 +15,7 @@ from UM.OutputDevice import OutputDeviceError #To show that something went wrong from UM.Scene.SceneNode import SceneNode #For typing. from UM.Version import Version #To check against firmware versions for support. +from cura.CuraApplication import CuraApplication from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState from cura.PrinterOutput.PrinterOutputModel import PrinterOutputModel from cura.PrinterOutput.PrintJobOutputModel import PrintJobOutputModel @@ -30,7 +28,7 @@ from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply from PyQt5.QtGui import QDesktopServices from PyQt5.QtCore import pyqtSlot, QUrl, pyqtSignal, pyqtProperty, QObject -from time import time, sleep +from time import time from datetime import datetime from typing import Optional, Dict, List @@ -55,7 +53,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._number_of_extruders = 2 - self._dummy_lambdas = set() # type: Set[Tuple[str, Dict, Union[io.StringIO, io.BytesIO]]] + self._dummy_lambdas = ("", {}, io.BytesIO()) #type: Tuple[str, Dict, Union[io.StringIO, io.BytesIO]] self._print_jobs = [] # type: List[PrintJobOutputModel] @@ -65,18 +63,18 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): # See comments about this hack with the clusterPrintersChanged signal self.printersChanged.connect(self.clusterPrintersChanged) - self._accepts_commands = True + self._accepts_commands = True #type: bool # Cluster does not have authentication, so default to authenticated self._authentication_state = AuthState.Authenticated - self._error_message = None - self._write_job_progress_message = None - self._progress_message = None + self._error_message = None #type: Optional[Message] + self._write_job_progress_message = None #type: Optional[Message] + self._progress_message = None #type: Optional[Message] self._active_printer = None # type: Optional[PrinterOutputModel] - self._printer_selection_dialog = None + self._printer_selection_dialog = None #type: QObject self.setPriority(3) # Make sure the output device gets selected above local file output self.setName(self._id) @@ -91,7 +89,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._cluster_size = int(properties.get(b"cluster_size", 0)) - self._latest_reply_handler = None + self._latest_reply_handler = None #type: Optional[QNetworkReply] def requestWrite(self, nodes: List[SceneNode], file_name: Optional[str] = None, limit_mimetypes: bool = False, file_handler: Optional[FileHandler] = None, **kwargs: str) -> None: self.writeStarted.emit(self) @@ -100,10 +98,10 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): if file_handler: file_formats = file_handler.getSupportedFileTypesWrite() else: - file_formats = Application.getInstance().getMeshFileHandler().getSupportedFileTypesWrite() + file_formats = CuraApplication.getInstance().getMeshFileHandler().getSupportedFileTypesWrite() #Create a list from the supported file formats string. - machine_file_formats = Application.getInstance().getGlobalContainerStack().getMetaDataEntry("file_formats").split(";") + machine_file_formats = CuraApplication.getInstance().getGlobalContainerStack().getMetaDataEntry("file_formats").split(";") machine_file_formats = [file_type.strip() for file_type in machine_file_formats] #Exception for UM3 firmware version >=4.4: UFP is now supported and should be the preferred file format. if "application/x-ufp" not in machine_file_formats and self.printerType == "ultimaker3" and Version(self.firmwareVersion) >= Version("4.4"): @@ -120,9 +118,9 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): #Just take the first file format available. if file_handler is not None: - writer = file_handler.getWriterByMimeType(preferred_format["mime_type"]) + writer = file_handler.getWriterByMimeType(cast(str, preferred_format["mime_type"])) else: - writer = Application.getInstance().getMeshFileHandler().getWriterByMimeType(preferred_format["mime_type"]) + writer = CuraApplication.getInstance().getMeshFileHandler().getWriterByMimeType(cast(str, preferred_format["mime_type"])) #This function pauses with the yield, waiting on instructions on which printer it needs to print with. self._sending_job = self._sendPrintJob(writer, preferred_format, nodes) @@ -138,7 +136,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): def _spawnPrinterSelectionDialog(self): if self._printer_selection_dialog is None: path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "PrintWindow.qml") - self._printer_selection_dialog = Application.getInstance().createQmlComponent(path, {"OutputDevice": self}) + self._printer_selection_dialog = CuraApplication.getInstance().createQmlComponent(path, {"OutputDevice": self}) if self._printer_selection_dialog is not None: self._printer_selection_dialog.show() @@ -173,8 +171,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._error_message = Message( i18n_catalog.i18nc("@info:status", "Sending new jobs (temporarily) blocked, still sending the previous print job.")) - - assert(self._error_message is not None) self._error_message.show() yield #Wait on the user to select a target printer. yield #Wait for the write job to be finished. @@ -195,8 +191,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._write_job_progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), lifetime = 0, dismissable = False, progress = -1, title = i18n_catalog.i18nc("@info:title", "Sending Data"), use_inactivity_timer = False) - - assert(self._write_job_progress_message is not None) # use for typing purposes self._write_job_progress_message.show() self._dummy_lambdas = (target_printer, preferred_format, stream) @@ -207,9 +201,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): yield True #Return that we had success! yield #To prevent having to catch the StopIteration exception. - from cura.Utils.Threading import call_on_qt_thread - - def _sendPrintJobWaitOnWriteJobFinished(self, job): + def _sendPrintJobWaitOnWriteJobFinished(self, job: WriteFileJob) -> None: self._write_job_progress_message.hide() self._progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), lifetime = 0, dismissable = False, progress = -1, @@ -230,34 +222,35 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): # Add user name to the print_job parts.append(self._createFormPart("name=owner", bytes(self._getUserName(), "utf-8"), "text/plain")) - file_name = Application.getInstance().getPrintInformation().jobName + "." + preferred_format["extension"] + file_name = CuraApplication.getInstance().getPrintInformation().jobName + "." + preferred_format["extension"] output = stream.getvalue() #Either str or bytes depending on the output mode. if isinstance(stream, io.StringIO): - output = output.encode("utf-8") + output = cast(str, output).encode("utf-8") + output = cast(bytes, output) parts.append(self._createFormPart("name=\"file\"; filename=\"%s\"" % file_name, output)) - self._latest_reply_handler = self.postFormWithParts("print_jobs/", parts, onFinished=self._onPostPrintJobFinished, onProgress=self._onUploadPrintJobProgress) + self._latest_reply_handler = self.postFormWithParts("print_jobs/", parts, on_finished = self._onPostPrintJobFinished, on_progress = self._onUploadPrintJobProgress) - @pyqtProperty(QObject, notify=activePrinterChanged) + @pyqtProperty(QObject, notify = activePrinterChanged) def activePrinter(self) -> Optional[PrinterOutputModel]: return self._active_printer @pyqtSlot(QObject) - def setActivePrinter(self, printer: Optional[PrinterOutputModel]): + def setActivePrinter(self, printer: Optional[PrinterOutputModel]) -> None: if self._active_printer != printer: if self._active_printer and self._active_printer.camera: self._active_printer.camera.stop() self._active_printer = printer self.activePrinterChanged.emit() - def _onPostPrintJobFinished(self, reply): + def _onPostPrintJobFinished(self, reply: QNetworkReply) -> None: self._progress_message.hide() self._compressing_gcode = False self._sending_gcode = False - def _onUploadPrintJobProgress(self, bytes_sent:int, bytes_total:int): + def _onUploadPrintJobProgress(self, bytes_sent: int, bytes_total: int) -> None: if bytes_total > 0: new_progress = bytes_sent / bytes_total * 100 # Treat upload progress as response. Uploading can take more than 10 seconds, so if we don't, we can get @@ -284,14 +277,14 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._progress_message.setProgress(0) self._progress_message.hide() - def _progressMessageActionTriggered(self, message_id: Optional[str]=None, action_id: Optional[str]=None) -> None: + def _progressMessageActionTriggered(self, message_id: Optional[str] = None, action_id: Optional[str] = None) -> None: if action_id == "Abort": Logger.log("d", "User aborted sending print to remote.") if self._progress_message is not None: self._progress_message.hide() self._compressing_gcode = False self._sending_gcode = False - Application.getInstance().getController().setActiveStage("PrepareStage") + CuraApplication.getInstance().getController().setActiveStage("PrepareStage") # After compressing the sliced model Cura sends data to printer, to stop receiving updates from the request # the "reply" should be disconnected @@ -299,9 +292,9 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): self._latest_reply_handler.disconnect() self._latest_reply_handler = None - def _successMessageActionTriggered(self, message_id: Optional[str]=None, action_id: Optional[str]=None) -> None: + def _successMessageActionTriggered(self, message_id: Optional[str] = None, action_id: Optional[str] = None) -> None: if action_id == "View": - Application.getInstance().getController().setActiveStage("MonitorStage") + CuraApplication.getInstance().getController().setActiveStage("MonitorStage") @pyqtSlot() def openPrintJobControlPanel(self) -> None: @@ -313,19 +306,19 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): Logger.log("d", "Opening printer control panel...") QDesktopServices.openUrl(QUrl("http://" + self._address + "/printers")) - @pyqtProperty("QVariantList", notify=printJobsChanged) - def printJobs(self)-> List[PrintJobOutputModel] : + @pyqtProperty("QVariantList", notify = printJobsChanged) + def printJobs(self)-> List[PrintJobOutputModel]: return self._print_jobs - @pyqtProperty("QVariantList", notify=printJobsChanged) + @pyqtProperty("QVariantList", notify = printJobsChanged) def queuedPrintJobs(self) -> List[PrintJobOutputModel]: return [print_job for print_job in self._print_jobs if print_job.state == "queued"] - @pyqtProperty("QVariantList", notify=printJobsChanged) + @pyqtProperty("QVariantList", notify = printJobsChanged) def activePrintJobs(self) -> List[PrintJobOutputModel]: return [print_job for print_job in self._print_jobs if print_job.assignedPrinter is not None and print_job.state != "queued"] - @pyqtProperty("QVariantList", notify=clusterPrintersChanged) + @pyqtProperty("QVariantList", notify = clusterPrintersChanged) def connectedPrintersTypeCount(self) -> List[Dict[str, str]]: printer_count = {} # type: Dict[str, int] for printer in self._printers: @@ -338,17 +331,17 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): result.append({"machine_type": machine_type, "count": str(printer_count[machine_type])}) return result - @pyqtSlot(int, result=str) + @pyqtSlot(int, result = str) def formatDuration(self, seconds: int) -> str: return Duration(seconds).getDisplayString(DurationFormat.Format.Short) - @pyqtSlot(int, result=str) + @pyqtSlot(int, result = str) def getTimeCompleted(self, time_remaining: int) -> str: current_time = time() datetime_completed = datetime.fromtimestamp(current_time + time_remaining) return "{hour:02d}:{minute:02d}".format(hour=datetime_completed.hour, minute=datetime_completed.minute) - @pyqtSlot(int, result=str) + @pyqtSlot(int, result = str) def getDateCompleted(self, time_remaining: int) -> str: current_time = time() datetime_completed = datetime.fromtimestamp(current_time + time_remaining) @@ -379,8 +372,8 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): def _update(self) -> None: super()._update() - self.get("printers/", on_finished=self._onGetPrintersDataFinished) - self.get("print_jobs/", on_finished=self._onGetPrintJobsFinished) + self.get("printers/", on_finished = self._onGetPrintersDataFinished) + self.get("print_jobs/", on_finished = self._onGetPrintJobsFinished) def _onGetPrintJobsFinished(self, reply: QNetworkReply) -> None: if not checkValidGetReply(reply): @@ -419,7 +412,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): removed_jobs = [print_job for print_job in self._print_jobs if print_job not in print_jobs_seen] for removed_job in removed_jobs: - job_list_changed |= self._removeJob(removed_job) + job_list_changed = job_list_changed or self._removeJob(removed_job) if job_list_changed: self.printJobsChanged.emit() # Do a single emit for all print job changes. @@ -453,27 +446,27 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): if removed_printers or printer_list_changed: self.printersChanged.emit() - def _createPrinterModel(self, data: Dict) -> PrinterOutputModel: - printer = PrinterOutputModel(output_controller=ClusterUM3PrinterOutputController(self), - number_of_extruders=self._number_of_extruders) + def _createPrinterModel(self, data: Dict[str, Any]) -> PrinterOutputModel: + printer = PrinterOutputModel(output_controller = ClusterUM3PrinterOutputController(self), + number_of_extruders = self._number_of_extruders) printer.setCamera(NetworkCamera("http://" + data["ip_address"] + ":8080/?action=stream")) self._printers.append(printer) return printer - def _createPrintJobModel(self, data: Dict) -> PrintJobOutputModel: + def _createPrintJobModel(self, data: Dict[str, Any]) -> PrintJobOutputModel: print_job = PrintJobOutputModel(output_controller=ClusterUM3PrinterOutputController(self), key=data["uuid"], name= data["name"]) print_job.stateChanged.connect(self._printJobStateChanged) self._print_jobs.append(print_job) return print_job - def _updatePrintJob(self, print_job: PrintJobOutputModel, data: Dict) -> None: + def _updatePrintJob(self, print_job: PrintJobOutputModel, data: Dict[str, Any]) -> None: print_job.updateTimeTotal(data["time_total"]) print_job.updateTimeElapsed(data["time_elapsed"]) print_job.updateState(data["status"]) print_job.updateOwner(data["owner"]) - def _updatePrinter(self, printer: PrinterOutputModel, data: Dict) -> None: + def _updatePrinter(self, printer: PrinterOutputModel, data: Dict[str, Any]) -> None: # For some unknown reason the cluster wants UUID for everything, except for sending a job directly to a printer. # Then we suddenly need the unique name. So in order to not have to mess up all the other code, we save a mapping. self._printer_uuid_to_unique_name_mapping[data["uuid"]] = data["unique_name"] @@ -489,7 +482,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): printer.updateKey(data["uuid"]) printer.updateType(data["machine_variant"]) - # Do not store the buildplate information that comes from connect if the current printer has not buildplate information + # Do not store the build plate information that comes from connect if the current printer has not build plate information if "build_plate" in data and machine_definition.getMetaDataEntry("has_variant_buildplates", False): printer.updateBuildplateName(data["build_plate"]["type"]) if not data["enabled"]: @@ -528,7 +521,7 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): brand=brand, color=color, name=name) extruder.updateActiveMaterial(material) - def _removeJob(self, job: PrintJobOutputModel): + def _removeJob(self, job: PrintJobOutputModel) -> bool: if job not in self._print_jobs: return False @@ -539,23 +532,23 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice): return True - def _removePrinter(self, printer: PrinterOutputModel): + def _removePrinter(self, printer: PrinterOutputModel) -> None: self._printers.remove(printer) if self._active_printer == printer: self._active_printer = None self.activePrinterChanged.emit() -def loadJsonFromReply(reply): +def loadJsonFromReply(reply: QNetworkReply) -> Optional[List[Dict[str, Any]]]: try: result = json.loads(bytes(reply.readAll()).decode("utf-8")) except json.decoder.JSONDecodeError: Logger.logException("w", "Unable to decode JSON from reply.") - return + return None return result -def checkValidGetReply(reply): +def checkValidGetReply(reply: QNetworkReply) -> bool: status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) if status_code != 200: @@ -564,7 +557,8 @@ def checkValidGetReply(reply): return True -def findByKey(list, key): +def findByKey(list: List[Union[PrintJobOutputModel, PrinterOutputModel]], key: str) -> Optional[PrintJobOutputModel]: for item in list: if item.key == key: - return item \ No newline at end of file + return item + return None \ No newline at end of file diff --git a/plugins/UM3NetworkPrinting/DiscoverUM3Action.py b/plugins/UM3NetworkPrinting/DiscoverUM3Action.py index 9b25de7f42..c51092ed98 100644 --- a/plugins/UM3NetworkPrinting/DiscoverUM3Action.py +++ b/plugins/UM3NetworkPrinting/DiscoverUM3Action.py @@ -1,43 +1,47 @@ +# Copyright (c) 2018 Ultimaker B.V. +# Cura is released under the terms of the LGPLv3 or higher. + import os.path import time +from typing import Optional from PyQt5.QtCore import pyqtSignal, pyqtProperty, pyqtSlot, QObject -from UM.Application import Application from UM.PluginRegistry import PluginRegistry from UM.Logger import Logger from UM.i18n import i18nCatalog +from cura.CuraApplication import CuraApplication from cura.MachineAction import MachineAction +from .UM3OutputDevicePlugin import UM3OutputDevicePlugin + catalog = i18nCatalog("cura") class DiscoverUM3Action(MachineAction): discoveredDevicesChanged = pyqtSignal() - def __init__(self): + def __init__(self) -> None: super().__init__("DiscoverUM3Action", catalog.i18nc("@action","Connect via Network")) self._qml_url = "DiscoverUM3Action.qml" - self._network_plugin = None + self._network_plugin = None #type: Optional[UM3OutputDevicePlugin] - self.__additional_components_context = None - self.__additional_component = None - self.__additional_components_view = None + self.__additional_components_view = None #type: Optional[QObject] - Application.getInstance().engineCreatedSignal.connect(self._createAdditionalComponentsView) + CuraApplication.getInstance().engineCreatedSignal.connect(self._createAdditionalComponentsView) - self._last_zero_conf_event_time = time.time() + self._last_zero_conf_event_time = time.time() #type: float # Time to wait after a zero-conf service change before allowing a zeroconf reset - self._zero_conf_change_grace_period = 0.25 + self._zero_conf_change_grace_period = 0.25 #type: float @pyqtSlot() def startDiscovery(self): if not self._network_plugin: Logger.log("d", "Starting device discovery.") - self._network_plugin = Application.getInstance().getOutputDeviceManager().getOutputDevicePlugin("UM3NetworkPrinting") + self._network_plugin = CuraApplication.getInstance().getOutputDeviceManager().getOutputDevicePlugin("UM3NetworkPrinting") self._network_plugin.discoveredDevicesChanged.connect(self._onDeviceDiscoveryChanged) self.discoveredDevicesChanged.emit() @@ -93,16 +97,16 @@ class DiscoverUM3Action(MachineAction): return [] @pyqtSlot(str) - def setGroupName(self, group_name): + def setGroupName(self, group_name: str) -> None: Logger.log("d", "Attempting to set the group name of the active machine to %s", group_name) - global_container_stack = Application.getInstance().getGlobalContainerStack() + global_container_stack = CuraApplication.getInstance().getGlobalContainerStack() if global_container_stack: meta_data = global_container_stack.getMetaData() if "connect_group_name" in meta_data: previous_connect_group_name = meta_data["connect_group_name"] global_container_stack.setMetaDataEntry("connect_group_name", group_name) # Find all the places where there is the same group name and change it accordingly - Application.getInstance().getMachineManager().replaceContainersMetadata(key = "connect_group_name", value = previous_connect_group_name, new_value = group_name) + CuraApplication.getInstance().getMachineManager().replaceContainersMetadata(key = "connect_group_name", value = previous_connect_group_name, new_value = group_name) else: global_container_stack.addMetaDataEntry("connect_group_name", group_name) global_container_stack.addMetaDataEntry("hidden", False) @@ -112,9 +116,9 @@ class DiscoverUM3Action(MachineAction): self._network_plugin.reCheckConnections() @pyqtSlot(str) - def setKey(self, key): + def setKey(self, key: str) -> None: Logger.log("d", "Attempting to set the network key of the active machine to %s", key) - global_container_stack = Application.getInstance().getGlobalContainerStack() + global_container_stack = CuraApplication.getInstance().getGlobalContainerStack() if global_container_stack: meta_data = global_container_stack.getMetaData() if "um_network_key" in meta_data: @@ -124,7 +128,7 @@ class DiscoverUM3Action(MachineAction): Logger.log("d", "Removing old authentication id %s for device %s", global_container_stack.getMetaDataEntry("network_authentication_id", None), key) global_container_stack.removeMetaDataEntry("network_authentication_id") global_container_stack.removeMetaDataEntry("network_authentication_key") - Application.getInstance().getMachineManager().replaceContainersMetadata(key = "um_network_key", value = previous_network_key, new_value = key) + CuraApplication.getInstance().getMachineManager().replaceContainersMetadata(key = "um_network_key", value = previous_network_key, new_value = key) else: global_container_stack.addMetaDataEntry("um_network_key", key) @@ -134,7 +138,7 @@ class DiscoverUM3Action(MachineAction): @pyqtSlot(result = str) def getStoredKey(self) -> str: - global_container_stack = Application.getInstance().getGlobalContainerStack() + global_container_stack = CuraApplication.getInstance().getGlobalContainerStack() if global_container_stack: meta_data = global_container_stack.getMetaData() if "um_network_key" in meta_data: @@ -149,12 +153,12 @@ class DiscoverUM3Action(MachineAction): return "" @pyqtSlot(str, result = bool) - def existsKey(self, key) -> bool: - return Application.getInstance().getMachineManager().existNetworkInstances(network_key = key) + def existsKey(self, key: str) -> bool: + return CuraApplication.getInstance().getMachineManager().existNetworkInstances(network_key = key) @pyqtSlot() - def loadConfigurationFromPrinter(self): - machine_manager = Application.getInstance().getMachineManager() + def loadConfigurationFromPrinter(self) -> None: + machine_manager = CuraApplication.getInstance().getMachineManager() hotend_ids = machine_manager.printerOutputDevices[0].hotendIds for index in range(len(hotend_ids)): machine_manager.printerOutputDevices[0].hotendIdChanged.emit(index, hotend_ids[index]) @@ -162,16 +166,16 @@ class DiscoverUM3Action(MachineAction): for index in range(len(material_ids)): machine_manager.printerOutputDevices[0].materialIdChanged.emit(index, material_ids[index]) - def _createAdditionalComponentsView(self): + def _createAdditionalComponentsView(self) -> None: Logger.log("d", "Creating additional ui components for UM3.") # Create networking dialog path = os.path.join(PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"), "UM3InfoComponents.qml") - self.__additional_components_view = Application.getInstance().createQmlComponent(path, {"manager": self}) + self.__additional_components_view = CuraApplication.getInstance().createQmlComponent(path, {"manager": self}) if not self.__additional_components_view: Logger.log("w", "Could not create ui components for UM3.") return # Create extra components - Application.getInstance().addAdditionalComponent("monitorButtons", self.__additional_components_view.findChild(QObject, "networkPrinterConnectButton")) - Application.getInstance().addAdditionalComponent("machinesDetailPane", self.__additional_components_view.findChild(QObject, "networkPrinterConnectionInfo")) + CuraApplication.getInstance().addAdditionalComponent("monitorButtons", self.__additional_components_view.findChild(QObject, "networkPrinterConnectButton")) + CuraApplication.getInstance().addAdditionalComponent("machinesDetailPane", self.__additional_components_view.findChild(QObject, "networkPrinterConnectionInfo")) diff --git a/plugins/UM3NetworkPrinting/LegacyUM3OutputDevice.py b/plugins/UM3NetworkPrinting/LegacyUM3OutputDevice.py index 0111ad5e4f..9b90f8542d 100644 --- a/plugins/UM3NetworkPrinting/LegacyUM3OutputDevice.py +++ b/plugins/UM3NetworkPrinting/LegacyUM3OutputDevice.py @@ -2,6 +2,7 @@ from typing import List, Optional from UM.FileHandler.FileHandler import FileHandler from UM.Scene.SceneNode import SceneNode +from cura.CuraApplication import CuraApplication from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState from cura.PrinterOutput.PrinterOutputModel import PrinterOutputModel from cura.PrinterOutput.PrintJobOutputModel import PrintJobOutputModel @@ -13,12 +14,11 @@ from cura.Settings.ExtruderManager import ExtruderManager from UM.Logger import Logger from UM.Settings.ContainerRegistry import ContainerRegistry -from UM.Application import Application from UM.i18n import i18nCatalog from UM.Message import Message from PyQt5.QtNetwork import QNetworkRequest -from PyQt5.QtCore import QTimer, QCoreApplication +from PyQt5.QtCore import QTimer from PyQt5.QtWidgets import QMessageBox from .LegacyUM3PrinterOutputController import LegacyUM3PrinterOutputController @@ -129,7 +129,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): def connect(self): super().connect() self._setupMessages() - global_container = Application.getInstance().getGlobalContainerStack() + global_container = CuraApplication.getInstance().getGlobalContainerStack() if global_container: self._authentication_id = global_container.getMetaDataEntry("network_authentication_id", None) self._authentication_key = global_container.getMetaDataEntry("network_authentication_key", None) @@ -187,8 +187,8 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): self.writeStarted.emit(self) - gcode_dict = getattr(Application.getInstance().getController().getScene(), "gcode_dict", []) - active_build_plate_id = Application.getInstance().getMultiBuildPlateModel().activeBuildPlate + gcode_dict = getattr(CuraApplication.getInstance().getController().getScene(), "gcode_dict", []) + active_build_plate_id = CuraApplication.getInstance().getMultiBuildPlateModel().activeBuildPlate gcode_list = gcode_dict[active_build_plate_id] if not gcode_list: @@ -207,7 +207,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): for error in errors: detailed_text += error + "\n" - Application.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"), + CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"), text, informative_text, detailed_text, @@ -229,7 +229,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): for warning in warnings: detailed_text += warning + "\n" - Application.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"), + CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"), text, informative_text, detailed_text, @@ -243,7 +243,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): self._startPrint() # Notify the UI that a switch to the print monitor should happen - Application.getInstance().getController().setActiveStage("MonitorStage") + CuraApplication.getInstance().getController().setActiveStage("MonitorStage") def _startPrint(self): Logger.log("i", "Sending print job to printer.") @@ -268,7 +268,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): # Abort was called. return - file_name = "%s.gcode.gz" % Application.getInstance().getPrintInformation().jobName + file_name = "%s.gcode.gz" % CuraApplication.getInstance().getPrintInformation().jobName self.postForm("print_job", "form-data; name=\"file\";filename=\"%s\"" % file_name, compressed_gcode, onFinished=self._onPostPrintJobFinished) @@ -280,7 +280,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): self._progress_message.hide() self._compressing_gcode = False self._sending_gcode = False - Application.getInstance().getController().setActiveStage("PrepareStage") + CuraApplication.getInstance().getController().setActiveStage("PrepareStage") def _onPostPrintJobFinished(self, reply): self._progress_message.hide() @@ -305,7 +305,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): if button == QMessageBox.Yes: self._startPrint() else: - Application.getInstance().getController().setActiveStage("PrepareStage") + CuraApplication.getInstance().getController().setActiveStage("PrepareStage") # For some unknown reason Cura on OSX will hang if we do the call back code # immediately without first returning and leaving QML's event system. @@ -313,7 +313,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): def _checkForErrors(self): errors = [] - print_information = Application.getInstance().getPrintInformation() + print_information = CuraApplication.getInstance().getPrintInformation() if not print_information.materialLengths: Logger.log("w", "There is no material length information. Unable to check for errors.") return errors @@ -333,7 +333,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): def _checkForWarnings(self): warnings = [] - print_information = Application.getInstance().getPrintInformation() + print_information = CuraApplication.getInstance().getPrintInformation() if not print_information.materialLengths: Logger.log("w", "There is no material length information. Unable to check for warnings.") @@ -456,7 +456,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): self._authentication_failed_message.show() def _saveAuthentication(self): - global_container_stack = Application.getInstance().getGlobalContainerStack() + global_container_stack = CuraApplication.getInstance().getGlobalContainerStack() if global_container_stack: if "network_authentication_key" in global_container_stack.getMetaData(): global_container_stack.setMetaDataEntry("network_authentication_key", self._authentication_key) @@ -469,7 +469,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): global_container_stack.addMetaDataEntry("network_authentication_id", self._authentication_id) # Force save so we are sure the data is not lost. - Application.getInstance().saveStack(global_container_stack) + CuraApplication.getInstance().saveStack(global_container_stack) Logger.log("i", "Authentication succeeded for id %s and key %s", self._authentication_id, self._getSafeAuthKey()) else: @@ -500,7 +500,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): self._authentication_id = None self.post("auth/request", - json.dumps({"application": "Cura-" + Application.getInstance().getVersion(), + json.dumps({"application": "Cura-" + CuraApplication.getInstance().getVersion(), "user": self._getUserName()}).encode(), onFinished=self._onRequestAuthenticationFinished) @@ -546,7 +546,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice): "Got status code {status_code} while trying to get printer data".format(status_code=status_code)) def materialHotendChangedMessage(self, callback): - Application.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Sync with your printer"), + CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Sync with your printer"), i18n_catalog.i18nc("@label", "Would you like to use your current printer configuration in Cura?"), i18n_catalog.i18nc("@label",