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

Restructure codebase - part 1

Browse source
This commit is contained in:
ChrisTerBeke 2019-07-26 15:07:52 +02:00
parent 87517a77c2
commit 3c1b377308
46 changed files with 898 additions and 1725 deletions
Download patch file Download diff file Expand all files Collapse all files

5
plugins/UM3NetworkPrinting/__init__.py
View file

@ -1,6 +1,6 @@
# Copyright (c) 2017 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from .src import DiscoverUM3Action
# from .src import DiscoverUM3Action
from .src import UM3OutputDevicePlugin
@ -10,6 +10,5 @@ def getMetaData():
def register(app):
return {
"output_device": UM3OutputDevicePlugin.UM3OutputDevicePlugin(),
"machine_action": DiscoverUM3Action.DiscoverUM3Action()
"output_device": UM3OutputDevicePlugin.UM3OutputDevicePlugin()
}

4
plugins/UM3NetworkPrinting/plugin.json
View file

@ -1,7 +1,7 @@
{
"name": "UM3 Network Connection",
"name": "Ultimaker Network Connection",
"author": "Ultimaker B.V.",
"description": "Manages network connections to Ultimaker 3 printers.",
"description": "Manages network connections to Ultimaker networked printers.",
"version": "1.0.1",
"api": "6.0",
"i18n-catalog": "cura"

16
plugins/UM3NetworkPrinting/src/Cloud/CloudApiClient.py
View file

@ -12,17 +12,17 @@ from UM.Logger import Logger
from cura import UltimakerCloudAuthentication
from cura.API import Account
from .ToolPathUploader import ToolPathUploader
from ..Models import BaseModel
from .Models.CloudClusterResponse import CloudClusterResponse
from .Models.CloudError import CloudError
from .Models.CloudClusterStatus import CloudClusterStatus
from .Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from .Models.CloudPrintResponse import CloudPrintResponse
from .Models.CloudPrintJobResponse import CloudPrintJobResponse
from ..Models.BaseModel import BaseModel
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudError import CloudError
from plugins.UM3NetworkPrinting.src.Models.CloudClusterStatus import CloudClusterStatus
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from plugins.UM3NetworkPrinting.src.Models.CloudPrintResponse import CloudPrintResponse
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobResponse import CloudPrintJobResponse
## The generic type variable used to document the methods below.
CloudApiClientModel = TypeVar("CloudApiClientModel", bound = BaseModel)
CloudApiClientModel = TypeVar("CloudApiClientModel", bound=BaseModel)
## The cloud API client is responsible for handling the requests and responses from the cloud.

16
plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDevice.py
View file

@ -25,16 +25,16 @@ from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from .CloudOutputController import CloudOutputController
from ..MeshFormatHandler import MeshFormatHandler
from ..UM3PrintJobOutputModel import UM3PrintJobOutputModel
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
from .CloudProgressMessage import CloudProgressMessage
from .CloudApiClient import CloudApiClient
from .Models.CloudClusterResponse import CloudClusterResponse
from .Models.CloudClusterStatus import CloudClusterStatus
from .Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from .Models.CloudPrintResponse import CloudPrintResponse
from .Models.CloudPrintJobResponse import CloudPrintJobResponse
from .Models.CloudClusterPrinterStatus import CloudClusterPrinterStatus
from .Models.CloudClusterPrintJobStatus import CloudClusterPrintJobStatus
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudClusterStatus import CloudClusterStatus
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from plugins.UM3NetworkPrinting.src.Models.CloudPrintResponse import CloudPrintResponse
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobResponse import CloudPrintJobResponse
from plugins.UM3NetworkPrinting.src.Models.CloudClusterPrinterStatus import CloudClusterPrinterStatus
from plugins.UM3NetworkPrinting.src.Models.CloudClusterPrintJobStatus import CloudClusterPrintJobStatus
from .Utils import formatDateCompleted, formatTimeCompleted
I18N_CATALOG = i18nCatalog("cura")

22
plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py
View file

@ -13,8 +13,8 @@ from cura.CuraApplication import CuraApplication
from cura.Settings.GlobalStack import GlobalStack
from .CloudApiClient import CloudApiClient
from .CloudOutputDevice import CloudOutputDevice
from .Models.CloudClusterResponse import CloudClusterResponse
from .Models.CloudError import CloudError
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudError import CloudError
from .Utils import findChanges
@ -52,7 +52,11 @@ class CloudOutputDeviceManager:
self._running = False
# Called when the uses logs in or out
## Force refreshing connections.
def refreshConnections(self) -> None:
pass
## Called when the uses logs in or out
def _onLoginStateChanged(self, is_logged_in: bool) -> None:
Logger.log("d", "Log in state changed to %s", is_logged_in)
if is_logged_in:
@ -66,12 +70,12 @@ class CloudOutputDeviceManager:
# Notify that all clusters have disappeared
self._onGetRemoteClustersFinished([])
## Gets all remote clusters from the API.
## Gets all remote clusters from the API.
def _getRemoteClusters(self) -> None:
Logger.log("d", "Retrieving remote clusters")
self._api.getClusters(self._onGetRemoteClustersFinished)
## Callback for when the request for getting the clusters. is finished.
## Callback for when the request for getting the clusters. is finished.
def _onGetRemoteClustersFinished(self, clusters: List[CloudClusterResponse]) -> None:
online_clusters = {c.cluster_id: c for c in clusters if c.is_online} # type: Dict[str, CloudClusterResponse]
@ -115,19 +119,19 @@ class CloudOutputDeviceManager:
)
self._connectToActiveMachine()
def _createMachineFromDiscoveredPrinter(self, key: str) -> None:
device = self._remote_clusters[key] # type: CloudOutputDevice
if not device:
Logger.log("e", "Could not find discovered device with key [%s]", key)
return
group_name = device.clusterData.friendly_name
machine_type_id = device.printerType
Logger.log("i", "Creating machine from cloud device with key = [%s], group name = [%s], printer type = [%s]",
key, group_name, machine_type_id)
# The newly added machine is automatically activated.
self._application.getMachineManager().addMachine(machine_type_id, group_name)
active_machine = CuraApplication.getInstance().getGlobalContainerStack()

2
plugins/UM3NetworkPrinting/src/Cloud/Models/__init__.py
View file

@ -1,2 +0,0 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.

4
plugins/UM3NetworkPrinting/src/Cloud/ToolPathUploader.py
View file

@ -6,7 +6,7 @@ from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply, QNetworkAccessManage
from typing import Optional, Callable, Any, Tuple, cast
from UM.Logger import Logger
from .Models.CloudPrintJobResponse import CloudPrintJobResponse
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobResponse import CloudPrintJobResponse
## Class responsible for uploading meshes to the cloud in separate requests.
@ -53,7 +53,7 @@ class ToolPathUploader:
def _createRequest(self) -> QNetworkRequest:
request = QNetworkRequest(QUrl(self._print_job.upload_url))
request.setHeader(QNetworkRequest.ContentTypeHeader, self._print_job.content_type)
first_byte, last_byte = self._chunkRange()
content_range = "bytes {}-{}/{}".format(first_byte, last_byte - 1, len(self._data))
request.setRawHeader(b"Content-Range", content_range.encode())

179
plugins/UM3NetworkPrinting/src/DiscoverUM3Action.py
View file

@ -1,179 +0,0 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import os.path
import time
from typing import Optional, TYPE_CHECKING
from PyQt5.QtCore import pyqtSignal, pyqtProperty, pyqtSlot, QObject
from UM.PluginRegistry import PluginRegistry
from UM.Logger import Logger
from UM.i18n import i18nCatalog
from cura.CuraApplication import CuraApplication
from cura.MachineAction import MachineAction
from cura.Settings.CuraContainerRegistry import CuraContainerRegistry
from .UM3OutputDevicePlugin import UM3OutputDevicePlugin
if TYPE_CHECKING:
from cura.PrinterOutput.PrinterOutputDevice import PrinterOutputDevice
catalog = i18nCatalog("cura")
class DiscoverUM3Action(MachineAction):
discoveredDevicesChanged = pyqtSignal()
def __init__(self) -> None:
super().__init__("DiscoverUM3Action", catalog.i18nc("@action","Connect via Network"))
self._qml_url = "resources/qml/DiscoverUM3Action.qml"
self._network_plugin = None #type: Optional[UM3OutputDevicePlugin]
self.__additional_components_view = None #type: Optional[QObject]
CuraApplication.getInstance().engineCreatedSignal.connect(self._createAdditionalComponentsView)
self._last_zero_conf_event_time = time.time() #type: float
# Time to wait after a zero-conf service change before allowing a zeroconf reset
self._zero_conf_change_grace_period = 0.25 #type: float
# Overrides the one in MachineAction.
# This requires not attention from the user (any more), so we don't need to show any 'upgrade screens'.
def needsUserInteraction(self) -> bool:
return False
@pyqtSlot()
def startDiscovery(self):
if not self._network_plugin:
Logger.log("d", "Starting device discovery.")
self._network_plugin = CuraApplication.getInstance().getOutputDeviceManager().getOutputDevicePlugin("UM3NetworkPrinting")
self._network_plugin.discoveredDevicesChanged.connect(self._onDeviceDiscoveryChanged)
self.discoveredDevicesChanged.emit()
## Re-filters the list of devices.
@pyqtSlot()
def reset(self):
Logger.log("d", "Reset the list of found devices.")
if self._network_plugin:
self._network_plugin.resetLastManualDevice()
self.discoveredDevicesChanged.emit()
@pyqtSlot()
def restartDiscovery(self):
# Ensure that there is a bit of time after a printer has been discovered.
# This is a work around for an issue with Qt 5.5.1 up to Qt 5.7 which can segfault if we do this too often.
# It's most likely that the QML engine is still creating delegates, where the python side already deleted or
# garbage collected the data.
# Whatever the case, waiting a bit ensures that it doesn't crash.
if time.time() - self._last_zero_conf_event_time > self._zero_conf_change_grace_period:
if not self._network_plugin:
self.startDiscovery()
else:
self._network_plugin.startDiscovery()
@pyqtSlot(str, str)
def removeManualDevice(self, key, address):
if not self._network_plugin:
return
self._network_plugin.removeManualDevice(key, address)
@pyqtSlot(str, str)
def setManualDevice(self, key, address):
if key != "":
# This manual printer replaces a current manual printer
self._network_plugin.removeManualDevice(key)
if address != "":
self._network_plugin.addManualDevice(address)
def _onDeviceDiscoveryChanged(self, *args):
self._last_zero_conf_event_time = time.time()
self.discoveredDevicesChanged.emit()
@pyqtProperty("QVariantList", notify = discoveredDevicesChanged)
def foundDevices(self):
if self._network_plugin:
printers = list(self._network_plugin.getDiscoveredDevices().values())
printers.sort(key = lambda k: k.name)
return printers
else:
return []
@pyqtSlot(str)
def setGroupName(self, group_name: str) -> None:
Logger.log("d", "Attempting to set the group name of the active machine to %s", group_name)
global_container_stack = CuraApplication.getInstance().getGlobalContainerStack()
if global_container_stack:
# Update a GlobalStacks in the same group with the new group name.
group_id = global_container_stack.getMetaDataEntry("group_id")
machine_manager = CuraApplication.getInstance().getMachineManager()
for machine in machine_manager.getMachinesInGroup(group_id):
machine.setMetaDataEntry("group_name", group_name)
# Set the default value for "hidden", which is used when you have a group with multiple types of printers
global_container_stack.setMetaDataEntry("hidden", False)
if self._network_plugin:
# Ensure that the connection states are refreshed.
self._network_plugin.refreshConnections()
# Associates the currently active machine with the given printer device. The network connection information will be
# stored into the metadata of the currently active machine.
@pyqtSlot(QObject)
def associateActiveMachineWithPrinterDevice(self, printer_device: Optional["PrinterOutputDevice"]) -> None:
if self._network_plugin:
self._network_plugin.associateActiveMachineWithPrinterDevice(printer_device)
@pyqtSlot(result = str)
def getStoredKey(self) -> str:
global_container_stack = CuraApplication.getInstance().getGlobalContainerStack()
if global_container_stack:
meta_data = global_container_stack.getMetaData()
if "um_network_key" in meta_data:
return global_container_stack.getMetaDataEntry("um_network_key")
return ""
@pyqtSlot(result = str)
def getLastManualEntryKey(self) -> str:
if self._network_plugin:
return self._network_plugin.getLastManualDevice()
return ""
@pyqtSlot(str, result = bool)
def existsKey(self, key: str) -> bool:
metadata_filter = {"um_network_key": key}
containers = CuraContainerRegistry.getInstance().findContainerStacks(type="machine", **metadata_filter)
return bool(containers)
@pyqtSlot()
def loadConfigurationFromPrinter(self) -> None:
machine_manager = CuraApplication.getInstance().getMachineManager()
hotend_ids = machine_manager.printerOutputDevices[0].hotendIds
for index in range(len(hotend_ids)):
machine_manager.printerOutputDevices[0].hotendIdChanged.emit(index, hotend_ids[index])
material_ids = machine_manager.printerOutputDevices[0].materialIds
for index in range(len(material_ids)):
machine_manager.printerOutputDevices[0].materialIdChanged.emit(index, material_ids[index])
def _createAdditionalComponentsView(self) -> None:
Logger.log("d", "Creating additional ui components for UM3.")
# Create networking dialog
plugin_path = PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting")
if not plugin_path:
return
path = os.path.join(plugin_path, "resources/qml/UM3InfoComponents.qml")
self.__additional_components_view = CuraApplication.getInstance().createQmlComponent(path, {"manager": self})
if not self.__additional_components_view:
Logger.log("w", "Could not create ui components for UM3.")
return
# Create extra components
CuraApplication.getInstance().addAdditionalComponent("monitorButtons", self.__additional_components_view.findChild(QObject, "networkPrinterConnectButton"))

30
plugins/UM3NetworkPrinting/src/Factories/PrinterModelFactory.py Normal file
View file

@ -0,0 +1,30 @@
from typing import List, Any, Dict
from PyQt5.QtCore import QUrl
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from plugins.UM3NetworkPrinting.src.Models.ConfigurationChangeModel import ConfigurationChangeModel
class PrinterModelFactory:
CAMERA_URL_FORMAT = "http://{ip_address}:8080/?action=stream"
# Create a printer output model from some data.
@classmethod
def createPrinter(cls, output_controller: PrinterOutputController, ip_address: str, extruder_count: int = 2
) -> PrinterOutputModel:
printer = PrinterOutputModel(output_controller=output_controller, number_of_extruders=extruder_count)
printer.setCameraUrl(QUrl(cls.CAMERA_URL_FORMAT.format(ip_address=ip_address)))
return printer
# Create a list of configuration change models.
@classmethod
def createConfigurationChanges(cls, data: List[Dict[str, Any]]) -> List[ConfigurationChangeModel]:
return [ConfigurationChangeModel(
type_of_change=change.get("type_of_change"),
index=change.get("index"),
target_name=change.get("target_name"),
origin_name=change.get("origin_name")
) for change in data]

