Used NamedTuple from typing iso namedtuple from collections so we can at least give type hints

This commit is contained in:
Marijn Deé 2018-11-21 10:12:53 +01:00
parent c1c3f3abf3
commit 9e8be286af
3 changed files with 55 additions and 54 deletions

View file

@ -1,33 +1,33 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from collections import namedtuple
from typing import NamedTuple
ClusterMaterial = namedtuple('ClusterMaterial', [
'guid',
'material',
'brand',
'version',
'color',
'density'
ClusterMaterial = NamedTuple("ClusterMaterial", [
("guid", str),
("material", str),
("brand", str),
("version", int),
("color", str),
("density", str),
])
LocalMaterial = namedtuple('LocalMaterial', [
'GUID',
'id',
'type',
'status',
'base_file',
'setting_version',
'version',
'name',
'brand',
'material',
'color_name',
'color_code',
'description',
'adhesion_info',
'approximate_diameter',
'properties',
'definition',
'compatible'
LocalMaterial = NamedTuple("LocalMaterial", [
("GUID", str),
("id", str),
("type", str),
("status", str),
("base_file", str),
("setting_version", int),
("version", int),
("name", str),
("brand", str),
("material", str),
("color_name", str),
("color_code", str),
("description", str),
("adhesion_info", str),
("approximate_diameter", str),
("properties", str),
("definition", str),
("compatible", str),
])

View file

@ -2,7 +2,6 @@
# Cura is released under the terms of the LGPLv3 or higher.
import json
import os
import re
import urllib.parse
from typing import Dict, TYPE_CHECKING, Set
@ -13,13 +12,13 @@ from UM.Logger import Logger
from UM.MimeTypeDatabase import MimeTypeDatabase
from UM.Resources import Resources
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.
@ -86,7 +85,7 @@ class SendMaterialJob(Job):
return {
material.id
for guid, material in local_materials.items()
if guid not in remote_materials or int(material.version) > remote_materials[guid].version
if guid not in remote_materials or material.version > remote_materials[guid].version
}
## Send the materials to the printer.
@ -122,23 +121,23 @@ class SendMaterialJob(Job):
# \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 = []
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 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()))
# 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)
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
@ -174,16 +173,18 @@ class SendMaterialJob(Job):
# Find the latest version of all material containers in the registry.
for material in material_containers:
try:
material = LocalMaterial(**material)
# material version must be an int
if not re.match("\d+", material.version):
Logger.logException("w", "Local material {} has invalid version '{}'."
.format(material["id"], material.version))
continue
material["version"] = int(material["version"])
if material.GUID not in result or material.version > result.get(material.GUID).version:
result[material.GUID] = material
# 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"]))