Merge more stuff, re-use models for local networking as well

This commit is contained in:
ChrisTerBeke 2019-07-29 14:53:50 +02:00
parent 8f37c83b9c
commit 4b212d6c05
31 changed files with 688 additions and 975 deletions

View file

@ -1,55 +0,0 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from datetime import datetime, timezone
from typing import Dict, Union, TypeVar, Type, List, Any
from plugins.UM3NetworkPrinting.src.Models.BaseModel import BaseModel
## Base class for the models used in the interface with the Ultimaker cloud APIs.
class BaseCloudModel(BaseModel):
## Checks whether the two models are equal.
# \param other: The other model.
# \return True if they are equal, False if they are different.
def __eq__(self, other):
return type(self) == type(other) and self.toDict() == other.toDict()
## Checks whether the two models are different.
# \param other: The other model.
# \return True if they are different, False if they are the same.
def __ne__(self, other) -> bool:
return type(self) != type(other) or self.toDict() != other.toDict()
## Converts the model into a serializable dictionary
def toDict(self) -> Dict[str, Any]:
return self.__dict__
# Type variable used in the parse methods below, which should be a subclass of BaseModel.
T = TypeVar("T", bound=BaseModel)
## Parses a single model.
# \param model_class: The model class.
# \param values: The value of the model, which is usually a dictionary, but may also be already parsed.
# \return An instance of the model_class given.
@staticmethod
def parseModel(model_class: Type[T], values: Union[T, Dict[str, Any]]) -> T:
if isinstance(values, dict):
return model_class(**values)
return values
## Parses a list of models.
# \param model_class: The model class.
# \param values: The value of the list. Each value is usually a dictionary, but may also be already parsed.
# \return A list of instances of the model_class given.
@classmethod
def parseModels(cls, model_class: Type[T], values: List[Union[T, Dict[str, Any]]]) -> List[T]:
return [cls.parseModel(model_class, value) for value in values]
## Parses the given date string.
# \param date: The date to parse.
# \return The parsed date.
@staticmethod
def parseDate(date: Union[str, datetime]) -> datetime:
if isinstance(date, datetime):
return date
return datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)

View file

@ -1,5 +1,10 @@
## Base model that maps kwargs to instance attributes.
from datetime import datetime, timezone
from typing import TypeVar, Dict, List, Any, Type, Union
class BaseModel:
def __init__(self, **kwargs) -> None:
self.__dict__.update(kwargs)
self.validate()
@ -7,3 +12,49 @@ class BaseModel:
# Validates the model, raising an exception if the model is invalid.
def validate(self) -> None:
pass
## Checks whether the two models are equal.
# \param other: The other model.
# \return True if they are equal, False if they are different.
def __eq__(self, other):
return type(self) == type(other) and self.toDict() == other.toDict()
## Checks whether the two models are different.
# \param other: The other model.
# \return True if they are different, False if they are the same.
def __ne__(self, other) -> bool:
return type(self) != type(other) or self.toDict() != other.toDict()
## Converts the model into a serializable dictionary
def toDict(self) -> Dict[str, Any]:
return self.__dict__
# Type variable used in the parse methods below, which should be a subclass of BaseModel.
T = TypeVar("T", bound="BaseModel")
## Parses a single model.
# \param model_class: The model class.
# \param values: The value of the model, which is usually a dictionary, but may also be already parsed.
# \return An instance of the model_class given.
@staticmethod
def parseModel(model_class: Type[T], values: Union[T, Dict[str, Any]]) -> T:
if isinstance(values, dict):
return model_class(**values)
return values
## Parses a list of models.
# \param model_class: The model class.
# \param values: The value of the list. Each value is usually a dictionary, but may also be already parsed.
# \return A list of instances of the model_class given.
@classmethod
def parseModels(cls, model_class: Type[T], values: List[Union[T, Dict[str, Any]]]) -> List[T]:
return [cls.parseModel(model_class, value) for value in values]
## Parses the given date string.
# \param date: The date to parse.
# \return The parsed date.
@staticmethod
def parseDate(date: Union[str, datetime]) -> datetime:
if isinstance(date, datetime):
return date
return datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)

