diff --git a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py index 86af368a86..ff9fe4477a 100644 --- a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py +++ b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py @@ -2,17 +2,18 @@ # Cura is released under the terms of the LGPLv3 or higher. from UM.Application import Application +from UM.FileHandler.FileHandler import FileHandler #For typing. from UM.Logger import Logger +from UM.Scene.SceneNode import SceneNode #For typing. from cura.CuraApplication import CuraApplication from cura.PrinterOutputDevice import PrinterOutputDevice, ConnectionState -from PyQt5.QtNetwork import QHttpMultiPart, QHttpPart, QNetworkRequest, QNetworkAccessManager, QNetworkReply -from PyQt5.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, pyqtSignal, QUrl, QCoreApplication +from PyQt5.QtNetwork import QHttpMultiPart, QHttpPart, QNetworkRequest, QNetworkAccessManager, QNetworkReply, QAuthenticator +from PyQt5.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, QUrl, QCoreApplication from time import time -from typing import Callable, Any, Optional, Dict, Tuple +from typing import Any, Callable, Dict, List, Optional from enum import IntEnum -from typing import List import os # To get the username import gzip @@ -28,7 +29,7 @@ class AuthState(IntEnum): class NetworkedPrinterOutputDevice(PrinterOutputDevice): authenticationStateChanged = pyqtSignal() - def __init__(self, device_id, address: str, properties, parent = None) -> None: + def __init__(self, device_id, address: str, properties: Dict[bytes, bytes], parent = None) -> None: super().__init__(device_id = device_id, parent = parent) self._manager = None # type: QNetworkAccessManager self._last_manager_create_time = None # type: float @@ -68,16 +69,16 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): self._printer_type = value break - def requestWrite(self, nodes, file_name=None, filter_by_machine=False, file_handler=None, **kwargs) -> None: + def requestWrite(self, nodes: List[SceneNode], file_name: str = None, filter_by_machine: bool = False, file_handler: FileHandler = None, **kwargs: Dict[str, Any]) -> None: raise NotImplementedError("requestWrite needs to be implemented") - def setAuthenticationState(self, authentication_state) -> None: + def setAuthenticationState(self, authentication_state: AuthState) -> None: if self._authentication_state != authentication_state: self._authentication_state = authentication_state self.authenticationStateChanged.emit() - @pyqtProperty(int, notify=authenticationStateChanged) - def authenticationState(self) -> int: + @pyqtProperty(int, notify = authenticationStateChanged) + def authenticationState(self) -> AuthState: return self._authentication_state def _compressDataAndNotifyQt(self, data_to_append: str) -> bytes: @@ -154,7 +155,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): return True - def _createEmptyRequest(self, target, content_type: Optional[str] = "application/json") -> QNetworkRequest: + def _createEmptyRequest(self, target: str, content_type: Optional[str] = "application/json") -> QNetworkRequest: url = QUrl("http://" + self._address + self._api_prefix + target) request = QNetworkRequest(url) if content_type is not None: @@ -162,7 +163,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): request.setHeader(QNetworkRequest.UserAgentHeader, self._user_agent) return request - def _createFormPart(self, content_header, data, content_type = None) -> QHttpPart: + def _createFormPart(self, content_header: str, data: str, content_type: Optional[str] = None) -> QHttpPart: part = QHttpPart() if not content_header.startswith("form-data;"): @@ -188,33 +189,33 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if reply in self._kept_alive_multiparts: del self._kept_alive_multiparts[reply] - def put(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: + def put(self, target: str, data: str, on_finished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.put(request, data.encode()) - self._registerOnFinishedCallback(reply, onFinished) + self._registerOnFinishedCallback(reply, on_finished) - def get(self, target: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: + def get(self, target: str, on_finished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.get(request) - self._registerOnFinishedCallback(reply, onFinished) + self._registerOnFinishedCallback(reply, on_finished) - def post(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None) -> None: + def post(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]], on_progress: Callable = None) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) self._last_request_time = time() reply = self._manager.post(request, data) - if onProgress is not None: - reply.uploadProgress.connect(onProgress) + if on_progress is not None: + reply.uploadProgress.connect(on_progress) self._registerOnFinishedCallback(reply, onFinished) - def postFormWithParts(self, target:str, parts: List[QHttpPart], onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None) -> None: + def postFormWithParts(self, target:str, parts: List[QHttpPart], on_finished: Optional[Callable[[Any, QNetworkReply], None]], on_progress: Callable = None) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target, content_type=None) @@ -228,20 +229,20 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): self._kept_alive_multiparts[reply] = multi_post_part - if onProgress is not None: - reply.uploadProgress.connect(onProgress) - self._registerOnFinishedCallback(reply, onFinished) + if on_progress is not None: + reply.uploadProgress.connect(on_progress) + self._registerOnFinishedCallback(reply, on_finished) return reply - def postForm(self, target: str, header_data: str, body_data: bytes, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None) -> None: + def postForm(self, target: str, header_data: str, body_data: bytes, on_finished: Optional[Callable[[Any, QNetworkReply], None]], on_progress: Callable = None) -> None: post_part = QHttpPart() post_part.setHeader(QNetworkRequest.ContentDispositionHeader, header_data) post_part.setBody(body_data) - self.postFormWithParts(target, [post_part], onFinished, onProgress) + self.postFormWithParts(target, [post_part], on_finished, on_progress) - def _onAuthenticationRequired(self, reply, authenticator) -> None: + def _onAuthenticationRequired(self, reply: QNetworkReply, authenticator: QAuthenticator) -> None: Logger.log("w", "Request to {url} required authentication, which was not implemented".format(url = reply.url().toString())) def _createNetworkManager(self) -> None: @@ -258,9 +259,9 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): machine_manager = CuraApplication.getInstance().getMachineManager() machine_manager.checkCorrectGroupName(self.getId(), self.name) - def _registerOnFinishedCallback(self, reply: QNetworkReply, onFinished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: - if onFinished is not None: - self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished + def _registerOnFinishedCallback(self, reply: QNetworkReply, on_finished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: + if on_finished is not None: + self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = on_finished def __handleOnFinished(self, reply: QNetworkReply) -> None: # Due to garbage collection, we need to cache certain bits of post operations. @@ -297,30 +298,30 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): ## Get the unique key of this machine # \return key String containing the key of the machine. - @pyqtProperty(str, constant=True) + @pyqtProperty(str, constant = True) def key(self) -> str: return self._id ## The IP address of the printer. - @pyqtProperty(str, constant=True) + @pyqtProperty(str, constant = True) def address(self) -> str: return self._properties.get(b"address", b"").decode("utf-8") ## Name of the printer (as returned from the ZeroConf properties) - @pyqtProperty(str, constant=True) + @pyqtProperty(str, constant = True) def name(self) -> str: return self._properties.get(b"name", b"").decode("utf-8") ## Firmware version (as returned from the ZeroConf properties) - @pyqtProperty(str, constant=True) + @pyqtProperty(str, constant = True) def firmwareVersion(self) -> str: return self._properties.get(b"firmware_version", b"").decode("utf-8") - @pyqtProperty(str, constant=True) + @pyqtProperty(str, constant = True) def printerType(self) -> str: return self._printer_type - ## IPadress of this printer - @pyqtProperty(str, constant=True) + ## IP adress of this printer + @pyqtProperty(str, constant = True) def ipAddress(self) -> str: return self._address