Saltar a contenido

Log Module

Logger

Bases: LoggerABC

Class to handle logging functionality.

Source code in devices\raspberry_pi_5\src\log\__init__.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class Logger(LoggerABC):
    """
    Class to handle logging functionality.
    """

    def __init__(
        self,
        writer_messages_queue: Queue,
        tag: Optional[str] = None,
        debug: bool = False,
        unique_tag: bool = False
    ) -> None:
        """
        Initialize the Logger class.

        Args:
            writer_messages_queue (Queue): Queue to hold log messages.
            tag (Optional[str]): Tag to identify the logger instance.
            debug (Optional[bool]): Flag to indicate if the logger is in debug mode.
            unique_tag (Optional[bool]): Whether to generate a unique tag for the logger instance.
        """
        # Check the type of debug
        is_instance(debug, bool)
        self.__debug = debug

        # Initialize the messages queue and events
        self.__writer_messages_queue = writer_messages_queue

        # Check the type of tag
        is_instance(tag, (str, NoneType,))
        self.__tag = self.get_unique_tag(tag) if unique_tag or not tag else tag

        # Log the initialization if a tag is provided
        self.debug("Initializing new Logger") if self.__tag else None

    @final
    def get_unique_tag(self, tag: str):
        return f"{tag}_{id(self)}" if tag else f"Logger_{id(self)}"

    @final
    def log(self, content: str, category: Category = Category.INFO) -> None:
        # Check the type of content
        is_instance(content, str)

        # Check the type of category
        is_instance(category, Category)

        # Create a message object
        msg = Message(content, category, self.__tag)

        # Put the message in the queue
        self.__writer_messages_queue.put(msg)

    @final
    def info(self, content: str) -> None:
        self.log(content, Category.INFO)

    @final
    def error(self, content: str) -> None:
        self.log(content, Category.ERROR)

    @final
    def warning(self, content: str) -> None:
        self.log(content, Category.WARNING)

    @final
    def debug(self, content: str) -> None:
        self.log(content, Category.DEBUG) if self.__debug else None

__init__(writer_messages_queue, tag=None, debug=False, unique_tag=False)

Initialize the Logger class.

Parameters:

Name Type Description Default
writer_messages_queue Queue

Queue to hold log messages.

 required
tag Optional[str]

Tag to identify the logger instance.

 None
debug Optional[bool]

Flag to indicate if the logger is in debug mode.

 False
unique_tag Optional[bool]

Whether to generate a unique tag for the logger instance.

 False
Source code in devices\raspberry_pi_5\src\log\__init__.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    writer_messages_queue: Queue,
    tag: Optional[str] = None,
    debug: bool = False,
    unique_tag: bool = False
) -> None:
    """
    Initialize the Logger class.

    Args:
        writer_messages_queue (Queue): Queue to hold log messages.
        tag (Optional[str]): Tag to identify the logger instance.
        debug (Optional[bool]): Flag to indicate if the logger is in debug mode.
        unique_tag (Optional[bool]): Whether to generate a unique tag for the logger instance.
    """
    # Check the type of debug
    is_instance(debug, bool)
    self.__debug = debug

    # Initialize the messages queue and events
    self.__writer_messages_queue = writer_messages_queue

    # Check the type of tag
    is_instance(tag, (str, NoneType,))
    self.__tag = self.get_unique_tag(tag) if unique_tag or not tag else tag

    # Log the initialization if a tag is provided
    self.debug("Initializing new Logger") if self.__tag else None

abstracts

LoggerABC

Bases: ABC

Abstract class to handle logging functionality.

