diff --git a/moonraker/utils.py b/moonraker/utils.py index 0c9e803..04c3639 100644 --- a/moonraker/utils.py +++ b/moonraker/utils.py @@ -3,6 +3,8 @@ # Copyright (C) 2020 Eric Callahan # # This file may be distributed under the terms of the GNU GPLv3 license + +from __future__ import annotations import logging import logging.handlers import os @@ -11,11 +13,29 @@ import subprocess import asyncio from queue import SimpleQueue as Queue +# Annotation imports +from typing import ( + Optional, + ClassVar, + Tuple, + Dict +) + class ServerError(Exception): - def __init__(self, message, status_code=400): + def __init__(self, message: str, status_code: int = 400) -> None: Exception.__init__(self, message) self.status_code = status_code + +class SentinelClass: + _instance: ClassVar[Optional[SentinelClass]] = None + + @staticmethod + def get_instance() -> SentinelClass: + if SentinelClass._instance is None: + SentinelClass._instance = SentinelClass() + return SentinelClass._instance + # Coroutine friendly QueueHandler courtesy of Martjin Pieters: # https://www.zopatista.com/python/2019/05/11/asyncio-logging/ class LocalQueueHandler(logging.handlers.QueueHandler): @@ -30,9 +50,12 @@ class LocalQueueHandler(logging.handlers.QueueHandler): # Timed Rotating File Handler, based on Klipper's implementation class MoonrakerLoggingHandler(logging.handlers.TimedRotatingFileHandler): - def __init__(self, software_version, filename, **kwargs): + def __init__(self, + software_version: str, + filename: str, + **kwargs) -> None: super(MoonrakerLoggingHandler, self).__init__(filename, **kwargs) - self.rollover_info = { + self.rollover_info: Dict[str, str] = { 'header': f"{'-'*20}Moonraker Log Start{'-'*20}", 'version': f"Git Version: {software_version}", } @@ -40,10 +63,10 @@ class MoonrakerLoggingHandler(logging.handlers.TimedRotatingFileHandler): if self.stream is not None: self.stream.write("\n".join(lines) + "\n") - def set_rollover_info(self, name, item): + def set_rollover_info(self, name: str, item: str) -> None: self.rollover_info[name] = item - def doRollover(self): + def doRollover(self) -> None: super(MoonrakerLoggingHandler, self).doRollover() lines = [line for line in self.rollover_info.values() if line] if self.stream is not None: @@ -51,7 +74,7 @@ class MoonrakerLoggingHandler(logging.handlers.TimedRotatingFileHandler): # Parse the git version from the command line. This code # is borrowed from Klipper. -def get_software_version(): +def get_software_version() -> str: moonraker_path = os.path.join( os.path.dirname(__file__), '..') @@ -64,20 +87,20 @@ def get_software_version(): ver, err = process.communicate() retcode = process.wait() if retcode == 0: - version = ver.strip() - if isinstance(version, bytes): - version = version.decode() - return version + return ver.strip().decode() else: - logging.debug(f"Error getting git version: {err}") + logging.debug(f"Error getting git version: {err.decode()}") except OSError: logging.exception("Error runing git describe") return "?" -def setup_logging(log_file, software_version): +def setup_logging(log_file: str, + software_version: str + ) -> Tuple[logging.handlers.QueueListener, + Optional[MoonrakerLoggingHandler]]: root_logger = logging.getLogger() - queue = Queue() + queue: Queue = Queue() queue_handler = LocalQueueHandler(queue) root_logger.addHandler(queue_handler) root_logger.setLevel(logging.INFO)