From c977948c2c6723cbbaf6b5190d8592faa05edd1d Mon Sep 17 00:00:00 2001 From: Arksine Date: Fri, 14 May 2021 19:13:02 -0400 Subject: [PATCH] octoprint_compat: add annotations Signed-off-by: Eric Callahan --- moonraker/components/octoprint_compat.py | 73 +++++++++++++++++------- 1 file changed, 52 insertions(+), 21 deletions(-) diff --git a/moonraker/components/octoprint_compat.py b/moonraker/components/octoprint_compat.py index 11d75a0..099c4d6 100644 --- a/moonraker/components/octoprint_compat.py +++ b/moonraker/components/octoprint_compat.py @@ -3,9 +3,22 @@ # Copyright (C) 2021 Nickolas Grigoriadis # # This file may be distributed under the terms of the GNU GPLv3 license. + +from __future__ import annotations import logging -import utils +# Annotation imports +from typing import ( + TYPE_CHECKING, + Any, + Dict, + List, +) +if TYPE_CHECKING: + from confighelper import ConfigHelper + from websockets import WebRequest + from . import klippy_apis + APIComp = klippy_apis.KlippyAPI OCTO_VERSION = '1.5.0' @@ -23,14 +36,14 @@ class OctoprintCompat: * Heater temperatures """ - def __init__(self, config): + def __init__(self, config: ConfigHelper) -> None: self.server = config.get_server() self.software_version = config['system_args'].get('software_version') # Local variables - self.klippy_apis = self.server.lookup_component('klippy_apis') - self.heaters = {} - self.last_print_stats = {} + self.klippy_apis: APIComp = self.server.lookup_component('klippy_apis') + self.heaters: Dict[str, Dict[str, Any]] = {} + self.last_print_stats: Dict[str, Any] = {} # Register status update event self.server.register_event_handler( @@ -78,31 +91,33 @@ class OctoprintCompat: # System # TODO: shutdown/reboot/restart operations - async def _init(self): + async def _init(self) -> None: self.heaters = {} # Fetch heaters try: + result: Dict[str, Any] + sensors: List[str] result = await self.klippy_apis.query_objects({'heaters': None}) sensors = result.get('heaters', {}).get('available_sensors', []) except self.server.error as e: logging.info(f'Error Configuring heaters: {e}') sensors = [] # subscribe objects - sub = {s: None for s in sensors} + sub: Dict[str, Any] = {s: None for s in sensors} sub['print_stats'] = None result = await self.klippy_apis.subscribe_objects(sub) self.last_print_stats = result.get('print_stats', {}) if sensors: self.heaters = {name: result.get(name, {}) for name in sensors} - def _handle_status_update(self, status): + def _handle_status_update(self, status: Dict[str, Any]) -> None: if 'print_stats' in status: self.last_print_stats.update(status['print_stats']) for heater_name, data in self.heaters.items(): if heater_name in status: data.update(status[heater_name]) - def printer_state(self): + def printer_state(self) -> str: klippy_state = self.server.get_klippy_state() if klippy_state in ["disconnected", "startup"]: return 'Offline' @@ -115,8 +130,8 @@ class OctoprintCompat: 'complete': 'Operational' }.get(self.last_print_stats.get('state', 'standby'), 'Error') - def printer_temps(self): - temps = {} + def printer_temps(self) -> Dict[str, Any]: + temps: Dict[str, Any] = {} for heater, data in self.heaters.items(): name = 'bed' if heater.startswith('extruder'): @@ -134,7 +149,9 @@ class OctoprintCompat: } return temps - async def _get_version(self, web_request): + async def _get_version(self, + web_request: WebRequest + ) -> Dict[str, str]: """ Version information """ @@ -144,7 +161,9 @@ class OctoprintCompat: 'text': f'OctoPrint (Moonraker {self.software_version})', } - async def _get_server(self, web_request): + async def _get_server(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Server status """ @@ -155,7 +174,9 @@ class OctoprintCompat: None if klippy_state == 'ready' else 'settings') } - async def _post_login_user(self, web_request): + async def _post_login_user(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Confirm session login. @@ -174,7 +195,9 @@ class OctoprintCompat: 'groups': ['admins', 'users'], } - async def _get_settings(self, web_request): + async def _get_settings(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Used to parse Octoprint capabilities @@ -209,7 +232,9 @@ class OctoprintCompat: }, } - async def _get_job(self, web_request): + async def _get_job(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Get current job status """ @@ -230,7 +255,9 @@ class OctoprintCompat: 'state': self.printer_state() } - async def _get_printer(self, web_request): + async def _get_printer(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Get Printer status """ @@ -252,11 +279,13 @@ class OctoprintCompat: }, } - async def _post_command(self, web_request): + async def _post_command(self, + web_request: WebRequest + ) -> Dict: """ Request to run some gcode command """ - commands = web_request.get('commands', []) + commands: List[str] = web_request.get('commands', []) for command in commands: logging.info(f'Executing GCode: {command}') try: @@ -267,7 +296,9 @@ class OctoprintCompat: return {} - async def _get_printerprofiles(self, web_request): + async def _get_printerprofiles(self, + web_request: WebRequest + ) -> Dict[str, Any]: """ Get Printer profiles """ @@ -287,5 +318,5 @@ class OctoprintCompat: } -def load_component(config): +def load_component(config: ConfigHelper) -> OctoprintCompat: return OctoprintCompat(config)