View file

@ -2,12 +2,12 @@
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing a cloud connected cluster.
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterResponse(BaseCloudModel):
class CloudClusterResponse(BaseModel):
## Creates a new cluster response object.
# \param cluster_id: The secret unique ID, e.g. 'kBEeZWEifXbrXviO8mRYLx45P8k5lHVGs43XKvRniPg='.
# \param host_guid: The unique identifier of the print cluster host, e.g. 'e90ae0ac-1257-4403-91ee-a44c9b7e8050'.

View file

@ -3,24 +3,24 @@
from datetime import datetime
from typing import List, Dict, Union, Any
from .CloudClusterPrinterStatus import CloudClusterPrinterStatus
from .CloudClusterPrintJobStatus import CloudClusterPrintJobStatus
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
from .ClusterPrinterStatus import ClusterPrinterStatus
from .ClusterPrintJobStatus import ClusterPrintJobStatus
# Model that represents the status of the cluster for the cloud
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterStatus(BaseCloudModel):
class CloudClusterStatus(BaseModel):
## Creates a new cluster status model object.
# \param printers: The latest status of each printer in the cluster.
# \param print_jobs: The latest status of each print job in the cluster.
# \param generated_time: The datetime when the object was generated on the server-side.
def __init__(self,
printers: List[Union[CloudClusterPrinterStatus, Dict[str, Any]]],
print_jobs: List[Union[CloudClusterPrintJobStatus, Dict[str, Any]]],
printers: List[Union[ClusterPrinterStatus, Dict[str, Any]]],
print_jobs: List[Union[ClusterPrintJobStatus, Dict[str, Any]]],
generated_time: Union[str, datetime],
**kwargs) -> None:
self.generated_time = self.parseDate(generated_time)
self.printers = self.parseModels(CloudClusterPrinterStatus, printers)
self.print_jobs = self.parseModels(CloudClusterPrintJobStatus, print_jobs)
self.printers = self.parseModels(ClusterPrinterStatus, printers)
self.print_jobs = self.parseModels(ClusterPrintJobStatus, print_jobs)
super().__init__(**kwargs)

View file

@ -2,12 +2,12 @@
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Dict, Optional, Any
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing errors generated by the cloud servers, according to the JSON-API standard.
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudError(BaseCloudModel):
class CloudError(BaseModel):
## Creates a new error object.
# \param id: Unique identifier for this particular occurrence of the problem.
# \param title: A short, human-readable summary of the problem that SHOULD NOT change from occurrence to occurrence

View file

@ -2,12 +2,12 @@
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
# Model that represents the response received from the cloud after requesting to upload a print job
# Spec: https://api-staging.ultimaker.com/cura/v1/spec
class CloudPrintJobResponse(BaseCloudModel):
class CloudPrintJobResponse(BaseModel):
## Creates a new print job response model.
# \param job_id: The job unique ID, e.g. 'kBEeZWEifXbrXviO8mRYLx45P8k5lHVGs43XKvRniPg='.
# \param status: The status of the print job.

View file

@ -1,11 +1,11 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
# Model that represents the request to upload a print job to the cloud
# Spec: https://api-staging.ultimaker.com/cura/v1/spec
class CloudPrintJobUploadRequest(BaseCloudModel):
class CloudPrintJobUploadRequest(BaseModel):
## Creates a new print job upload request.
# \param job_name: The name of the print job.
# \param file_size: The size of the file in bytes.

View file

@ -3,12 +3,12 @@
from datetime import datetime
from typing import Optional, Union
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
# Model that represents the responses received from the cloud after requesting a job to be printed.
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudPrintResponse(BaseCloudModel):
class CloudPrintResponse(BaseModel):
## Creates a new print response object.
# \param job_id: The unique ID of a print job inside of the cluster. This ID is generated by Cura Connect.
# \param status: The status of the print request (queued or failed).

View file

