Cura/plugins/UFPWriter/UFPWriter.py

# Copyright (c) 2022 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json
from dataclasses import asdict
from typing import cast, List, Dict

from Charon.VirtualFile import VirtualFile  # To open UFP files.
from Charon.OpenMode import OpenMode  # To indicate that we want to write to UFP files.
from Charon.filetypes.OpenPackagingConvention import OPCError
from io import StringIO  # For converting g-code to bytes.

from PyQt6.QtCore import QBuffer

from UM.Application import Application
from UM.Logger import Logger
from UM.Settings.SettingFunction import SettingFunction
from UM.Mesh.MeshWriter import MeshWriter  # The writer we need to implement.
from UM.MimeTypeDatabase import MimeTypeDatabase, MimeType
from UM.PluginRegistry import PluginRegistry  # To get the g-code writer.

from UM.Scene.Iterator.DepthFirstIterator import DepthFirstIterator
from UM.Scene.SceneNode import SceneNode
from UM.Settings.InstanceContainer import InstanceContainer
from cura.CuraApplication import CuraApplication
from cura.Settings.GlobalStack import GlobalStack
from cura.Utils.Threading import call_on_qt_thread

from UM.i18n import i18nCatalog

METADATA_OBJECTS_PATH = "metadata/objects"
SLICE_METADATA_PATH = "Cura/slicemetadata.json"

catalog = i18nCatalog("cura")


