xref: /aosp_15_r20/external/pigweed/pw_console/py/pw_console/python_logging.py (revision 61c4878ac05f98d0ceed94b57d316916de578985) 
1# Copyright 2021 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Python logging helper fuctions."""
15
16import copy
17from datetime import datetime
18import json
19import logging
20import tempfile
21from typing import Any, Iterable, Iterator
22
23
24def all_loggers() -> Iterator[logging.Logger]:
25    """Iterates over all loggers known to Python logging."""
26    manager = logging.getLogger().manager  # type: ignore[attr-defined]
27
28    for logger_name in manager.loggerDict:  # pylint: disable=no-member
29        yield logging.getLogger(logger_name)
30
31
32def create_temp_log_file(
33    prefix: str | None = None, add_time: bool = True
34) -> str:
35    """Create a unique tempfile for saving logs.
36
37    Example format: /tmp/pw_console_2021-05-04_151807_8hem6iyq
38    """
39    if not prefix:
40        prefix = str(__package__)
41
42    # Grab the current system timestamp as a string.
43    isotime = datetime.now().isoformat(sep="_", timespec="seconds")
44    # Timestamp string should not have colons in it.
45    isotime = isotime.replace(":", "")
46
47    if add_time:
48        prefix += f"_{isotime}"
49
50    log_file_name = None
51    with tempfile.NamedTemporaryFile(
52        prefix=f"{prefix}_", delete=False
53    ) as log_file:
54        log_file_name = log_file.name
55
56    return log_file_name
57
58
59def set_logging_last_resort_file_handler(
60    file_name: str | None = None,
61) -> None:
62    log_file = file_name if file_name else create_temp_log_file()
63    logging.lastResort = logging.FileHandler(log_file)
64
65
66def disable_stdout_handlers(logger: logging.Logger) -> None:
67    """Remove all stdout and stdout & stderr logger handlers."""
68    for handler in copy.copy(logger.handlers):
69        # Must use type() check here since this returns True:
70        #   isinstance(logging.FileHandler, logging.StreamHandler)
71        # pylint: disable=unidiomatic-typecheck
72        if type(handler) == logging.StreamHandler:
73            logger.removeHandler(handler)
74        # pylint: enable=unidiomatic-typecheck
75
76
77def setup_python_logging(
78    last_resort_filename: str | None = None,
79    loggers_with_no_propagation: Iterable[logging.Logger] | None = None,
80) -> None:
81    """Disable log handlers for full screen prompt_toolkit applications."""
82    if not loggers_with_no_propagation:
83        loggers_with_no_propagation = []
84    disable_stdout_handlers(logging.getLogger())
85
86    if logging.lastResort is not None:
87        set_logging_last_resort_file_handler(last_resort_filename)
88
89    for logger in list(all_loggers()):
90        # Prevent stdout handlers from corrupting the prompt_toolkit UI.
91        disable_stdout_handlers(logger)
92        if logger in loggers_with_no_propagation:
93            continue
94        # Make sure all known loggers propagate to the root logger.
95        logger.propagate = True
96
97    # Prevent these loggers from propagating to the root logger.
98    hidden_host_loggers = [
99        "blib2to3.pgen2.driver",  # spammy and unhelpful
100        "pw_console",
101        "pw_console.plugins",
102        # prompt_toolkit triggered debug log messages
103        "prompt_toolkit",
104        "prompt_toolkit.buffer",
105        "parso.python.diff",
106        "parso.cache",
107        "pw_console.serial_debug_logger",
108        "websockets.server",
109    ]
110    for logger_name in hidden_host_loggers:
111        logging.getLogger(logger_name).propagate = False
112
113    # Set asyncio log level to WARNING
114    logging.getLogger("asyncio").setLevel(logging.WARNING)
115
116    # Always set DEBUG level for serial debug.
117    logging.getLogger("pw_console.serial_debug_logger").setLevel(logging.DEBUG)
118
119
120def log_record_to_dict(record: logging.LogRecord) -> dict[str, Any]:
121    """Convert a log record into a dict for use with json formatting."""
122    log_dict: dict[str, Any] = {}
123    log_dict["message"] = record.getMessage()
124    log_dict["levelno"] = record.levelno
125    log_dict["levelname"] = record.levelname
126    log_dict["args"] = ''
127    if record.args:
128        log_dict["args"] = [str(arg) for arg in record.args]
129    log_dict["time"] = str(record.created)
130    log_dict["time_string"] = datetime.fromtimestamp(record.created).isoformat(
131        timespec="seconds"
132    )
133
134    lineno = record.lineno
135    file_name = str(record.filename)
136    log_dict['py_file'] = f'{file_name}:{lineno}'
137    log_dict['py_logger'] = str(record.name)
138
139    if hasattr(record, "extra_metadata_fields") and (
140        record.extra_metadata_fields  # type: ignore
141    ):
142        fields = record.extra_metadata_fields  # type: ignore
143        log_dict["fields"] = {}
144        for key, value in fields.items():
145            if key == "msg":
146                log_dict["message"] = value
147                continue
148
149            log_dict["fields"][key] = str(value)
150
151    return log_dict
152
153
154def log_record_to_json(record: logging.LogRecord) -> str:
155    return json.dumps(log_record_to_dict(record))
156
157
158class JsonLogFormatter(logging.Formatter):
159    """Json Python logging Formatter
160
161    Use this formatter to log pw_console messages to a file in json
162    format. Column values normally shown in table view will be populated in the
163    'fields' key.
164
165    Example log entry:
166
167    .. code-block:: json
168
169       {
170         "message": "System init",
171         "levelno": 20,
172         "levelname": "INF",
173         "args": [
174           "0:00",
175           "pw_system ",
176           "System init"
177         ],
178         "time": "1692302986.4729185",
179         "time_string": "2023-08-17T13:09:46",
180         "fields": {
181           "module": "pw_system",
182           "file": "pw_system/init.cc",
183           "timestamp": "0:00"
184         },
185         "py_file": "script.py:1234",
186         "py_logger": "root"
187       }
188
189    Example usage:
190
191    .. code-block:: python
192
193       import logging
194       import pw_console.python_logging
195
196       _DEVICE_LOG = logging.getLogger('rpc_device')
197
198       json_filehandler = logging.FileHandler('logs.json', encoding='utf-8')
199       json_filehandler.setLevel(logging.DEBUG)
200       json_filehandler.setFormatter(
201           pw_console.python_logging.JsonLogFormatter())
202       _DEVICE_LOG.addHandler(json_filehandler)
203
204    """
205
206    def __init__(self, *args, **kwargs):
207        super().__init__(*args, **kwargs)
208
209    def format(self, record: logging.LogRecord) -> str:
210        return log_record_to_json(record)
211