feat: Improves logger to remove duplication

Tools/Logger.py

@@ -1,52 +1,75 @@
 #region imports
 from AlgorithmImports import *
 #endregion
-
-########################################################################################
-#                                                                                      #
-# Licensed under the Apache License, Version 2.0 (the "License");                      #
-# you may not use this file except in compliance with the License.                     #
-# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0   #
-#                                                                                      #
-# Unless required by applicable law or agreed to in writing, software                  #
-# distributed under the License is distributed on an "AS IS" BASIS,                    #
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.             #
-# See the License for the specific language governing permissions and                  #
-# limitations under the License.                                                       #
-#                                                                                      #
-# Copyright [2021] [Rocco Claudio Cannizzaro]                                          #
-#                                                                                      #
-########################################################################################
-
+import sys
+import pandas as pd
+from collections import deque
 
 class Logger:
-    def __init__(self, context, className=None, logLevel=0):
-        if logLevel is None:
-            logLevel = 0
-
+    def __init__(self, context, className=None, logLevel=0, buffer_size=100):
         self.context = context
         self.className = className
         self.logLevel = logLevel
+        self.log_buffer = deque(maxlen=buffer_size)
+        self.current_pattern = []
+        self.pattern_count = 0
 
     def Log(self, msg, trsh=0):
-        # Set the class name (if available)
-        if self.className is not None:
-            className = f"{self.className}."
-
-        # Set the prefix for the message
-        if trsh is None or trsh <= 0:
-            prefix = "ERROR"
-        elif trsh == 1:
-            prefix = "WARNING"
-        elif trsh == 2:
-            prefix = "INFO"
-        elif trsh == 3:
-            prefix = "DEBUG"
+        if self.logLevel < trsh:
+            return
+
+        className = f"{self.className}." if self.className else ""
+        prefix = ["ERROR", "WARNING", "INFO", "DEBUG", "TRACE"][min(trsh, 4)]
+        log_msg = f"{prefix} -> {className}{sys._getframe(2).f_code.co_name}: {msg}"
+
+        self.process_log(log_msg)
+
+    def process_log(self, log_msg):
+        if not self.current_pattern:
+            self.print_log(log_msg)
+            self.current_pattern.append(log_msg)
         else:
-            prefix = "TRACE"
+            pattern_index = self.find_pattern_start(log_msg)
+            if pattern_index == -1:
+                self.print_pattern()
+                self.print_log(log_msg)
+                self.current_pattern.append(log_msg)
+            else:
+                if pattern_index == 0:
+                    self.pattern_count += 1
+                else:
+                    self.print_pattern()
+                    self.print_log("--- New log cycle starts ---")
+                    self.current_pattern = self.current_pattern[pattern_index:]
+                    self.pattern_count = 1
 
-        if self.logLevel >= trsh:
-            self.context.Log(f" {prefix} -> {className}{sys._getframe(2).f_code.co_name}: {msg}")
+        self.log_buffer.append(log_msg)
+
+    def find_pattern_start(self, log_msg):
+        for i in range(len(self.current_pattern)):
+            if log_msg == self.current_pattern[i]:
+                if self.is_pattern_repeating(i):
+                    return i
+        return -1
+
+    def is_pattern_repeating(self, start_index):
+        pattern_length = len(self.current_pattern) - start_index
+        if len(self.log_buffer) < pattern_length:
+            return False
+        return list(self.log_buffer)[-pattern_length:] == self.current_pattern[start_index:]
+
+    def print_pattern(self):
+        if self.pattern_count > 1:
+            self.print_log(f"The following pattern repeated {self.pattern_count} times:")
+            for msg in self.current_pattern:
+                self.print_log(f"  {msg}")
+        elif self.pattern_count == 1:
+            for msg in self.current_pattern:
+                self.print_log(msg)
+        self.pattern_count = 0
+
+    def print_log(self, msg):
+        self.context.Log(msg)
 
     def error(self, msg):
         self.Log(msg, trsh=0)
@@ -64,9 +87,6 @@ def trace(self, msg):
         self.Log(msg, trsh=4)
 
     def dataframe(self, data):
-        """
-        Should be used to print out to the log as an info the data sent as a dictionary via the data.
-        """
         if isinstance(data, list):
             columns = list(data[0].keys())
         else:
@@ -76,3 +96,7 @@ def dataframe(self, data):
 
         if df.shape[0] > 0:
             self.info(f"\n{df.to_string(index=False)}")
+
+    def __del__(self):
+        self.print_pattern()
+