647
plugins/UM3NetworkPrinting/src/LegacyUM3OutputDevice.py
View file

@ -1,647 +0,0 @@
from typing import List, Optional
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.Models.PrintJobOutputModel import PrintJobOutputModel
from cura.PrinterOutput.Models.MaterialOutputModel import MaterialOutputModel
from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from cura.Settings.ContainerManager import ContainerManager
from cura.Settings.ExtruderManager import ExtruderManager
from UM.FileHandler.FileHandler import FileHandler
from UM.i18n import i18nCatalog
from UM.Logger import Logger
from UM.Message import Message
from UM.PluginRegistry import PluginRegistry
from UM.Scene.SceneNode import SceneNode
from UM.Settings.ContainerRegistry import ContainerRegistry
from PyQt5.QtNetwork import QNetworkRequest
from PyQt5.QtCore import QTimer, QUrl
from PyQt5.QtWidgets import QMessageBox
from .LegacyUM3PrinterOutputController import LegacyUM3PrinterOutputController
from time import time
import json
import os
i18n_catalog = i18nCatalog("cura")
## This is the output device for the "Legacy" API of the UM3. All firmware before 4.0.1 uses this API.
# Everything after that firmware uses the ClusterUM3Output.
# The Legacy output device can only have one printer (whereas the cluster can have 0 to n).
#
# Authentication is done in a number of steps;
# 1. Request an id / key pair by sending the application & user name. (state = authRequested)
# 2. Machine sends this back and will display an approve / deny message on screen. (state = AuthReceived)
# 3. OutputDevice will poll if the button was pressed.
# 4. At this point the machine either has the state Authenticated or AuthenticationDenied.
# 5. As a final step, we verify the authentication, as this forces the QT manager to setup the authenticator.
class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice):
def __init__(self, device_id, address: str, properties, parent = None) -> None:
super().__init__(device_id = device_id, address = address, properties = properties, connection_type = ConnectionType.NetworkConnection, parent = parent)
self._api_prefix = "/api/v1/"
self._number_of_extruders = 2
self._authentication_id = None
self._authentication_key = None
self._authentication_counter = 0
self._max_authentication_counter = 5 * 60 # Number of attempts before authentication timed out (5 min)
self._authentication_timer = QTimer()
self._authentication_timer.setInterval(1000) # TODO; Add preference for update interval
self._authentication_timer.setSingleShot(False)
self._authentication_timer.timeout.connect(self._onAuthenticationTimer)
# The messages are created when connect is called the first time.
# This ensures that the messages are only created for devices that actually want to connect.
self._authentication_requested_message = None
self._authentication_failed_message = None
self._authentication_succeeded_message = None
self._not_authenticated_message = None
self.authenticationStateChanged.connect(self._onAuthenticationStateChanged)
self.setPriority(3) # Make sure the output device gets selected above local file output
self.setName(self._id)
self.setShortDescription(i18n_catalog.i18nc("@action:button Preceded by 'Ready to'.", "Print over network"))
self.setDescription(i18n_catalog.i18nc("@properties:tooltip", "Print over network"))
self.setIconName("print")
self._output_controller = LegacyUM3PrinterOutputController(self)
def _createMonitorViewFromQML(self) -> None:
if self._monitor_view_qml_path is None and PluginRegistry.getInstance() is not None:
self._monitor_view_qml_path = os.path.join(
PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"),
"resources", "qml", "MonitorStage.qml"
)
super()._createMonitorViewFromQML()
def _onAuthenticationStateChanged(self):
# We only accept commands if we are authenticated.
self._setAcceptsCommands(self._authentication_state == AuthState.Authenticated)
if self._authentication_state == AuthState.Authenticated:
self.setConnectionText(i18n_catalog.i18nc("@info:status", "Connected over the network."))
elif self._authentication_state == AuthState.AuthenticationRequested:
self.setConnectionText(i18n_catalog.i18nc("@info:status",
"Connected over the network. Please approve the access request on the printer."))
elif self._authentication_state == AuthState.AuthenticationDenied:
self.setConnectionText(i18n_catalog.i18nc("@info:status", "Connected over the network. No access to control the printer."))
def _setupMessages(self):
self._authentication_requested_message = Message(i18n_catalog.i18nc("@info:status",
"Access to the printer requested. Please approve the request on the printer"),
lifetime=0, dismissable=False, progress=0,
title=i18n_catalog.i18nc("@info:title",
"Authentication status"))
self._authentication_failed_message = Message("", title=i18n_catalog.i18nc("@info:title", "Authentication Status"))
self._authentication_failed_message.addAction("Retry", i18n_catalog.i18nc("@action:button", "Retry"), None,
i18n_catalog.i18nc("@info:tooltip", "Re-send the access request"))
self._authentication_failed_message.actionTriggered.connect(self._messageCallback)
self._authentication_succeeded_message = Message(
i18n_catalog.i18nc("@info:status", "Access to the printer accepted"),
title=i18n_catalog.i18nc("@info:title", "Authentication Status"))
self._not_authenticated_message = Message(
i18n_catalog.i18nc("@info:status", "No access to print with this printer. Unable to send print job."),
title=i18n_catalog.i18nc("@info:title", "Authentication Status"))
self._not_authenticated_message.addAction("Request", i18n_catalog.i18nc("@action:button", "Request Access"),
None, i18n_catalog.i18nc("@info:tooltip",
"Send access request to the printer"))
self._not_authenticated_message.actionTriggered.connect(self._messageCallback)
def _messageCallback(self, message_id=None, action_id="Retry"):
if action_id == "Request" or action_id == "Retry":
if self._authentication_failed_message:
self._authentication_failed_message.hide()
if self._not_authenticated_message:
self._not_authenticated_message.hide()
self._requestAuthentication()
def connect(self):
super().connect()
self._setupMessages()
global_container = CuraApplication.getInstance().getGlobalContainerStack()
if global_container:
self._authentication_id = global_container.getMetaDataEntry("network_authentication_id", None)
self._authentication_key = global_container.getMetaDataEntry("network_authentication_key", None)
def close(self):
super().close()
if self._authentication_requested_message:
self._authentication_requested_message.hide()
if self._authentication_failed_message:
self._authentication_failed_message.hide()
if self._authentication_succeeded_message:
self._authentication_succeeded_message.hide()
self._sending_gcode = False
self._compressing_gcode = False
self._authentication_timer.stop()
## Send all material profiles to the printer.
def _sendMaterialProfiles(self):
Logger.log("i", "Sending material profiles to printer")
# TODO: Might want to move this to a job...
for container in ContainerRegistry.getInstance().findInstanceContainers(type="material"):
try:
xml_data = container.serialize()
if xml_data == "" or xml_data is None:
continue
names = ContainerManager.getInstance().getLinkedMaterials(container.getId())
if names:
# There are other materials that share this GUID.
if not container.isReadOnly():
continue # If it's not readonly, it's created by user, so skip it.
file_name = "none.xml"
self.postForm("materials", "form-data; name=\"file\";filename=\"%s\"" % file_name, xml_data.encode(), on_finished=None)
except NotImplementedError:
# If the material container is not the most "generic" one it can't be serialized an will raise a
# NotImplementedError. We can simply ignore these.
pass
def requestWrite(self, nodes: List["SceneNode"], file_name: Optional[str] = None, limit_mimetypes: bool = False,
file_handler: Optional["FileHandler"] = None, filter_by_machine: bool = False, **kwargs) -> None:
if not self.activePrinter:
# No active printer. Unable to write
return
if self.activePrinter.state not in ["idle", ""]:
# Printer is not able to accept commands.
return
if self._authentication_state != AuthState.Authenticated:
# Not authenticated, so unable to send job.
return
self.writeStarted.emit(self)
gcode_dict = getattr(CuraApplication.getInstance().getController().getScene(), "gcode_dict", [])
active_build_plate_id = CuraApplication.getInstance().getMultiBuildPlateModel().activeBuildPlate
gcode_list = gcode_dict[active_build_plate_id]
if not gcode_list:
# Unable to find g-code. Nothing to send
return
self._gcode = gcode_list
errors = self._checkForErrors()
if errors:
text = i18n_catalog.i18nc("@label", "Unable to start a new print job.")
informative_text = i18n_catalog.i18nc("@label",
"There is an issue with the configuration of your Ultimaker, which makes it impossible to start the print. "
"Please resolve this issues before continuing.")
detailed_text = ""
for error in errors:
detailed_text += error + "\n"
CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"),
text,
informative_text,
detailed_text,
buttons=QMessageBox.Ok,
icon=QMessageBox.Critical,
callback = self._messageBoxCallback
)
return # Don't continue; Errors must block sending the job to the printer.
# There might be multiple things wrong with the configuration. Check these before starting.
warnings = self._checkForWarnings()
if warnings:
text = i18n_catalog.i18nc("@label", "Are you sure you wish to print with the selected configuration?")
informative_text = i18n_catalog.i18nc("@label",
"There is a mismatch between the configuration or calibration of the printer and Cura. "
"For the best result, always slice for the PrintCores and materials that are inserted in your printer.")
detailed_text = ""
for warning in warnings:
detailed_text += warning + "\n"
CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Mismatched configuration"),
text,
informative_text,
detailed_text,
buttons=QMessageBox.Yes + QMessageBox.No,
icon=QMessageBox.Question,
callback=self._messageBoxCallback
)
return
# No warnings or errors, so we're good to go.
self._startPrint()
# Notify the UI that a switch to the print monitor should happen
CuraApplication.getInstance().getController().setActiveStage("MonitorStage")
def _startPrint(self):
Logger.log("i", "Sending print job to printer.")
if self._sending_gcode:
self._error_message = Message(
i18n_catalog.i18nc("@info:status",
"Sending new jobs (temporarily) blocked, still sending the previous print job."))
self._error_message.show()
return
self._sending_gcode = True
self._send_gcode_start = time()
self._progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), 0, False, -1,
i18n_catalog.i18nc("@info:title", "Sending Data"))
self._progress_message.addAction("Abort", i18n_catalog.i18nc("@action:button", "Cancel"), None, "")
self._progress_message.actionTriggered.connect(self._progressMessageActionTriggered)
self._progress_message.show()
compressed_gcode = self._compressGCode()
if compressed_gcode is None:
# Abort was called.
return
file_name = "%s.gcode.gz" % CuraApplication.getInstance().getPrintInformation().jobName
self.postForm("print_job", "form-data; name=\"file\";filename=\"%s\"" % file_name, compressed_gcode,
on_finished=self._onPostPrintJobFinished)
return
def _progressMessageActionTriggered(self, message_id=None, action_id=None):
if action_id == "Abort":
Logger.log("d", "User aborted sending print to remote.")
self._progress_message.hide()
self._compressing_gcode = False
self._sending_gcode = False
CuraApplication.getInstance().getController().setActiveStage("PrepareStage")
def _onPostPrintJobFinished(self, reply):
self._progress_message.hide()
self._sending_gcode = False
def _onUploadPrintJobProgress(self, bytes_sent, bytes_total):
if bytes_total > 0:
new_progress = bytes_sent / bytes_total * 100
# Treat upload progress as response. Uploading can take more than 10 seconds, so if we don't, we can get
# timeout responses if this happens.
self._last_response_time = time()
if new_progress > self._progress_message.getProgress():
self._progress_message.show() # Ensure that the message is visible.
self._progress_message.setProgress(bytes_sent / bytes_total * 100)
else:
self._progress_message.setProgress(0)
self._progress_message.hide()
def _messageBoxCallback(self, button):
def delayedCallback():
if button == QMessageBox.Yes:
self._startPrint()
else:
CuraApplication.getInstance().getController().setActiveStage("PrepareStage")
# For some unknown reason Cura on OSX will hang if we do the call back code
# immediately without first returning and leaving QML's event system.
QTimer.singleShot(100, delayedCallback)
def _checkForErrors(self):
errors = []
print_information = CuraApplication.getInstance().getPrintInformation()
if not print_information.materialLengths:
Logger.log("w", "There is no material length information. Unable to check for errors.")
return errors
for index, extruder in enumerate(self.activePrinter.extruders):
# Due to airflow issues, both slots must be loaded, regardless if they are actually used or not.
if extruder.hotendID == "":
# No Printcore loaded.
errors.append(i18n_catalog.i18nc("@info:status", "No Printcore loaded in slot {slot_number}".format(slot_number=index + 1)))
if index < len(print_information.materialLengths) and print_information.materialLengths[index] != 0:
# The extruder is by this print.
if extruder.activeMaterial is None:
# No active material
errors.append(i18n_catalog.i18nc("@info:status", "No material loaded in slot {slot_number}".format(slot_number=index + 1)))
return errors
def _checkForWarnings(self):
warnings = []
print_information = CuraApplication.getInstance().getPrintInformation()
if not print_information.materialLengths:
Logger.log("w", "There is no material length information. Unable to check for warnings.")
return warnings
extruder_manager = ExtruderManager.getInstance()
for index, extruder in enumerate(self.activePrinter.extruders):
if index < len(print_information.materialLengths) and print_information.materialLengths[index] != 0:
# The extruder is by this print.
# TODO: material length check
# Check if the right Printcore is active.
variant = extruder_manager.getExtruderStack(index).findContainer({"type": "variant"})
if variant:
if variant.getName() != extruder.hotendID:
warnings.append(i18n_catalog.i18nc("@label", "Different PrintCore (Cura: {cura_printcore_name}, Printer: {remote_printcore_name}) selected for extruder {extruder_id}".format(cura_printcore_name = variant.getName(), remote_printcore_name = extruder.hotendID, extruder_id = index + 1)))
else:
Logger.log("w", "Unable to find variant.")
# Check if the right material is loaded.
local_material = extruder_manager.getExtruderStack(index).findContainer({"type": "material"})
if local_material:
if extruder.activeMaterial.guid != local_material.getMetaDataEntry("GUID"):
Logger.log("w", "Extruder %s has a different material (%s) as Cura (%s)", index + 1, extruder.activeMaterial.guid, local_material.getMetaDataEntry("GUID"))
warnings.append(i18n_catalog.i18nc("@label", "Different material (Cura: {0}, Printer: {1}) selected for extruder {2}").format(local_material.getName(), extruder.activeMaterial.name, index + 1))
else:
Logger.log("w", "Unable to find material.")
return warnings
def _update(self):
if not super()._update():
return
if self._authentication_state == AuthState.NotAuthenticated:
if self._authentication_id is None and self._authentication_key is None:
# This machine doesn't have any authentication, so request it.
self._requestAuthentication()
elif self._authentication_id is not None and self._authentication_key is not None:
# We have authentication info, but we haven't checked it out yet. Do so now.
self._verifyAuthentication()
elif self._authentication_state == AuthState.AuthenticationReceived:
# We have an authentication, but it's not confirmed yet.
self._checkAuthentication()
# We don't need authentication for requesting info, so we can go right ahead with requesting this.
self.get("printer", on_finished=self._onGetPrinterDataFinished)
self.get("print_job", on_finished=self._onGetPrintJobFinished)
def _resetAuthenticationRequestedMessage(self):
if self._authentication_requested_message:
self._authentication_requested_message.hide()
self._authentication_timer.stop()
self._authentication_counter = 0
def _onAuthenticationTimer(self):
self._authentication_counter += 1
self._authentication_requested_message.setProgress(
self._authentication_counter / self._max_authentication_counter * 100)
if self._authentication_counter > self._max_authentication_counter:
self._authentication_timer.stop()
Logger.log("i", "Authentication timer ended. Setting authentication to denied for printer: %s" % self._id)
self.setAuthenticationState(AuthState.AuthenticationDenied)
self._resetAuthenticationRequestedMessage()
self._authentication_failed_message.show()
def _verifyAuthentication(self):
Logger.log("d", "Attempting to verify authentication")
# This will ensure that the "_onAuthenticationRequired" is triggered, which will setup the authenticator.
self.get("auth/verify", on_finished=self._onVerifyAuthenticationCompleted)
def _onVerifyAuthenticationCompleted(self, reply):
status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
if status_code == 401:
# Something went wrong; We somehow tried to verify authentication without having one.
Logger.log("d", "Attempted to verify auth without having one.")
self._authentication_id = None
self._authentication_key = None
self.setAuthenticationState(AuthState.NotAuthenticated)
elif status_code == 403 and self._authentication_state != AuthState.Authenticated:
# If we were already authenticated, we probably got an older message back all of the sudden. Drop that.
Logger.log("d",
"While trying to verify the authentication state, we got a forbidden response. Our own auth state was %s. ",
self._authentication_state)
self.setAuthenticationState(AuthState.AuthenticationDenied)
self._authentication_failed_message.show()
elif status_code == 200:
self.setAuthenticationState(AuthState.Authenticated)
def _checkAuthentication(self):
Logger.log("d", "Checking if authentication is correct for id %s and key %s", self._authentication_id, self._getSafeAuthKey())
self.get("auth/check/" + str(self._authentication_id), on_finished=self._onCheckAuthenticationFinished)
def _onCheckAuthenticationFinished(self, reply):
if str(self._authentication_id) not in reply.url().toString():
Logger.log("w", "Got an old id response.")
# Got response for old authentication ID.
return
try:
data = json.loads(bytes(reply.readAll()).decode("utf-8"))
except json.decoder.JSONDecodeError:
Logger.log("w", "Received an invalid authentication check from printer: Not valid JSON.")
return
if data.get("message", "") == "authorized":
Logger.log("i", "Authentication was approved")
self.setAuthenticationState(AuthState.Authenticated)
self._saveAuthentication()
# Double check that everything went well.
self._verifyAuthentication()
# Notify the user.
self._resetAuthenticationRequestedMessage()
self._authentication_succeeded_message.show()
elif data.get("message", "") == "unauthorized":
Logger.log("i", "Authentication was denied.")
self.setAuthenticationState(AuthState.AuthenticationDenied)
self._authentication_failed_message.show()
def _saveAuthentication(self) -> None:
global_container_stack = CuraApplication.getInstance().getGlobalContainerStack()
if self._authentication_key is None:
Logger.log("e", "Authentication key is None, nothing to save.")
return
if self._authentication_id is None:
Logger.log("e", "Authentication id is None, nothing to save.")
return
if global_container_stack:
global_container_stack.setMetaDataEntry("network_authentication_key", self._authentication_key)
global_container_stack.setMetaDataEntry("network_authentication_id", self._authentication_id)
# Force save so we are sure the data is not lost.
CuraApplication.getInstance().saveStack(global_container_stack)
Logger.log("i", "Authentication succeeded for id %s and key %s", self._authentication_id,
self._getSafeAuthKey())
else:
Logger.log("e", "Unable to save authentication for id %s and key %s", self._authentication_id,
self._getSafeAuthKey())
def _onRequestAuthenticationFinished(self, reply):
try:
data = json.loads(bytes(reply.readAll()).decode("utf-8"))
except json.decoder.JSONDecodeError:
Logger.log("w", "Received an invalid authentication request reply from printer: Not valid JSON.")
self.setAuthenticationState(AuthState.NotAuthenticated)
return
self.setAuthenticationState(AuthState.AuthenticationReceived)
self._authentication_id = data["id"]
self._authentication_key = data["key"]
Logger.log("i", "Got a new authentication ID (%s) and KEY (%s). Waiting for authorization.",
self._authentication_id, self._getSafeAuthKey())
def _requestAuthentication(self):
self._authentication_requested_message.show()
self._authentication_timer.start()
# Reset any previous authentication info. If this isn't done, the "Retry" action on the failed message might
# give issues.
self._authentication_key = None
self._authentication_id = None
self.post("auth/request",
json.dumps({"application": "Cura-" + CuraApplication.getInstance().getVersion(),
"user": self._getUserName()}),
on_finished=self._onRequestAuthenticationFinished)
self.setAuthenticationState(AuthState.AuthenticationRequested)
def _onAuthenticationRequired(self, reply, authenticator):
if self._authentication_id is not None and self._authentication_key is not None:
Logger.log("d",
"Authentication was required for printer: %s. Setting up authenticator with ID %s and key %s",
self._id, self._authentication_id, self._getSafeAuthKey())
authenticator.setUser(self._authentication_id)
authenticator.setPassword(self._authentication_key)
else:
Logger.log("d", "No authentication is available to use for %s, but we did got a request for it.", self._id)
def _onGetPrintJobFinished(self, reply):
status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
if not self._printers:
return # Ignore the data for now, we don't have info about a printer yet.
printer = self._printers[0]
if status_code == 200:
try:
result = json.loads(bytes(reply.readAll()).decode("utf-8"))
except json.decoder.JSONDecodeError:
Logger.log("w", "Received an invalid print job state message: Not valid JSON.")
return
if printer.activePrintJob is None:
print_job = PrintJobOutputModel(output_controller=self._output_controller)
printer.updateActivePrintJob(print_job)
else:
print_job = printer.activePrintJob
print_job.updateState(result["state"])
print_job.updateTimeElapsed(result["time_elapsed"])
print_job.updateTimeTotal(result["time_total"])
print_job.updateName(result["name"])
elif status_code == 404:
# No job found, so delete the active print job (if any!)
printer.updateActivePrintJob(None)
else:
Logger.log("w",
"Got status code {status_code} while trying to get printer data".format(status_code=status_code))
def materialHotendChangedMessage(self, callback):
CuraApplication.getInstance().messageBox(i18n_catalog.i18nc("@window:title", "Sync with your printer"),
i18n_catalog.i18nc("@label",
"Would you like to use your current printer configuration in Cura?"),
i18n_catalog.i18nc("@label",
"The PrintCores and/or materials on your printer differ from those within your current project. For the best result, always slice for the PrintCores and materials that are inserted in your printer."),
buttons=QMessageBox.Yes + QMessageBox.No,
icon=QMessageBox.Question,
callback=callback
)
def _onGetPrinterDataFinished(self, reply):
status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
if status_code == 200:
try:
result = json.loads(bytes(reply.readAll()).decode("utf-8"))
except json.decoder.JSONDecodeError:
Logger.log("w", "Received an invalid printer state message: Not valid JSON.")
return
if not self._printers:
# Quickest way to get the firmware version is to grab it from the zeroconf.
firmware_version = self._properties.get(b"firmware_version", b"").decode("utf-8")
self._printers = [PrinterOutputModel(output_controller=self._output_controller, number_of_extruders=self._number_of_extruders, firmware_version=firmware_version)]
self._printers[0].setCameraUrl(QUrl("http://" + self._address + ":8080/?action=stream"))
for extruder in self._printers[0].extruders:
extruder.activeMaterialChanged.connect(self.materialIdChanged)
extruder.hotendIDChanged.connect(self.hotendIdChanged)
self.printersChanged.emit()
# LegacyUM3 always has a single printer.
printer = self._printers[0]
printer.updateBedTemperature(result["bed"]["temperature"]["current"])
printer.updateTargetBedTemperature(result["bed"]["temperature"]["target"])
printer.updateState(result["status"])
try:
# If we're still handling the request, we should ignore remote for a bit.
if not printer.getController().isPreheatRequestInProgress():
printer.updateIsPreheating(result["bed"]["pre_heat"]["active"])
except KeyError:
# Older firmwares don't support preheating, so we need to fake it.
pass
head_position = result["heads"][0]["position"]
printer.updateHeadPosition(head_position["x"], head_position["y"], head_position["z"])
for index in range(0, self._number_of_extruders):
temperatures = result["heads"][0]["extruders"][index]["hotend"]["temperature"]
extruder = printer.extruders[index]
extruder.updateTargetHotendTemperature(temperatures["target"])
extruder.updateHotendTemperature(temperatures["current"])
material_guid = result["heads"][0]["extruders"][index]["active_material"]["guid"]
if extruder.activeMaterial is None or extruder.activeMaterial.guid != material_guid:
# Find matching material (as we need to set brand, type & color)
containers = ContainerRegistry.getInstance().findInstanceContainers(type="material",
GUID=material_guid)
if containers:
color = containers[0].getMetaDataEntry("color_code")
brand = containers[0].getMetaDataEntry("brand")
material_type = containers[0].getMetaDataEntry("material")
name = containers[0].getName()
else:
# Unknown material.
color = "#00000000"
brand = "Unknown"
material_type = "Unknown"
name = "Unknown"
material = MaterialOutputModel(guid=material_guid, type=material_type,
brand=brand, color=color, name = name)
extruder.updateActiveMaterial(material)
try:
hotend_id = result["heads"][0]["extruders"][index]["hotend"]["id"]
except KeyError:
hotend_id = ""
printer.extruders[index].updateHotendID(hotend_id)
else:
Logger.log("w",
"Got status code {status_code} while trying to get printer data".format(status_code = status_code))
## Convenience function to "blur" out all but the last 5 characters of the auth key.
# This can be used to debug print the key, without it compromising the security.
def _getSafeAuthKey(self):
if self._authentication_key is not None:
result = self._authentication_key[-5:]
result = "********" + result
return result
return self._authentication_key

