3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-12 09:17:50 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #4816 from Ultimaker/tests-for-um3networkplugin

Browse source 
Tests for um3networkplugin
This commit is contained in:
Diego Prado Gesto 2018-11-27 12:59:44 +01:00 • committed by GitHub
commit c56098d82c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 400 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

43
plugins/UM3NetworkPrinting/src/Models.py Normal file
View file

@ -0,0 +1,43 @@
# 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()
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:
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")

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

@ -1,99 +1,197 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json
import os
import urllib.parse
from typing import Dict, TYPE_CHECKING, Set
import json #To understand the list of materials from the printer reply.
import os #To walk over material files.
import os.path #To filter on material files.
from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest #To listen to the reply from the printer.
from typing import Any, Dict, Set, TYPE_CHECKING
import urllib.parse #For getting material IDs from their file names.
from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest
from UM.Job import Job #The interface we're implementing.
from UM.Application import Application
from UM.Job import Job
from UM.Logger import Logger
from UM.MimeTypeDatabase import MimeTypeDatabase #To strip the extensions of the material profile files.
from UM.MimeTypeDatabase import MimeTypeDatabase
from UM.Resources import Resources
from UM.Settings.ContainerRegistry import ContainerRegistry #To find the GUIDs of materials.
from cura.CuraApplication import CuraApplication #For the resource types.
from cura.CuraApplication import CuraApplication
# Absolute imports don't work in plugins
from .Models import ClusterMaterial, LocalMaterial
if TYPE_CHECKING:
from .ClusterUM3OutputDevice import ClusterUM3OutputDevice
## Asynchronous job to send material profiles to the printer.
#
# This way it won't freeze up the interface while sending those materials.
class SendMaterialJob(Job):
def __init__(self, device: "ClusterUM3OutputDevice") -> None:
super().__init__()
self.device = device #type: ClusterUM3OutputDevice
self.device = device # type: ClusterUM3OutputDevice
## Send the request to the printer and register a callback
def run(self) -> None:
self.device.get("materials/", on_finished = self.sendMissingMaterials)
self.device.get("materials/", on_finished = self._onGetRemoteMaterials)
def sendMissingMaterials(self, reply: QNetworkReply) -> None:
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200: #Got an error from the HTTP request.
Logger.log("e", "Couldn't request current material storage on printer. Not syncing materials.")
## Process the materials reply from the printer.
#
# \param reply The reply from the printer, a json file.
def _onGetRemoteMaterials(self, reply: QNetworkReply) -> None:
# Got an error from the HTTP request. If we did not receive a 200 something happened.
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200:
Logger.log("e", "Error fetching materials from printer: %s", reply.errorString())
return
remote_materials_list = reply.readAll().data().decode("utf-8")
try:
remote_materials_list = json.loads(remote_materials_list)
except json.JSONDecodeError:
Logger.log("e", "Request material storage on printer: I didn't understand the printer's answer.")
return
try:
remote_materials_by_guid = {material["guid"]: material for material in remote_materials_list} #Index by GUID.
except KeyError:
Logger.log("e", "Request material storage on printer: Printer's answer was missing GUIDs.")
# Collect materials from the printer's reply and send the missing ones if needed.
remote_materials_by_guid = self._parseReply(reply)
if remote_materials_by_guid:
self._sendMissingMaterials(remote_materials_by_guid)
## Determine which materials should be updated and send them to the printer.
#
# \param remote_materials_by_guid The remote materials by GUID.
def _sendMissingMaterials(self, remote_materials_by_guid: Dict[str, ClusterMaterial]) -> None:
# Collect local materials
local_materials_by_guid = self._getLocalMaterials()
if len(local_materials_by_guid) == 0:
Logger.log("d", "There are no local materials to synchronize with the printer.")
return
container_registry = ContainerRegistry.getInstance()
local_materials_list = filter(lambda material: ("GUID" in material and "version" in material and "id" in material), container_registry.findContainersMetadata(type = "material"))
local_materials_by_guid = {material["GUID"]: material for material in local_materials_list if material["id"] == material["base_file"]}
for material in local_materials_list: #For each GUID get the material with the highest version number.
try:
if int(material["version"]) > local_materials_by_guid[material["GUID"]]["version"]:
local_materials_by_guid[material["GUID"]] = material
except ValueError:
Logger.log("e", "Material {material_id} has invalid version number {number}.".format(material_id = material["id"], number = material["version"]))
continue
# Find out what materials are new or updated and must be sent to the printer
material_ids_to_send = self._determineMaterialsToSend(local_materials_by_guid, remote_materials_by_guid)
if len(material_ids_to_send) == 0:
Logger.log("d", "There are no remote materials to update.")
return
materials_to_send = set() #type: Set[Dict[str, Any]]
for guid, material in local_materials_by_guid.items():
if guid not in remote_materials_by_guid:
materials_to_send.add(material["id"])
continue
try:
if int(material["version"]) > remote_materials_by_guid[guid]["version"]:
materials_to_send.add(material["id"])
continue
except KeyError:
Logger.log("e", "Current material storage on printer was an invalid reply (missing version).")
return
# Send materials to the printer
self._sendMaterials(material_ids_to_send)
for file_path in Resources.getAllResourcesOfType(CuraApplication.ResourceTypes.MaterialInstanceContainer):
## From the local and remote materials, determine which ones should be synchronized.
#
# Makes a Set of id's containing only the id's of the materials that are not on the printer yet or the ones that
# are newer in Cura.
#
# \param local_materials The local materials by GUID.
# \param remote_materials The remote materials by GUID.
@staticmethod
def _determineMaterialsToSend(local_materials: Dict[str, LocalMaterial],
remote_materials: Dict[str, ClusterMaterial]) -> Set[str]:
return {
material.id
for guid, material in local_materials.items()
if guid not in remote_materials or material.version > remote_materials[guid].version
}
## Send the materials to the printer.
#
# The given materials will be loaded from disk en sent to to printer.
# The given id's will be matched with filenames of the locally stored materials.
#
# \param materials_to_send A set with id's of materials that must be sent.
def _sendMaterials(self, materials_to_send: Set[str]) -> None:
file_paths = Resources.getAllResourcesOfType(CuraApplication.ResourceTypes.MaterialInstanceContainer)
# Find all local material files and send them if needed.
for file_path in file_paths:
try:
mime_type = MimeTypeDatabase.getMimeTypeForFile(file_path)
except MimeTypeDatabase.MimeTypeNotFoundError:
continue #Not the sort of file we'd like to send then.
_, file_name = os.path.split(file_path)
material_id = urllib.parse.unquote_plus(mime_type.stripExtension(file_name))
if material_id not in materials_to_send:
continue
parts = []
with open(file_path, "rb") as f:
parts.append(self.device._createFormPart("name=\"file\"; filename=\"{file_name}\"".format(file_name = file_name), f.read()))
signature_file_path = file_path + ".sig"
if os.path.exists(signature_file_path):
_, signature_file_name = os.path.split(signature_file_path)
with open(signature_file_path, "rb") as f:
parts.append(self.device._createFormPart("name=\"signature_file\"; filename=\"{file_name}\"".format(file_name = signature_file_name), f.read()))
file_name = os.path.basename(file_path)
material_id = urllib.parse.unquote_plus(mime_type.stripExtension(file_name))
if material_id not in materials_to_send:
# If the material does not have to be sent we skip it.
continue
Logger.log("d", "Syncing material {material_id} with cluster.".format(material_id = material_id))
self.device.postFormWithParts(target = "materials/", parts = parts, on_finished = self.sendingFinished)
self._sendMaterialFile(file_path, file_name, material_id)
def sendingFinished(self, reply: QNetworkReply):
## Send a single material file to the printer.
#
# Also add the material signature file if that is available.
#
# \param file_path The path of the material file.
# \param file_name The name of the material file.
# \param material_id The ID of the material in the file.
def _sendMaterialFile(self, file_path: str, file_name: str, material_id: str) -> None:
parts = []
# Add the material file.
with open(file_path, "rb") as f:
parts.append(self.device.createFormPart("name=\"file\"; filename=\"{file_name}\""
.format(file_name = file_name), f.read()))
# Add the material signature file if needed.
signature_file_path = "{}.sig".format(file_path)
if os.path.exists(signature_file_path):
signature_file_name = os.path.basename(signature_file_path)
with open(signature_file_path, "rb") as f:
parts.append(self.device.createFormPart("name=\"signature_file\"; filename=\"{file_name}\""
.format(file_name = signature_file_name), f.read()))
Logger.log("d", "Syncing material {material_id} with cluster.".format(material_id = material_id))
self.device.postFormWithParts(target = "materials/", parts = parts, on_finished = self.sendingFinished)
## Check a reply from an upload to the printer and log an error when the call failed
@staticmethod
def sendingFinished(reply: QNetworkReply) -> None:
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200:
Logger.log("e", "Received error code from printer when syncing material: {code}".format(code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)))
Logger.log("e", reply.readAll().data().decode("utf-8"))
Logger.log("e", "Received error code from printer when syncing material: {code}, {text}".format(
code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute),
text = reply.errorString()
))
## Parse the reply from the printer
#
# Parses the reply to a "/materials" request to the printer
#
# \return a dictionary of ClusterMaterial objects by GUID
# \throw KeyError Raised when on of the materials does not include a valid guid
@classmethod
def _parseReply(cls, reply: QNetworkReply) -> Dict[str, ClusterMaterial]:
try:
remote_materials = json.loads(reply.readAll().data().decode("utf-8"))
return {material["guid"]: ClusterMaterial(**material) for material in remote_materials}
except UnicodeDecodeError:
Logger.log("e", "Request material storage on printer: I didn't understand the printer's answer.")
except json.JSONDecodeError:
Logger.log("e", "Request material storage on printer: I didn't understand the printer's answer.")
except ValueError:
Logger.log("e", "Request material storage on printer: Printer's answer had an incorrect value.")
except TypeError:
Logger.log("e", "Request material storage on printer: Printer's answer was missing a required value.")
## Retrieves a list of local materials
#
# Only the new newest version of the local materials is returned
#
# \return a dictionary of LocalMaterial objects by GUID
def _getLocalMaterials(self) -> Dict[str, LocalMaterial]:
result = {} # type: Dict[str, LocalMaterial]
container_registry = Application.getInstance().getContainerRegistry()
material_containers = container_registry.findContainersMetadata(type = "material")
# Find the latest version of all material containers in the registry.
for material in material_containers:
try:
# material version must be an int
material["version"] = int(material["version"])
# Create a new local material
local_material = LocalMaterial(**material)
if local_material.GUID not in result or \
local_material.version > result.get(local_material.GUID).version:
result[local_material.GUID] = local_material
except KeyError:
Logger.logException("w", "Local material {} has missing values.".format(material["id"]))
except ValueError:
Logger.logException("w", "Local material {} has invalid values.".format(material["id"]))
except TypeError:
Logger.logException("w", "Local material {} has invalid values.".format(material["id"]))
return result