3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-08 23:46:22 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Start inheriting both output devices from a base device

Browse source
This commit is contained in:
ChrisTerBeke 2019-07-26 17:05:39 +02:00
parent bd4f0c4e25
commit 4268c011a7
5 changed files with 79 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3,7 +3,7 @@
import os
from time import time
from typing import Dict, List, Optional, Set, cast
from typing import List, Optional, Set, cast
from PyQt5.QtCore import QObject, QUrl, pyqtProperty, pyqtSignal, pyqtSlot
from PyQt5.QtGui import QDesktopServices
@ -14,14 +14,13 @@ from UM.FileHandler.FileHandler import FileHandler
from UM.Logger import Logger
from UM.Message import Message
from UM.PluginRegistry import PluginRegistry
from UM.Qt.Duration import Duration, DurationFormat
from UM.Scene.SceneNode import SceneNode
from UM.Version import Version
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState, NetworkedPrinterOutputDevice
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState
from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from plugins.UM3NetworkPrinting.src.UltimakerNetworkedPrinterOutputDevice import UltimakerNetworkedPrinterOutputDevice
from .CloudOutputController import CloudOutputController
from ..MeshFormatHandler import MeshFormatHandler
@ -35,7 +34,7 @@ from plugins.UM3NetworkPrinting.src.Models.CloudPrintResponse import CloudPrintR
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")
@ -44,7 +43,8 @@ I18N_CATALOG = i18nCatalog("cura")
# Currently it only supports viewing the printer and print job status and adding a new job to the queue.
# As such, those methods have been implemented here.
# Note that this device represents a single remote cluster, not a list of multiple clusters.
class CloudOutputDevice(NetworkedPrinterOutputDevice):
class CloudOutputDevice(UltimakerNetworkedPrinterOutputDevice):
# The interval with which the remote clusters are checked
CHECK_CLUSTER_INTERVAL = 10.0 # seconds
@ -81,12 +81,11 @@ class CloudOutputDevice(NetworkedPrinterOutputDevice):
super().__init__(device_id=cluster.cluster_id, address="",
connection_type=ConnectionType.CloudConnection, properties=properties, parent=parent)
self._api = api_client
self._account = api_client.account
self._cluster = cluster
self._setInterfaceElements()
self._account = api_client.account
# We use the Cura Connect monitor tab to get most functionality right away.
if PluginRegistry.getInstance() is not None:
plugin_path = PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting")
@ -98,13 +97,6 @@ class CloudOutputDevice(NetworkedPrinterOutputDevice):
# Trigger the printersChanged signal when the private signal is triggered.
self.printersChanged.connect(self._clusterPrintersChanged)
# We keep track of which printer is visible in the monitor page.
self._active_printer = None # type: Optional[PrinterOutputModel]
# Properties to populate later on with received cloud data.
self._print_jobs = [] # type: List[UM3PrintJobOutputModel]
self._number_of_extruders = 2 # All networked printers are dual-extrusion Ultimaker machines.
# We only allow a single upload at a time.
self._progress = CloudProgressMessage()
@ -382,98 +374,27 @@ class CloudOutputDevice(NetworkedPrinterOutputDevice):
firmware_version = Version([version_number[0], version_number[1], version_number[2]])
return firmware_version >= self.PRINT_JOB_ACTIONS_MIN_VERSION
## Gets the number of printers in the cluster.
# We use a minimum of 1 because cloud devices are always a cluster and printer discovery needs it.
@pyqtProperty(int, notify=_clusterPrintersChanged)
def clusterSize(self) -> int:
return max(1, len(self._printers))
## Gets the remote printers.
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def printers(self) -> List[PrinterOutputModel]:
return self._printers
## Get the active printer in the UI (monitor page).
@pyqtProperty(QObject, notify=activePrinterChanged)
def activePrinter(self) -> Optional[PrinterOutputModel]:
return self._active_printer
## Set the active printer in the UI (monitor page).
@pyqtSlot(QObject)
def setActivePrinter(self, printer: Optional[PrinterOutputModel] = None) -> None:
if printer != self._active_printer:
self._active_printer = printer
self.activePrinterChanged.emit()
## Get remote print jobs.
@pyqtProperty("QVariantList", notify=printJobsChanged)
def printJobs(self) -> List[UM3PrintJobOutputModel]:
return self._print_jobs
## Get remote print jobs that are still in the print queue.
@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"]
## Get remote print jobs that are assigned to a printer.
@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"]
## Set the remote print job state.
def setJobState(self, print_job_uuid: str, state: str) -> None:
self._api.doPrintJobAction(self._cluster.cluster_id, print_job_uuid, state)
@pyqtSlot(str)
@pyqtSlot(str, name="sendJobToTop")
def sendJobToTop(self, print_job_uuid: str) -> None:
self._api.doPrintJobAction(self._cluster.cluster_id, print_job_uuid, "move",
{"list": "queued", "to_position": 0})
@pyqtSlot(str)
@pyqtSlot(str, name="deleteJobFromQueue")
def deleteJobFromQueue(self, print_job_uuid: str) -> None:
self._api.doPrintJobAction(self._cluster.cluster_id, print_job_uuid, "remove")
@pyqtSlot(str)
@pyqtSlot(str, name="forceSendJob")
def forceSendJob(self, print_job_uuid: str) -> None:
self._api.doPrintJobAction(self._cluster.cluster_id, print_job_uuid, "force")
@pyqtSlot(int, result=str)
def formatDuration(self, seconds: int) -> str:
return Duration(seconds).getDisplayString(DurationFormat.Format.Short)
@pyqtSlot(int, result=str)
def getTimeCompleted(self, time_remaining: int) -> str:
return formatTimeCompleted(time_remaining)
@pyqtSlot(int, result=str)
def getDateCompleted(self, time_remaining: int) -> str:
return formatDateCompleted(time_remaining)
@pyqtProperty(bool, notify=printJobsChanged)
def receivedPrintJobs(self) -> bool:
return bool(self._print_jobs)
@pyqtSlot()
@pyqtSlot(name="openPrintJobControlPanel")
def openPrintJobControlPanel(self) -> None:
QDesktopServices.openUrl(QUrl("https://mycloud.ultimaker.com"))
@pyqtSlot()
@pyqtSlot(name="openPrinterControlPanel")
def openPrinterControlPanel(self) -> None:
QDesktopServices.openUrl(QUrl("https://mycloud.ultimaker.com"))
## TODO: The following methods are required by the monitor page QML, but are not actually available using cloud.
# TODO: We fake the methods here to not break the monitor page.
@pyqtProperty(QUrl, notify=_clusterPrintersChanged)
def activeCameraUrl(self) -> "QUrl":
return QUrl()
@pyqtSlot(QUrl)
def setActiveCameraUrl(self, camera_url: "QUrl") -> None:
pass
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def connectedPrintersTypeCount(self) -> List[Dict[str, str]]:
return []

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