96
plugins/UM3NetworkPrinting/src/LegacyUM3PrinterOutputController.py
View file

@ -1,96 +0,0 @@
# Copyright (c) 2019 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from PyQt5.QtCore import QTimer
from UM.Version import Version
MYPY = False
if MYPY:
from cura.PrinterOutput.Models.PrintJobOutputModel import PrintJobOutputModel
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
class LegacyUM3PrinterOutputController(PrinterOutputController):
def __init__(self, output_device):
super().__init__(output_device)
self._preheat_bed_timer = QTimer()
self._preheat_bed_timer.setSingleShot(True)
self._preheat_bed_timer.timeout.connect(self._onPreheatBedTimerFinished)
self._preheat_printer = None
self.can_control_manually = False
self.can_send_raw_gcode = False
# Are we still waiting for a response about preheat?
# We need this so we can already update buttons, so it feels more snappy.
self._preheat_request_in_progress = False
def isPreheatRequestInProgress(self):
return self._preheat_request_in_progress
def setJobState(self, job: "PrintJobOutputModel", state: str):
data = "{\"target\": \"%s\"}" % state
self._output_device.put("print_job/state", data, on_finished=None)
def setTargetBedTemperature(self, printer: "PrinterOutputModel", temperature: float):
data = str(temperature)
self._output_device.put("printer/bed/temperature/target", data, on_finished = self._onPutBedTemperatureCompleted)
def _onPutBedTemperatureCompleted(self, reply):
if Version(self._preheat_printer.firmwareVersion) < Version("3.5.92"):
# If it was handling a preheat, it isn't anymore.
self._preheat_request_in_progress = False
def _onPutPreheatBedCompleted(self, reply):
self._preheat_request_in_progress = False
def moveHead(self, printer: "PrinterOutputModel", x, y, z, speed):
head_pos = printer._head_position
new_x = head_pos.x + x
new_y = head_pos.y + y
new_z = head_pos.z + z
data = "{\n\"x\":%s,\n\"y\":%s,\n\"z\":%s\n}" %(new_x, new_y, new_z)
self._output_device.put("printer/heads/0/position", data, on_finished=None)
def homeBed(self, printer):
self._output_device.put("printer/heads/0/position/z", "0", on_finished=None)
def _onPreheatBedTimerFinished(self):
self.setTargetBedTemperature(self._preheat_printer, 0)
self._preheat_printer.updateIsPreheating(False)
self._preheat_request_in_progress = True
def cancelPreheatBed(self, printer: "PrinterOutputModel"):
self.preheatBed(printer, temperature=0, duration=0)
self._preheat_bed_timer.stop()
printer.updateIsPreheating(False)
def preheatBed(self, printer: "PrinterOutputModel", temperature, duration):
try:
temperature = round(temperature) # The API doesn't allow floating point.
duration = round(duration)
except ValueError:
return # Got invalid values, can't pre-heat.
if duration > 0:
data = """{"temperature": "%i", "timeout": "%i"}""" % (temperature, duration)
else:
data = """{"temperature": "%i"}""" % temperature
# Real bed pre-heating support is implemented from 3.5.92 and up.
if Version(printer.firmwareVersion) < Version("3.5.92"):
# No firmware-side duration support then, so just set target bed temp and set a timer.
self.setTargetBedTemperature(printer, temperature=temperature)
self._preheat_bed_timer.setInterval(duration * 1000)
self._preheat_bed_timer.start()
self._preheat_printer = printer
printer.updateIsPreheating(True)
return
self._output_device.put("printer/bed/pre_heat", data, on_finished = self._onPutPreheatBedCompleted)
printer.updateIsPreheating(True)
self._preheat_request_in_progress = True

46
plugins/UM3NetworkPrinting/src/Models.py
View file

@ -1,46 +0,0 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
## Base model that maps kwargs to instance attributes.
class BaseModel:
def __init__(self, **kwargs) -> None:
self.__dict__.update(kwargs)
self.validate()
# Validates the model, raising an exception if the model is invalid.
def validate(self) -> None:
pass
## Class representing a material that was fetched from the cluster API.
class ClusterMaterial(BaseModel):
def __init__(self, guid: str, version: int, **kwargs) -> None:
self.guid = guid # type: str
self.version = version # type: int
super().__init__(**kwargs)
def validate(self) -> None:
if not self.guid:
raise ValueError("guid is required on ClusterMaterial")
if not self.version:
raise ValueError("version is required on ClusterMaterial")
## Class representing a local material that was fetched from the container registry.
class LocalMaterial(BaseModel):
def __init__(self, GUID: str, id: str, version: int, **kwargs) -> None:
self.GUID = GUID # type: str
self.id = id # type: str
self.version = version # type: int
super().__init__(**kwargs)
#
def validate(self) -> None:
super().validate()
if not self.GUID:
raise ValueError("guid is required on LocalMaterial")
if not self.version:
raise ValueError("version is required on LocalMaterial")
if not self.id:
raise ValueError("id is required on LocalMaterial")

