From 96e69240caa3bc6b5a7be93f88ea90a0e8e38035 Mon Sep 17 00:00:00 2001
From: Arksine <arksine.code@gmail.com>
Date: Tue, 11 May 2021 18:14:33 -0400
Subject: [PATCH] utils: add annotations

Add a SentinelClass singleton.  This can be used to generate Sentinels by
modules that need to use the Sentinel pattern for default values.  This
allows for more robust type checking, as the Sentinel type should not be
included as a return value.

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
---
 moonraker/utils.py | 49 ++++++++++++++++++++++++++++++++++------------
 1 file changed, 36 insertions(+), 13 deletions(-)

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 <arksine.code@gmail.com>
 #
 # 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)