3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-15 02:37:49 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

STAR-322: Using composition rather than inheritance

Browse source
This commit is contained in:
Daniel Schiavini 2018-12-05 12:02:04 +01:00
parent d99e2d1533
commit 163226f940
4 changed files with 97 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

9
cura/PrinterOutputDevice.py
View file

@ -131,7 +131,8 @@ class PrinterOutputDevice(QObject, OutputDevice):
return None
def requestWrite(self, nodes: List["SceneNode"], file_name: Optional[str] = None, limit_mimetypes: bool = False, file_handler: Optional["FileHandler"] = None, **kwargs: str) -> None:
def requestWrite(self, nodes: List["SceneNode"], file_name: Optional[str] = None, limit_mimetypes: bool = False,
file_handler: Optional["FileHandler"] = None, **kwargs: str) -> None:
raise NotImplementedError("requestWrite needs to be implemented")
@pyqtProperty(QObject, notify = printersChanged)
@ -207,8 +208,10 @@ class PrinterOutputDevice(QObject, OutputDevice):
return self._unique_configurations
def _updateUniqueConfigurations(self) -> None:
self._unique_configurations = list(set([printer.printerConfiguration for printer in self._printers if printer.printerConfiguration is not None]))
self._unique_configurations.sort(key = lambda k: k.printerType)
self._unique_configurations = sorted(
{printer.printerConfiguration for printer in self._printers if printer.printerConfiguration is not None},
key=lambda config: config.printerType,
)
self.uniqueConfigurationsChanged.emit()
# Returns the unique configurations of the printers within this output device

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