2
plugins/UM3NetworkPrinting/src/Cloud/Models/BaseCloudModel.py → plugins/UM3NetworkPrinting/src/Models/BaseCloudModel.py
View file

@ -3,7 +3,7 @@
from datetime import datetime, timezone
from typing import Dict, Union, TypeVar, Type, List, Any
from ...Models import BaseModel
from plugins.UM3NetworkPrinting.src.Models.BaseModel import BaseModel
## Base class for the models used in the interface with the Ultimaker cloud APIs.

9
plugins/UM3NetworkPrinting/src/Models/BaseModel.py Normal file
View file

@ -0,0 +1,9 @@
## Base model that maps kwargs to instance attributes.
class BaseModel:
def __init__(self, **kwargs) -> None:
self.__dict__.update(kwargs)
self.validate()
# Validates the model, raising an exception if the model is invalid.
def validate(self) -> None:
pass

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterBuildPlate.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterBuildPlate.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrintCoreConfiguration.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrintCoreConfiguration.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrintJobConfigurationChange.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrintJobConfigurationChange.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrintJobConstraint.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrintJobConstraint.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrintJobImpediment.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrintJobImpediment.py
View file

6
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrintJobStatus.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrintJobStatus.py
View file

@ -3,9 +3,9 @@
from typing import List, Optional, Union, Dict, Any
from cura.PrinterOutput.Models.PrinterConfigurationModel import PrinterConfigurationModel
from ...UM3PrintJobOutputModel import UM3PrintJobOutputModel
from ...ConfigurationChangeModel import ConfigurationChangeModel
from ..CloudOutputController import CloudOutputController
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
from plugins.UM3NetworkPrinting.src.Models.ConfigurationChangeModel import ConfigurationChangeModel
from plugins.UM3NetworkPrinting.src.Cloud.CloudOutputController import CloudOutputController
from .BaseCloudModel import BaseCloudModel
from .CloudClusterBuildPlate import CloudClusterBuildPlate
from .CloudClusterPrintJobConfigurationChange import CloudClusterPrintJobConfigurationChange

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrinterConfigurationMaterial.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrinterConfigurationMaterial.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterPrinterStatus.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterPrinterStatus.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterResponse.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterResponse.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudClusterStatus.py → plugins/UM3NetworkPrinting/src/Models/CloudClusterStatus.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudError.py → plugins/UM3NetworkPrinting/src/Models/CloudError.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudPrintJobResponse.py → plugins/UM3NetworkPrinting/src/Models/CloudPrintJobResponse.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudPrintJobUploadRequest.py → plugins/UM3NetworkPrinting/src/Models/CloudPrintJobUploadRequest.py
View file

0
plugins/UM3NetworkPrinting/src/Cloud/Models/CloudPrintResponse.py → plugins/UM3NetworkPrinting/src/Models/CloudPrintResponse.py
View file

15
plugins/UM3NetworkPrinting/src/Models/ClusterMaterial.py Normal file
View file

@ -0,0 +1,15 @@
## Class representing a material that was fetched from the cluster API.
from plugins.UM3NetworkPrinting.src.Models.BaseModel import BaseModel
class ClusterMaterial(BaseModel):
def __init__(self, guid: str, version: int, **kwargs) -> None:
self.guid = guid # type: str
self.version = version # type: int
super().__init__(**kwargs)
def validate(self) -> None:
if not self.guid:
raise ValueError("guid is required on ClusterMaterial")
if not self.version:
raise ValueError("version is required on ClusterMaterial")

5
plugins/UM3NetworkPrinting/src/ConfigurationChangeModel.py → plugins/UM3NetworkPrinting/src/Models/ConfigurationChangeModel.py
View file

@ -1,12 +1,13 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from PyQt5.QtCore import pyqtSignal, pyqtProperty, QObject, pyqtSlot
from PyQt5.QtCore import pyqtProperty, QObject
BLOCKING_CHANGE_TYPES = [
"material_insert", "buildplate_change"
]
class ConfigurationChangeModel(QObject):
def __init__(self, type_of_change: str, index: int, target_name: str, origin_name: str) -> None:
super().__init__()
@ -35,4 +36,4 @@ class ConfigurationChangeModel(QObject):
@pyqtProperty(bool, constant = True)
def canOverride(self) -> bool:
return self._can_override
return self._can_override

20
plugins/UM3NetworkPrinting/src/Models/LocalMaterial.py Normal file
View file

@ -0,0 +1,20 @@
## Class representing a local material that was fetched from the container registry.
from plugins.UM3NetworkPrinting.src.Models.BaseModel import BaseModel
class LocalMaterial(BaseModel):
def __init__(self, GUID: str, id: str, version: int, **kwargs) -> None:
self.GUID = GUID # type: str
self.id = id # type: str
self.version = version # type: int
super().__init__(**kwargs)
#
def validate(self) -> None:
super().validate()
if not self.GUID:
raise ValueError("guid is required on LocalMaterial")
if not self.version:
raise ValueError("version is required on LocalMaterial")
if not self.id:
raise ValueError("id is required on LocalMaterial")

3
plugins/UM3NetworkPrinting/src/UM3PrintJobOutputModel.py → plugins/UM3NetworkPrinting/src/Models/UM3PrintJobOutputModel.py
View file

@ -1,13 +1,12 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import List
from PyQt5.QtCore import pyqtProperty, pyqtSignal
from cura.PrinterOutput.Models.PrintJobOutputModel import PrintJobOutputModel
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from .ConfigurationChangeModel import ConfigurationChangeModel
from plugins.UM3NetworkPrinting.src.Models.ConfigurationChangeModel import ConfigurationChangeModel
class UM3PrintJobOutputModel(PrintJobOutputModel):

0
plugins/UM3NetworkPrinting/src/Models/__init__.py Normal file
View file

101
plugins/UM3NetworkPrinting/src/ClusterUM3OutputDevice.py → plugins/UM3NetworkPrinting/src/Network/ClusterUM3OutputDevice.py
View file

@ -21,17 +21,17 @@ from UM.Settings.ContainerRegistry import ContainerRegistry
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.Models.PrinterConfigurationModel import PrinterConfigurationModel
from cura.PrinterOutput.Models.ExtruderConfigurationModel import ExtruderConfigurationModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState, NetworkedPrinterOutputDevice
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.Models.MaterialOutputModel import MaterialOutputModel
from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from plugins.UM3NetworkPrinting.src.Factories.PrinterModelFactory import PrinterModelFactory
from plugins.UM3NetworkPrinting.src.UltimakerNetworkedPrinterOutputDevice import UltimakerNetworkedPrinterOutputDevice
from .Cloud.Utils import formatTimeCompleted, formatDateCompleted
from .ClusterUM3PrinterOutputController import ClusterUM3PrinterOutputController
from .ConfigurationChangeModel import ConfigurationChangeModel
from .MeshFormatHandler import MeshFormatHandler
from .SendMaterialJob import SendMaterialJob
from .UM3PrintJobOutputModel import UM3PrintJobOutputModel
from plugins.UM3NetworkPrinting.src.Cloud.Utils import formatTimeCompleted, formatDateCompleted
from plugins.UM3NetworkPrinting.src.Network.ClusterUM3PrinterOutputController import ClusterUM3PrinterOutputController
from plugins.UM3NetworkPrinting.src.MeshFormatHandler import MeshFormatHandler
from plugins.UM3NetworkPrinting.src.SendMaterialJob import SendMaterialJob
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply
from PyQt5.QtGui import QDesktopServices, QImage
@ -40,16 +40,11 @@ from PyQt5.QtCore import pyqtSlot, QUrl, pyqtSignal, pyqtProperty, QObject
i18n_catalog = i18nCatalog("cura")
class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
printJobsChanged = pyqtSignal()
activePrinterChanged = pyqtSignal()
class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
activeCameraUrlChanged = pyqtSignal()
receivedPrintJobsChanged = pyqtSignal()
# Notify can only use signals that are defined by the class that they are in, not inherited ones.
# Therefore we create a private signal used to trigger the printersChanged signal.
_clusterPrintersChanged = pyqtSignal()
def __init__(self, device_id, address, properties, parent = None) -> None:
super().__init__(device_id = device_id, address = address, properties=properties, connection_type = ConnectionType.NetworkConnection, parent = parent)
self._api_prefix = "/cluster-api/v1/"
@ -62,7 +57,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
"", {}, io.BytesIO()
) # type: Tuple[Optional[str], Dict[str, Union[str, int, bool]], Union[io.StringIO, io.BytesIO]]
self._print_jobs = [] # type: List[UM3PrintJobOutputModel]
self._received_print_jobs = False # type: bool
if PluginRegistry.getInstance() is not None:
@ -77,15 +71,10 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
self._accepts_commands = True # type: bool
# Cluster does not have authentication, so default to authenticated
self._authentication_state = AuthState.Authenticated
self._error_message = None # type: Optional[Message]
self._write_job_progress_message = None # type: Optional[Message]
self._progress_message = None # type: Optional[Message]
self._active_printer = None # type: Optional[PrinterOutputModel]
self._printer_selection_dialog = None # type: QObject
self.setPriority(3) # Make sure the output device gets selected above local file output
@ -145,10 +134,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
def supportsPrintJobActions(self) -> bool:
return True
@pyqtProperty(int, constant=True)
def clusterSize(self) -> int:
return self._cluster_size
## Allows the user to choose a printer to print with from the printer
# selection dialogue.
# \param target_printer The name of the printer to target.
@ -240,16 +225,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
on_finished = self._onPostPrintJobFinished,
on_progress = self._onUploadPrintJobProgress)
@pyqtProperty(QObject, notify = activePrinterChanged)
def activePrinter(self) -> Optional[PrinterOutputModel]:
return self._active_printer
@pyqtSlot(QObject)
def setActivePrinter(self, printer: Optional[PrinterOutputModel]) -> None:
if self._active_printer != printer:
self._active_printer = printer
self.activePrinterChanged.emit()
@pyqtProperty(QUrl, notify = activeCameraUrlChanged)
def activeCameraUrl(self) -> "QUrl":
return self._active_camera_url
@ -317,49 +292,20 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
if action_id == "View":
self._application.getController().setActiveStage("MonitorStage")
@pyqtSlot()
@pyqtSlot(name="openPrintJobControlPanel")
def openPrintJobControlPanel(self) -> None:
Logger.log("d", "Opening print job control panel...")
QDesktopServices.openUrl(QUrl("http://" + self._address + "/print_jobs"))
@pyqtSlot()
@pyqtSlot(name="openPrinterControlPanel")
def openPrinterControlPanel(self) -> None:
Logger.log("d", "Opening printer control panel...")
QDesktopServices.openUrl(QUrl("http://" + self._address + "/printers"))
@pyqtProperty("QVariantList", notify = printJobsChanged)
def printJobs(self)-> List[UM3PrintJobOutputModel]:
return self._print_jobs
@pyqtProperty(bool, notify = receivedPrintJobsChanged)
def receivedPrintJobs(self) -> bool:
return self._received_print_jobs
@pyqtProperty("QVariantList", notify = printJobsChanged)
def queuedPrintJobs(self) -> List[UM3PrintJobOutputModel]:
return [print_job for print_job in self._print_jobs if print_job.state == "queued" or print_job.state == "error"]
@pyqtProperty("QVariantList", notify = printJobsChanged)
def activePrintJobs(self) -> List[UM3PrintJobOutputModel]:
return [print_job for print_job in self._print_jobs if print_job.assignedPrinter is not None and print_job.state != "queued"]
@pyqtProperty("QVariantList", notify = _clusterPrintersChanged)
def connectedPrintersTypeCount(self) -> List[Dict[str, str]]:
printer_count = {} # type: Dict[str, int]
for printer in self._printers:
if printer.type in printer_count:
printer_count[printer.type] += 1
else:
printer_count[printer.type] = 1
result = []
for machine_type in printer_count:
result.append({"machine_type": machine_type, "count": str(printer_count[machine_type])})
return result
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def printers(self):
return self._printers
@pyqtSlot(int, result = str)
def getTimeCompleted(self, time_remaining: int) -> str:
return formatTimeCompleted(time_remaining)
@ -510,7 +456,11 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
printer = findByKey(self._printers, printer_data["uuid"])
if printer is None:
printer = self._createPrinterModel(printer_data)
output_controller = ClusterUM3PrinterOutputController(self)
printer = PrinterModelFactory.createPrinter(output_controller=output_controller,
ip_address=printer_data.get("ip_address", ""),
extruder_count=self._number_of_extruders)
self._printers.append(printer)
printer_list_changed = True
printers_seen.append(printer)
@ -524,13 +474,6 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
if removed_printers or printer_list_changed:
self.printersChanged.emit()
def _createPrinterModel(self, data: Dict[str, Any]) -> PrinterOutputModel:
printer = PrinterOutputModel(output_controller = ClusterUM3PrinterOutputController(self),
number_of_extruders = self._number_of_extruders)
printer.setCameraUrl(QUrl("http://" + data["ip_address"] + ":8080/?action=stream"))
self._printers.append(printer)
return printer
def _createPrintJobModel(self, data: Dict[str, Any]) -> UM3PrintJobOutputModel:
print_job = UM3PrintJobOutputModel(output_controller=ClusterUM3PrinterOutputController(self),
key=data["uuid"], name= data["name"])
@ -569,16 +512,8 @@ class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
if not status_set_by_impediment:
print_job.updateState(data["status"])
print_job.updateConfigurationChanges(self._createConfigurationChanges(data["configuration_changes_required"]))
def _createConfigurationChanges(self, data: List[Dict[str, Any]]) -> List[ConfigurationChangeModel]:
result = []
for change in data:
result.append(ConfigurationChangeModel(type_of_change=change["type_of_change"],
index=change["index"],
target_name=change["target_name"],
origin_name=change["origin_name"]))
return result
configuration_changes = PrinterModelFactory.createConfigurationChanges(data.get("configuration_changes_required"))
print_job.updateConfigurationChanges(configuration_changes)
def _createMaterialOutputModel(self, material_data: Dict[str, Any]) -> "MaterialOutputModel":
material_manager = self._application.getMaterialManager()

0
plugins/UM3NetworkPrinting/src/ClusterUM3PrinterOutputController.py → plugins/UM3NetworkPrinting/src/Network/ClusterUM3PrinterOutputController.py
View file

13
plugins/UM3NetworkPrinting/src/Network/ManualPrinterRequest.py Normal file
View file

@ -0,0 +1,13 @@
from typing import Optional, Callable
## Represents a request for adding a manual printer. It has the following fields:
# - address: The string of the (IP) address of the manual printer
# - callback: (Optional) Once the HTTP request to the printer to get printer information is done, whether successful
# or not, this callback will be invoked to notify about the result. The callback must have a signature of
# func(success: bool, address: str) -> None
class ManualPrinterRequest:
def __init__(self, address: str, callback: Optional[Callable[[bool, str], None]] = None) -> None:
self.address = address
self.callback = callback

23
plugins/UM3NetworkPrinting/src/Network/NetworkApiClient.py Normal file
View file