Source code in devices\raspberry_pi_5\src\log\abstracts.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class LoggerABC(ABC):
    """
    Abstract class to handle logging functionality.
    """

    @abstractmethod
    def get_unique_tag(self, tag: str):
        """
        Generate a unique tag for the logger instance.

        Args:
            tag (str): The tag to identify the logger instance.

        Returns:
            str: A unique tag for the logger instance.
        """

    @abstractmethod
    def log(self, content: str, category: Category = Category.INFO) -> None:
        """
        Put a log message in the queue.

        Args:
            content (str): Content of the log message.
            category (Category): Category of the log message.
        """
        pass

    @abstractmethod
    def info(self, content: str) -> None:
        """
        Log an informational message.

        Args:
            content (str): Content of the log message.
        """
        pass

    @abstractmethod
    def error(self, content: str) -> None:
        """
        Log an error message.

        Args:
            content (str): Content of the log message.
        """
        pass

    @abstractmethod
    def warning(self, content: str) -> None:
        """
        Log a warning message.

        Args:
            content (str): Content of the log message.
        """
        pass

    @abstractmethod
    def debug(self, content: str) -> None:
        """
        Log a debug message.

        Args:
            content (str): Content of the log message.
        """
        pass

debug(content) abstractmethod

Log a debug message.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
67
68
69
70
71
72
73
74
75
@abstractmethod
def debug(self, content: str) -> None:
    """
    Log a debug message.

    Args:
        content (str): Content of the log message.
    """
    pass

error(content) abstractmethod

Log an error message.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
47
48
49
50
51
52
53
54
55
@abstractmethod
def error(self, content: str) -> None:
    """
    Log an error message.

    Args:
        content (str): Content of the log message.
    """
    pass

get_unique_tag(tag) abstractmethod

Generate a unique tag for the logger instance.

Parameters:

Name Type Description Default
tag str

The tag to identify the logger instance.

 required

Returns:

Name Type Description
str

A unique tag for the logger instance.
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
14
15
16
17
18
19
20
21
22
23
24
@abstractmethod
def get_unique_tag(self, tag: str):
    """
    Generate a unique tag for the logger instance.

    Args:
        tag (str): The tag to identify the logger instance.

    Returns:
        str: A unique tag for the logger instance.
    """

info(content) abstractmethod

Log an informational message.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
37
38
39
40
41
42
43
44
45
@abstractmethod
def info(self, content: str) -> None:
    """
    Log an informational message.

    Args:
        content (str): Content of the log message.
    """
    pass

log(content, category=Category.INFO) abstractmethod

Put a log message in the queue.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
category Category

Category of the log message.

 INFO
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
26
27
28
29
30
31
32
33
34
35
@abstractmethod
def log(self, content: str, category: Category = Category.INFO) -> None:
    """
    Put a log message in the queue.

    Args:
        content (str): Content of the log message.
        category (Category): Category of the log message.
    """
    pass

warning(content) abstractmethod

Log a warning message.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
57
58
59
60
61
62
63
64
65
@abstractmethod
def warning(self, content: str) -> None:
    """
    Log a warning message.

    Args:
        content (str): Content of the log message.
    """
    pass

WriterABC

Bases: ABC

Abstract class to handle writing log messages to a file.

Source code in devices\raspberry_pi_5\src\log\abstracts.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class WriterABC(ABC):
    """
    Abstract class to handle writing log messages to a file.
    """

    @classmethod
    def _write(cls, file: TextIO, msg: Message) -> None:
        """
        Write a message to the log file.

        Args:
            file (TextIO): The file to write the message to.
            msg (Message): Message to log.
        """
        # Check if the file is open
        if not file:
            print(f"Log file is not open. Must open it first.")
            return

        # Check if the message is an instance of Message
        if not isinstance(msg, Message):
            cls._write(
                file, Message(
                    f"Invalid message type: {type(msg)}. Expected Message.",
                    Category.ERROR
                )
            )
            return

        # Write the message to the log file
        file.write(f"{msg}\n")

        # Ensure immediate write
        file.flush()

    @abstractmethod
    def _write_last_message(self) -> None:
        """
        Write the last message to the log file if available.

        Raises:
            RuntimeError: If the log file is not open.
        """
        pass

    @abstractmethod
    def _start(self) -> None:
        """
        Start the writer.

        Raises:
            RuntimeError: If the writer fails to start.
        """
        pass

    @abstractmethod
    def _stop(self) -> None:
        """
        Stop the writer.
        """
        pass

    @abstractmethod
    def run(self, file_path: str = Files.get_log_file_path()) -> None:
        """
        Main loop for the logger to write messages to the log file.

        Args:
            file_path (str): Path to the log file.
        """
        pass

run(file_path=Files.get_log_file_path()) abstractmethod