class UFPWriter(MeshWriter):
    def __init__(self):
        super().__init__(add_to_recent_files = False)

        MimeTypeDatabase.addMimeType(
            MimeType(
                name = "application/x-ufp",
                comment = "UltiMaker Format Package",
                suffixes = ["ufp"]
            )
        )

    # This needs to be called on the main thread (Qt thread) because the serialization of material containers can
    # trigger loading other containers. Because those loaded containers are QtObjects, they must be created on the
    # Qt thread. The File read/write operations right now are executed on separated threads because they are scheduled
    # by the Job class.
    @call_on_qt_thread
    def write(self, stream, nodes, mode = MeshWriter.OutputMode.BinaryMode):
        archive = VirtualFile()
        archive.openStream(stream, "application/x-ufp", OpenMode.WriteOnly)

        try:
            self._writeObjectList(archive)

            # Store the g-code from the scene.
            archive.addContentType(extension = "gcode", mime_type = "text/x-gcode")
        except EnvironmentError as e:
            error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
            self.setInformation(error_msg)
            Logger.error(error_msg)
            return False
        gcode_textio = StringIO()  # We have to convert the g-code into bytes.
        gcode_writer = cast(MeshWriter, PluginRegistry.getInstance().getPluginObject("GCodeWriter"))
        success = gcode_writer.write(gcode_textio, None)
        if not success:  # Writing the g-code failed. Then I can also not write the gzipped g-code.
            self.setInformation(gcode_writer.getInformation())
            return False
        try:
            gcode = archive.getStream("/3D/model.gcode")
            gcode.write(gcode_textio.getvalue().encode("UTF-8"))
            archive.addRelation(virtual_path = "/3D/model.gcode",
                                relation_type = "http://schemas.ultimaker.org/package/2018/relationships/gcode")
        except EnvironmentError as e:
            error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
            self.setInformation(error_msg)
            Logger.error(error_msg)
            return False

        # Write settings
        try:
            archive.addContentType(extension="json", mime_type="application/json")
            setting_textio = StringIO()
            json.dump(self._getSliceMetadata(), setting_textio, separators=(", ", ": "), indent=4)
            steam = archive.getStream(SLICE_METADATA_PATH)
            steam.write(setting_textio.getvalue().encode("UTF-8"))
        except EnvironmentError as e:
            error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
            self.setInformation(error_msg)
            Logger.error(error_msg)
            return False

        # Attempt to store the thumbnail, if any:
        backend = CuraApplication.getInstance().getBackend()
        snapshot = None if getattr(backend, "getLatestSnapshot", None) is None else backend.getLatestSnapshot()
        if snapshot:
            try:
                archive.addContentType(extension = "png", mime_type = "image/png")
                thumbnail = archive.getStream("/Metadata/thumbnail.png")

                thumbnail_buffer = QBuffer()
                thumbnail_buffer.open(QBuffer.OpenModeFlag.ReadWrite)
                snapshot.save(thumbnail_buffer, "PNG")

                thumbnail.write(thumbnail_buffer.data())
                archive.addRelation(virtual_path = "/Metadata/thumbnail.png",
                                    relation_type = "http://schemas.openxmlformats.org/package/2006/relationships/metadata/thumbnail",
                                    origin = "/3D/model.gcode")
            except EnvironmentError as e:
                error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
                self.setInformation(error_msg)
                Logger.error(error_msg)
                return False
        else:
            Logger.log("w", "Thumbnail not created, cannot save it")

        # Store the material.
        application = CuraApplication.getInstance()
        machine_manager = application.getMachineManager()
        container_registry = application.getContainerRegistry()
        global_stack = machine_manager.activeMachine

        material_extension = "xml.fdm_material"
        material_mime_type = "application/x-ultimaker-material-profile"

        try:
            archive.addContentType(extension = material_extension, mime_type = material_mime_type)
        except OPCError:
            Logger.log("w", "The material extension: %s was already added", material_extension)

        added_materials = []
        for extruder_stack in global_stack.extruderList:
            material = extruder_stack.material
            try:
                material_file_name = material.getMetaData()["base_file"] + ".xml.fdm_material"
            except KeyError:
                Logger.log("w", "Unable to get base_file for the material %s", material.getId())
                continue
            material_file_name = "/Materials/" + material_file_name

            # The same material should not be added again.
            if material_file_name in added_materials:
                continue

            material_root_id = material.getMetaDataEntry("base_file")
            material_root_query = container_registry.findContainers(id = material_root_id)
            if not material_root_query:
                Logger.log("e", "Cannot find material container with root id {root_id}".format(root_id = material_root_id))
                return False
            material_container = material_root_query[0]

            try:
                serialized_material = material_container.serialize()
            except NotImplementedError:
                Logger.log("e", "Unable serialize material container with root id: %s", material_root_id)
                return False

            try:
                material_file = archive.getStream(material_file_name)
                material_file.write(serialized_material.encode("UTF-8"))
                archive.addRelation(virtual_path = material_file_name,
                                    relation_type = "http://schemas.ultimaker.org/package/2018/relationships/material",
                                    origin = "/3D/model.gcode")
            except EnvironmentError as e:
                error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
                self.setInformation(error_msg)
                Logger.error(error_msg)
                return False

            added_materials.append(material_file_name)

        try:
            archive.close()
        except EnvironmentError as e:
            error_msg = catalog.i18nc("@info:error", "Can't write to UFP file:") + " " + str(e)
            self.setInformation(error_msg)
            Logger.error(error_msg)
            return False
        return True

    @staticmethod
    def _writeObjectList(archive):
        """Write a json list of object names to the METADATA_OBJECTS_PATH metadata field

        To retrieve, use: `archive.getMetadata(METADATA_OBJECTS_PATH)`
        """

        objects_model = CuraApplication.getInstance().getObjectsModel()
        object_metas = []

        for item in objects_model.items:
            object_metas.extend(UFPWriter._getObjectMetadata(item["node"]))

        data = {METADATA_OBJECTS_PATH: object_metas}
        archive.setMetadata(data)

    @staticmethod
    def _getObjectMetadata(node: SceneNode) -> List[Dict[str, str]]:
        """Get object metadata to write for a Node.

        :return: List of object metadata dictionaries.
                 Might contain > 1 element in case of a group node.
                 Might be empty in case of nonPrintingMesh
        """

        return [{"name": item.getName()}
                for item in DepthFirstIterator(node)
                if item.getMeshData() is not None and not item.callDecoration("isNonPrintingMesh")]

    def _getSliceMetadata(self) -> Dict[str, Dict[str, Dict[str, str]]]:
        """Get all changed settings and all settings. For each extruder and the global stack"""
        print_information = CuraApplication.getInstance().getPrintInformation()
        machine_manager = CuraApplication.getInstance().getMachineManager()
        settings = {
            "material": {
                "length": print_information.materialLengths,
                "weight": print_information.materialWeights,
                "cost": print_information.materialCosts,
            },
            "global": {
                "changes": {},
                "all_settings": {},
            },
            "quality": asdict(machine_manager.activeQualityDisplayNameMap()),
        }

        def _retrieveValue(container: InstanceContainer, setting_: str):
            value_ = container.getProperty(setting_, "value")
            for _ in range(0, 1024):  # Prevent possibly endless loop by not using a limit.
                if not isinstance(value_, SettingFunction):
                    return value_  # Success!
                value_ = value_(container)
            return 0  # Fallback value after breaking possibly endless loop.

        global_stack = cast(GlobalStack, Application.getInstance().getGlobalContainerStack())

        # Add global user or quality changes
        global_flattened_changes = InstanceContainer.createMergedInstanceContainer(global_stack.userChanges, global_stack.qualityChanges)
        for setting in global_flattened_changes.getAllKeys():
            settings["global"]["changes"][setting] = _retrieveValue(global_flattened_changes, setting)

        # Get global all settings values without user or quality changes
        for setting in global_stack.getAllKeys():
            settings["global"]["all_settings"][setting] = _retrieveValue(global_stack, setting)

        for i, extruder in enumerate(global_stack.extruderList):
            # Add extruder fields to settings dictionary
            settings[f"extruder_{i}"] = {
                "changes": {},
                "all_settings": {},
            }

            # Add extruder user or quality changes
            extruder_flattened_changes = InstanceContainer.createMergedInstanceContainer(extruder.userChanges, extruder.qualityChanges)
            for setting in extruder_flattened_changes.getAllKeys():
                settings[f"extruder_{i}"]["changes"][setting] = _retrieveValue(extruder_flattened_changes, setting)

            # Get extruder all settings values without user or quality changes
            for setting in extruder.getAllKeys():
                settings[f"extruder_{i}"]["all_settings"][setting] = _retrieveValue(extruder, setting)

        return settings