@ -15,7 +15,7 @@ from .CloudApiClient import CloudApiClient
from .CloudOutputDevice import CloudOutputDevice
from plugins.UM3NetworkPrinting.src.Models.CloudClusterResponse import CloudClusterResponse
from plugins.UM3NetworkPrinting.src.Models.CloudError import CloudError
from .Utils import findChanges
from plugins.UM3NetworkPrinting.src.Utils import findChanges
## The cloud output device manager is responsible for using the Ultimaker Cloud APIs to manage remote clusters.

60
plugins/UM3NetworkPrinting/src/Network/ClusterUM3OutputDevice.py
View file

@ -1,6 +1,5 @@
# Copyright (c) 2019 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Any, cast, Tuple, Union, Optional, Dict, List
from time import time
@ -14,7 +13,6 @@ from UM.i18n import i18nCatalog
from UM.Logger import Logger
from UM.Message import Message
from UM.PluginRegistry import PluginRegistry
from UM.Qt.Duration import Duration, DurationFormat
from UM.Scene.SceneNode import SceneNode # For typing.
from UM.Settings.ContainerRegistry import ContainerRegistry
@ -27,7 +25,6 @@ from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from plugins.UM3NetworkPrinting.src.Factories.PrinterModelFactory import PrinterModelFactory
from plugins.UM3NetworkPrinting.src.UltimakerNetworkedPrinterOutputDevice import UltimakerNetworkedPrinterOutputDevice
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
@ -43,7 +40,6 @@ i18n_catalog = i18nCatalog("cura")
class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
activeCameraUrlChanged = pyqtSignal()
receivedPrintJobsChanged = 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)
@ -57,8 +53,6 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
"", {}, io.BytesIO()
) # type: Tuple[Optional[str], Dict[str, Union[str, int, bool]], Union[io.StringIO, io.BytesIO]]
self._received_print_jobs = False # type: bool
if PluginRegistry.getInstance() is not None:
plugin_path = PluginRegistry.getInstance().getPluginPath("UM3NetworkPrinting")
if plugin_path is None:
@ -113,10 +107,9 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
if len(self._printers) > 1: # We need to ask the user.
self._spawnPrinterSelectionDialog()
is_job_sent = True
else: # Just immediately continue.
self._sending_job.send("") # No specifically selected printer.
is_job_sent = self._sending_job.send(None)
self._sending_job.send(None)
def _spawnPrinterSelectionDialog(self):
if self._printer_selection_dialog is None:
@ -129,11 +122,6 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
if self._printer_selection_dialog is not None:
self._printer_selection_dialog.show()
## Whether the printer that this output device represents supports print job actions via the local network.
@pyqtProperty(bool, constant=True)
def supportsPrintJobActions(self) -> bool:
return True
## 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.
@ -225,27 +213,27 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
on_finished = self._onPostPrintJobFinished,
on_progress = self._onUploadPrintJobProgress)
@pyqtProperty(QUrl, notify = activeCameraUrlChanged)
def activeCameraUrl(self) -> "QUrl":
@pyqtProperty(QUrl, notify=activeCameraUrlChanged)
def activeCameraUrl(self) -> QUrl:
return self._active_camera_url
@pyqtSlot(QUrl)
def setActiveCameraUrl(self, camera_url: "QUrl") -> None:
@pyqtSlot(QUrl, name="setActiveCameraUrl")
def setActiveCameraUrl(self, camera_url: QUrl) -> None:
if self._active_camera_url != camera_url:
self._active_camera_url = camera_url
self.activeCameraUrlChanged.emit()
## The IP address of the printer.
@pyqtProperty(str, constant = True)
def address(self) -> str:
return self._address
def _onPostPrintJobFinished(self, reply: QNetworkReply) -> None:
if self._progress_message:
self._progress_message.hide()
self._compressing_gcode = False
self._sending_gcode = False
## The IP address of the printer.
@pyqtProperty(str, constant = True)
def address(self) -> str:
return self._address
def _onUploadPrintJobProgress(self, bytes_sent: int, bytes_total: int) -> None:
if bytes_total > 0:
new_progress = bytes_sent / bytes_total * 100
@ -294,44 +282,26 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
@pyqtSlot(name="openPrintJobControlPanel")
def openPrintJobControlPanel(self) -> None:
Logger.log("d", "Opening print job control panel...")
QDesktopServices.openUrl(QUrl("http://" + self._address + "/print_jobs"))
@pyqtSlot(name="openPrinterControlPanel")
def openPrinterControlPanel(self) -> None:
Logger.log("d", "Opening printer control panel...")
QDesktopServices.openUrl(QUrl("http://" + self._address + "/printers"))
@pyqtProperty(bool, notify = receivedPrintJobsChanged)
def receivedPrintJobs(self) -> bool:
return self._received_print_jobs
@pyqtSlot(int, result = str)
def getTimeCompleted(self, time_remaining: int) -> str:
return formatTimeCompleted(time_remaining)
@pyqtSlot(int, result = str)
def getDateCompleted(self, time_remaining: int) -> str:
return formatDateCompleted(time_remaining)
@pyqtSlot(int, result = str)
def formatDuration(self, seconds: int) -> str:
return Duration(seconds).getDisplayString(DurationFormat.Format.Short)
@pyqtSlot(str)
@pyqtSlot(str, name="sendJobToTop")
def sendJobToTop(self, print_job_uuid: str) -> None:
# This function is part of the output device (and not of the printjob output model) as this type of operation
# is a modification of the cluster queue and not of the actual job.
data = "{\"to_position\": 0}"
self.put("print_jobs/{uuid}/move_to_position".format(uuid = print_job_uuid), data, on_finished=None)
@pyqtSlot(str)
@pyqtSlot(str, name="deleteJobFromQueue")
def deleteJobFromQueue(self, print_job_uuid: str) -> None:
# This function is part of the output device (and not of the printjob output model) as this type of operation
# is a modification of the cluster queue and not of the actual job.
self.delete("print_jobs/{uuid}".format(uuid = print_job_uuid), on_finished=None)
@pyqtSlot(str)
@pyqtSlot(str, name="forceSendJob")
def forceSendJob(self, print_job_uuid: str) -> None:
data = "{\"force\": true}"
self.put("print_jobs/{uuid}".format(uuid=print_job_uuid), data, on_finished=None)
@ -392,9 +362,6 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
self.get("print_jobs/{uuid}/preview_image".format(uuid=print_job.key), on_finished=self._onGetPreviewImageFinished)
def _onGetPrintJobsFinished(self, reply: QNetworkReply) -> None:
self._received_print_jobs = True
self.receivedPrintJobsChanged.emit()
if not checkValidGetReply(reply):
return
@ -634,6 +601,7 @@ class ClusterUM3OutputDevice(UltimakerNetworkedPrinterOutputDevice):
job = SendMaterialJob(device = self)
job.run()
def loadJsonFromReply(reply: QNetworkReply) -> Optional[List[Dict[str, Any]]]:
try:
result = json.loads(bytes(reply.readAll()).decode("utf-8"))