Main loop for the logger to write messages to the log file.

Parameters:

Name Type Description Default
file_path str

Path to the log file.

 get_log_file_path()
Source code in devices\raspberry_pi_5\src\log\abstracts.py 
140
141
142
143
144
145
146
147
148
@abstractmethod
def run(self, file_path: str = Files.get_log_file_path()) -> None:
    """
    Main loop for the logger to write messages to the log file.

    Args:
        file_path (str): Path to the log file.
    """
    pass

decorators

log_on_error()

Decorator to log errors in a method.

Returns:

Name Type Description
function

The decorated function that logs errors.
Source code in devices\raspberry_pi_5\src\log\decorators.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def log_on_error():
    """
    Decorator to log errors in a method.

    Returns:
        function: The decorated function that logs errors.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            # Check if the instance has a logger attribute
            is_instance(self, LoggerConsumerProtocol)

            try:
                return func(self, *args, **kwargs)

            except Exception as e:
                tb = traceback.extract_tb(e.__traceback__)
                if tb:
                    filename, lineno, _, _ = tb[-1]
                    self.logger.error(
                        f"Error in {func.__name__} at {filename}:{lineno}: {e}"
                    )
                else:
                    self.logger.error(f"Error in {func.__name__}: {e}")

        return wrapper

    return decorator

enums

Category

Bases: Enum

Enum to define the category of log message.

Source code in devices\raspberry_pi_5\src\log\enums.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
@unique
class Category(Enum):
    """
    Enum to define the category of log message.
    """

    INFO = 1
    WARNING = 2
    ERROR = 3
    DEBUG = 4

message

Message

Class to handle log messages.

Source code in devices\raspberry_pi_5\src\log\message.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class Message:
    """
    Class to handle log messages.
    """

    def __init__(
        self,
        content: str,
        category: Category = Category.INFO,
        tag: Optional[str] = None
    ):
        """
        Initialize the Message class.

        Args:
            content (str): Content of the log message.
            category (Category): Category of the log message.
            tag (Optional[str]): Optional tag for the log message.
        """
        self.content = content
        self.category = category
        self.tag = tag

        # Get the formatted time
        self.__formatted_time = dt.now().strftime('%H:%M:%S')

    def __str__(self):
        """
        String representation of the log message.

        Returns:
            str: The formatted log message.
        """
        if self.tag:
            return f"[{self.__formatted_time}] [{self.tag}] {self.category.name}: {self.content}"
        return f"[{self.__formatted_time}] {self.category.name}: {self.content}"

    def __repr__(self):
        """
        Representation of the log message.

        Returns:
            str: The formatted log message.
        """
        return f"Message(category={self.category}, tag={self.tag}, content={self.content})"

    @property
    def content(self) -> str:
        """
        Get the content of the log message.

        Returns:
            str: The content of the log message.
        """
        return self.__content

    @content.setter
    def content(self, content: str):
        """
        Set the content of the log message.

        Args:
            content (str): The new content for the log message.
        """
        is_instance(content, str)
        self.__content = content

    @property
    def category(self) -> Category:
        """
        Get the category of the log message.

        Returns:
            Category: The category of the log message.
        """
        return self.__category

    @category.setter
    def category(self, category: Category):
        """
        Set the category of the log message.

        Args:
            category (Category): The new category for the log message.
        """
        is_instance(category, Category)
        self.__category = category

    @property
    def tag(self) -> Optional[str]:
        """
        Get the tag of the log message.

        Returns:
            Optional[str]: The tag of the log message, or None if not set.
        """
        return self.__tag

    @tag.setter
    def tag(self, tag: Optional[str]):
        """
        Set the tag of the log message.

        Args:
            tag (Optional[str]): The new tag for the log message.
        """
        if tag:
            is_instance(tag, str)
        self.__tag = tag

category property writable

Get the category of the log message.

Returns:

Name Type Description
Category Category

The category of the log message.

content property writable

Get the content of the log message.

Returns:

Name Type Description
str str

The content of the log message.

tag property writable

Get the tag of the log message.

Returns:

Type Description
Optional[str]

Optional[str]: The tag of the log message, or None if not set.

__init__(content, category=Category.INFO, tag=None)

Initialize the Message class.

Parameters:

Name Type Description Default
content str

Content of the log message.

 required
category Category

Category of the log message.

 INFO
tag Optional[str]

Optional tag for the log message.

 None
Source code in devices\raspberry_pi_5\src\log\message.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(
    self,
    content: str,
    category: Category = Category.INFO,
    tag: Optional[str] = None
):
    """
    Initialize the Message class.

    Args:
        content (str): Content of the log message.
        category (Category): Category of the log message.
        tag (Optional[str]): Optional tag for the log message.
    """
    self.content = content
    self.category = category
    self.tag = tag

    # Get the formatted time
    self.__formatted_time = dt.now().strftime('%H:%M:%S')

__repr__()

Representation of the log message.

Returns:

Name Type Description
str

The formatted log message.
Source code in devices\raspberry_pi_5\src\log\message.py 
45
46
47
48
49
50
51
52
def __repr__(self):
    """
    Representation of the log message.

    Returns:
        str: The formatted log message.
    """
    return f"Message(category={self.category}, tag={self.tag}, content={self.content})"

__str__()

String representation of the log message.

Returns:

Name Type Description
str

The formatted log message.
Source code in devices\raspberry_pi_5\src\log\message.py 
34
35
36
37
38
39
40
41
42
43
def __str__(self):
    """
    String representation of the log message.

    Returns:
        str: The formatted log message.
    """
    if self.tag:
        return f"[{self.__formatted_time}] [{self.tag}] {self.category.name}: {self.content}"
    return f"[{self.__formatted_time}] {self.category.name}: {self.content}"

multiprocessing

writer_target(debug, messages_queue, stop_event)

Target function for a multiprocessing process that handles writing log messages.

Parameters:

Name Type Description Default
debug bool

Flag to indicate if the writer is in debug mode.

 required
messages_queue Queue

Queue to hold log messages.

 required
stop_event Event

Event to signal when the process should stop.

 required
Source code in devices\raspberry_pi_5\src\log\multiprocessing.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def writer_target(
    debug: bool,
    messages_queue: Queue,
    stop_event: EventCls
) -> None:
    """
    Target function for a multiprocessing process that handles writing log messages.

    Args:
        debug (bool): Flag to indicate if the writer is in debug mode.
        messages_queue (Queue): Queue to hold log messages.
        stop_event (EventCls): Event to signal when the process should stop.
    """
    print(
        "Initializing Writer in multiprocessing mode. Process ID: ",
        os.getpid()
    )

    # Initialize the writer
    writer = Writer(
        debug=debug,
        messages_queue=messages_queue,
        stop_event=stop_event
    )

    # Run the writer
    writer.run()

protocols

LoggerConsumerProtocol

Bases: Protocol

Protocol for classes that consume a logger instance.

Source code in devices\raspberry_pi_5\src\log\protocols.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
@runtime_checkable
class LoggerConsumerProtocol(Protocol):
    """
    Protocol for classes that consume a logger instance.
    """

    @property
    def logger(self) -> Logger:
        """
        Get the logger instance for the consumer.

        Returns:
            Logger: The logger instance.
        """
        pass

logger property

Get the logger instance for the consumer.

Returns:

Name Type Description
Logger Logger

The logger instance.

writer

Writer

Bases: WriterABC

Class to handle writing log messages to a file.

Source code in devices\raspberry_pi_5\src\log\writer.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class Writer(WriterABC):
    """
    Class to handle writing log messages to a file.
    """

    # Wait timeout for processing messages
    WAIT_TIMEOUT = 0.1

    def __init__(
        self,
        debug: bool,
        messages_queue: Queue,
        stop_event: EventCls
        ) -> None:
        """
        Initialize the Logger class.

        Args:
            debug (bool): Flag to indicate if the logger is in debug mode.
            messages_queue (Queue): Queue to hold log messages.
            stop_event (EventCls): Event to signal when the logger should stop.
        """
        # Initialize the debug flag
        self.__debug = debug

        # Initialize the messages queue and events
        self.__messages_queue = messages_queue
        self.__started_event = Event()
        self.__deleted_event = Event()
        self.__stop_event = stop_event

        # Initialize the reentrant lock
        self.__rlock = RLock()

        # Initialize the file path and file
        self.__file_path: str = ""
        self.__file: TextIO | None = None

    @final
    def _write_last_message(self) -> None:
        try:
            # Process any remaining messages in the queue
            msg = self.__messages_queue.get(timeout=self.WAIT_TIMEOUT)

        except Empty:
            return None

        # Check if the message is an instance of Message
        is_instance(msg, Message)

        # Write the message to the log file
        if not self.__file:
            raise RuntimeError("Log file is not open. Must open it first.")
        self._write(self.__file, msg)

    @final
    def _start(self) -> None:
        with self.__rlock:
            # Check if the stop event is set
            if self.__stop_event.is_set():
                raise RuntimeError("Stop event is set. Logger will not run.")

            # Check if the logger is already running
            if self.__started_event.is_set():
                raise RuntimeError(
                    "Logger is already running. Cannot start again."
                    )

            # Set the started event
            self.__started_event.set()

        # Log
        print("Writer initialized.")

    @final
    def _stop(self) -> None:
        # Check if there are any remaining messages in the queue
        while not self.__messages_queue.empty():
            self._write_last_message()

        with self.__rlock:
            # Clear the started event
            self.__started_event.clear()

            # Clear the deleted event
            self.__deleted_event.clear()

        # Write the stop message to the log file
        self._write(self.__file, Message("Writer stopped.", Category.DEBUG))

    @final
    @ignore_sigint
    def run(self, file_path: str = Files.get_log_file_path()) -> None:
        try:
            # Check the type of file_path
            is_instance(file_path, str)
            self.__file_path = file_path

            # Ensure the file exists
            Files.ensure_file_exists(self.__file_path)

            # Start the writer
            self._start()

        except Exception as e:
            print(f"An error occurred while starting the Writer: {e}")
            raise e

        try:
            # Open the log file in append mode
            print(f"Opening log file at {self.__file_path}...")
            with open(self.__file_path, 'a') as file:
                # Set the file
                self.__file = file

                # Process messages from the queue
                while not self.__stop_event.is_set() and not self.__deleted_event.is_set():
                    # Write the last message if available
                    self._write_last_message()

                # Stop the writer
                self._stop()

        except Exception as e:
            # Stop the writer in case of an exception
            self._stop()

            # Log any exceptions that occur
            print(f"An error occurred while running the Writer: {e}")
            raise e

    def __del__(self):
        """
        Destructor to clean up resources when the photographer is no longer needed.
        """
        self.__deleted_event.set()

        # Log
        print("Writer instance is being deleted. Resources will be cleaned up.")

__del__()

Destructor to clean up resources when the photographer is no longer needed.

Source code in devices\raspberry_pi_5\src\log\writer.py 
145
146
147
148
149
150
151
152
def __del__(self):
    """
    Destructor to clean up resources when the photographer is no longer needed.
    """
    self.__deleted_event.set()

    # Log
    print("Writer instance is being deleted. Resources will be cleaned up.")

__init__(debug, messages_queue, stop_event)

Initialize the Logger class.

Parameters:

Name Type Description Default
debug bool

Flag to indicate if the logger is in debug mode.

 required
messages_queue Queue

Queue to hold log messages.

 required
stop_event Event

Event to signal when the logger should stop.

 required
Source code in devices\raspberry_pi_5\src\log\writer.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    debug: bool,
    messages_queue: Queue,
    stop_event: EventCls
    ) -> None:
    """
    Initialize the Logger class.

    Args:
        debug (bool): Flag to indicate if the logger is in debug mode.
        messages_queue (Queue): Queue to hold log messages.
        stop_event (EventCls): Event to signal when the logger should stop.
    """
    # Initialize the debug flag
    self.__debug = debug

    # Initialize the messages queue and events
    self.__messages_queue = messages_queue
    self.__started_event = Event()
    self.__deleted_event = Event()
    self.__stop_event = stop_event

    # Initialize the reentrant lock
    self.__rlock = RLock()

    # Initialize the file path and file
    self.__file_path: str = ""
    self.__file: TextIO | None = None