@ -0,0 +1,23 @@
# Copyright (c) 2019 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
## The network API client is responsible for handling requests and responses to printer over the local network (LAN).
class NetworkApiClient:
API_PREFIX = "/cluster-api/v1/"
def __init__(self) -> None:
pass
def getPrinters(self):
pass
def getPrintJobs(self):
pass
def requestPrint(self):
pass
def doPrintJobAction(self):
pass

425
plugins/UM3NetworkPrinting/src/Network/NetworkOutputDeviceManager.py Normal file
View file

@ -0,0 +1,425 @@
from queue import Queue
from threading import Thread, Event
from time import time
from typing import Dict, Optional, Callable, List
from PyQt5.QtCore import QUrl
from PyQt5.QtNetwork import QNetworkAccessManager, QNetworkReply, QNetworkRequest
from zeroconf import Zeroconf, ServiceBrowser, ServiceStateChange, ServiceInfo
from UM.Logger import Logger
from UM.Signal import Signal
from UM.Version import Version
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.PrinterOutputDevice import PrinterOutputDevice
from plugins.UM3NetworkPrinting.src.Network.ClusterUM3OutputDevice import ClusterUM3OutputDevice
from plugins.UM3NetworkPrinting.src.Network.ManualPrinterRequest import ManualPrinterRequest
## The NetworkOutputDeviceManager is responsible for discovering and managing local networked clusters.
class NetworkOutputDeviceManager:
PRINTER_API_VERSION = "1"
PRINTER_API_PREFIX = "/api/v" + PRINTER_API_VERSION
CLUSTER_API_VERSION = "1"
CLUSTER_API_PREFIX = "/cluster-api/v" + CLUSTER_API_VERSION
ZERO_CONF_NAME = u"_ultimaker._tcp.local."
MANUAL_DEVICES_PREFERENCE_KEY = "um3networkprinting/manual_instances"
MIN_SUPPORTED_CLUSTER_VERSION = Version("4.0.0")
discoveredDevicesChanged = Signal()
addedNetworkCluster = Signal()
removedNetworkCluster = Signal()
def __init__(self):
# Persistent dict containing the networked clusters.
self._discovered_devices = {} # type: Dict[str, ClusterUM3OutputDevice]
self._output_device_manager = CuraApplication.getInstance().getOutputDeviceManager()
self._network_manager = QNetworkAccessManager()
# In order to avoid garbage collection we keep the callbacks in this list.
self._anti_gc_callbacks = [] # type: List[Callable[[], None]]
self._zero_conf = None # type: Optional[Zeroconf]
self._zero_conf_browser = None # type: Optional[ServiceBrowser]
self._service_changed_request_queue = None # type: Optional[Queue]
self._service_changed_request_event = None # type: Optional[Event]
self._service_changed_request_thread = None # type: Optional[Thread]
# Persistent dict containing manually connected clusters.
self._manual_instances = {} # type: Dict[str, ManualPrinterRequest]
self._last_manual_entry_key = None # type: Optional[str]
# Hook up the signals for discovery.
self.addedNetworkCluster.connect(self._onAddDevice)
self.removedNetworkCluster.connect(self._onRemoveDevice)
# # Get all discovered devices in the local network.
# def getDiscoveredDevices(self) -> Dict[str, ClusterUM3OutputDevice]:
# return self._discovered_devices
# ## Get the key of the last manually added device.
# def getLastManualDevice(self) -> str:
# return self._last_manual_entry_key
# ## Reset the last manually added device key.
# def resetLastManualDevice(self) -> None:
# self._last_manual_entry_key = ""
## Force reset all network device connections.
def refreshConnections(self):
active_machine = CuraApplication.getInstance().getGlobalContainerStack()
if not active_machine:
return
um_network_key = active_machine.getMetaDataEntry("um_network_key")
for key in self._discovered_devices:
if key == um_network_key:
if not self._discovered_devices[key].isConnected():
Logger.log("d", "Attempting to connect with [%s]" % key)
# It should already be set, but if it actually connects we know for sure it's supported!
active_machine.addConfiguredConnectionType(self._discovered_devices[key].connectionType.value)
self._discovered_devices[key].connect()
self._discovered_devices[key].connectionStateChanged.connect(self._onDeviceConnectionStateChanged)
else:
self._onDeviceConnectionStateChanged(key)
else:
if self._discovered_devices[key].isConnected():
Logger.log("d", "Attempting to close connection with [%s]" % key)
self._discovered_devices[key].close()
self._discovered_devices[key].connectionStateChanged.disconnect(self._onDeviceConnectionStateChanged)
## Start the network discovery.
def start(self):
# The ZeroConf service changed requests are handled in a separate thread so we don't block the UI.
# We can also re-schedule the requests when they fail to get detailed service info.
# Any new or re-reschedule requests will be appended to the request queue and the thread will process them.
self._service_changed_request_queue = Queue()
self._service_changed_request_event = Event()
self._service_changed_request_thread = Thread(target=self._handleOnServiceChangedRequests, daemon=True)
self._service_changed_request_thread.start()
# Start network discovery.
self.stop()
self._zero_conf = Zeroconf()
self._zero_conf_browser = ServiceBrowser(self._zero_conf, self.ZERO_CONF_NAME, [
self._appendServiceChangedRequest
])
# Load all manual devices.
self._manual_instances = self._getStoredManualInstances()
for address in self._manual_instances:
if address:
self.addManualDevice(address)
# TODO: self.resetLastManualDevice()
## Stop network discovery and clean up discovered devices.
def stop(self):
# Cleanup ZeroConf resources.
if self._zero_conf is not None:
self._zero_conf.close()
self._zero_conf = None
if self._zero_conf_browser is not None:
self._zero_conf_browser.cancel()
self._zero_conf_browser = None
# Cleanup all manual devices.
for instance_name in list(self._discovered_devices):
self._onRemoveDevice(instance_name)
## Add a networked printer manually by address.
def addManualDevice(self, address: str, callback: Optional[Callable[[bool, str], None]] = None) -> None:
self._manual_instances[address] = ManualPrinterRequest(address, callback=callback)
new_manual_devices = ",".join(self._manual_instances.keys())
CuraApplication.getInstance().getPreferences().setValue(self.MANUAL_DEVICES_PREFERENCE_KEY, new_manual_devices)
key = f"manual:{address}"
if key not in self._discovered_devices:
self._onAddDevice(key, address, {
b"name": address.encode("utf-8"),
b"address": address.encode("utf-8"),
b"manual": b"true",
b"incomplete": b"true",
b"temporary": b"true"
})
self._last_manual_entry_key = key
response_callback = lambda status_code, response: self._onCheckManualDeviceResponse(status_code, address)
self._checkManualDevice(address, response_callback)
## Remove a manually added networked printer.
def removeManualDevice(self, key: str, address: Optional[str] = None) -> None:
if key not in self._discovered_devices and address is not None:
key = f"manual:{address}"
if key in self._discovered_devices:
if not address:
address = self._discovered_devices[key].ipAddress
self._onRemoveDevice(key)
# TODO: self.resetLastManualDevice()
if address in self._manual_instances:
manual_printer_request = self._manual_instances.pop(address)
new_manual_devices = ",".join(self._manual_instances.keys())
CuraApplication.getInstance().getPreferences().setValue(self.MANUAL_DEVICES_PREFERENCE_KEY,
new_manual_devices)
if manual_printer_request.callback is not None:
CuraApplication.getInstance().callLater(manual_printer_request.callback, False, address)
## Checks if a networked printer exists at the given address.
# If the printer responds it will replace the preliminary printer created from the stored manual instances.
def _checkManualDevice(self, address: str, on_finished: Callable) -> None:
Logger.log("d", "checking manual device: {}".format(address))
url = QUrl(f"http://{address}/{self.PRINTER_API_PREFIX}/system")
request = QNetworkRequest(url)
reply = self._network_manager.get(request)
self._addCallback(reply, on_finished)
## Callback for when a manual device check request was responded to.
def _onCheckManualDeviceResponse(self, status_code: int, address: str) -> None:
Logger.log("d", "manual device check response: {} {}".format(status_code, address))
if address in self._manual_instances:
callback = self._manual_instances[address].callback
if callback:
CuraApplication.getInstance().callLater(callback, status_code == 200, address)
## Returns a dict of printer BOM numbers to machine types.
# These numbers are available in the machine definition already so we just search for them here.
@staticmethod
def _getPrinterTypeIdentifiers() -> Dict[str, str]:
container_registry = CuraApplication.getInstance().getContainerRegistry()
ultimaker_machines = container_registry.findContainersMetadata(type="machine", manufacturer="Ultimaker B.V.")
found_machine_type_identifiers = {} # type: Dict[str, str]
for machine in ultimaker_machines:
machine_bom_number = machine.get("firmware_update_info", {}).get("id", None)
machine_type = machine.get("id", None)
if machine_bom_number and machine_type:
found_machine_type_identifiers[str(machine_bom_number)] = machine_type
return found_machine_type_identifiers
## Add a new device.
def _onAddDevice(self, key: str, address: str, properties: Dict[bytes, bytes]) -> None:
cluster_size = int(properties.get(b"cluster_size", -1))
printer_type = properties.get(b"machine", b"").decode("utf-8")
printer_type_identifiers = self._getPrinterTypeIdentifiers()
# Detect the machine type based on the BOM number that is sent over the network.
for bom, p_type in printer_type_identifiers.items():
if printer_type.startswith(bom):
properties[b"printer_type"] = bytes(p_type, encoding="utf8")
break
else:
properties[b"printer_type"] = b"Unknown"
# We no longer support legacy devices, so check that here.
if cluster_size == -1:
return
device = ClusterUM3OutputDevice(key, address, properties)
CuraApplication.getInstance().getDiscoveredPrintersModel().addDiscoveredPrinter(
ip_address=address,
key=device.getId(),
name=properties[b"name"].decode("utf-8"),
create_callback=self._createMachineFromDiscoveredPrinter,
machine_type=properties[b"printer_type"].decode("utf-8"),
device=device
)
self._discovered_devices[device.getId()] = device
self.discoveredDevicesChanged.emit()
global_container_stack = CuraApplication.getInstance().getGlobalContainerStack()
if global_container_stack and device.getId() == global_container_stack.getMetaDataEntry("um_network_key"):
# Ensure that the configured connection type is set.
global_container_stack.addConfiguredConnectionType(device.connectionType.value)
device.connect()
device.connectionStateChanged.connect(self._onDeviceConnectionStateChanged)
## Remove a device.
def _onRemoveDevice(self, device_id: str) -> None:
device = self._discovered_devices.pop(device_id, None)
if device:
if device.isConnected():
device.disconnect()
try:
device.connectionStateChanged.disconnect(self._onDeviceConnectionStateChanged)
except TypeError:
# Disconnect already happened.
pass
CuraApplication.getInstance().getDiscoveredPrintersModel().removeDiscoveredPrinter(device.address)
self.discoveredDevicesChanged.emit()
## Appends a service changed request so later the handling thread will pick it up and processes it.
def _appendServiceChangedRequest(self, zeroconf: Zeroconf, service_type, name: str,
state_change: ServiceStateChange) -> None:
item = (zeroconf, service_type, name, state_change)
self._service_changed_request_queue.put(item)
self._service_changed_request_event.set()
def _handleOnServiceChangedRequests(self) -> None:
while True:
# Wait for the event to be set
self._service_changed_request_event.wait(timeout=5.0)
# Stop if the application is shutting down
if CuraApplication.getInstance().isShuttingDown():
return
self._service_changed_request_event.clear()
# Handle all pending requests
reschedule_requests = [] # A list of requests that have failed so later they will get re-scheduled
while not self._service_changed_request_queue.empty():
request = self._service_changed_request_queue.get()
zeroconf, service_type, name, state_change = request
try:
result = self._onServiceChanged(zeroconf, service_type, name, state_change)
if not result:
reschedule_requests.append(request)
except Exception:
Logger.logException("e", "Failed to get service info for [%s] [%s], the request will be rescheduled",
service_type, name)
reschedule_requests.append(request)
# Re-schedule the failed requests if any
if reschedule_requests:
for request in reschedule_requests:
self._service_changed_request_queue.put(request)
## Callback handler for when the connection state of a networked device has changed.
def _onDeviceConnectionStateChanged(self, key: str) -> None:
if key not in self._discovered_devices:
return
if self._discovered_devices[key].isConnected():
um_network_key = CuraApplication.getInstance().getGlobalContainerStack().getMetaDataEntry("um_network_key")
if key != um_network_key:
return
CuraApplication.getInstance().getOutputDeviceManager().addOutputDevice(self._discovered_devices[key])
# TODO: self.checkCloudFlowIsPossible(None)
else:
CuraApplication.getInstance().getOutputDeviceManager().removeOutputDevice(key)
## Handler for zeroConf detection.
# Return True or False indicating if the process succeeded.
# Note that this function can take over 3 seconds to complete. Be careful calling it from the main thread.
def _onServiceChanged(self, zero_conf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
) -> bool:
if state_change == ServiceStateChange.Added:
return self._onServiceAdded(zero_conf, service_type, name)
elif state_change == ServiceStateChange.Removed:
return self._onServiceRemoved(name)
return True
## Handler for when a ZeroConf service was added.
def _onServiceAdded(self, zero_conf: Zeroconf, service_type: str, name: str) -> bool:
# First try getting info from zero-conf cache
info = ServiceInfo(service_type, name, properties={})
for record in zero_conf.cache.entries_with_name(name.lower()):
info.update_record(zero_conf, time(), record)
for record in zero_conf.cache.entries_with_name(info.server):
info.update_record(zero_conf, time(), record)
if info.address:
break
# Request more data if info is not complete
if not info.address:
info = zero_conf.get_service_info(service_type, name)
if info:
type_of_device = info.properties.get(b"type", None)
if type_of_device:
if type_of_device == b"printer":
address = '.'.join(map(lambda n: str(n), info.address))
self.addedNetworkCluster.emit(str(name), address, info.properties)
else:
Logger.log("w",
"The type of the found device is '%s', not 'printer'! Ignoring.." % type_of_device)
else:
Logger.log("w", "Could not get information about %s" % name)
return False
return True
## Handler for when a ZeroConf service was removed.
def _onServiceRemoved(self, name: str) -> bool:
Logger.log("d", "Bonjour service removed: %s" % name)
self.removedNetworkCluster.emit(str(name))
return True
def _associateActiveMachineWithPrinterDevice(self, printer_device: Optional["PrinterOutputDevice"]) -> None:
if not printer_device:
return
Logger.log("d", "Attempting to set the network key of the active machine to %s", printer_device.key)
machine_manager = CuraApplication.getInstance().getMachineManager()
global_container_stack = machine_manager.activeMachine
if not global_container_stack:
return
for machine in machine_manager.getMachinesInGroup(global_container_stack.getMetaDataEntry("group_id")):
machine.setMetaDataEntry("um_network_key", printer_device.key)
machine.setMetaDataEntry("group_name", printer_device.name)
# Delete old authentication data.
Logger.log("d", "Removing old authentication id %s for device %s",
global_container_stack.getMetaDataEntry("network_authentication_id", None), printer_device.key)
machine.removeMetaDataEntry("network_authentication_id")
machine.removeMetaDataEntry("network_authentication_key")
# Ensure that these containers do know that they are configured for network connection
machine.addConfiguredConnectionType(printer_device.connectionType.value)
self.refreshConnections()
## Create a machine instance based on the discovered network printer.
def _createMachineFromDiscoveredPrinter(self, key: str) -> None:
discovered_device = self._discovered_devices.get(key)
if discovered_device is None:
Logger.log("e", "Could not find discovered device with key [%s]", key)
return
group_name = discovered_device.getProperty("name")
machine_type_id = discovered_device.getProperty("printer_type")
Logger.log("i", "Creating machine from network device with key = [%s], group name = [%s], printer type = [%s]",
key, group_name, machine_type_id)
CuraApplication.getInstance().getMachineManager().addMachine(machine_type_id, group_name)
# connect the new machine to that network printer
self._associateActiveMachineWithPrinterDevice(discovered_device)
# ensure that the connection states are refreshed.
self.refreshConnections()
## Creates a callback function so that it includes the parsing of the response into the correct model.
# The callback is added to the 'finished' signal of the reply.
# \param reply: The reply that should be listened to.
# \param on_finished: The callback in case the response is successful.
def _addCallback(self, reply: QNetworkReply, on_finished: Callable) -> None:
def parse() -> None:
# Don't try to parse the reply if we didn't get one
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) is None:
return
status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
response = bytes(reply.readAll()).decode()
self._anti_gc_callbacks.remove(parse)
on_finished(int(status_code), response)
return
self._anti_gc_callbacks.append(parse)
reply.finished.connect(parse)
## Load the user-configured manual devices from Cura preferences.
def _getStoredManualInstances(self) -> Dict[str, ManualPrinterRequest]:
preferences = CuraApplication.getInstance().getPreferences()
preferences.addPreference(self.MANUAL_DEVICES_PREFERENCE_KEY, "")
manual_instances = preferences.getValue(self.MANUAL_DEVICES_PREFERENCE_KEY).split(",")
return {address: ManualPrinterRequest(address) for address in manual_instances}

