/
/
/
1"""
2Logging utilities.
3
4A lot in this file has been copied from Home Assistant:
5https://github.com/home-assistant/core/blob/e5ccd85e7e26c167d0b73669a88bc3a7614dd456/homeassistant/util/logging.py#L78
6
7All rights reserved.
8"""
9
10from __future__ import annotations
11
12import inspect
13import logging
14import logging.handlers
15import queue
16import traceback
17from collections.abc import Callable, Coroutine
18from functools import partial, wraps
19from typing import Any, cast, overload
20
21from music_assistant.helpers.diagnostics import DiagnosticsLogHandler
22
23
24class LoggingQueueHandler(logging.handlers.QueueHandler):
25 """Process the log in another thread."""
26
27 listener: logging.handlers.QueueListener | None = None
28
29 def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
30 """
31 Prepare a record for queuing.
32
33 This is added as a workaround for https://bugs.python.org/issue46755
34 """
35 record = super().prepare(record)
36 record.stack_info = None
37 return record
38
39 def handle(self, record: logging.LogRecord) -> Any:
40 """
41 Conditionally emit the specified logging record.
42
43 Depending on which filters have been added to the handler, push the new
44 records onto the backing Queue.
45
46 The default python logger Handler acquires a lock
47 in the parent class which we do not need as
48 SimpleQueue is already thread safe.
49
50 See https://bugs.python.org/issue24645
51 """
52 return_value = self.filter(record)
53 if return_value:
54 self.emit(record)
55 return return_value
56
57 def close(self) -> None:
58 """
59 Tidy up any resources used by the handler.
60
61 This adds shutdown of the QueueListener
62 """
63 super().close()
64 if not self.listener:
65 return
66 self.listener.stop()
67 self.listener = None
68
69
70def activate_log_queue_handler() -> None:
71 """
72 Migrate the existing log handlers to use the queue.
73
74 This allows us to avoid blocking I/O and formatting messages
75 in the event loop as log messages are written in another thread.
76 """
77 simple_queue: queue.SimpleQueue[logging.Handler] = queue.SimpleQueue()
78 queue_handler = LoggingQueueHandler(simple_queue)
79 logging.root.addHandler(queue_handler)
80
81 migrated_handlers: list[logging.Handler] = []
82 for handler in logging.root.handlers[:]:
83 if handler is queue_handler:
84 continue
85 # the diagnostics capture handler stays attached directly to the root logger:
86 # the queue handler copies records and strips exc_info, which would break
87 # its exception aggregation (and it never blocks anyway)
88 if isinstance(handler, DiagnosticsLogHandler):
89 continue
90 logging.root.removeHandler(handler)
91 migrated_handlers.append(handler)
92
93 listener = logging.handlers.QueueListener(simple_queue, *migrated_handlers)
94 queue_handler.listener = listener
95
96 listener.start()
97
98
99def log_exception(format_err: Callable[..., Any], *args: Any) -> None:
100 """Log an exception with additional context."""
101 module = inspect.getmodule(inspect.stack(context=0)[1].frame)
102 if module is not None: # noqa: SIM108
103 module_name = module.__name__
104 else:
105 # If Python is unable to access the sources files, the call stack frame
106 # will be missing information, so let's guard.
107 # https://github.com/home-assistant/core/issues/24982
108 module_name = __name__
109
110 # Do not print the wrapper in the traceback
111 frames = len(inspect.trace()) - 1
112 exc_msg = traceback.format_exc(-frames)
113 friendly_msg = format_err(*args)
114 logging.getLogger(module_name).error("%s\n%s", friendly_msg, exc_msg)
115
116
117@overload
118def catch_log_exception(
119 func: Callable[..., Coroutine[Any, Any, Any]], format_err: Callable[..., Any]
120) -> Callable[..., Coroutine[Any, Any, None]]: ...
121
122
123@overload
124def catch_log_exception(
125 func: Callable[..., Any], format_err: Callable[..., Any]
126) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]: ...
127
128
129def catch_log_exception(
130 func: Callable[..., Any], format_err: Callable[..., Any]
131) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]:
132 """
133 Decorate a function func to catch and log exceptions.
134
135 If func is a coroutine function, a coroutine function will be returned.
136 If func is a callback, a callback will be returned.
137 """
138 # Check for partials to properly determine if coroutine function
139 check_func = func
140 while isinstance(check_func, partial):
141 check_func = check_func.func
142
143 wrapper_func: Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]
144 if inspect.iscoroutinefunction(check_func):
145 async_func = cast("Callable[..., Coroutine[Any, Any, None]]", func)
146
147 @wraps(async_func)
148 async def async_wrapper(*args: Any) -> None:
149 """Catch and log exception."""
150 try:
151 await async_func(*args)
152 except Exception:
153 log_exception(format_err, *args)
154
155 wrapper_func = async_wrapper
156
157 else:
158
159 @wraps(func)
160 def wrapper(*args: Any) -> None:
161 """Catch and log exception."""
162 try:
163 func(*args)
164 except Exception:
165 log_exception(format_err, *args)
166
167 wrapper_func = wrapper
168 return wrapper_func
169