CelestialSurveyor

1
import datetime
2
import inspect
3
import logging
4
import os
5
import sys
6
import threading
7
import wx.lib.newevent
8

9
from functools import wraps
10
from multiprocessing import Queue
11
from logging.handlers import QueueListener
12
from typing import Callable
13

14
# Define a custom event for logging
15
LogEvent, EVT_LOG_EVENT = wx.lib.newevent.NewEvent()
16

17

18
def get_function_and_class_name(func: Callable) -> str:
19
    """
20
    Get the function and class name of the provided function.
21

22
    Args:
23
        func: The function to extract the name from.
24

25
    Returns:
26
        str: The combined function and class name if applicable.
27
    """
28
    # Check if the provided object is a function
29
    function_name = func.__name__
30
    if inspect.ismethod(func):
31
        class_name = func.__self__.__class__.__name__
32
        res =  f"{class_name}.{function_name}"
33
    else:
34
        res = function_name
35
    return res
36

37

38
def arg_logger(func: Callable) -> Callable:
39
    """
40
    Decorator for logging function arguments and return values.
41

42
    Args:
43
        func: The function to decorate.
44

45
    Returns:
46
        Callable: The decorated function.
47
    """
48
    @wraps(func)
49
    def new_func(*args, **kwargs):
50
        saved_args = locals()
51
        logger = get_logger()
52
        func_name = get_function_and_class_name(func)
53

54
        logger.log.debug(f"{func_name}({saved_args})")
55
        return func(*args, **kwargs)
56
    return new_func
57

58

59
class LogHandler(logging.Handler):
60
    """
61
    Custom logging handler that appends log messages to a wx.TextCtrl widget.
62

63
    Args:
64
        text_ctrl: The wx.TextCtrl widget to append log messages to.
65
    """
66
    def __init__(self, text_ctrl):
67
        logging.Handler.__init__(self)
68
        self._text_ctrl = text_ctrl
69

70
    @property
71
    def text_ctrl(self):
72
        return self._text_ctrl
73

74
    @text_ctrl.setter
75
    def text_ctrl(self, value):
76
        self._text_ctrl = value
77

78
    def emit(self, record: logging.LogRecord) -> None:
79
        """
80
        Append log message to the text control.
81
        """
82
        msg = self.format(record)
83
        if self.text_ctrl is not None:
84
            wx.CallAfter(self.text_ctrl.AppendText, msg + '\n')  # Append log message to the text control
85

86

87
class LogFileHandler(logging.FileHandler):
88
    """
89
    Custom logging handler that logs into file.
90

91
    """
92
    def __init__(self):
93
        fp = self.__get_log_file_path()
94
        super(LogFileHandler, self).__init__(fp)
95

96
    @staticmethod
97
    def __get_log_file_path() -> str:
98
        """
99
        Generates the log file path based on the current timestamp.
100

101
        Returns:
102
        str: The log file path.
103
        """
104
        folder = os.path.join(sys.path[1], "log")
105
        os.makedirs(folder, exist_ok=True)
106
        current_time = datetime.datetime.now()
107
        file_name = "{}.log".format(current_time.strftime("%y%m%d%H%M"))
108
        fp = os.path.join(folder, file_name)
109
        return fp
110

111

112
class Logger:
113
    """
114
    Custom logger class that manages logging messages. Singleton class.
115
    """
116

117
    log_level = logging.INFO
118
    _instance_lock = threading.Lock()
119
    formatter = None
120

121
    def __new__(cls, text_ctrl=None):
122
        with cls._instance_lock:
123
            if not hasattr(cls, '_instance'):
124
                cls._instance = super(Logger, cls).__new__(cls)
125
                cls._instance.log = logging.getLogger('MyLogger')
126
                cls._instance.log.setLevel(cls.log_level)
127
                if cls.log_level == logging.DEBUG:
128
                    cls.formatter = logging.Formatter(
129
                        '[%(asctime)s] p%(process)s {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s',
130
                        '%m-%d %H:%M:%S')
131
                else:
132
                    cls.formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
133

134
                gui_handler = LogHandler(text_ctrl)
135
                gui_handler.setFormatter(cls.formatter)
136
                cls._instance.log.addHandler(gui_handler)
137
                console_handler = logging.StreamHandler()
138
                console_handler.setLevel(cls.log_level)
139
                console_handler.setFormatter(cls.formatter)
140
                cls._instance.log.addHandler(console_handler)
141
                file_handler = LogFileHandler()
142
                file_handler.setLevel(cls.log_level)
143
                file_handler.setFormatter(cls.formatter)
144
                cls._instance.log.addHandler(file_handler)
145
                cls._instance.text_ctrl = text_ctrl
146

147
        return cls._instance
148

149
    @property
150
    def text_ctrl(self):
151
        return self.log.handlers[0].text_ctrl
152

153
    @text_ctrl.setter
154
    def text_ctrl(self, value):
155
        self.log.handlers[0].text_ctrl = value
156

157
    def start_process_listener(self, queue: Queue) -> None:
158
        """Start listening for log messages from the given queue. Used to get logs from child processes."""
159
        self.listener = QueueListener(queue, *self.log.handlers)
160
        self.listener.start()
161

162
    def stop_process_listener(self) -> None:
163
        """Stop listening for log messages from the queue. Used to stop getting logs from child processes."""
164
        self.listener.stop()
165

166

167
def get_logger() -> Logger:
168
    return Logger()
169