0
plugins/UM3NetworkPrinting/src/Network/__init__.py Normal file
View file

7
plugins/UM3NetworkPrinting/src/SendMaterialJob.py
View file

@ -9,12 +9,11 @@ from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest
from UM.Job import Job
from UM.Logger import Logger
from cura.CuraApplication import CuraApplication
# Absolute imports don't work in plugins
from .Models import ClusterMaterial, LocalMaterial
from plugins.UM3NetworkPrinting.src.Models.ClusterMaterial import ClusterMaterial
from plugins.UM3NetworkPrinting.src.Models.LocalMaterial import LocalMaterial
if TYPE_CHECKING:
from .ClusterUM3OutputDevice import ClusterUM3OutputDevice
from plugins.UM3NetworkPrinting.src.Network.ClusterUM3OutputDevice import ClusterUM3OutputDevice
## Asynchronous job to send material profiles to the printer.

801
plugins/UM3NetworkPrinting/src/UM3OutputDevicePlugin.py
View file

@ -1,657 +1,226 @@
# Copyright (c) 2019 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json
import os
from queue import Queue
from threading import Event, Thread
from time import time
from typing import Optional, TYPE_CHECKING, Dict, Callable, Union, Any
from zeroconf import Zeroconf, ServiceBrowser, ServiceStateChange, ServiceInfo
from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply, QNetworkAccessManager
from PyQt5.QtCore import QUrl
from PyQt5.QtGui import QDesktopServices
from typing import Optional, TYPE_CHECKING, Callable
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from UM.i18n import i18nCatalog
from UM.Logger import Logger
from UM.Message import Message
from UM.OutputDevice.OutputDeviceManager import ManualDeviceAdditionAttempt
from UM.OutputDevice.OutputDevicePlugin import OutputDevicePlugin
from UM.PluginRegistry import PluginRegistry
from UM.Signal import Signal, signalemitter
from UM.Version import Version
from plugins.UM3NetworkPrinting.src.Network.NetworkOutputDeviceManager import NetworkOutputDeviceManager
from . import ClusterUM3OutputDevice, LegacyUM3OutputDevice
from .Cloud.CloudOutputDeviceManager import CloudOutputDeviceManager
from .Cloud.CloudOutputDevice import CloudOutputDevice # typing
if TYPE_CHECKING:
from PyQt5.QtNetwork import QNetworkReply
from UM.OutputDevice.OutputDevicePlugin import OutputDevicePlugin
from cura.PrinterOutput.PrinterOutputDevice import PrinterOutputDevice
from cura.Settings.GlobalStack import GlobalStack
i18n_catalog = i18nCatalog("cura")
#
# Represents a request for adding a manual printer. It has the following fields:
# - address: The string of the (IP) address of the manual printer
# - callback: (Optional) Once the HTTP request to the printer to get printer information is done, whether successful
# or not, this callback will be invoked to notify about the result. The callback must have a signature of
# func(success: bool, address: str) -> None
# - network_reply: This is the QNetworkReply instance for this request if the request has been issued and still in
# progress. It is kept here so we can cancel a request when needed.
#
class ManualPrinterRequest:
def __init__(self, address: str, callback: Optional[Callable[[bool, str], None]] = None) -> None:
self.address = address
self.callback = callback
self.network_reply = None # type: Optional["QNetworkReply"]
## This plugin handles the connection detection & creation of output device objects for the UM3 printer.
# Zero-Conf is used to detect printers, which are saved in a dict.
# If we discover a printer that has the same key as the active machine instance a connection is made.
@signalemitter
## This plugin handles the discovery and networking for Ultimaker 3D printers that support network and cloud printing.
class UM3OutputDevicePlugin(OutputDevicePlugin):
addDeviceSignal = Signal() # Called '...Signal' to avoid confusion with function-names.
removeDeviceSignal = Signal() # Ditto ^^^.
discoveredDevicesChanged = Signal()
cloudFlowIsPossible = Signal()
def __init__(self):
# cloudFlowIsPossible = Signal()
def __init__(self) -> None:
super().__init__()
self._zero_conf = None
self._zero_conf_browser = None
self._application = CuraApplication.getInstance()
# Create a network output device manager that abstracts all network connection logic away.
self._network_output_device_manager = NetworkOutputDeviceManager()
# Create a cloud output device manager that abstracts all cloud connection logic away.
self._cloud_output_device_manager = CloudOutputDeviceManager()
# Because the model needs to be created in the same thread as the QMLEngine, we use a signal.
self.addDeviceSignal.connect(self._onAddDevice)
self.removeDeviceSignal.connect(self._onRemoveDevice)
# Refresh network connections when another machine was selected in Cura.
CuraApplication.getInstance().globalContainerStackChanged.connect(self.refreshConnections)
self._application.globalContainerStackChanged.connect(self.refreshConnections)
self._discovered_devices = {}
self._network_manager = QNetworkAccessManager()
self._network_manager.finished.connect(self._onNetworkRequestFinished)
self._min_cluster_version = Version("4.0.0")
self._min_cloud_version = Version("5.2.0")
self._api_version = "1"
self._api_prefix = "/api/v" + self._api_version + "/"
self._cluster_api_version = "1"
self._cluster_api_prefix = "/cluster-api/v" + self._cluster_api_version + "/"
# Get list of manual instances from preferences
self._preferences = CuraApplication.getInstance().getPreferences()
self._preferences.addPreference("um3networkprinting/manual_instances",
"") # A comma-separated list of ip adresses or hostnames
manual_instances = self._preferences.getValue("um3networkprinting/manual_instances").split(",")
self._manual_instances = {address: ManualPrinterRequest(address)
for address in manual_instances} # type: Dict[str, ManualPrinterRequest]
# Store the last manual entry key
self._last_manual_entry_key = "" # type: str
# The zero-conf service changed requests are handled in a separate thread, so we can re-schedule the requests
# which fail to get detailed service info.
# Any new or re-scheduled requests will be appended to the request queue, and the handling thread will pick
# them up and process them.
self._service_changed_request_queue = Queue()
self._service_changed_request_event = Event()
self._service_changed_request_thread = Thread(target=self._handleOnServiceChangedRequests, daemon=True)
self._service_changed_request_thread.start()
self._account = self._application.getCuraAPI().account
# TODO: re-write cloud messaging
# self._account = self._application.getCuraAPI().account
# Check if cloud flow is possible when user logs in
self._account.loginStateChanged.connect(self.checkCloudFlowIsPossible)
# self._account.loginStateChanged.connect(self.checkCloudFlowIsPossible)
# Check if cloud flow is possible when user switches machines
self._application.globalContainerStackChanged.connect(self._onMachineSwitched)
# self._application.globalContainerStackChanged.connect(self._onMachineSwitched)
# Listen for when cloud flow is possible
self.cloudFlowIsPossible.connect(self._onCloudFlowPossible)
# self.cloudFlowIsPossible.connect(self._onCloudFlowPossible)
# Listen if cloud cluster was added
self._cloud_output_device_manager.addedCloudCluster.connect(self._onCloudPrintingConfigured)
# self._start_cloud_flow_message = None # type: Optional[Message]
# self._cloud_flow_complete_message = None # type: Optional[Message]
# Listen if cloud cluster was removed
self._cloud_output_device_manager.removedCloudCluster.connect(self.checkCloudFlowIsPossible)
# self._cloud_output_device_manager.addedCloudCluster.connect(self._onCloudPrintingConfigured)
# self._cloud_output_device_manager.removedCloudCluster.connect(self.checkCloudFlowIsPossible)
self._start_cloud_flow_message = None # type: Optional[Message]
self._cloud_flow_complete_message = None # type: Optional[Message]
def getDiscoveredDevices(self):
return self._discovered_devices
def getLastManualDevice(self) -> str:
return self._last_manual_entry_key
def resetLastManualDevice(self) -> None:
self._last_manual_entry_key = ""
## Start looking for devices on network.
## Start looking for devices in the network and cloud.
def start(self):
self.startDiscovery()
self._network_output_device_manager.start()
self._cloud_output_device_manager.start()
def startDiscovery(self):
self.stop()
if self._zero_conf_browser:
self._zero_conf_browser.cancel()
self._zero_conf_browser = None # Force the old ServiceBrowser to be destroyed.
for instance_name in list(self._discovered_devices):
self._onRemoveDevice(instance_name)
self._zero_conf = Zeroconf()
self._zero_conf_browser = ServiceBrowser(self._zero_conf, u'_ultimaker._tcp.local.',
[self._appendServiceChangedRequest])
# Look for manual instances from preference
for address in self._manual_instances:
if address:
self.addManualDevice(address)
self.resetLastManualDevice()
# TODO: CHANGE TO HOSTNAME
def refreshConnections(self):
active_machine = CuraApplication.getInstance().getGlobalContainerStack()
if not active_machine:
return
um_network_key = active_machine.getMetaDataEntry("um_network_key")
for key in self._discovered_devices:
if key == um_network_key:
if not self._discovered_devices[key].isConnected():
Logger.log("d", "Attempting to connect with [%s]" % key)
# It should already be set, but if it actually connects we know for sure it's supported!
active_machine.addConfiguredConnectionType(self._discovered_devices[key].connectionType.value)
self._discovered_devices[key].connect()
self._discovered_devices[key].connectionStateChanged.connect(self._onDeviceConnectionStateChanged)
else:
self._onDeviceConnectionStateChanged(key)
else:
if self._discovered_devices[key].isConnected():
Logger.log("d", "Attempting to close connection with [%s]" % key)
self._discovered_devices[key].close()
self._discovered_devices[key].connectionStateChanged.disconnect(self._onDeviceConnectionStateChanged)
def _onDeviceConnectionStateChanged(self, key):
if key not in self._discovered_devices:
return
if self._discovered_devices[key].isConnected():
# Sometimes the status changes after changing the global container and maybe the device doesn't belong to this machine
um_network_key = CuraApplication.getInstance().getGlobalContainerStack().getMetaDataEntry("um_network_key")
if key == um_network_key:
self.getOutputDeviceManager().addOutputDevice(self._discovered_devices[key])
self.checkCloudFlowIsPossible(None)
else:
self.getOutputDeviceManager().removeOutputDevice(key)
def stop(self):
if self._zero_conf is not None:
Logger.log("d", "zeroconf close...")
self._zero_conf.close()
# Stop network and cloud discovery.
def stop(self) -> None:
self._network_output_device_manager.stop()
self._cloud_output_device_manager.stop()
## Force refreshing the network connections.
def refreshConnections(self) -> None:
self._network_output_device_manager.refreshConnections()
self._cloud_output_device_manager.refreshConnections()
## Indicate that this plugin supports adding networked printers manually.
def canAddManualDevice(self, address: str = "") -> ManualDeviceAdditionAttempt:
# This plugin should always be the fallback option (at least try it):
return ManualDeviceAdditionAttempt.POSSIBLE
def removeManualDevice(self, key: str, address: Optional[str] = None) -> None:
if key not in self._discovered_devices and address is not None:
key = "manual:%s" % address
if key in self._discovered_devices:
if not address:
address = self._discovered_devices[key].ipAddress
self._onRemoveDevice(key)
self.resetLastManualDevice()
if address in self._manual_instances:
manual_printer_request = self._manual_instances.pop(address)
self._preferences.setValue("um3networkprinting/manual_instances", ",".join(self._manual_instances.keys()))
if manual_printer_request.network_reply is not None:
manual_printer_request.network_reply.abort()
if manual_printer_request.callback is not None:
self._application.callLater(manual_printer_request.callback, False, address)
## Add a networked printer manually based on its network address.
def addManualDevice(self, address: str, callback: Optional[Callable[[bool, str], None]] = None) -> None:
self._manual_instances[address] = ManualPrinterRequest(address, callback = callback)
self._preferences.setValue("um3networkprinting/manual_instances", ",".join(self._manual_instances.keys()))
instance_name = "manual:%s" % address
properties = {
b"name": address.encode("utf-8"),
b"address": address.encode("utf-8"),
b"manual": b"true",
b"incomplete": b"true",
b"temporary": b"true" # Still a temporary device until all the info is retrieved in _onNetworkRequestFinished
}
if instance_name not in self._discovered_devices:
# Add a preliminary printer instance
self._onAddDevice(instance_name, address, properties)
self._last_manual_entry_key = instance_name
reply = self._checkManualDevice(address)
self._manual_instances[address].network_reply = reply
def _createMachineFromDiscoveredPrinter(self, key: str) -> None:
discovered_device = self._discovered_devices.get(key)
if discovered_device is None:
Logger.log("e", "Could not find discovered device with key [%s]", key)
return
group_name = discovered_device.getProperty("name")
machine_type_id = discovered_device.getProperty("printer_type")
Logger.log("i", "Creating machine from network device with key = [%s], group name = [%s], printer type = [%s]",
key, group_name, machine_type_id)
self._application.getMachineManager().addMachine(machine_type_id, group_name)
# connect the new machine to that network printer
self.associateActiveMachineWithPrinterDevice(discovered_device)
# ensure that the connection states are refreshed.
self.refreshConnections()
def associateActiveMachineWithPrinterDevice(self, printer_device: Optional["PrinterOutputDevice"]) -> None:
if not printer_device:
return
Logger.log("d", "Attempting to set the network key of the active machine to %s", printer_device.key)
machine_manager = CuraApplication.getInstance().getMachineManager()
global_container_stack = machine_manager.activeMachine
if not global_container_stack:
return
for machine in machine_manager.getMachinesInGroup(global_container_stack.getMetaDataEntry("group_id")):
machine.setMetaDataEntry("um_network_key", printer_device.key)
machine.setMetaDataEntry("group_name", printer_device.name)
# Delete old authentication data.
Logger.log("d", "Removing old authentication id %s for device %s",
global_container_stack.getMetaDataEntry("network_authentication_id", None), printer_device.key)
machine.removeMetaDataEntry("network_authentication_id")
machine.removeMetaDataEntry("network_authentication_key")
# Ensure that these containers do know that they are configured for network connection
machine.addConfiguredConnectionType(printer_device.connectionType.value)
self.refreshConnections()
def _checkManualDevice(self, address: str) -> "QNetworkReply":
# Check if a UM3 family device exists at this address.
# If a printer responds, it will replace the preliminary printer created above
# origin=manual is for tracking back the origin of the call
url = QUrl("http://" + address + self._api_prefix + "system")
name_request = QNetworkRequest(url)
return self._network_manager.get(name_request)
def _onNetworkRequestFinished(self, reply: "QNetworkReply") -> None:
reply_url = reply.url().toString()
address = reply.url().host()
device = None
properties = {} # type: Dict[bytes, bytes]
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200:
# Either:
# - Something went wrong with checking the firmware version!
# - Something went wrong with checking the amount of printers the cluster has!
# - Couldn't find printer at the address when trying to add it manually.
if address in self._manual_instances:
key = "manual:" + address
self.removeManualDevice(key, address)
return
if "system" in reply_url:
try:
system_info = json.loads(bytes(reply.readAll()).decode("utf-8"))
except:
Logger.log("e", "Something went wrong converting the JSON.")
return
if address in self._manual_instances:
manual_printer_request = self._manual_instances[address]
manual_printer_request.network_reply = None
if manual_printer_request.callback is not None:
self._application.callLater(manual_printer_request.callback, True, address)
has_cluster_capable_firmware = Version(system_info["firmware"]) > self._min_cluster_version
instance_name = "manual:%s" % address
properties = {
b"name": (system_info["name"] + " (manual)").encode("utf-8"),
b"address": address.encode("utf-8"),
b"firmware_version": system_info["firmware"].encode("utf-8"),
b"manual": b"true",
b"machine": str(system_info['hardware']["typeid"]).encode("utf-8")
}
if has_cluster_capable_firmware:
# Cluster needs an additional request, before it's completed.
properties[b"incomplete"] = b"true"
# Check if the device is still in the list & re-add it with the updated
# information.
if instance_name in self._discovered_devices:
self._onRemoveDevice(instance_name)
self._onAddDevice(instance_name, address, properties)
if has_cluster_capable_firmware:
# We need to request more info in order to figure out the size of the cluster.
cluster_url = QUrl("http://" + address + self._cluster_api_prefix + "printers/")
cluster_request = QNetworkRequest(cluster_url)
self._network_manager.get(cluster_request)
elif "printers" in reply_url:
# So we confirmed that the device is in fact a cluster printer, and we should now know how big it is.
try:
cluster_printers_list = json.loads(bytes(reply.readAll()).decode("utf-8"))
except:
Logger.log("e", "Something went wrong converting the JSON.")
return
instance_name = "manual:%s" % address
if instance_name in self._discovered_devices:
device = self._discovered_devices[instance_name]
properties = device.getProperties().copy()
if b"incomplete" in properties:
del properties[b"incomplete"]
properties[b"cluster_size"] = str(len(cluster_printers_list)).encode("utf-8")
self._onRemoveDevice(instance_name)
self._onAddDevice(instance_name, address, properties)
def _onRemoveDevice(self, device_id: str) -> None:
device = self._discovered_devices.pop(device_id, None)
if device:
if device.isConnected():
device.disconnect()
try:
device.connectionStateChanged.disconnect(self._onDeviceConnectionStateChanged)
except TypeError:
# Disconnect already happened.
pass
self._application.getDiscoveredPrintersModel().removeDiscoveredPrinter(device.address)
self.discoveredDevicesChanged.emit()
## Returns a dict of printer BOM numbers to machine types.
# These numbers are available in the machine definition already so we just search for them here.
def _getPrinterTypeIdentifiers(self) -> Dict[str, str]:
container_registry = self._application.getContainerRegistry()
ultimaker_machines = container_registry.findContainersMetadata(type="machine", manufacturer="Ultimaker B.V.")
found_machine_type_identifiers = {} # type: Dict[str, str]
for machine in ultimaker_machines:
machine_bom_number = machine.get("firmware_update_info", {}).get("id", None)
machine_type = machine.get("id", None)
if machine_bom_number and machine_type:
found_machine_type_identifiers[str(machine_bom_number)] = machine_type
return found_machine_type_identifiers
def _onAddDevice(self, name, address, properties):
# Check what kind of device we need to add; Depending on the firmware we either add a "Connect"/"Cluster"
# or "Legacy" UM3 device.
cluster_size = int(properties.get(b"cluster_size", -1))
printer_type = properties.get(b"machine", b"").decode("utf-8")
printer_type_identifiers = self._getPrinterTypeIdentifiers()
for key, value in printer_type_identifiers.items():
if printer_type.startswith(key):
properties[b"printer_type"] = bytes(value, encoding="utf8")
break
else:
properties[b"printer_type"] = b"Unknown"
if cluster_size >= 0:
device = ClusterUM3OutputDevice.ClusterUM3OutputDevice(name, address, properties)
else:
device = LegacyUM3OutputDevice.LegacyUM3OutputDevice(name, address, properties)
self._application.getDiscoveredPrintersModel().addDiscoveredPrinter(
address, device.getId(), properties[b"name"].decode("utf-8"), self._createMachineFromDiscoveredPrinter,
properties[b"printer_type"].decode("utf-8"), device)
self._discovered_devices[device.getId()] = device
self.discoveredDevicesChanged.emit()
global_container_stack = CuraApplication.getInstance().getGlobalContainerStack()
if global_container_stack and device.getId() == global_container_stack.getMetaDataEntry("um_network_key"):
# Ensure that the configured connection type is set.
global_container_stack.addConfiguredConnectionType(device.connectionType.value)
device.connect()
device.connectionStateChanged.connect(self._onDeviceConnectionStateChanged)
## Appends a service changed request so later the handling thread will pick it up and processes it.
def _appendServiceChangedRequest(self, zeroconf, service_type, name, state_change):
# append the request and set the event so the event handling thread can pick it up
item = (zeroconf, service_type, name, state_change)
self._service_changed_request_queue.put(item)
self._service_changed_request_event.set()
def _handleOnServiceChangedRequests(self):
while True:
# Wait for the event to be set
self._service_changed_request_event.wait(timeout = 5.0)
# Stop if the application is shutting down
if CuraApplication.getInstance().isShuttingDown():
return
self._service_changed_request_event.clear()
# Handle all pending requests
reschedule_requests = [] # A list of requests that have failed so later they will get re-scheduled
while not self._service_changed_request_queue.empty():
request = self._service_changed_request_queue.get()
zeroconf, service_type, name, state_change = request
try:
result = self._onServiceChanged(zeroconf, service_type, name, state_change)
if not result:
reschedule_requests.append(request)
except Exception:
Logger.logException("e", "Failed to get service info for [%s] [%s], the request will be rescheduled",
service_type, name)
reschedule_requests.append(request)
# Re-schedule the failed requests if any
if reschedule_requests:
for request in reschedule_requests:
self._service_changed_request_queue.put(request)
## Handler for zeroConf detection.
# Return True or False indicating if the process succeeded.
# Note that this function can take over 3 seconds to complete. Be careful
# calling it from the main thread.
def _onServiceChanged(self, zero_conf, service_type, name, state_change):
if state_change == ServiceStateChange.Added:
# First try getting info from zero-conf cache
info = ServiceInfo(service_type, name, properties = {})
for record in zero_conf.cache.entries_with_name(name.lower()):
info.update_record(zero_conf, time(), record)
for record in zero_conf.cache.entries_with_name(info.server):
info.update_record(zero_conf, time(), record)
if info.address:
break
# Request more data if info is not complete
if not info.address:
info = zero_conf.get_service_info(service_type, name)
if info:
type_of_device = info.properties.get(b"type", None)
if type_of_device:
if type_of_device == b"printer":
address = '.'.join(map(lambda n: str(n), info.address))
self.addDeviceSignal.emit(str(name), address, info.properties)
else:
Logger.log("w",
"The type of the found device is '%s', not 'printer'! Ignoring.." % type_of_device)
else:
Logger.log("w", "Could not get information about %s" % name)
return False
elif state_change == ServiceStateChange.Removed:
Logger.log("d", "Bonjour service removed: %s" % name)
self.removeDeviceSignal.emit(str(name))
return True
## Check if the prerequsites are in place to start the cloud flow
def checkCloudFlowIsPossible(self, cluster: Optional[CloudOutputDevice]) -> None:
Logger.log("d", "Checking if cloud connection is possible...")
# Pre-Check: Skip if active machine already has been cloud connected or you said don't ask again
active_machine = self._application.getMachineManager().activeMachine # type: Optional[GlobalStack]
if active_machine:
# Check 1A: Printer isn't already configured for cloud
if ConnectionType.CloudConnection.value in active_machine.configuredConnectionTypes:
Logger.log("d", "Active machine was already configured for cloud.")
return
# Check 1B: Printer isn't already configured for cloud
if active_machine.getMetaDataEntry("cloud_flow_complete", False):
Logger.log("d", "Active machine was already configured for cloud.")
return
# Check 2: User did not already say "Don't ask me again"
if active_machine.getMetaDataEntry("do_not_show_cloud_message", False):
Logger.log("d", "Active machine shouldn't ask about cloud anymore.")
return
# Check 3: User is logged in with an Ultimaker account
if not self._account.isLoggedIn:
Logger.log("d", "Cloud Flow not possible: User not logged in!")
return
# Check 4: Machine is configured for network connectivity
if not self._application.getMachineManager().activeMachineHasNetworkConnection:
Logger.log("d", "Cloud Flow not possible: Machine is not connected!")
return
# Check 5: Machine has correct firmware version
firmware_version = self._application.getMachineManager().activeMachineFirmwareVersion # type: str
if not Version(firmware_version) > self._min_cloud_version:
Logger.log("d", "Cloud Flow not possible: Machine firmware (%s) is too low! (Requires version %s)",
firmware_version,
self._min_cloud_version)
return
Logger.log("d", "Cloud flow is possible!")
self.cloudFlowIsPossible.emit()
def _onCloudFlowPossible(self) -> None:
# Cloud flow is possible, so show the message
if not self._start_cloud_flow_message:
self._createCloudFlowStartMessage()
if self._start_cloud_flow_message and not self._start_cloud_flow_message.visible:
self._start_cloud_flow_message.show()
def _onCloudPrintingConfigured(self, device) -> None:
# Hide the cloud flow start message if it was hanging around already
# For example: if the user already had the browser openen and made the association themselves
if self._start_cloud_flow_message and self._start_cloud_flow_message.visible:
self._start_cloud_flow_message.hide()
# Cloud flow is complete, so show the message
if not self._cloud_flow_complete_message:
self._createCloudFlowCompleteMessage()
if self._cloud_flow_complete_message and not self._cloud_flow_complete_message.visible:
self._cloud_flow_complete_message.show()
# Set the machine's cloud flow as complete so we don't ask the user again and again for cloud connected printers
active_machine = self._application.getMachineManager().activeMachine
if active_machine:
# The active machine _might_ not be the machine that was in the added cloud cluster and
# then this will hide the cloud message for the wrong machine. So we only set it if the
# host names match between the active machine and the newly added cluster
saved_host_name = active_machine.getMetaDataEntry("um_network_key", "").split('.')[0]
added_host_name = device.toDict()["host_name"]
if added_host_name == saved_host_name:
active_machine.setMetaDataEntry("do_not_show_cloud_message", True)
return
def _onDontAskMeAgain(self, checked: bool) -> None:
active_machine = self._application.getMachineManager().activeMachine # type: Optional[GlobalStack]
if active_machine:
active_machine.setMetaDataEntry("do_not_show_cloud_message", checked)
if checked:
Logger.log("d", "Will not ask the user again to cloud connect for current printer.")
return
def _onCloudFlowStarted(self, messageId: str, actionId: str) -> None:
address = self._application.getMachineManager().activeMachineAddress # type: str
if address:
QDesktopServices.openUrl(QUrl("http://" + address + "/cloud_connect"))
if self._start_cloud_flow_message:
self._start_cloud_flow_message.hide()
self._start_cloud_flow_message = None
return
def _onReviewCloudConnection(self, messageId: str, actionId: str) -> None:
address = self._application.getMachineManager().activeMachineAddress # type: str
if address:
QDesktopServices.openUrl(QUrl("http://" + address + "/settings"))
return
def _onMachineSwitched(self) -> None:
# Hide any left over messages
if self._start_cloud_flow_message is not None and self._start_cloud_flow_message.visible:
self._start_cloud_flow_message.hide()
if self._cloud_flow_complete_message is not None and self._cloud_flow_complete_message.visible:
self._cloud_flow_complete_message.hide()
# Check for cloud flow again with newly selected machine
self.checkCloudFlowIsPossible(None)
def _createCloudFlowStartMessage(self):
self._start_cloud_flow_message = Message(
text = i18n_catalog.i18nc("@info:status", "Send and monitor print jobs from anywhere using your Ultimaker account."),
lifetime = 0,
image_source = QUrl.fromLocalFile(os.path.join(
PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"),
"resources", "svg", "cloud-flow-start.svg"
)),
image_caption = i18n_catalog.i18nc("@info:status Ultimaker Cloud is a brand name and shouldn't be translated.", "Connect to Ultimaker Cloud"),
option_text = i18n_catalog.i18nc("@action", "Don't ask me again for this printer."),
option_state = False
)
self._start_cloud_flow_message.addAction("", i18n_catalog.i18nc("@action", "Get started"), "", "")
self._start_cloud_flow_message.optionToggled.connect(self._onDontAskMeAgain)
self._start_cloud_flow_message.actionTriggered.connect(self._onCloudFlowStarted)
def _createCloudFlowCompleteMessage(self):
self._cloud_flow_complete_message = Message(
text = i18n_catalog.i18nc("@info:status", "You can now send and monitor print jobs from anywhere using your Ultimaker account."),
lifetime = 30,
image_source = QUrl.fromLocalFile(os.path.join(
PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"),
"resources", "svg", "cloud-flow-completed.svg"
)),
image_caption = i18n_catalog.i18nc("@info:status", "Connected!")
)
self._cloud_flow_complete_message.addAction("", i18n_catalog.i18nc("@action", "Review your connection"), "", "", 1) # TODO: Icon
self._cloud_flow_complete_message.actionTriggered.connect(self._onReviewCloudConnection)
self._network_output_device_manager.addManualDevice(address, callback)
## Remove a manually connected networked printer.
def removeManualDevice(self, key: str, address: Optional[str] = None) -> None:
self._network_output_device_manager.removeManualDevice(key, address)
# ## Get the last manual device attempt.
# # Used by the DiscoverUM3Action.
# def getLastManualDevice(self) -> str:
# return self._network_output_device_manager.getLastManualDevice()
# ## Reset the last manual device attempt.
# # Used by the DiscoverUM3Action.
# def resetLastManualDevice(self) -> None:
# self._network_output_device_manager.resetLastManualDevice()
# ## Check if the prerequsites are in place to start the cloud flow
# def checkCloudFlowIsPossible(self, cluster: Optional[CloudOutputDevice]) -> None:
# Logger.log("d", "Checking if cloud connection is possible...")
#
# # Pre-Check: Skip if active machine already has been cloud connected or you said don't ask again
# active_machine = self._application.getMachineManager().activeMachine # type: Optional[GlobalStack]
# if active_machine:
# # Check 1A: Printer isn't already configured for cloud
# if ConnectionType.CloudConnection.value in active_machine.configuredConnectionTypes:
# Logger.log("d", "Active machine was already configured for cloud.")
# return
#
# # Check 1B: Printer isn't already configured for cloud
# if active_machine.getMetaDataEntry("cloud_flow_complete", False):
# Logger.log("d", "Active machine was already configured for cloud.")
# return
#
# # Check 2: User did not already say "Don't ask me again"
# if active_machine.getMetaDataEntry("do_not_show_cloud_message", False):
# Logger.log("d", "Active machine shouldn't ask about cloud anymore.")
# return
#
# # Check 3: User is logged in with an Ultimaker account
# if not self._account.isLoggedIn:
# Logger.log("d", "Cloud Flow not possible: User not logged in!")
# return
#
# # Check 4: Machine is configured for network connectivity
# if not self._application.getMachineManager().activeMachineHasNetworkConnection:
# Logger.log("d", "Cloud Flow not possible: Machine is not connected!")
# return
#
# # Check 5: Machine has correct firmware version
# firmware_version = self._application.getMachineManager().activeMachineFirmwareVersion # type: str
# if not Version(firmware_version) > self._min_cloud_version:
# Logger.log("d", "Cloud Flow not possible: Machine firmware (%s) is too low! (Requires version %s)",
# firmware_version,
# self._min_cloud_version)
# return
#
# Logger.log("d", "Cloud flow is possible!")
# self.cloudFlowIsPossible.emit()
# def _onCloudFlowPossible(self) -> None:
# # Cloud flow is possible, so show the message
# if not self._start_cloud_flow_message:
# self._createCloudFlowStartMessage()
# if self._start_cloud_flow_message and not self._start_cloud_flow_message.visible:
# self._start_cloud_flow_message.show()
# def _onCloudPrintingConfigured(self, device) -> None:
# # Hide the cloud flow start message if it was hanging around already
# # For example: if the user already had the browser openen and made the association themselves
# if self._start_cloud_flow_message and self._start_cloud_flow_message.visible:
# self._start_cloud_flow_message.hide()
#
# # Cloud flow is complete, so show the message
# if not self._cloud_flow_complete_message:
# self._createCloudFlowCompleteMessage()
# if self._cloud_flow_complete_message and not self._cloud_flow_complete_message.visible:
# self._cloud_flow_complete_message.show()
#
# # Set the machine's cloud flow as complete so we don't ask the user again and again for cloud connected printers
# active_machine = self._application.getMachineManager().activeMachine
# if active_machine:
#
# # The active machine _might_ not be the machine that was in the added cloud cluster and
# # then this will hide the cloud message for the wrong machine. So we only set it if the
# # host names match between the active machine and the newly added cluster
# saved_host_name = active_machine.getMetaDataEntry("um_network_key", "").split('.')[0]
# added_host_name = device.toDict()["host_name"]
#
# if added_host_name == saved_host_name:
# active_machine.setMetaDataEntry("do_not_show_cloud_message", True)
#
# return
# def _onDontAskMeAgain(self, checked: bool) -> None:
# active_machine = self._application.getMachineManager().activeMachine # type: Optional[GlobalStack]
# if active_machine:
# active_machine.setMetaDataEntry("do_not_show_cloud_message", checked)
# if checked:
# Logger.log("d", "Will not ask the user again to cloud connect for current printer.")
# return
# def _onCloudFlowStarted(self, messageId: str, actionId: str) -> None:
# address = self._application.getMachineManager().activeMachineAddress # type: str
# if address:
# QDesktopServices.openUrl(QUrl("http://" + address + "/cloud_connect"))
# if self._start_cloud_flow_message:
# self._start_cloud_flow_message.hide()
# self._start_cloud_flow_message = None
# return
# def _onReviewCloudConnection(self, messageId: str, actionId: str) -> None:
# address = self._application.getMachineManager().activeMachineAddress # type: str
# if address:
# QDesktopServices.openUrl(QUrl("http://" + address + "/settings"))
# return
# def _onMachineSwitched(self) -> None:
# # Hide any left over messages
# if self._start_cloud_flow_message is not None and self._start_cloud_flow_message.visible:
# self._start_cloud_flow_message.hide()
# if self._cloud_flow_complete_message is not None and self._cloud_flow_complete_message.visible:
# self._cloud_flow_complete_message.hide()
#
# # Check for cloud flow again with newly selected machine
# self.checkCloudFlowIsPossible(None)
# def _createCloudFlowStartMessage(self):
# self._start_cloud_flow_message = Message(
# text = i18n_catalog.i18nc("@info:status", "Send and monitor print jobs from anywhere using your Ultimaker account."),
# lifetime = 0,
# image_source = QUrl.fromLocalFile(os.path.join(
# PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"),
# "resources", "svg", "cloud-flow-start.svg"
# )),
# image_caption = i18n_catalog.i18nc("@info:status Ultimaker Cloud is a brand name and shouldn't be translated.", "Connect to Ultimaker Cloud"),
# option_text = i18n_catalog.i18nc("@action", "Don't ask me again for this printer."),
# option_state = False
# )
# self._start_cloud_flow_message.addAction("", i18n_catalog.i18nc("@action", "Get started"), "", "")
# self._start_cloud_flow_message.optionToggled.connect(self._onDontAskMeAgain)
# self._start_cloud_flow_message.actionTriggered.connect(self._onCloudFlowStarted)
# def _createCloudFlowCompleteMessage(self):
# self._cloud_flow_complete_message = Message(
# text = i18n_catalog.i18nc("@info:status", "You can now send and monitor print jobs from anywhere using your Ultimaker account."),
# lifetime = 30,
# image_source = QUrl.fromLocalFile(os.path.join(
# PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting"),
# "resources", "svg", "cloud-flow-completed.svg"
# )),
# image_caption = i18n_catalog.i18nc("@info:status", "Connected!")
# )
# self._cloud_flow_complete_message.addAction("", i18n_catalog.i18nc("@action", "Review your connection"), "", "", 1) # TODO: Icon
# self._cloud_flow_complete_message.actionTriggered.connect(self._onReviewCloudConnection)

