STAR-322: Extracting file handler methods

This commit is contained in:
Daniel Schiavini 2018-12-05 11:21:17 +01:00
parent 3c5e74a72c
commit d99e2d1533
3 changed files with 83 additions and 106 deletions

View file

@ -1,8 +1,72 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional, Dict, Union
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.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
## this is the base class for the UM3 output devices (via connect or cloud)
## Class that contains all the translations for this module.
class T:
# The translation catalog for this device.
_I18N_CATALOG = i18nCatalog("cura")
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):
pass
## Gets the default file handler
@property
def defaultFileHandler(self) -> FileHandler:
return CuraApplication.getInstance().getMeshFileHandler()
## 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]]]:
# 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()
global_stack = application.getGlobalContainerStack()
# Create a list from the supported file formats string.
if not global_stack:
Logger.log("e", "Missing global stack!")
return
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"):
machine_file_formats = ["application/x-ufp"] + machine_file_formats
# Take the intersection between file_formats and machine_file_formats.
format_by_mimetype = {f["mime_type"]: f for f in file_formats}
# Keep them ordered according to the preference in machine_file_formats.
file_formats = [format_by_mimetype[mimetype] for mimetype in machine_file_formats]
if len(file_formats) == 0:
Logger.log("e", "There are no file formats available to write with!")
raise OutputDeviceError.WriteRequestFailedError(T.NO_FORMATS_AVAILABLE)
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]:
# Just take the first file format available.
file_handler = file_handler or self.defaultFileHandler
return file_handler.getWriterByMimeType(mime_type)

View file

