mirror of
https://github.com/Ultimaker/Cura.git
synced 2025-07-15 10:47:49 -06:00
Add function typing for PrinterOutputDevice
This causes a lot of typing errors to surface. We'll fix this later though, when we get to it. Contributes to issue CURA-5330.
This commit is contained in:
parent
a55cf0678e
commit
a946a8aaed
1 changed files with 58 additions and 55 deletions
|
@ -1,17 +1,20 @@
|
||||||
# Copyright (c) 2018 Ultimaker B.V.
|
# Copyright (c) 2018 Ultimaker B.V.
|
||||||
# Cura is released under the terms of the LGPLv3 or higher.
|
# Cura is released under the terms of the LGPLv3 or higher.
|
||||||
|
|
||||||
from UM.Decorators import deprecated
|
from UM.Decorators import deprecated
|
||||||
from UM.i18n import i18nCatalog
|
from UM.i18n import i18nCatalog
|
||||||
from UM.OutputDevice.OutputDevice import OutputDevice
|
from UM.OutputDevice.OutputDevice import OutputDevice
|
||||||
from PyQt5.QtCore import pyqtProperty, QObject, QTimer, pyqtSignal, QVariant
|
from PyQt5.QtCore import pyqtProperty, QObject, QTimer, pyqtSignal
|
||||||
from PyQt5.QtWidgets import QMessageBox
|
from PyQt5.QtWidgets import QMessageBox
|
||||||
|
|
||||||
from UM.Logger import Logger
|
from UM.Logger import Logger
|
||||||
|
from UM.FileHandler.FileHandler import FileHandler #For typing.
|
||||||
|
from UM.Scene.SceneNode import SceneNode #For typing.
|
||||||
from UM.Signal import signalemitter
|
from UM.Signal import signalemitter
|
||||||
from UM.Application import Application
|
from UM.Application import Application
|
||||||
|
|
||||||
from enum import IntEnum # For the connection state tracking.
|
from enum import IntEnum # For the connection state tracking.
|
||||||
from typing import List, Optional
|
from typing import Callable, List, Optional
|
||||||
|
|
||||||
MYPY = False
|
MYPY = False
|
||||||
if MYPY:
|
if MYPY:
|
||||||
|
@ -20,6 +23,16 @@ if MYPY:
|
||||||
|
|
||||||
i18n_catalog = i18nCatalog("cura")
|
i18n_catalog = i18nCatalog("cura")
|
||||||
|
|
||||||
|
|
||||||
|
## The current processing state of the backend.
|
||||||
|
class ConnectionState(IntEnum):
|
||||||
|
closed = 0
|
||||||
|
connecting = 1
|
||||||
|
connected = 2
|
||||||
|
busy = 3
|
||||||
|
error = 4
|
||||||
|
|
||||||
|
|
||||||
## Printer output device adds extra interface options on top of output device.
|
## Printer output device adds extra interface options on top of output device.
|
||||||
#
|
#
|
||||||
# The assumption is made the printer is a FDM printer.
|
# The assumption is made the printer is a FDM printer.
|
||||||
|
@ -47,38 +60,37 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
# Signal to indicate that the configuration of one of the printers has changed.
|
# Signal to indicate that the configuration of one of the printers has changed.
|
||||||
uniqueConfigurationsChanged = pyqtSignal()
|
uniqueConfigurationsChanged = pyqtSignal()
|
||||||
|
|
||||||
def __init__(self, device_id, parent = None):
|
def __init__(self, device_id: str, parent: QObject = None) -> None:
|
||||||
super().__init__(device_id = device_id, parent = parent)
|
super().__init__(device_id = device_id, parent = parent)
|
||||||
|
|
||||||
self._printers = [] # type: List[PrinterOutputModel]
|
self._printers = [] # type: List[PrinterOutputModel]
|
||||||
self._unique_configurations = [] # type: List[ConfigurationModel]
|
self._unique_configurations = [] # type: List[ConfigurationModel]
|
||||||
|
|
||||||
self._monitor_view_qml_path = ""
|
self._monitor_view_qml_path = "" #type: str
|
||||||
self._monitor_component = None
|
self._monitor_component = None #type: Optional[QObject]
|
||||||
self._monitor_item = None
|
self._monitor_item = None #type: Optional[QObject]
|
||||||
|
|
||||||
self._control_view_qml_path = ""
|
self._control_view_qml_path = "" #type: str
|
||||||
self._control_component = None
|
self._control_component = None #type: Optional[QObject]
|
||||||
self._control_item = None
|
self._control_item = None #type: Optional[QObject]
|
||||||
|
|
||||||
self._qml_context = None
|
self._accepts_commands = False #type: bool
|
||||||
self._accepts_commands = False
|
|
||||||
|
|
||||||
self._update_timer = QTimer()
|
self._update_timer = QTimer() #type: QTimer
|
||||||
self._update_timer.setInterval(2000) # TODO; Add preference for update interval
|
self._update_timer.setInterval(2000) # TODO; Add preference for update interval
|
||||||
self._update_timer.setSingleShot(False)
|
self._update_timer.setSingleShot(False)
|
||||||
self._update_timer.timeout.connect(self._update)
|
self._update_timer.timeout.connect(self._update)
|
||||||
|
|
||||||
self._connection_state = ConnectionState.closed
|
self._connection_state = ConnectionState.closed #type: ConnectionState
|
||||||
|
|
||||||
self._firmware_name = None
|
self._firmware_name = None #type: Optional[str]
|
||||||
self._address = ""
|
self._address = "" #type: str
|
||||||
self._connection_text = ""
|
self._connection_text = "" #type: str
|
||||||
self.printersChanged.connect(self._onPrintersChanged)
|
self.printersChanged.connect(self._onPrintersChanged)
|
||||||
Application.getInstance().getOutputDeviceManager().outputDevicesChanged.connect(self._updateUniqueConfigurations)
|
Application.getInstance().getOutputDeviceManager().outputDevicesChanged.connect(self._updateUniqueConfigurations)
|
||||||
|
|
||||||
@pyqtProperty(str, notify = connectionTextChanged)
|
@pyqtProperty(str, notify = connectionTextChanged)
|
||||||
def address(self):
|
def address(self) -> str:
|
||||||
return self._address
|
return self._address
|
||||||
|
|
||||||
def setConnectionText(self, connection_text):
|
def setConnectionText(self, connection_text):
|
||||||
|
@ -87,36 +99,36 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
self.connectionTextChanged.emit()
|
self.connectionTextChanged.emit()
|
||||||
|
|
||||||
@pyqtProperty(str, constant=True)
|
@pyqtProperty(str, constant=True)
|
||||||
def connectionText(self):
|
def connectionText(self) -> str:
|
||||||
return self._connection_text
|
return self._connection_text
|
||||||
|
|
||||||
def materialHotendChangedMessage(self, callback):
|
def materialHotendChangedMessage(self, callback: Callable[[int], None]) -> None:
|
||||||
Logger.log("w", "materialHotendChangedMessage needs to be implemented, returning 'Yes'")
|
Logger.log("w", "materialHotendChangedMessage needs to be implemented, returning 'Yes'")
|
||||||
callback(QMessageBox.Yes)
|
callback(QMessageBox.Yes)
|
||||||
|
|
||||||
def isConnected(self):
|
def isConnected(self) -> bool:
|
||||||
return self._connection_state != ConnectionState.closed and self._connection_state != ConnectionState.error
|
return self._connection_state != ConnectionState.closed and self._connection_state != ConnectionState.error
|
||||||
|
|
||||||
def setConnectionState(self, connection_state):
|
def setConnectionState(self, connection_state: ConnectionState) -> None:
|
||||||
if self._connection_state != connection_state:
|
if self._connection_state != connection_state:
|
||||||
self._connection_state = connection_state
|
self._connection_state = connection_state
|
||||||
self.connectionStateChanged.emit(self._id)
|
self.connectionStateChanged.emit(self._id)
|
||||||
|
|
||||||
@pyqtProperty(str, notify = connectionStateChanged)
|
@pyqtProperty(str, notify = connectionStateChanged)
|
||||||
def connectionState(self):
|
def connectionState(self) -> ConnectionState:
|
||||||
return self._connection_state
|
return self._connection_state
|
||||||
|
|
||||||
def _update(self):
|
def _update(self) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def _getPrinterByKey(self, key) -> Optional["PrinterOutputModel"]:
|
def _getPrinterByKey(self, key: str) -> Optional["PrinterOutputModel"]:
|
||||||
for printer in self._printers:
|
for printer in self._printers:
|
||||||
if printer.key == key:
|
if printer.key == key:
|
||||||
return printer
|
return printer
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def requestWrite(self, nodes, file_name = None, filter_by_machine = False, file_handler = None, **kwargs):
|
def requestWrite(self, nodes: List[SceneNode], file_name: Optional[str] = None, limit_mimetypes: bool = False, file_handler: Optional[FileHandler] = None, **kwargs: str) -> None:
|
||||||
raise NotImplementedError("requestWrite needs to be implemented")
|
raise NotImplementedError("requestWrite needs to be implemented")
|
||||||
|
|
||||||
@pyqtProperty(QObject, notify = printersChanged)
|
@pyqtProperty(QObject, notify = printersChanged)
|
||||||
|
@ -126,11 +138,11 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@pyqtProperty("QVariantList", notify = printersChanged)
|
@pyqtProperty("QVariantList", notify = printersChanged)
|
||||||
def printers(self):
|
def printers(self) -> List["PrinterOutputModel"]:
|
||||||
return self._printers
|
return self._printers
|
||||||
|
|
||||||
@pyqtProperty(QObject, constant=True)
|
@pyqtProperty(QObject, constant = True)
|
||||||
def monitorItem(self):
|
def monitorItem(self) -> QObject:
|
||||||
# Note that we specifically only check if the monitor component is created.
|
# Note that we specifically only check if the monitor component is created.
|
||||||
# It could be that it failed to actually create the qml item! If we check if the item was created, it will try to
|
# It could be that it failed to actually create the qml item! If we check if the item was created, it will try to
|
||||||
# create the item (and fail) every time.
|
# create the item (and fail) every time.
|
||||||
|
@ -138,19 +150,19 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
self._createMonitorViewFromQML()
|
self._createMonitorViewFromQML()
|
||||||
return self._monitor_item
|
return self._monitor_item
|
||||||
|
|
||||||
@pyqtProperty(QObject, constant=True)
|
@pyqtProperty(QObject, constant = True)
|
||||||
def controlItem(self):
|
def controlItem(self) -> QObject:
|
||||||
if not self._control_component:
|
if not self._control_component:
|
||||||
self._createControlViewFromQML()
|
self._createControlViewFromQML()
|
||||||
return self._control_item
|
return self._control_item
|
||||||
|
|
||||||
def _createControlViewFromQML(self):
|
def _createControlViewFromQML(self) -> None:
|
||||||
if not self._control_view_qml_path:
|
if not self._control_view_qml_path:
|
||||||
return
|
return
|
||||||
if self._control_item is None:
|
if self._control_item is None:
|
||||||
self._control_item = Application.getInstance().createQmlComponent(self._control_view_qml_path, {"OutputDevice": self})
|
self._control_item = Application.getInstance().createQmlComponent(self._control_view_qml_path, {"OutputDevice": self})
|
||||||
|
|
||||||
def _createMonitorViewFromQML(self):
|
def _createMonitorViewFromQML(self) -> None:
|
||||||
if not self._monitor_view_qml_path:
|
if not self._monitor_view_qml_path:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -158,29 +170,29 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
self._monitor_item = Application.getInstance().createQmlComponent(self._monitor_view_qml_path, {"OutputDevice": self})
|
self._monitor_item = Application.getInstance().createQmlComponent(self._monitor_view_qml_path, {"OutputDevice": self})
|
||||||
|
|
||||||
## Attempt to establish connection
|
## Attempt to establish connection
|
||||||
def connect(self):
|
def connect(self) -> None:
|
||||||
self.setConnectionState(ConnectionState.connecting)
|
self.setConnectionState(ConnectionState.connecting)
|
||||||
self._update_timer.start()
|
self._update_timer.start()
|
||||||
|
|
||||||
## Attempt to close the connection
|
## Attempt to close the connection
|
||||||
def close(self):
|
def close(self) -> None:
|
||||||
self._update_timer.stop()
|
self._update_timer.stop()
|
||||||
self.setConnectionState(ConnectionState.closed)
|
self.setConnectionState(ConnectionState.closed)
|
||||||
|
|
||||||
## Ensure that close gets called when object is destroyed
|
## Ensure that close gets called when object is destroyed
|
||||||
def __del__(self):
|
def __del__(self) -> None:
|
||||||
self.close()
|
self.close()
|
||||||
|
|
||||||
@pyqtProperty(bool, notify=acceptsCommandsChanged)
|
@pyqtProperty(bool, notify = acceptsCommandsChanged)
|
||||||
def acceptsCommands(self):
|
def acceptsCommands(self) -> bool:
|
||||||
return self._accepts_commands
|
return self._accepts_commands
|
||||||
|
|
||||||
@deprecated("Please use the protected function instead", "3.2")
|
@deprecated("Please use the protected function instead", "3.2")
|
||||||
def setAcceptsCommands(self, accepts_commands):
|
def setAcceptsCommands(self, accepts_commands: bool) -> None:
|
||||||
self._setAcceptsCommands(accepts_commands)
|
self._setAcceptsCommands(accepts_commands)
|
||||||
|
|
||||||
## Set a flag to signal the UI that the printer is not (yet) ready to receive commands
|
## Set a flag to signal the UI that the printer is not (yet) ready to receive commands
|
||||||
def _setAcceptsCommands(self, accepts_commands):
|
def _setAcceptsCommands(self, accepts_commands: bool) -> None:
|
||||||
if self._accepts_commands != accepts_commands:
|
if self._accepts_commands != accepts_commands:
|
||||||
self._accepts_commands = accepts_commands
|
self._accepts_commands = accepts_commands
|
||||||
|
|
||||||
|
@ -188,15 +200,15 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
|
|
||||||
# Returns the unique configurations of the printers within this output device
|
# Returns the unique configurations of the printers within this output device
|
||||||
@pyqtProperty("QVariantList", notify = uniqueConfigurationsChanged)
|
@pyqtProperty("QVariantList", notify = uniqueConfigurationsChanged)
|
||||||
def uniqueConfigurations(self):
|
def uniqueConfigurations(self) -> List[ConfigurationModel]:
|
||||||
return self._unique_configurations
|
return self._unique_configurations
|
||||||
|
|
||||||
def _updateUniqueConfigurations(self):
|
def _updateUniqueConfigurations(self) -> None:
|
||||||
self._unique_configurations = list(set([printer.printerConfiguration for printer in self._printers if printer.printerConfiguration is not None]))
|
self._unique_configurations = list(set([printer.printerConfiguration for printer in self._printers if printer.printerConfiguration is not None]))
|
||||||
self._unique_configurations.sort(key = lambda k: k.printerType)
|
self._unique_configurations.sort(key = lambda k: k.printerType)
|
||||||
self.uniqueConfigurationsChanged.emit()
|
self.uniqueConfigurationsChanged.emit()
|
||||||
|
|
||||||
def _onPrintersChanged(self):
|
def _onPrintersChanged(self) -> None:
|
||||||
for printer in self._printers:
|
for printer in self._printers:
|
||||||
printer.configurationChanged.connect(self._updateUniqueConfigurations)
|
printer.configurationChanged.connect(self._updateUniqueConfigurations)
|
||||||
|
|
||||||
|
@ -205,21 +217,12 @@ class PrinterOutputDevice(QObject, OutputDevice):
|
||||||
|
|
||||||
## Set the device firmware name
|
## Set the device firmware name
|
||||||
#
|
#
|
||||||
# \param name \type{str} The name of the firmware.
|
# \param name The name of the firmware.
|
||||||
def _setFirmwareName(self, name):
|
def _setFirmwareName(self, name: str) -> None:
|
||||||
self._firmware_name = name
|
self._firmware_name = name
|
||||||
|
|
||||||
## Get the name of device firmware
|
## Get the name of device firmware
|
||||||
#
|
#
|
||||||
# This name can be used to define device type
|
# This name can be used to define device type
|
||||||
def getFirmwareName(self):
|
def getFirmwareName(self) -> str:
|
||||||
return self._firmware_name
|
return self._firmware_name
|
||||||
|
|
||||||
|
|
||||||
## The current processing state of the backend.
|
|
||||||
class ConnectionState(IntEnum):
|
|
||||||
closed = 0
|
|
||||||
connecting = 1
|
|
||||||
connected = 2
|
|
||||||
busy = 3
|
|
||||||
error = 4
|
|
Loading…
Add table
Add a link
Reference in a new issue