@ -1,13 +1,13 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing a cluster printer
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterBuildPlate(BaseCloudModel):
## Class representing a cluster printer
class ClusterBuildPlate(BaseModel):
## Create a new build plate
# \param type: The type of buildplate glass or aluminium
# \param type: The type of build plate glass or aluminium
def __init__(self, type: str = "glass", **kwargs) -> None:
self.type = type
super().__init__(**kwargs)

View file

@ -4,23 +4,23 @@ from typing import Union, Dict, Optional, Any
from cura.PrinterOutput.Models.ExtruderConfigurationModel import ExtruderConfigurationModel
from cura.PrinterOutput.Models.ExtruderOutputModel import ExtruderOutputModel
from .CloudClusterPrinterConfigurationMaterial import CloudClusterPrinterConfigurationMaterial
from .BaseCloudModel import BaseCloudModel
from .ClusterPrinterConfigurationMaterial import ClusterPrinterConfigurationMaterial
from ..BaseModel import BaseModel
## Class representing a cloud cluster printer configuration
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrintCoreConfiguration(BaseCloudModel):
class ClusterPrintCoreConfiguration(BaseModel):
## Creates a new cloud cluster printer configuration object
# \param extruder_index: The position of the extruder on the machine as list index. Numbered from left to right.
# \param material: The material of a configuration object in a cluster printer. May be in a dict or an object.
# \param nozzle_diameter: The diameter of the print core at this position in millimeters, e.g. '0.4'.
# \param print_core_id: The type of print core inserted at this position, e.g. 'AA 0.4'.
def __init__(self, extruder_index: int,
material: Union[None, Dict[str, Any], CloudClusterPrinterConfigurationMaterial],
material: Union[None, Dict[str, Any], ClusterPrinterConfigurationMaterial],
print_core_id: Optional[str] = None, **kwargs) -> None:
self.extruder_index = extruder_index
self.material = self.parseModel(CloudClusterPrinterConfigurationMaterial, material) if material else None
self.material = self.parseModel(ClusterPrinterConfigurationMaterial, material) if material else None
self.print_core_id = print_core_id
super().__init__(**kwargs)

View file

@ -2,12 +2,12 @@
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Model for the types of changes that are needed before a print job can start
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrintJobConfigurationChange(BaseCloudModel):
class ClusterPrintJobConfigurationChange(BaseModel):
## Creates a new print job constraint.
# \param type_of_change: The type of configuration change, one of: "material", "print_core_change"
# \param index: The hotend slot or extruder index to change

View file

@ -2,12 +2,12 @@
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing a cloud cluster print job constraint
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrintJobConstraints(BaseCloudModel):
class ClusterPrintJobConstraints(BaseModel):
## Creates a new print job constraint.
# \param require_printer_name: Unique name of the printer that this job should be printed on.
# Should be one of the unique_name field values in the cluster, e.g. 'ultimakersystem-ccbdd30044ec'

View file

@ -1,13 +1,14 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing the reasons that prevent this job from being printed on the associated printer
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrintJobImpediment(BaseCloudModel):
class ClusterPrintJobImpediment(BaseModel):
## Creates a new print job constraint.
# \param translation_key: A string indicating a reason the print cannot be printed, such as 'does_not_fit_in_build_volume'
# \param translation_key: A string indicating a reason the print cannot be printed,
# such as 'does_not_fit_in_build_volume'
# \param severity: A number indicating the severity of the problem, with higher being more severe
def __init__(self, translation_key: str, severity: int, **kwargs) -> None:
self.translation_key = translation_key

View file

