Coverage for src/backend/InvenTree/machine/machine_type.py: 30%
163 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Base machine type/base driver."""
3from typing import TYPE_CHECKING, Any, Literal, TypedDict
5from generic.states import StatusCode
6from InvenTree.helpers_mixin import (
7 ClassProviderMixin,
8 ClassValidationMixin,
9 get_shared_class_instance_state_mixin,
10)
12# Import only for typechecking, otherwise this throws cyclic import errors
13if TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true
14 from common.models import SettingsKeyType
15 from machine.models import MachineConfig
16else: # pragma: no cover
18 class MachineConfig:
19 """Only used if not typechecking currently."""
21 class SettingsKeyType:
22 """Only used if not typechecking currently."""
25class MachineStatus(StatusCode):
26 """Base class for representing a set of machine status codes.
28 Use enum syntax to define the status codes, e.g.
29 ```python
30 CONNECTED = 200, _("Connected"), 'success'
31 ```
33 The values of the status can be accessed with `MachineStatus.CONNECTED.value`.
35 Additionally there are helpers to access all additional attributes `text`, `label`, `color`.
37 Available colors:
38 primary, secondary, warning, danger, success, warning, info
40 Status code ranges:
41 ```
42 1XX - Everything fine
43 2XX - Warnings (e.g. ink is about to become empty)
44 3XX - Something wrong with the machine (e.g. no labels are remaining on the spool)
45 4XX - Something wrong with the driver (e.g. cannot connect to the machine)
46 5XX - Unknown issues
47 ```
48 """
51MachinePropertyType = Literal['str', 'bool', 'progress', 'int', 'float']
54class MachineProperty(TypedDict, total=False):
55 """Type definition for machine properties.
57 Attributes:
58 key: Key of the property (required)
59 value: Value of the property (required)
60 group: Grouping of the property
61 type: Type of the property (one of 'str', 'bool', 'progress', 'int', 'float') default = 'str'
62 max_progress: Maximum value for progress type (required if type is 'progress')
63 """
65 key: str
66 value: str | bool | int | float
67 group: str
68 type: MachinePropertyType
69 max_progress: int | None
72class BaseDriver(
73 ClassValidationMixin,
74 ClassProviderMixin,
75 get_shared_class_instance_state_mixin(lambda x: f'machine:driver:{x.SLUG}'),
76):
77 """Base class for all machine drivers.
79 Attributes:
80 SLUG: Slug string for identifying the driver in format /[a-z-]+/ (required)
81 NAME: User friendly name for displaying (required)
82 DESCRIPTION: Description of what this driver does (required)
84 MACHINE_SETTINGS: Driver specific settings dict
85 """
87 SLUG: str
88 NAME: str
89 DESCRIPTION: str
91 MACHINE_SETTINGS: dict[str, SettingsKeyType]
93 machine_type: str
95 required_attributes = ['SLUG', 'NAME', 'DESCRIPTION', 'machine_type']
97 def __init__(self) -> None:
98 """Base driver __init__ method."""
99 super().__init__()
101 def init_driver(self):
102 """This method gets called after all machines are created and can be used to initialize the driver.
104 After the driver is initialized, the self.init_machine function is
105 called for each machine associated with that driver.
106 """
108 def init_machine(self, machine: 'BaseMachineType'):
109 """This method gets called for each active machine using that driver while initialization.
111 If this function raises an Exception, it gets added to the machine.errors
112 list and the machine does not initialize successfully.
114 Arguments:
115 machine: Machine instance
116 """
118 def update_machine(
119 self, old_machine_state: dict[str, Any], machine: 'BaseMachineType'
120 ):
121 """This method gets called for each update of a machine.
123 Note:
124 machine.restart_required can be set to True here if the machine needs a manual restart to apply the changes
126 Arguments:
127 old_machine_state: Dict holding the old machine state before update
128 machine: Machine instance with the new state
129 """
131 def restart_machine(self, machine: 'BaseMachineType'):
132 """This method gets called on manual machine restart e.g. by using the restart machine action in the Admin Center.
134 Note:
135 `machine.restart_required` gets set to False again before this function is called
137 Arguments:
138 machine: Machine instance
139 """
141 def ping_machines(self):
142 """Ping all machines using this driver to check if they are online.
144 This is called periodically by a background task if the setting 'MACHINE_PING_ENABLED' is active.
145 """
147 def get_machines(self, **kwargs):
148 """Return all machines using this driver (By default only initialized machines).
150 Keyword Arguments:
151 name (str): Machine name
152 machine_type (BaseMachineType): Machine type definition (class)
153 initialized (bool | None): use None to get all machines (default: True)
154 active (bool): machine needs to be active
155 base_driver (BaseDriver): base driver (class)
156 """
157 from machine import registry
159 kwargs.pop('driver', None)
161 return registry.get_machines(driver=self, **kwargs)
163 def handle_error(self, error: Exception | str):
164 """Handle driver error.
166 Arguments:
167 error: Exception or string
168 """
169 if error not in self.errors:
170 self.set_shared_state('errors', [*self.errors, error])
172 # --- state getters/setters
173 @property
174 def errors(self) -> list[str | Exception]:
175 """List of driver errors."""
176 return self.get_shared_state('errors', [])
179class BaseMachineType(
180 ClassValidationMixin,
181 ClassProviderMixin,
182 get_shared_class_instance_state_mixin(lambda x: f'machine:machine:{x.pk}'),
183):
184 """Base class for machine types.
186 Attributes:
187 SLUG: Slug string for identifying the machine type in format /[a-z-]+/ (required)
188 NAME: User friendly name for displaying (required)
189 DESCRIPTION: Description of what this machine type can do (required)
191 base_driver: Reference to the base driver for this machine type
193 MACHINE_SETTINGS: Machine type specific settings dict (optional)
195 MACHINE_STATUS: Set of status codes this machine type can have
196 default_machine_status: Default machine status with which this machine gets initialized
197 """
199 SLUG: str
200 NAME: str
201 DESCRIPTION: str
203 base_driver: type[BaseDriver]
205 MACHINE_SETTINGS: dict[str, SettingsKeyType]
207 MACHINE_STATUS: type[MachineStatus]
208 default_machine_status: MachineStatus
210 # used by the ClassValidationMixin
211 required_attributes = [
212 'SLUG',
213 'NAME',
214 'DESCRIPTION',
215 'base_driver',
216 'MACHINE_STATUS',
217 'default_machine_status',
218 ]
220 def __init__(self, machine_config: MachineConfig) -> None:
221 """Base machine type __init__ function."""
222 from machine import registry
223 from machine.models import MachineSetting
225 self.pk = machine_config.pk
226 self.driver = registry.get_driver_instance(machine_config.driver)
228 if not self.driver:
229 self.handle_error(f"Driver '{machine_config.driver}' not found")
230 if self.driver and not isinstance(self.driver, self.base_driver):
231 self.handle_error(
232 f"'{self.driver.NAME}' is incompatible with machine type '{self.NAME}'"
233 )
235 self.machine_settings: dict[str, SettingsKeyType] = getattr(
236 self, 'MACHINE_SETTINGS', {}
237 )
238 self.driver_settings: dict[str, SettingsKeyType] = getattr(
239 self.driver, 'MACHINE_SETTINGS', {}
240 )
242 self.setting_types: list[
243 tuple[dict[str, SettingsKeyType], MachineSetting.ConfigType]
244 ] = [
245 (self.machine_settings, MachineSetting.ConfigType.MACHINE),
246 (self.driver_settings, MachineSetting.ConfigType.DRIVER),
247 ]
249 def __str__(self):
250 """String representation of a machine."""
251 return f'{self.name}'
253 def __repr__(self):
254 """Python representation of a machine."""
255 return f'<{self.__class__.__name__}: {self.name}>'
257 # --- properties
258 @property
259 def machine_config(self):
260 """Machine_config property which is a reference to the database entry."""
261 # always fetch the machine_config if needed to ensure we get the newest reference
262 from .models import MachineConfig
264 return MachineConfig.objects.filter(pk=self.pk).first()
266 @property
267 def name(self):
268 """The machines name."""
269 if config := self.machine_config:
270 return config.name
272 @property
273 def active(self):
274 """The machines active status."""
275 if config := self.machine_config:
276 return config.active
278 return False
280 # --- hook functions
281 def initialize(self):
282 """Machine initialization function, gets called after all machines are loaded."""
283 if self.driver is None:
284 return
286 # check if all required settings are defined before continue with init process
287 settings_valid, missing_settings = self.check_settings()
288 if not settings_valid:
289 error_parts = []
290 for config_type, missing in missing_settings.items():
291 if len(missing) > 0:
292 error_parts.append(
293 f'{config_type.name} settings: ' + ', '.join(missing)
294 )
295 self.handle_error(f'Missing {" and ".join(error_parts)}')
296 return
298 try:
299 self.driver.init_machine(self)
300 self.initialized = True
301 except Exception as e:
302 self.handle_error(e)
304 def update(self, old_state: dict[str, Any]):
305 """Machine update function, gets called if the machine itself changes or their settings.
307 Arguments:
308 old_state: Dict holding the old machine state before update
309 """
310 if self.driver is None:
311 return
313 try:
314 self.driver.update_machine(old_state, self)
316 # check if the active state has changed and initialize the machine if necessary
317 if old_state['active'] != self.active:
318 if self.initialized is False and self.active is True:
319 self.initialize()
320 elif self.initialized is True and self.active is False:
321 self.initialized = False
322 except Exception as e:
323 self.handle_error(e)
325 def restart(self):
326 """Machine restart function, can be used to manually restart the machine from the admin ui.
328 This will first reset the machines state (errors, status, status_text) and then call the drivers restart function.
329 """
330 if self.driver is None:
331 return
333 try:
334 # reset the machine state
335 self.restart_required = False
336 self.reset_errors()
337 self.set_status(self.default_machine_status)
338 self.set_status_text('')
340 # call the driver restart function
341 self.driver.restart_machine(self)
342 except Exception as e:
343 self.handle_error(e)
345 # --- helper functions
346 def handle_error(self, error: Exception | str):
347 """Helper function for capturing errors with the machine.
349 Arguments:
350 error: Exception or string
351 """
352 if error not in self.errors:
353 self.set_shared_state('errors', [*self.errors, error])
355 def reset_errors(self):
356 """Helper function for resetting the error list for a machine."""
357 self.set_shared_state('errors', [])
359 def get_setting(
360 self, key: str, config_type_str: Literal['M', 'D'], cache: bool = False
361 ):
362 """Return the 'value' of the setting associated with this machine.
364 Arguments:
365 key: The 'name' of the setting value to be retrieved
366 config_type_str: Either "M" (machine scoped settings) or "D" (driver scoped settings)
367 cache: Whether to use RAM cached value (default = False)
368 """
369 from machine.models import MachineSetting
371 config_type = MachineSetting.get_config_type(config_type_str)
372 return MachineSetting.get_setting(
373 key,
374 machine_config=self.machine_config,
375 config_type=config_type,
376 cache=cache,
377 )
379 def set_setting(self, key: str, config_type_str: Literal['M', 'D'], value: Any):
380 """Set plugin setting value by key.
382 Arguments:
383 key: The 'name' of the setting to set
384 config_type_str: Either "M" (machine scoped settings) or "D" (driver scoped settings)
385 value: The 'value' of the setting
386 """
387 from machine.models import MachineSetting
389 config_type = MachineSetting.get_config_type(config_type_str)
390 MachineSetting.set_setting(
391 key,
392 value,
393 None,
394 machine_config=self.machine_config,
395 config_type=config_type,
396 )
398 def check_settings(self):
399 """Check if all required settings for this machine are defined.
401 Returns:
402 is_valid: Are all required settings defined
403 missing_settings: dict[ConfigType, list[str]] of all settings that are missing (empty if is_valid is 'True')
404 """
405 from machine.models import MachineSetting
407 missing_settings: dict[MachineSetting.ConfigType, list[str]] = {}
408 for settings, config_type in self.setting_types:
409 _nbr, missing = MachineSetting.check_all_settings(
410 settings_definition=settings,
411 machine_config=self.machine_config,
412 config_type=config_type,
413 )
414 missing_settings[config_type] = missing
416 return all(
417 len(missing) == 0 for missing in missing_settings.values()
418 ), missing_settings
420 def set_status(self, status: MachineStatus):
421 """Set the machine status code. There are predefined ones for each MachineType.
423 Import the MachineType to access it's `MACHINE_STATUS` enum.
425 Arguments:
426 status: The new MachineStatus code to set
427 """
428 self.set_shared_state('status', status.value)
430 def set_status_text(self, status_text: str):
431 """Set the machine status text. It can be any arbitrary text.
433 Arguments:
434 status_text: The new status text to set
435 """
436 self.set_shared_state('status_text', status_text)
438 def set_properties(self, properties: list[MachineProperty]):
439 """Set the machine properties. This can be any arbitrary dict with model information, etc.
441 Arguments:
442 properties: The new properties dict to set
443 """
444 for p in properties:
445 if 'type' not in p:
446 p['type'] = 'str'
448 if 'group' not in p:
449 p['group'] = ''
451 if p['type'] == 'progress' and 'max_progress' not in p:
452 p['max_progress'] = 100
453 if 'max_progress' not in p:
454 p['max_progress'] = None
456 self.set_shared_state('properties', properties)
458 # --- state getters/setters
459 @property
460 def initialized(self) -> bool:
461 """Initialized state of the machine."""
462 return self.get_shared_state('initialized', False)
464 @initialized.setter
465 def initialized(self, value: bool):
466 self.set_shared_state('initialized', value)
468 @property
469 def restart_required(self) -> bool:
470 """Restart required state of the machine."""
471 return self.get_shared_state('restart_required', False)
473 @restart_required.setter
474 def restart_required(self, value: bool):
475 self.set_shared_state('restart_required', value)
477 @property
478 def errors(self) -> list[str | Exception]:
479 """List of machine errors."""
480 return self.get_shared_state('errors', [])
482 @property
483 def status(self) -> MachineStatus:
484 """Machine status code."""
485 status_code = self.get_shared_state('status', self.default_machine_status.value)
486 return self.MACHINE_STATUS(status_code)
488 @property
489 def status_text(self) -> str:
490 """Machine status text."""
491 return self.get_shared_state('status_text', '')
493 @property
494 def properties(self) -> list[MachineProperty]:
495 """Return a dict of all relevant machine properties."""
496 return self.get_shared_state('properties', [])
498 @property
499 def properties_dict(self) -> dict[str, MachineProperty]:
500 """Return a dict of all machine properties with key as dict key."""
501 return {prop['key']: prop for prop in self.properties}