102
plugins/UM3NetworkPrinting/src/UltimakerNetworkedPrinterOutputDevice.py Normal file
View file

@ -0,0 +1,102 @@
# Copyright (c) 2019 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import List, Optional, Dict
from PyQt5.QtCore import pyqtProperty, pyqtSignal, QObject, pyqtSlot
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
## Output device class that forms the basis of Ultimaker networked printer output devices.
# Currently used for local networking and cloud printing using Ultimaker Connect.
# This base class primarily contains all the Qt properties and slots needed for the monitor page to work.
class UltimakerNetworkedPrinterOutputDevice(NetworkedPrinterOutputDevice):
# Signal emitted when the status of the print jobs for this cluster were changed over the network.
printJobsChanged = pyqtSignal()
# Signal emitted when the currently visible printer card in the UI was changed by the user.
activePrinterChanged = pyqtSignal()
# Notify can only use signals that are defined by the class that they are in, not inherited ones.
# Therefore we create a private signal used to trigger the printersChanged signal.
_clusterPrintersChanged = pyqtSignal()
# States indicating if a print job is queued.
QUEUED_PRINT_JOBS_STATES = {"queued", "error"}
def __init__(self, device_id, address, properties, connection_type, parent=None) -> None:
super().__init__(device_id=device_id, address=address, properties=properties, connection_type=connection_type,
parent=parent)
# Keeps track of all print jobs in the cluster.
self._print_jobs = [] # type: List[UM3PrintJobOutputModel]
# Keep track of the printer currently selected in the UI.
self._active_printer = None # type: Optional[PrinterOutputModel]
# By default we are not authenticated. This state will be changed later.
self._authentication_state = AuthState.NotAuthenticated
# Get all print jobs for this cluster.
@pyqtProperty("QVariantList", notify=printJobsChanged)
def printJobs(self) -> List[UM3PrintJobOutputModel]:
return self._print_jobs
# Get all print jobs for this cluster that are queued.
@pyqtProperty("QVariantList", notify=printJobsChanged)
def queuedPrintJobs(self) -> List[UM3PrintJobOutputModel]:
return [print_job for print_job in self._print_jobs if print_job.state in self.QUEUED_PRINT_JOBS_STATES]
# Get all print jobs for this cluster that are currently printing.
@pyqtProperty("QVariantList", notify=printJobsChanged)
def activePrintJobs(self) -> List[UM3PrintJobOutputModel]:
return [print_job for print_job in self._print_jobs if
print_job.assignedPrinter is not None and print_job.state not in self.QUEUED_PRINT_JOBS_STATES]
# Get the amount of printers in the cluster.
@pyqtProperty(int, notify=_clusterPrintersChanged)
def clusterSize(self) -> int:
return self._cluster_size
# Get the amount of printer in the cluster per type.
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def connectedPrintersTypeCount(self) -> List[Dict[str, str]]:
printer_count = {} # type: Dict[str, int]
for printer in self._printers:
if printer.type in printer_count:
printer_count[printer.type] += 1
else:
printer_count[printer.type] = 1
result = []
for machine_type in printer_count:
result.append({"machine_type": machine_type, "count": str(printer_count[machine_type])})
return result
# Get a list of all printers.
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def printers(self) -> List[PrinterOutputModel]:
return self._printers
# Get the currently active printer in the UI.
@pyqtProperty(QObject, notify=activePrinterChanged)
def activePrinter(self) -> Optional[PrinterOutputModel]:
return self._active_printer
# Set the currently active printer from the UI.
@pyqtSlot(QObject, name="setActivePrinter")
def setActivePrinter(self, printer: Optional[PrinterOutputModel]) -> None:
if self.activePrinter == printer:
return
self._active_printer = printer
self.activePrinterChanged.emit()
@pyqtSlot(name="openPrintJobControlPanel")
def openPrintJobControlPanel(self) -> None:
raise NotImplementedError("openPrintJobControlPanel must be implemented")
@pyqtSlot(name="openPrinterControlPanel")
def openPrinterControlPanel(self) -> None:
raise NotImplementedError("openPrinterControlPanel must be implemented")