@ -1,14 +1,12 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import io
import os
from time import time
from typing import List, Optional, Dict, Union, Set
from typing import List, Optional, Dict, Set
from PyQt5.QtCore import QObject, pyqtSignal, QUrl, pyqtProperty, pyqtSlot
from UM import i18nCatalog
from UM.FileHandler.FileWriter import FileWriter
from UM.FileHandler.FileHandler import FileHandler
from UM.Logger import Logger
from UM.Message import Message
@ -16,10 +14,10 @@ from UM.Scene.SceneNode import SceneNode
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from cura.PrinterOutput.MaterialOutputModel import MaterialOutputModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState, NetworkedPrinterOutputDevice
from cura.PrinterOutput.PrinterOutputModel import PrinterOutputModel
from plugins.UM3NetworkPrinting.src.BaseCuraConnectDevice import BaseCuraConnectDevice
from plugins.UM3NetworkPrinting.src.Cloud.CloudApiClient import CloudApiClient
from plugins.UM3NetworkPrinting.src.MeshFormatHandler import MeshFormatHandler
from plugins.UM3NetworkPrinting.src.UM3PrintJobOutputModel import UM3PrintJobOutputModel
from .Models import (
CloudClusterPrinter, CloudClusterPrintJob, CloudJobUploadRequest, CloudJobResponse, CloudClusterStatus,
@ -59,7 +57,7 @@ class T:
# Note that this device represents a single remote cluster, not a list of multiple clusters.
#
# TODO: figure our how the QML interface for the cluster networking should operate with this limited functionality.
class CloudOutputDevice(BaseCuraConnectDevice):
class CloudOutputDevice(NetworkedPrinterOutputDevice):
# The interval with which the remote clusters are checked
CHECK_CLUSTER_INTERVAL = 2.0 # seconds
@ -118,17 +116,20 @@ class CloudOutputDevice(BaseCuraConnectDevice):
self._sending_job = True
self.writeStarted.emit(self)
file_format = self._getPreferredFormat(file_handler)
writer = self._getWriter(file_handler, file_format["mime_type"])
if not writer:
mesh_format = MeshFormatHandler(file_handler, self.firmwareVersion)
if not mesh_format.is_valid:
Logger.log("e", "Missing file or mesh writer!")
return self._onUploadError(T.COULD_NOT_EXPORT)
stream = io.StringIO() if file_format["mode"] == FileWriter.OutputMode.TextMode else io.BytesIO()
writer.write(stream, nodes)
mesh_bytes = mesh_format.getBytes(nodes)
# TODO: Remove extension from the file name, since we are using content types now
self._sendPrintJob(file_name + "." + file_format["extension"], file_format["mime_type"], stream)
request = CloudJobUploadRequest(
job_name = file_name + "." + mesh_format.file_extension,
file_size = len(mesh_bytes),
content_type = mesh_format.mime_type,
)
self._api.requestUpload(request, lambda response: self._onPrintJobCreated(mesh_bytes, response))
## Get remote printers.
@pyqtProperty("QVariantList", notify = clusterPrintersChanged)
@ -292,16 +293,6 @@ class CloudOutputDevice(BaseCuraConnectDevice):
model.updateOwner(job.owner)
model.updateState(job.status)
def _sendPrintJob(self, file_name: str, content_type: str, stream: Union[io.StringIO, io.BytesIO]) -> None:
mesh = stream.getvalue()
request = CloudJobUploadRequest()
request.job_name = file_name
request.file_size = len(mesh)
request.content_type = content_type
self._api.requestUpload(request, lambda response: self._onPrintJobCreated(mesh, response))
def _onPrintJobCreated(self, mesh: bytes, job_response: CloudJobResponse) -> None:
self._api.uploadMesh(job_response, mesh, self._onPrintJobUploaded, self._updateUploadProgress,
lambda _: self._onUploadError(T.UPLOAD_ERROR))

45
plugins/UM3NetworkPrinting/src/ClusterUM3OutputDevice.py
View file

@ -10,7 +10,6 @@ import json
import os
from UM.FileHandler.FileHandler import FileHandler
from UM.FileHandler.FileWriter import FileWriter # To choose based on the output file mode (text vs. binary).
from UM.FileHandler.WriteFileJob import WriteFileJob # To call the file writer asynchronously.
from UM.Logger import Logger
from UM.Settings.ContainerRegistry import ContainerRegistry
@ -23,10 +22,10 @@ from UM.Scene.SceneNode import SceneNode # For typing.
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.ConfigurationModel import ConfigurationModel
from cura.PrinterOutput.ExtruderConfigurationModel import ExtruderConfigurationModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState, NetworkedPrinterOutputDevice
from cura.PrinterOutput.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.MaterialOutputModel import MaterialOutputModel
from plugins.UM3NetworkPrinting.src.BaseCuraConnectDevice import BaseCuraConnectDevice
from plugins.UM3NetworkPrinting.src.MeshFormatHandler import MeshFormatHandler
from .ClusterUM3PrinterOutputController import ClusterUM3PrinterOutputController
from .SendMaterialJob import SendMaterialJob
@ -40,7 +39,7 @@ from PyQt5.QtCore import pyqtSlot, QUrl, pyqtSignal, pyqtProperty, QObject
i18n_catalog = i18nCatalog("cura")
class ClusterUM3OutputDevice(BaseCuraConnectDevice):
class ClusterUM3OutputDevice(NetworkedPrinterOutputDevice):
printJobsChanged = pyqtSignal()
activePrinterChanged = pyqtSignal()
activeCameraUrlChanged = pyqtSignal()
@ -103,14 +102,13 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
self.sendMaterialProfiles()
preferred_format = self._getPreferredFormat(file_handler)
writer = self._getWriter(file_handler, preferred_format["mime_type"])
mesh_format = MeshFormatHandler(file_handler, self.firmwareVersion)
# This function pauses with the yield, waiting on instructions on which printer it needs to print with.
if not writer:
if not mesh_format.is_valid:
Logger.log("e", "Missing file or mesh writer!")
return
self._sending_job = self._sendPrintJob(writer, preferred_format, nodes)
self._sending_job = self._sendPrintJob(mesh_format, nodes)
if self._sending_job is not None:
self._sending_job.send(None) # Start the generator.
@ -150,11 +148,8 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
# greenlet in order to optionally wait for selectPrinter() to select a
# printer.
# The greenlet yields exactly three times: First time None,
# \param writer The file writer to use to create the data.
# \param preferred_format A dictionary containing some information about
# what format to write to. This is necessary to create the correct buffer
# types and file extension and such.
def _sendPrintJob(self, writer: FileWriter, preferred_format: Dict, nodes: List[SceneNode]):
# \param mesh_format Object responsible for choosing the right kind of format to write with.
def _sendPrintJob(self, mesh_format: MeshFormatHandler, nodes: List[SceneNode]):
Logger.log("i", "Sending print job to printer.")
if self._sending_gcode:
self._error_message = Message(
@ -172,17 +167,17 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
# Using buffering greatly reduces the write time for many lines of gcode
stream = io.BytesIO() # type: Union[io.BytesIO, io.StringIO]# Binary mode.
if preferred_format["mode"] == FileWriter.OutputMode.TextMode:
stream = io.StringIO()
stream = mesh_format.createStream()
job = WriteFileJob(writer, stream, nodes, preferred_format["mode"])
job = WriteFileJob(mesh_format.writer, stream, nodes, mesh_format.file_mode)
self._write_job_progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), lifetime = 0, dismissable = False, progress = -1,
title = i18n_catalog.i18nc("@info:title", "Sending Data"), use_inactivity_timer = False)
self._write_job_progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"),
lifetime = 0, dismissable = False, progress = -1,
title = i18n_catalog.i18nc("@info:title", "Sending Data"),
use_inactivity_timer = False)
self._write_job_progress_message.show()
self._dummy_lambdas = (target_printer, preferred_format, stream)
self._dummy_lambdas = (target_printer, mesh_format.preferred_format, stream)
job.finished.connect(self._sendPrintJobWaitOnWriteJobFinished)
job.start()
@ -194,9 +189,11 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
if self._write_job_progress_message:
self._write_job_progress_message.hide()
self._progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), lifetime = 0, dismissable = False, progress = -1,
self._progress_message = Message(i18n_catalog.i18nc("@info:status", "Sending data to printer"), lifetime = 0,
dismissable = False, progress = -1,
title = i18n_catalog.i18nc("@info:title", "Sending Data"))
self._progress_message.addAction("Abort", i18n_catalog.i18nc("@action:button", "Cancel"), icon = None, description = "")
self._progress_message.addAction("Abort", i18n_catalog.i18nc("@action:button", "Cancel"), icon = None,
description = "")
self._progress_message.actionTriggered.connect(self._progressMessageActionTriggered)
self._progress_message.show()
parts = []
@ -220,7 +217,9 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
parts.append(self._createFormPart("name=\"file\"; filename=\"%s\"" % file_name, output))
self._latest_reply_handler = self.postFormWithParts("print_jobs/", parts, on_finished = self._onPostPrintJobFinished, on_progress = self._onUploadPrintJobProgress)
self._latest_reply_handler = self.postFormWithParts("print_jobs/", parts,
on_finished = self._onPostPrintJobFinished,
on_progress = self._onUploadPrintJobProgress)
@pyqtProperty(QObject, notify = activePrinterChanged)
def activePrinter(self) -> Optional[PrinterOutputModel]:

74
plugins/UM3NetworkPrinting/src/BaseCuraConnectDevice.py → plugins/UM3NetworkPrinting/src/MeshFormatHandler.py
View file

@ -1,15 +1,16 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional, Dict, Union
import io
from typing import Optional, Dict, Union, List
from UM.FileHandler.FileHandler import FileHandler
from UM.FileHandler.FileWriter import FileWriter
from UM.Logger import Logger
from UM.OutputDevice import OutputDeviceError # To show that something went wrong when writing.
from UM.Scene.SceneNode import SceneNode
from UM.Version import Version # To check against firmware versions for support.
from UM.i18n import i18nCatalog
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice
## Class that contains all the translations for this module.
@ -20,24 +21,63 @@ class T:
NO_FORMATS_AVAILABLE = _I18N_CATALOG.i18nc("@info:status", "There are no file formats available to write with!")
## This is the base class for the UM3 output devices (via connect or cloud)
class BaseCuraConnectDevice(NetworkedPrinterOutputDevice):
## This class is responsible for choosing the formats used by the connected clusters.
class MeshFormatHandler:
def __init__(self, file_handler: Optional[FileHandler], firmware_version: str) -> None:
self._file_handler = file_handler or CuraApplication.getInstance().getMeshFileHandler()
self._preferred_format = self._getPreferredFormat(firmware_version)
self._writer = self._getWriter(self._preferred_format["mime_type"]) if self._preferred_format else None
## Gets the default file handler
@property
def defaultFileHandler(self) -> FileHandler:
return CuraApplication.getInstance().getMeshFileHandler()
def is_valid(self) -> bool:
return bool(self._writer)
## Chooses the preferred file format.
# \return A dict with the file format details, with the following keys:
# {id: str, extension: str, description: str, mime_type: str, mode: int, hide_in_file_dialog: bool}
@property
def preferred_format(self) -> Optional[Dict[str, Union[str, int, bool]]]:
return self._preferred_format
## Gets the file writer for the given file handler and mime type.
# \return A file writer.
@property
def writer(self) -> Optional[FileWriter]:
return self._writer
@property
def mime_type(self) -> str:
return self._preferred_format["mime_type"]
## Gets the file mode (FileWriter.OutputMode.TextMode or FileWriter.OutputMode.BinaryMode)
@property
def file_mode(self) -> int:
return self._preferred_format["mode"]
## Gets the file extension
@property
def file_extension(self) -> str:
return self._preferred_format["extension"]
## Creates the right kind of stream based on the preferred format.
def createStream(self) -> Union[io.BytesIO, io.StringIO]:
return io.StringIO() if self.file_mode == FileWriter.OutputMode.TextMode else io.BytesIO()
## Writes the mesh and returns its value.
def getBytes(self, nodes: List[SceneNode]) -> bytes:
stream = self.createStream()
self.writer.write(stream, nodes)
return stream.getvalue()
## Chooses the preferred file format for the given file handler.
# \param file_handler: The file handler.
# \return A dict with the file format details, with format:
# {id: str, extension: str, description: str, mime_type: str, mode: int, hide_in_file_dialog: bool}
def _getPreferredFormat(self, file_handler: Optional[FileHandler]) -> Optional[Dict[str, Union[str, int, bool]]]:
# \param firmware_version: The version of the firmware.
# \return A dict with the file format details.
def _getPreferredFormat(self, firmware_version: str) -> Optional[Dict[str, Union[str, int, bool]]]:
# Formats supported by this application (file types that we can actually write).
application = CuraApplication.getInstance()
file_handler = file_handler or self.defaultFileHandler
file_formats = file_handler.getSupportedFileTypesWrite()
file_formats = self._file_handler.getSupportedFileTypesWrite()
global_stack = application.getGlobalContainerStack()
# Create a list from the supported file formats string.
@ -48,7 +88,7 @@ class BaseCuraConnectDevice(NetworkedPrinterOutputDevice):
machine_file_formats = global_stack.getMetaDataEntry("file_formats").split(";")
machine_file_formats = [file_type.strip() for file_type in machine_file_formats]
# Exception for UM3 firmware version >=4.4: UFP is now supported and should be the preferred file format.
if "application/x-ufp" not in machine_file_formats and Version(self.firmwareVersion) >= Version("4.4"):
if "application/x-ufp" not in machine_file_formats and Version(firmware_version) >= Version("4.4"):
machine_file_formats = ["application/x-ufp"] + machine_file_formats
# Take the intersection between file_formats and machine_file_formats.
@ -63,10 +103,8 @@ class BaseCuraConnectDevice(NetworkedPrinterOutputDevice):
return file_formats[0]
## Gets the file writer for the given file handler and mime type.
# \param file_handler: The file handler.
# \param mime_type: The mine type.
# \return A file writer.
def _getWriter(self, file_handler: Optional[FileHandler], mime_type: str) -> Optional[FileWriter]:
def _getWriter(self, mime_type: str) -> Optional[FileWriter]:
# Just take the first file format available.
file_handler = file_handler or self.defaultFileHandler
return file_handler.getWriterByMimeType(mime_type)
return self._file_handler.getWriterByMimeType(mime_type)