diff --git a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py index 607d23aa53..84e186dacb 100644 --- a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py +++ b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py @@ -9,7 +9,7 @@ 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 time import time -from typing import Callable, Any, Optional +from typing import Callable, Any, Optional, Dict, Tuple from enum import IntEnum from typing import List @@ -27,45 +27,45 @@ class AuthState(IntEnum): class NetworkedPrinterOutputDevice(PrinterOutputDevice): authenticationStateChanged = pyqtSignal() - def __init__(self, device_id, address: str, properties, parent = None): + def __init__(self, device_id, address: str, properties, parent = None) -> None: super().__init__(device_id = device_id, parent = parent) - self._manager = None - self._last_manager_create_time = None + self._manager = None # type: QNetworkAccessManager + self._last_manager_create_time = None # type: float self._recreate_network_manager_time = 30 self._timeout_time = 10 # After how many seconds of no response should a timeout occur? - self._last_response_time = None - self._last_request_time = None + self._last_response_time = None # type: float + self._last_request_time = None # type: float self._api_prefix = "" self._address = address self._properties = properties self._user_agent = "%s/%s " % (Application.getInstance().getApplicationName(), Application.getInstance().getVersion()) - self._onFinishedCallbacks = {} + self._onFinishedCallbacks = {} # type: Dict[str, Callable[[QNetworkReply], None]] self._authentication_state = AuthState.NotAuthenticated - self._cached_multiparts = {} + self._cached_multiparts = {} # type: Dict[int, Tuple[QHttpMultiPart, QNetworkReply]] self._sending_gcode = False self._compressing_gcode = False - self._gcode = [] + self._gcode = [] # type: List[str] - self._connection_state_before_timeout = None + self._connection_state_before_timeout = None # type: Optional[ConnectionState] - def requestWrite(self, nodes, file_name=None, filter_by_machine=False, file_handler=None, **kwargs): + def requestWrite(self, nodes, file_name=None, filter_by_machine=False, file_handler=None, **kwargs) -> None: raise NotImplementedError("requestWrite needs to be implemented") - def setAuthenticationState(self, authentication_state): + def setAuthenticationState(self, authentication_state) -> None: if self._authentication_state != authentication_state: self._authentication_state = authentication_state self.authenticationStateChanged.emit() @pyqtProperty(int, notify=authenticationStateChanged) - def authenticationState(self): + def authenticationState(self) -> int: return self._authentication_state - def _compressDataAndNotifyQt(self, data_to_append): + def _compressDataAndNotifyQt(self, data_to_append: str) -> bytes: compressed_data = gzip.compress(data_to_append.encode("utf-8")) self._progress_message.setProgress(-1) # Tickle the message so that it's clear that it's still being used. QCoreApplication.processEvents() # Ensure that the GUI does not freeze. @@ -75,7 +75,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): self._last_response_time = time() return compressed_data - def _compressGCode(self): + def _compressGCode(self) -> Optional[bytes]: self._compressing_gcode = True ## Mash the data into single string @@ -87,7 +87,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if not self._compressing_gcode: self._progress_message.hide() # Stop trying to zip / send as abort was called. - return + return None batched_line += line # if the gcode was read from a gcode file, self._gcode will be a list of all lines in that file. # Compressing line by line in this case is extremely slow, so we need to batch them. @@ -103,7 +103,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): self._compressing_gcode = False return byte_array_file_data - def _update(self): + def _update(self) -> bool: if self._last_response_time: time_since_last_response = time() - self._last_response_time else: @@ -135,7 +135,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): return True - def _createEmptyRequest(self, target, content_type: Optional[str] = "application/json"): + def _createEmptyRequest(self, target, 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: @@ -143,7 +143,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): request.setHeader(QNetworkRequest.UserAgentHeader, self._user_agent) return request - def _createFormPart(self, content_header, data, content_type = None): + def _createFormPart(self, content_header, data, content_type = None) -> QHttpPart: part = QHttpPart() if not content_header.startswith("form-data;"): @@ -158,18 +158,18 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): ## Convenience function to get the username from the OS. # The code was copied from the getpass module, as we try to use as little dependencies as possible. - def _getUserName(self): + def _getUserName(self) -> str: for name in ("LOGNAME", "USER", "LNAME", "USERNAME"): user = os.environ.get(name) if user: return user return "Unknown User" # Couldn't find out username. - def _clearCachedMultiPart(self, reply): + def _clearCachedMultiPart(self, reply: QNetworkReply) -> None: if id(reply) in self._cached_multiparts: del self._cached_multiparts[id(reply)] - def put(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]): + def put(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) @@ -178,7 +178,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if onFinished is not None: self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished - def get(self, target: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]): + def get(self, target: str, onFinished: Optional[Callable[[QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) @@ -187,13 +187,12 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if onFinished is not None: self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished - def delete(self, target: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]]): + def delete(self, target: str, onFinished: Optional[Callable[[QNetworkReply], None]]) -> None: if self._manager is None: self._createNetworkManager() self._last_request_time = time() - pass - def post(self, target: str, data: str, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None): + def post(self, target: str, data: str, onFinished: Optional[Callable[[QNetworkReply], None]], onProgress: Callable = None) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target) @@ -204,7 +203,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if onFinished is not None: self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished - def postFormWithParts(self, target:str, parts: List[QHttpPart], onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None): + def postFormWithParts(self, target:str, parts: List[QHttpPart], onFinished: Optional[Callable[[QNetworkReply], None]], onProgress: Callable = None) -> None: if self._manager is None: self._createNetworkManager() request = self._createEmptyRequest(target, content_type=None) @@ -223,17 +222,17 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if onFinished is not None: self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished - def postForm(self, target: str, header_data: str, body_data: bytes, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None): + def postForm(self, target: str, header_data: str, body_data: bytes, onFinished: Optional[Callable[[QNetworkReply], None]], onProgress: 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) - def _onAuthenticationRequired(self, reply, authenticator): + def _onAuthenticationRequired(self, reply, authenticator) -> None: Logger.log("w", "Request to {url} required authentication, which was not implemented".format(url = reply.url().toString())) - def _createNetworkManager(self): + def _createNetworkManager(self) -> None: Logger.log("d", "Creating network manager") if self._manager: self._manager.finished.disconnect(self.__handleOnFinished) @@ -246,7 +245,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): self._manager.authenticationRequired.connect(self._onAuthenticationRequired) #self._manager.networkAccessibleChanged.connect(self._onNetworkAccesibleChanged) # for debug purposes - def __handleOnFinished(self, reply: QNetworkReply): + def __handleOnFinished(self, reply: QNetworkReply) -> None: # Due to garbage collection, we need to cache certain bits of post operations. # As we don't want to keep them around forever, delete them if we get a reply. if reply.operation() == QNetworkAccessManager.PostOperation: @@ -269,10 +268,10 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): Logger.logException("w", "something went wrong with callback") @pyqtSlot(str, result=str) - def getProperty(self, key): - key = key.encode("utf-8") - if key in self._properties: - return self._properties.get(key, b"").decode("utf-8") + def getProperty(self, key: str) -> str: + bytes_key = key.encode("utf-8") + if bytes_key in self._properties: + return self._properties.get(bytes_key, b"").decode("utf-8") else: return "" @@ -282,25 +281,25 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): ## Get the unique key of this machine # \return key String containing the key of the machine. @pyqtProperty(str, constant=True) - def key(self): + def key(self) -> str: return self._id ## The IP address of the printer. @pyqtProperty(str, constant=True) - def address(self): + 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) - def name(self): + 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) - def firmwareVersion(self): + def firmwareVersion(self) -> str: return self._properties.get(b"firmware_version", b"").decode("utf-8") ## IPadress of this printer @pyqtProperty(str, constant=True) - def ipAddress(self): - return self._address \ No newline at end of file + def ipAddress(self) -> str: + return self._address