16
plugins/UM3NetworkPrinting/tests/Cloud/TestCloudApiClient.py
View file

@ -7,11 +7,11 @@ from unittest.mock import patch, MagicMock
from cura.UltimakerCloudAuthentication import CuraCloudAPIRoot
from ...src.Cloud import CloudApiClient
from ...src.Cloud.Models.CloudClusterResponse import CloudClusterResponse
from ...src.Cloud.Models.CloudClusterStatus import CloudClusterStatus
from ...src.Cloud.Models.CloudPrintJobResponse import CloudPrintJobResponse
from ...src.Cloud.Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from ...src.Cloud.Models.CloudError import CloudError
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudClusterStatus import CloudClusterStatus
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobResponse import CloudPrintJobResponse
from plugins.UM3NetworkPrinting.src.Models.CloudPrintJobUploadRequest import CloudPrintJobUploadRequest
from plugins.UM3NetworkPrinting.src.Models.CloudError import CloudError
from .Fixtures import readFixture, parseFixture
from .NetworkManagerMock import NetworkManagerMock
@ -60,7 +60,7 @@ class TestCloudApiClient(TestCase):
self.assertEqual([CloudClusterStatus(**data)], result)
def test_requestUpload(self):
results = []
response = readFixture("putJobUploadResponse")
@ -74,7 +74,7 @@ class TestCloudApiClient(TestCase):
self.assertEqual(["uploading"], [r.status for r in results])
def test_uploadToolPath(self):
results = []
progress = MagicMock()
@ -94,7 +94,7 @@ class TestCloudApiClient(TestCase):
self.assertEqual(["sent"], results)
def test_requestPrint(self):
results = []
response = readFixture("postJobPrintResponse")

6
plugins/UM3NetworkPrinting/tests/Cloud/TestCloudOutputDevice.py
View file

@ -9,7 +9,7 @@ from cura.UltimakerCloudAuthentication import CuraCloudAPIRoot
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from ...src.Cloud import CloudApiClient
from ...src.Cloud.CloudOutputDevice import CloudOutputDevice
from ...src.Cloud.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from .Fixtures import readFixture, parseFixture
from .NetworkManagerMock import NetworkManagerMock
@ -46,7 +46,7 @@ class TestCloudOutputDevice(TestCase):
self.onError = MagicMock()
with patch.object(CloudApiClient, "QNetworkAccessManager", return_value = self.network):
self._api = CloudApiClient.CloudApiClient(self.account, self.onError)
self.device = CloudOutputDevice(self._api, self.cluster)
self.cluster_status = parseFixture("getClusterStatusResponse")
self.network.prepareReply("GET", self.STATUS_URL, 200, readFixture("getClusterStatusResponse"))
@ -138,7 +138,7 @@ class TestCloudOutputDevice(TestCase):
}, {
"extension": "gcode.gz",
"mime_type": "application/gzip",
"mode": 2,
"mode": 2,
}]
file_handler.getWriterByMimeType.return_value.write.side_effect = \
lambda stream, nodes: stream.write(str(nodes).encode())

2
plugins/UM3NetworkPrinting/tests/Cloud/TestCloudOutputDeviceManager.py
View file

@ -7,7 +7,7 @@ from UM.OutputDevice.OutputDeviceManager import OutputDeviceManager
from cura.UltimakerCloudAuthentication import CuraCloudAPIRoot
from ...src.Cloud import CloudApiClient
from ...src.Cloud import CloudOutputDeviceManager
from ...src.Cloud.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from .Fixtures import parseFixture, readFixture
from .NetworkManagerMock import NetworkManagerMock, FakeSignal

0
plugins/__init__.py Normal file
View file