Prepare 7.1.006 ( December release)

Classes/LoggingManagement.py

@@ -10,12 +10,14 @@
 
 """
 
+import inspect
 import json
 import logging
 import os
 import os.path
 import threading
 import time
+import traceback
 from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
 from pathlib import Path
 from queue import PriorityQueue, Queue
@@ -43,15 +45,19 @@ def __init__(self, pluginconf, PluginHealth, HardwareID, ListOfDevices, permitTo
         self._startTime = int(time.time())
 
         self.debugzigpy = False
-        zigpy_loging_mode("warning")
         self.debugZNP = False
-        zigpy_logging_znp("warning")
         self.debugEZSP = False
-        zigpy_logging_ezsp("warning")
         self.debugZigate = False
-        zigpy_logging_zigate("warning")
         self.debugdeconz = False
-        zigpy_logging_deconz("warning")
+
+        configure_zigpy_loggers("warning")
+        configure_zigpy_znp_loggers("warning")
+        configure_zigpy_ezsp_loggers("warning")
+        configure_zigpy_zigate_loggers("warning")
+        configure_zigpy_deconz_loggers("warning")
+
+
+
 
         self.zigpy_login()
 
@@ -73,6 +79,13 @@ def reset_new_error(self):
     def is_new_error(self):
         return bool(self._newError and bool(self.LogErrorHistory))
 
+    def zigpy_login(self):
+        _configure_debug_mode(self, self.debugzigpy, "Zigpy", configure_zigpy_loggers)
+        _configure_debug_mode(self, self.debugZNP, "ZigpyZNP", configure_zigpy_znp_loggers)
+        _configure_debug_mode(self, self.debugEZSP, "ZigpyEZSP", configure_zigpy_ezsp_loggers)
+        _configure_debug_mode(self, self.debugZigate, "ZigpyZigate", configure_zigpy_zigate_loggers)
+        _configure_debug_mode(self, self.debugdeconz, "ZigpydeCONZ", configure_zigpy_deconz_loggers)
+
     def loggingUpdatePluginVersion(self, Version):
         self.PluginVersion = Version
         if (
@@ -83,6 +96,7 @@ def loggingUpdatePluginVersion(self, Version):
         ):
             self.LogErrorHistory[str(self.LogErrorHistory["LastLog"])]["PluginVersion"] = Version
 
+
     def loggingUpdateFirmware(self, FirmwareVersion, FirmwareMajorVersion):
         if self.FirmwareVersion and self.FirmwareMajorVersion:
             return
@@ -97,51 +111,12 @@ def loggingUpdateFirmware(self, FirmwareVersion, FirmwareMajorVersion):
             self.LogErrorHistory[str(self.LogErrorHistory["LastLog"])]["FirmwareVersion"] = FirmwareVersion
             self.LogErrorHistory[str(self.LogErrorHistory["LastLog"])]["FirmwareMajorVersion"] = FirmwareMajorVersion
 
-    def zigpy_login(self):
-        if not self.debugzigpy and self.pluginconf.pluginConf["Zigpy"]:
-            self.debugzigpy = True
-            zigpy_loging_mode("debug")
-        elif self.debugzigpy and not self.pluginconf.pluginConf["Zigpy"]:
-            self.debugzigpy = False
-            zigpy_loging_mode("warning")
-
-        # Debug ZNP
-        if not self.debugZNP and self.pluginconf.pluginConf["ZigpyZNP"]:
-            self.debugZNP = True
-            zigpy_logging_znp("debug")
-        elif self.debugZNP and not self.pluginconf.pluginConf["ZigpyZNP"]:
-            self.debugZNP = False
-            zigpy_logging_znp("warning")
-
-        # Debug Bellows/Ezsp
-        if not self.debugEZSP and self.pluginconf.pluginConf["ZigpyEZSP"]:
-            self.debugEZSP = True
-            zigpy_logging_ezsp("debug")
-        elif self.debugEZSP and not self.pluginconf.pluginConf["ZigpyEZSP"]:
-            self.debugEZSP = False
-            zigpy_logging_ezsp("warning")
-
-        # Debug Zigate
-        if not self.debugZigate and self.pluginconf.pluginConf["ZigpyZigate"]:
-            self.debugZigate = True
-            zigpy_logging_zigate("debug")
-        elif self.debugZigate and not self.pluginconf.pluginConf["ZigpyZigate"]:
-            self.debugZigate = False
-            zigpy_logging_zigate("warning")
-
-        # Debug deConz
-        if not self.debugdeconz and self.pluginconf.pluginConf["ZigpydeCONZ"]:
-            self.debugdeconz = True
-            zigpy_logging_deconz("debug")
-        elif self.debugdeconz and not self.pluginconf.pluginConf["ZigpydeCONZ"]:
-            self.debugdeconz = False
-            zigpy_logging_deconz("warning")
-
 
     def openLogFile(self):
         self.open_logging_mode()
         self.open_log_history()
 
+
     def open_logging_mode(self):
         import sys
 
@@ -195,7 +170,6 @@ def open_logging_mode(self):
                 os.chmod(_logfilename, self.pluginconf.pluginConf["PluginLogMode"])
 
 
-
     def open_log_history(self):
         _pluginlogs = Path( self.pluginconf.pluginConf["pluginLogs"] )
         jsonLogHistory = _pluginlogs / ( LOG_ERROR_HISTORY + "%02d.json" % self.HardwareID) 
@@ -222,6 +196,7 @@ def open_log_history(self):
 
         handle.close()
 
+
     def closeLogFile(self):
         if self.logging_thread is None:
             Domoticz.Error("closeLogFile - logging_thread is None")
@@ -245,6 +220,7 @@ def closeLogFile(self):
         loggingWriteErrorHistory(self)
         Domoticz.Log("Logging Thread shutdown")
 
+
     def loggingCleaningErrorHistory(self):
         if len(self.LogErrorHistory) > 1:
             idx = list(self.LogErrorHistory.keys())[1]
@@ -262,10 +238,12 @@ def loggingCleaningErrorHistory(self):
         if len(self.LogErrorHistory) == 1:
             self.LogErrorHistory.clear()
 
+
     def loggingClearErrorHistory(self):
         self.LogErrorHistory.clear()
         self._newError = False
 
+
     def logging(self, module, logType, message, nwkid=None, context=None):
         #Domoticz.Log("%s %s %s %s %s %s %s" % (module, logType, message, nwkid, context, self.logging_thread, self.logging_queue))
         # Set python3 modules logging if required
@@ -277,6 +255,12 @@ def logging(self, module, logType, message, nwkid=None, context=None):
         except AttributeError:
             thread_id = 0
 
+        if logType == "Error":
+            if context:
+                context["StackTrace"] = get_stack_trace()
+            else:
+                context = { "StackTrace": get_stack_trace() }
+
         if self.logging_thread and self.logging_queue:
             logging_tuple = [
                 str(time.time()),
@@ -395,30 +379,40 @@ def loggingError(self, thread_name, module, message, nwkid, context):
     loggingWriteErrorHistory(self)
 
 
-def loggingBuildContext(self, thread_name, module, message, nwkid, context):
+def get_stack_trace():
+    # Get the current stack frame
+    current_frame = inspect.currentframe()
 
-    if not self.PluginHealth:
-        _txt = "Not Started"
-    if "Txt" not in self.PluginHealth:
-        _txt = "Not Started"
-    else:
-        _txt = self.PluginHealth["Txt"]
+    # Get the call stack ( -2 to exclude the get_stack_trace() and logging()
+    stack = traceback.extract_stack(current_frame)[:-2]
+
+    # Format the stack trace
+    stack_trace = traceback.format_list(stack)
+
+    # Alternatively, you can return the formatted stack trace as a string
+    return ''.join(stack_trace)
+
+
+def loggingBuildContext(self, thread_name, module, message, nwkid, context=None):
+
+    _txt = self.PluginHealth.get("Txt", "Not Started")
+
     _context = {
         "Time": int(time.time()),
-        "Module": module,
-        "nwkid": nwkid,
-        "PluginHealth": _txt,
-        "message": message,
         "PermitToJoin": self.permitTojoin,
+        "PluginHealth": _txt,
         "Thread": thread_name,
+        "nwkid": nwkid,
+        "Module": module,
+        "message": message,
     }
-    if nwkid and nwkid in self.ListOfDevices:
-        _context["DeviceInfos"] = dict(self.ListOfDevices[nwkid])
+
+    if nwkid in self.ListOfDevices:
+        _context["DeviceInfos"] = dict(self.ListOfDevices.get(nwkid, {}))
+
     if context is not None:
-        if isinstance(context, dict):
-            _context["context"] = context.copy()
-        elif isinstance(context, (str, int)):
-            _context["context"] = str(context)
+        _context["context"] = context.copy() if isinstance(context, dict) else str(context)
+
     return _context
 
 
@@ -501,56 +495,73 @@ def logging_thread(self):
             Domoticz.Error("logging_thread unexpected tuple %s" % (str(logging_tuple)))
     Domoticz.Log("logging_thread - ended")
 
-
-def zigpy_loging_mode(mode):
+def configure_loggers(logger_names, mode):
+    #Domoticz.Log(f"configure_loggers({logger_names} with {_set_logging_level})")
     _set_logging_level = logging.DEBUG if mode == "debug" else logging.WARNING
-    Domoticz.Log( "zigpy_login_mode(%s)" %_set_logging_level)
-    logging.getLogger("zigpy.application").setLevel(_set_logging_level)
-    logging.getLogger("zigpy").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.zdo").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.zcl").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.profiles").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.quirks").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.ota").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.appdb_schemas").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.backups").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.device").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.application").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.appdb").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.endpoint").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.group").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.neighbor").setLevel(_set_logging_level)
-    logging.getLogger("zigpy.topology").setLevel(_set_logging_level)
 
+    for logger_name in logger_names:
+        logging.getLogger(logger_name).setLevel(_set_logging_level)
 
-def zigpy_logging_znp(mode):
-    _set_logging_level = logging.DEBUG if mode == "debug" else logging.WARNING
-    logging.getLogger("zigpy_znp").setLevel(_set_logging_level)
-    logging.getLogger("zigpy_znp.zigbee").setLevel(_set_logging_level)
-    logging.getLogger("zigpy_znp.zigbee.application").setLevel(_set_logging_level)
-    logging.getLogger("zigpy_znp.zigbee.device").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppZnp").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppGeneric").setLevel(_set_logging_level)
-
-
-def zigpy_logging_ezsp(mode): 
-    _set_logging_level = logging.DEBUG if mode == "debug" else logging.WARNING  
-    logging.getLogger("bellows").setLevel(_set_logging_level)
-    logging.getLogger("bellows.zigbee").setLevel(_set_logging_level)
-    logging.getLogger("bellows.zigbee.application").setLevel(_set_logging_level)
-    logging.getLogger("bellows.zigbee.device").setLevel(_set_logging_level)
-    logging.getLogger("bellows.uart").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppBellows").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppGeneric").setLevel(_set_logging_level)
-
-
-def zigpy_logging_zigate(mode):
-    _set_logging_level = logging.DEBUG if mode == "debug" else logging.WARNING 
-    logging.getLogger("zigpy_zigate").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppZigate").setLevel(_set_logging_level)
-
-
-def zigpy_logging_deconz(mode):
-    _set_logging_level = logging.DEBUG if mode == "debug" else logging.WARNING 
-    logging.getLogger("zigpy_deconz").setLevel(_set_logging_level)
-    logging.getLogger("Classes.ZigpyTransport.AppDeconz").setLevel(_set_logging_level)
+
+# Loggers configurations
+def configure_zigpy_loggers(mode):
+    logger_names = [
+        "zigpy.application", "zigpy", "zigpy.zdo", "zigpy.zcl",
+        "zigpy.profiles", "zigpy.quirks", "zigpy.ota",
+        "zigpy.appdb_schemas", "zigpy.backups", "zigpy.device",
+        "zigpy.application", "zigpy.appdb", "zigpy.endpoint",
+        "zigpy.group", "zigpy.neighbor", "zigpy.topology"
+    ]
+    configure_loggers(logger_names, mode)
+
+
+def configure_zigpy_znp_loggers(mode):
+    logger_names = [
+        "zigpy_znp", 
+        "zigpy_znp.zigbee", 
+        "zigpy_znp.zigbee.application", 
+        "zigpy_znp.zigbee.device", 
+        "Classes.ZigpyTransport.AppZnp", 
+        "Classes.ZigpyTransport.AppGeneric"
+    ]
+    configure_loggers(logger_names, mode)
+
+
+def configure_zigpy_ezsp_loggers(mode):
+    logger_names = [
+        "bellows", 
+        "bellows.zigbee", 
+        "bellows.zigbee.application", 
+        "bellows.zigbee.device", 
+        "bellows.uart", 
+        "Classes.ZigpyTransport.AppBellows", 
+        "Classes.ZigpyTransport.AppGeneric"
+    ]
+    configure_loggers(logger_names, mode)
+
+
+def configure_zigpy_zigate_loggers(mode):
+    logger_names = [
+        "zigpy_zigate",
+        "Classes.ZigpyTransport.AppZigate"
+    ]
+    configure_loggers(logger_names, mode)
+
+
+def configure_zigpy_deconz_loggers(mode):
+    logger_names = [
+        "zigpy_deconz",
+        "Classes.ZigpyTransport.AppDeconz"
+    ]
+    configure_loggers(logger_names, mode)
+
+
+# Main configuration function
+
+def _configure_debug_mode(self, debug_flag, config_name, config_function):
+    if not debug_flag and self.pluginconf.pluginConf[config_name]:
+        debug_flag = True
+        config_function("debug")
+    elif debug_flag and not self.pluginconf.pluginConf[config_name]:
+        debug_flag = False
+        config_function("warning")