@ -2,9 +2,8 @@
# Cura is released under the terms of the LGPLv3 or higher.
import io
import os
from datetime import datetime, timedelta
from time import time
from typing import List, Optional, Dict, cast, Union, Set
from typing import List, Optional, Dict, Union, Set
from PyQt5.QtCore import QObject, pyqtSignal, QUrl, pyqtProperty, pyqtSlot
@ -13,9 +12,7 @@ from UM.FileHandler.FileWriter import FileWriter
from UM.FileHandler.FileHandler import FileHandler
from UM.Logger import Logger
from UM.Message import Message
from UM.OutputDevice import OutputDeviceError
from UM.Scene.SceneNode import SceneNode
from UM.Version import Version
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from cura.PrinterOutput.MaterialOutputModel import MaterialOutputModel
@ -45,7 +42,6 @@ class T:
"the previous print job.")
COULD_NOT_EXPORT = _I18N_CATALOG.i18nc("@info:status", "Could not export print job.")
WRITE_FAILED = _I18N_CATALOG.i18nc("@info:status", "There are no file formats available to write with!")
SENDING_DATA_TEXT = _I18N_CATALOG.i18nc("@info:status", "Sending data to remote cluster")
SENDING_DATA_TITLE = _I18N_CATALOG.i18nc("@info:status", "Sending data to remote cluster")
@ -69,7 +65,7 @@ class CloudOutputDevice(BaseCuraConnectDevice):
CHECK_CLUSTER_INTERVAL = 2.0 # seconds
# Signal triggered when the printers in the remote cluster were changed.
printersChanged = pyqtSignal()
clusterPrintersChanged = pyqtSignal()
# Signal triggered when the print jobs in the queue were changed.
printJobsChanged = pyqtSignal()
@ -122,8 +118,8 @@ class CloudOutputDevice(BaseCuraConnectDevice):
self._sending_job = True
self.writeStarted.emit(self)
file_format = self._determineFileFormat(file_handler)
writer = self._determineWriter(file_handler, file_format)
file_format = self._getPreferredFormat(file_handler)
writer = self._getWriter(file_handler, file_format["mime_type"])
if not writer:
Logger.log("e", "Missing file or mesh writer!")
return self._onUploadError(T.COULD_NOT_EXPORT)
@ -134,56 +130,8 @@ class CloudOutputDevice(BaseCuraConnectDevice):
# 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)
# TODO: This is yanked right out of ClusterUM3OutputDevice, great candidate for a utility or base class
def _determineFileFormat(self, file_handler) -> Optional[Dict[str, Union[str, int]]]:
# Formats supported by this application (file types that we can actually write).
if file_handler:
file_formats = file_handler.getSupportedFileTypesWrite()
else:
file_formats = CuraApplication.getInstance().getMeshFileHandler().getSupportedFileTypesWrite()
global_stack = CuraApplication.getInstance().getGlobalContainerStack()
# Create a list from the supported file formats string.
if not global_stack:
Logger.log("e", "Missing global stack!")
return
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"):
machine_file_formats = ["application/x-ufp"] + machine_file_formats
# Take the intersection between file_formats and machine_file_formats.
format_by_mimetype = {f["mime_type"]: f for f in file_formats}
# Keep them ordered according to the preference in machine_file_formats.
file_formats = [format_by_mimetype[mimetype] for mimetype in machine_file_formats]
if len(file_formats) == 0:
Logger.log("e", "There are no file formats available to write with!")
raise OutputDeviceError.WriteRequestFailedError(T.WRITE_FAILED)
return file_formats[0]
# TODO: This is yanked right out of ClusterUM3OutputDevice, great candidate for a utility or base class
@staticmethod
def _determineWriter(file_handler, file_format) -> Optional[FileWriter]:
# Just take the first file format available.
if file_handler is not None:
writer = file_handler.getWriterByMimeType(cast(str, file_format["mime_type"]))
else:
writer = CuraApplication.getInstance().getMeshFileHandler().getWriterByMimeType(
cast(str, file_format["mime_type"])
)
if not writer:
Logger.log("e", "Unexpected error when trying to get the FileWriter")
return
return writer
## Get remote printers.
@pyqtProperty("QVariantList", notify = printersChanged)
@pyqtProperty("QVariantList", notify = clusterPrintersChanged)
def printers(self):
return self._printers
@ -244,7 +192,7 @@ class CloudOutputDevice(BaseCuraConnectDevice):
for printer_guid in updated_printer_ids:
self._updatePrinter(current_printers[printer_guid], remote_printers[printer_guid])
self.printersChanged.emit()
self.clusterPrintersChanged.emit()
def _addPrinter(self, printer: CloudClusterPrinter) -> None:
model = PrinterOutputModel(
@ -409,7 +357,7 @@ class CloudOutputDevice(BaseCuraConnectDevice):
## 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(QObject, notify = printersChanged)
@pyqtProperty(QObject, notify = clusterPrintersChanged)
def activePrinter(self) -> Optional[PrinterOutputModel]:
if not self._printers:
return None
@ -419,7 +367,7 @@ class CloudOutputDevice(BaseCuraConnectDevice):
def setActivePrinter(self, printer: Optional[PrinterOutputModel]) -> None:
pass
@pyqtProperty(QUrl, notify = printersChanged)
@pyqtProperty(QUrl, notify = clusterPrintersChanged)
def activeCameraUrl(self) -> "QUrl":
return QUrl()

View file

@ -18,14 +18,12 @@ from UM.i18n import i18nCatalog
from UM.Message import Message
from UM.Qt.Duration import Duration, DurationFormat
from UM.OutputDevice import OutputDeviceError # To show that something went wrong when writing.
from UM.Scene.SceneNode import SceneNode # For typing.
from UM.Version import Version # To check against firmware versions for support.
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.ConfigurationModel import ConfigurationModel
from cura.PrinterOutput.ExtruderConfigurationModel import ExtruderConfigurationModel
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState
from cura.PrinterOutput.NetworkedPrinterOutputDevice import AuthState
from cura.PrinterOutput.PrinterOutputModel import PrinterOutputModel
from cura.PrinterOutput.MaterialOutputModel import MaterialOutputModel
from plugins.UM3NetworkPrinting.src.BaseCuraConnectDevice import BaseCuraConnectDevice
@ -50,7 +48,7 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
# This is a bit of a hack, as the notify can only use signals that are defined by the class that they are in.
# Inheritance doesn't seem to work. Tying them together does work, but i'm open for better suggestions.
clusterPrintersChanged = pyqtSignal()
_clusterPrintersChanged = pyqtSignal()
def __init__(self, device_id, address, properties, parent = None) -> None:
super().__init__(device_id = device_id, address = address, properties=properties, parent = parent)
@ -66,7 +64,7 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
self._monitor_view_qml_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../resources/qml/ClusterMonitorItem.qml")
# See comments about this hack with the clusterPrintersChanged signal
self.printersChanged.connect(self.clusterPrintersChanged)
self.printersChanged.connect(self._clusterPrintersChanged)
self._accepts_commands = True # type: bool
@ -99,47 +97,14 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
self._active_camera_url = QUrl() # type: QUrl
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:
self.writeStarted.emit(self)
self.sendMaterialProfiles()
# Formats supported by this application (file types that we can actually write).
if file_handler:
file_formats = file_handler.getSupportedFileTypesWrite()
else:
file_formats = CuraApplication.getInstance().getMeshFileHandler().getSupportedFileTypesWrite()
global_stack = CuraApplication.getInstance().getGlobalContainerStack()
# Create a list from the supported file formats string.
if not global_stack:
Logger.log("e", "Missing global stack!")
return
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"):
machine_file_formats = ["application/x-ufp"] + machine_file_formats
# Take the intersection between file_formats and machine_file_formats.
format_by_mimetype = {format["mime_type"]: format for format in file_formats}
file_formats = [format_by_mimetype[mimetype] for mimetype in machine_file_formats] #Keep them ordered according to the preference in machine_file_formats.
if len(file_formats) == 0:
Logger.log("e", "There are no file formats available to write with!")
raise OutputDeviceError.WriteRequestFailedError(i18n_catalog.i18nc("@info:status", "There are no file formats available to write with!"))
preferred_format = file_formats[0]
# Just take the first file format available.
if file_handler is not None:
writer = file_handler.getWriterByMimeType(cast(str, preferred_format["mime_type"]))
else:
writer = CuraApplication.getInstance().getMeshFileHandler().getWriterByMimeType(cast(str, preferred_format["mime_type"]))
if not writer:
Logger.log("e", "Unexpected error when trying to get the FileWriter")
return
preferred_format = self._getPreferredFormat(file_handler)
writer = self._getWriter(file_handler, preferred_format["mime_type"])
# This function pauses with the yield, waiting on instructions on which printer it needs to print with.
if not writer:
@ -355,7 +320,7 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
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)
@pyqtProperty("QVariantList", notify = _clusterPrintersChanged)
def connectedPrintersTypeCount(self) -> List[Dict[str, str]]:
printer_count = {} # type: Dict[str, int]
for printer in self._printers:
@ -368,7 +333,7 @@ class ClusterUM3OutputDevice(BaseCuraConnectDevice):
result.append({"machine_type": machine_type, "count": str(printer_count[machine_type])})
return result
@pyqtProperty("QVariantList", notify=clusterPrintersChanged)
@pyqtProperty("QVariantList", notify=_clusterPrintersChanged)
def printers(self):
return self._printers