3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-07 06:57:28 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Add missing function types

Browse source 
The class is typed now. There's some bugs though.

Contributes to issue CURA-5330.
This commit is contained in:
Ghostkeeper 2018-06-01 11:18:58 +02:00
parent 4157636552
commit a55cf0678e
No known key found for this signature in database
GPG key ID: 5252B696FB5E7C7A
1 changed files with 36 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

55
cura/PrinterOutput/NetworkedPrinterOutputDevice.py
View file

@ -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:
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.