Fill out the type hints in NetworkedPrinterOutputDevice.py

CL-541

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._manager = None    # type: QNetworkAccessManager
-        self._last_manager_create_time = None
+        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_response_time = None     # type: float
-        self._last_request_time = None
+        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):
+    def getProperty(self, key: str) -> str:
-        key = key.encode("utf-8")
+        bytes_key = key.encode("utf-8")
-        if key in self._properties:
+        if bytes_key in self._properties:
-            return self._properties.get(key, b"").decode("utf-8")
+            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):
+    def ipAddress(self) -> str:
-        return self._address
+        return self._address