@ -3,20 +3,21 @@
from typing import List, Optional, Union, Dict, Any
from cura.PrinterOutput.Models.PrinterConfigurationModel import PrinterConfigurationModel
from plugins.UM3NetworkPrinting.src.Models.UM3PrintJobOutputModel import UM3PrintJobOutputModel
from plugins.UM3NetworkPrinting.src.Models.ConfigurationChangeModel import ConfigurationChangeModel
from plugins.UM3NetworkPrinting.src.Cloud.CloudOutputController import CloudOutputController
from .BaseCloudModel import BaseCloudModel
from .CloudClusterBuildPlate import CloudClusterBuildPlate
from .CloudClusterPrintJobConfigurationChange import CloudClusterPrintJobConfigurationChange
from .CloudClusterPrintJobImpediment import CloudClusterPrintJobImpediment
from .CloudClusterPrintCoreConfiguration import CloudClusterPrintCoreConfiguration
from .CloudClusterPrintJobConstraint import CloudClusterPrintJobConstraints
from .ClusterBuildPlate import ClusterBuildPlate
from .ClusterPrintJobConfigurationChange import ClusterPrintJobConfigurationChange
from .ClusterPrintJobImpediment import ClusterPrintJobImpediment
from .ClusterPrintCoreConfiguration import ClusterPrintCoreConfiguration
from .ClusterPrintJobConstraint import ClusterPrintJobConstraints
from ..UM3PrintJobOutputModel import UM3PrintJobOutputModel
from ..ConfigurationChangeModel import ConfigurationChangeModel
from ..BaseModel import BaseModel
from ...ClusterOutputController import ClusterOutputController
## Model for the status of a single print job in a cluster.
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrintJobStatus(BaseCloudModel):
class ClusterPrintJobStatus(BaseModel):
## Creates a new cloud print job status model.
# \param assigned_to: The name of the printer this job is assigned to while being queued.
# \param configuration: The required print core configurations of this print job.
@ -45,21 +46,21 @@ class CloudClusterPrintJobStatus(BaseCloudModel):
# printer
def __init__(self, created_at: str, force: bool, machine_variant: str, name: str, started: bool, status: str,
time_total: int, uuid: str,
configuration: List[Union[Dict[str, Any], CloudClusterPrintCoreConfiguration]],
constraints: List[Union[Dict[str, Any], CloudClusterPrintJobConstraints]],
configuration: List[Union[Dict[str, Any], ClusterPrintCoreConfiguration]],
constraints: List[Union[Dict[str, Any], ClusterPrintJobConstraints]],
last_seen: Optional[float] = None, network_error_count: Optional[int] = None,
owner: Optional[str] = None, printer_uuid: Optional[str] = None, time_elapsed: Optional[int] = None,
assigned_to: Optional[str] = None, deleted_at: Optional[str] = None,
printed_on_uuid: Optional[str] = None,
configuration_changes_required: List[
Union[Dict[str, Any], CloudClusterPrintJobConfigurationChange]] = None,
build_plate: Union[Dict[str, Any], CloudClusterBuildPlate] = None,
Union[Dict[str, Any], ClusterPrintJobConfigurationChange]] = None,
build_plate: Union[Dict[str, Any], ClusterBuildPlate] = None,
compatible_machine_families: List[str] = None,
impediments_to_printing: List[Union[Dict[str, Any], CloudClusterPrintJobImpediment]] = None,
impediments_to_printing: List[Union[Dict[str, Any], ClusterPrintJobImpediment]] = None,
**kwargs) -> None:
self.assigned_to = assigned_to
self.configuration = self.parseModels(CloudClusterPrintCoreConfiguration, configuration)
self.constraints = self.parseModels(CloudClusterPrintJobConstraints, constraints)
self.configuration = self.parseModels(ClusterPrintCoreConfiguration, configuration)
self.constraints = self.parseModels(ClusterPrintJobConstraints, constraints)
self.created_at = created_at
self.force = force
self.last_seen = last_seen
@ -76,19 +77,19 @@ class CloudClusterPrintJobStatus(BaseCloudModel):
self.deleted_at = deleted_at
self.printed_on_uuid = printed_on_uuid
self.configuration_changes_required = self.parseModels(CloudClusterPrintJobConfigurationChange,
self.configuration_changes_required = self.parseModels(ClusterPrintJobConfigurationChange,
configuration_changes_required) \
if configuration_changes_required else []
self.build_plate = self.parseModel(CloudClusterBuildPlate, build_plate) if build_plate else None
self.build_plate = self.parseModel(ClusterBuildPlate, build_plate) if build_plate else None
self.compatible_machine_families = compatible_machine_families if compatible_machine_families else []
self.impediments_to_printing = self.parseModels(CloudClusterPrintJobImpediment, impediments_to_printing) \
self.impediments_to_printing = self.parseModels(ClusterPrintJobImpediment, impediments_to_printing) \
if impediments_to_printing else []
super().__init__(**kwargs)
## Creates an UM3 print job output model based on this cloud cluster print job.
# \param printer: The output model of the printer
def createOutputModel(self, controller: CloudOutputController) -> UM3PrintJobOutputModel:
def createOutputModel(self, controller: ClusterOutputController) -> UM3PrintJobOutputModel:
model = UM3PrintJobOutputModel(controller, self.uuid, self.name)
self.updateOutputModel(model)
return model

View file

@ -3,12 +3,13 @@ from typing import Optional
from UM.Logger import Logger
from cura.CuraApplication import CuraApplication
from cura.PrinterOutput.Models.MaterialOutputModel import MaterialOutputModel
from .BaseCloudModel import BaseCloudModel
from ..BaseModel import BaseModel
## Class representing a cloud cluster printer configuration
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrinterConfigurationMaterial(BaseCloudModel):
## Class representing a cloud cluster printer configuration
class ClusterPrinterConfigurationMaterial(BaseModel):
## Creates a new material configuration model.
# \param brand: The brand of material in this print core, e.g. 'Ultimaker'.
# \param color: The color of material in this print core, e.g. 'Blue'.

View file

@ -4,14 +4,14 @@ from typing import List, Union, Dict, Optional, Any
from cura.PrinterOutput.PrinterOutputController import PrinterOutputController
from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
from .CloudClusterBuildPlate import CloudClusterBuildPlate
from .CloudClusterPrintCoreConfiguration import CloudClusterPrintCoreConfiguration
from .BaseCloudModel import BaseCloudModel
from .ClusterBuildPlate import ClusterBuildPlate
from .ClusterPrintCoreConfiguration import ClusterPrintCoreConfiguration
from ..BaseModel import BaseModel
## Class representing a cluster printer
# Spec: https://api-staging.ultimaker.com/connect/v1/spec
class CloudClusterPrinterStatus(BaseCloudModel):
class ClusterPrinterStatus(BaseModel):
## Creates a new cluster printer status
# \param enabled: A printer can be disabled if it should not receive new jobs. By default every printer is enabled.
# \param firmware_version: Firmware version installed on the printer. Can differ for each printer in a cluster.
@ -30,12 +30,12 @@ class CloudClusterPrinterStatus(BaseCloudModel):
# \param build_plate: The build plate that is on the printer
def __init__(self, enabled: bool, firmware_version: str, friendly_name: str, ip_address: str, machine_variant: str,
status: str, unique_name: str, uuid: str,
configuration: List[Union[Dict[str, Any], CloudClusterPrintCoreConfiguration]],
configuration: List[Union[Dict[str, Any], ClusterPrintCoreConfiguration]],
reserved_by: Optional[str] = None, maintenance_required: Optional[bool] = None,
firmware_update_status: Optional[str] = None, latest_available_firmware: Optional[str] = None,
build_plate: Union[Dict[str, Any], CloudClusterBuildPlate] = None, **kwargs) -> None:
build_plate: Union[Dict[str, Any], ClusterBuildPlate] = None, **kwargs) -> None:
self.configuration = self.parseModels(CloudClusterPrintCoreConfiguration, configuration)
self.configuration = self.parseModels(ClusterPrintCoreConfiguration, configuration)
self.enabled = enabled
self.firmware_version = firmware_version
self.friendly_name = friendly_name
@ -48,7 +48,7 @@ class CloudClusterPrinterStatus(BaseCloudModel):
self.maintenance_required = maintenance_required
self.firmware_update_status = firmware_update_status
self.latest_available_firmware = latest_available_firmware
self.build_plate = self.parseModel(CloudClusterBuildPlate, build_plate) if build_plate else None
self.build_plate = self.parseModel(ClusterBuildPlate, build_plate) if build_plate else None
super().__init__(**kwargs)
## Creates a new output model.

View file

@ -3,13 +3,13 @@ from plugins.UM3NetworkPrinting.src.Models.BaseModel import BaseModel
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:
super().validate()
if not self.GUID: