Module livekit.agents.cli.log

Formatter instances are used to convert a LogRecord to text.

Formatters need to know how a LogRecord is constructed. They are responsible for converting a LogRecord to (usually) a string which can be interpreted by either a human or an external system. The base Formatter allows a formatting string to be specified. If none is supplied, the style-dependent default value, "%(message)s", "{message}", or "${message}", is used.

The Formatter can be initialized with a format string which makes use of knowledge of the LogRecord attributes - e.g. the default value mentioned above makes use of the fact that the user's message and arguments are pre- formatted into a LogRecord's message attribute. Currently, the useful attributes in a LogRecord are described by:

%(name)s Name of the logger (logging channel) %(levelno)s Numeric logging level for the message (DEBUG, INFO, WARNING, ERROR, CRITICAL) %(levelname)s Text logging level for the message ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") %(pathname)s Full pathname of the source file where the logging call was issued (if available) %(filename)s Filename portion of pathname %(module)s Module (name portion of filename) %(lineno)d Source line number where the logging call was issued (if available) %(funcName)s Function name %(created)f Time when the LogRecord was created (time.time() return value) %(asctime)s Textual time when the LogRecord was created %(msecs)d Millisecond portion of the creation time %(relativeCreated)d Time in milliseconds when the LogRecord was created, relative to the time the logging module was loaded (typically at application startup time) %(thread)d Thread ID (if available) %(threadName)s Thread name (if available) %(taskName)s Task name (if available) %(process)d Process ID (if available) %(message)s The result of record.getMessage(), computed just as the record is emitted

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of '%', '{' or '$' to specify that you want to use one of %-formatting, :meth:str.format ({}) formatting or :class:string.Template formatting in your format string.

class ColoredFormatter(logging.Formatter):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._esc_codes = {
            "esc_reset": self._esc(0),
            "esc_red": self._esc(31),
            "esc_green": self._esc(32),
            "esc_yellow": self._esc(33),
            "esc_blue": self._esc(34),
            "esc_purple": self._esc(35),
            "esc_cyan": self._esc(36),
            "esc_gray": self._esc(90),
            "esc_bold_red": self._esc(1, 31),
        }

        self._level_colors = {
            "DEBUG": self._esc_codes["esc_cyan"],
            "INFO": self._esc_codes["esc_green"],
            "WARNING": self._esc_codes["esc_yellow"],
            "ERROR": self._esc_codes["esc_red"],
            "CRITICAL": self._esc_codes["esc_bold_red"],
            "DEV": self._esc_codes["esc_purple"],
        }

        self._required_fields = _parse_style(self)

    @classmethod
    def _esc(cls, *codes: int) -> str:
        return "\033[" + ";".join(str(code) for code in codes) + "m"

    def formatMessage(self, record: logging.LogRecord) -> str:
        """Formats a log record with colors"""

        extra: Dict[Any, Any] = {}
        _merge_record_extra(record, extra)

        args = {}
        for field in self._required_fields:
            args[field] = record.__dict__.get(field)

        args["esc_levelcolor"] = self._level_colors.get(record.levelname, "")
        args["extra"] = ""
        args.update(self._esc_codes)

        if extra:
            args["extra"] = json.dumps(
                extra, cls=JsonFormatter.JsonEncoder, ensure_ascii=True
            )

        for field in self._required_fields:
            if field in extra:
                del extra[field]

        msg = self._style._fmt % args
        return msg + self._esc_codes["esc_reset"]

Formatter instances are used to convert a LogRecord to text.

Formatters need to know how a LogRecord is constructed. They are responsible for converting a LogRecord to (usually) a string which can be interpreted by either a human or an external system. The base Formatter allows a formatting string to be specified. If none is supplied, the style-dependent default value, "%(message)s", "{message}", or "${message}", is used.

The Formatter can be initialized with a format string which makes use of knowledge of the LogRecord attributes - e.g. the default value mentioned above makes use of the fact that the user's message and arguments are pre- formatted into a LogRecord's message attribute. Currently, the useful attributes in a LogRecord are described by:

%(name)s Name of the logger (logging channel) %(levelno)s Numeric logging level for the message (DEBUG, INFO, WARNING, ERROR, CRITICAL) %(levelname)s Text logging level for the message ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") %(pathname)s Full pathname of the source file where the logging call was issued (if available) %(filename)s Filename portion of pathname %(module)s Module (name portion of filename) %(lineno)d Source line number where the logging call was issued (if available) %(funcName)s Function name %(created)f Time when the LogRecord was created (time.time() return value) %(asctime)s Textual time when the LogRecord was created %(msecs)d Millisecond portion of the creation time %(relativeCreated)d Time in milliseconds when the LogRecord was created, relative to the time the logging module was loaded (typically at application startup time) %(thread)d Thread ID (if available) %(threadName)s Thread name (if available) %(taskName)s Task name (if available) %(process)d Process ID (if available) %(message)s The result of record.getMessage(), computed just as the record is emitted

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of '%', '{' or '$' to specify that you want to use one of %-formatting, :meth:str.format ({}) formatting or :class:string.Template formatting in your format string.

class JsonFormatter(logging.Formatter):
    class JsonEncoder(json.JSONEncoder):
        def default(self, o: Any):
            if isinstance(o, (date, datetime, time)):
                return o.isoformat()
            elif istraceback(o):
                return "".join(traceback.format_tb(o)).strip()
            elif type(o) is Exception or isinstance(o, Exception) or type(o) is type:
                return str(o)

            # extra values are formatted as str() if the encoder raises TypeError
            try:
                return super().default(o)
            except TypeError:
                try:
                    return str(o)
                except Exception:
                    return None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._required_fields = _parse_style(self)

    def format(self, record: logging.LogRecord) -> str:
        """Formats a log record and serializes to json"""
        message_dict: Dict[str, Any] = {}
        message_dict["level"] = record.levelname
        message_dict["name"] = record.name

        if isinstance(record.msg, dict):
            message_dict = record.msg
            record.message = ""
        else:
            record.message = record.getMessage()

        if "asctime" in self._required_fields:
            record.asctime = self.formatTime(record, self.datefmt)

        if record.exc_info and not message_dict.get("exc_info"):
            message_dict["exc_info"] = self.formatException(record.exc_info)
        if not message_dict.get("exc_info") and record.exc_text:
            message_dict["exc_info"] = record.exc_text
        if record.stack_info and not message_dict.get("stack_info"):
            message_dict["stack_info"] = self.formatStack(record.stack_info)

        log_record: Dict[str, Any] = OrderedDict()

        for field in self._required_fields:
            log_record[field] = record.__dict__.get(field)

        log_record.update(message_dict)
        _merge_record_extra(record, log_record)

        log_record["timestamp"] = datetime.fromtimestamp(
            record.created, tz=timezone.utc
        )

        return json.dumps(log_record, cls=JsonFormatter.JsonEncoder, ensure_ascii=True)