54
plugins/UM3NetworkPrinting/src/UltimakerNetworkedPrinterOutputDevice.py
View file

@ -2,10 +2,12 @@
# 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 PyQt5.QtCore import pyqtProperty, pyqtSignal, QObject, pyqtSlot, QUrl
from UM.Qt.Duration import Duration, DurationFormat
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState
from plugins.UM3NetworkPrinting.src.Utils import formatTimeCompleted, formatDateCompleted
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
@ -31,6 +33,9 @@ class UltimakerNetworkedPrinterOutputDevice(NetworkedPrinterOutputDevice):
super().__init__(device_id=device_id, address=address, properties=properties, connection_type=connection_type,
parent=parent)
# Trigger the printersChanged signal when the private signal is triggered.
self.printersChanged.connect(self._clusterPrintersChanged)
# Keeps track of all print jobs in the cluster.
self._print_jobs = [] # type: List[UM3PrintJobOutputModel]
@ -56,10 +61,14 @@ class UltimakerNetworkedPrinterOutputDevice(NetworkedPrinterOutputDevice):
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]
@pyqtProperty(bool, notify=printJobsChanged)
def receivedPrintJobs(self) -> bool:
return bool(self._print_jobs)
# Get the amount of printers in the cluster.
@pyqtProperty(int, notify=_clusterPrintersChanged)
def clusterSize(self) -> int:
return self._cluster_size
return max(1, len(self._printers))
# Get the amount of printer in the cluster per type.
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
@ -93,6 +102,27 @@ class UltimakerNetworkedPrinterOutputDevice(NetworkedPrinterOutputDevice):
self._active_printer = printer
self.activePrinterChanged.emit()
## Whether the printer that this output device represents supports print job actions via the local network.
@pyqtProperty(bool, constant=True)
def supportsPrintJobActions(self) -> bool:
return True
## Set the remote print job state.
def setJobState(self, print_job_uuid: str, state: str) -> None:
raise NotImplementedError("setJobState must be implemented")
@pyqtSlot(str, name="sendJobToTop")
def sendJobToTop(self, print_job_uuid: str) -> None:
raise NotImplementedError("sendJobToTop must be implemented")
@pyqtSlot(str, name="deleteJobFromQueue")
def deleteJobFromQueue(self, print_job_uuid: str) -> None:
raise NotImplementedError("deleteJobFromQueue must be implemented")
@pyqtSlot(str, name="forceSendJob")
def forceSendJob(self, print_job_uuid: str) -> None:
raise NotImplementedError("forceSendJob must be implemented")
@pyqtSlot(name="openPrintJobControlPanel")
def openPrintJobControlPanel(self) -> None:
raise NotImplementedError("openPrintJobControlPanel must be implemented")
@ -100,3 +130,23 @@ class UltimakerNetworkedPrinterOutputDevice(NetworkedPrinterOutputDevice):
@pyqtSlot(name="openPrinterControlPanel")
def openPrinterControlPanel(self) -> None:
raise NotImplementedError("openPrinterControlPanel must be implemented")
@pyqtProperty(QUrl, notify=_clusterPrintersChanged)
def activeCameraUrl(self) -> QUrl:
return QUrl()
@pyqtSlot(QUrl, name="setActiveCameraUrl")
def setActiveCameraUrl(self, camera_url: QUrl) -> None:
pass
@pyqtSlot(int, result=str, name="getTimeCompleted")
def getTimeCompleted(self, time_remaining: int) -> str:
return formatTimeCompleted(time_remaining)
@pyqtSlot(int, result=str, name="getDateCompleted")
def getDateCompleted(self, time_remaining: int) -> str:
return formatDateCompleted(time_remaining)
@pyqtSlot(int, result=str, name="formatDuration")
def formatDuration(self, seconds: int) -> str:
return Duration(seconds).getDisplayString(DurationFormat.Format.Short)

0
plugins/UM3NetworkPrinting/src/Cloud/Utils.py → plugins/UM3NetworkPrinting/src/Utils.py
View file