jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-20 03:37:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Shift ESPHome log parsing to the library (#146349)

Browse Source
This commit is contained in:
J. Nick Koston 2025-06-10 08:30:19 -05:00 committed by GitHub
parent 6c5f7eabff
commit 7da1671b06
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 10 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
homeassistant/components/esphome/manager.py
View File

@ -5,7 +5,6 @@ from __future__ import annotations
import asyncio import asyncio
from functools import partial from functools import partial
import logging import logging
import re
from typing import TYPE_CHECKING, Any, NamedTuple from typing import TYPE_CHECKING, Any, NamedTuple
from aioesphomeapi import ( from aioesphomeapi import (
@ -23,6 +22,7 @@ from aioesphomeapi import (
RequiresEncryptionAPIError, RequiresEncryptionAPIError,
UserService, UserService,
UserServiceArgType, UserServiceArgType,
parse_log_message,
) )
from awesomeversion import AwesomeVersion from awesomeversion import AwesomeVersion
import voluptuous as vol import voluptuous as vol
@ -110,11 +110,6 @@ LOGGER_TO_LOG_LEVEL = {
logging.ERROR: LogLevel.LOG_LEVEL_ERROR, logging.ERROR: LogLevel.LOG_LEVEL_ERROR,
logging.CRITICAL: LogLevel.LOG_LEVEL_ERROR, logging.CRITICAL: LogLevel.LOG_LEVEL_ERROR,
} }
# 7-bit and 8-bit C1 ANSI sequences
# https://stackoverflow.com/questions/14693701/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python
ANSI_ESCAPE_78BIT = re.compile(
rb"(?:\x1B[@-Z\\-_]|[\x80-\x9A\x9C-\x9F]|(?:\x1B\[|\x9B)[0-?]*[ -/]*[@-~])"
)
@callback @callback
@ -387,13 +382,15 @@ class ESPHomeManager:
def _async_on_log(self, msg: SubscribeLogsResponse) -> None: def _async_on_log(self, msg: SubscribeLogsResponse) -> None:
"""Handle a log message from the API.""" """Handle a log message from the API."""
log: bytes = msg.message for line in parse_log_message(
_LOGGER.log( msg.message.decode("utf-8", "backslashreplace"), "", strip_ansi_escapes=True
LOG_LEVEL_TO_LOGGER.get(msg.level, logging.DEBUG), ):
"%s: %s", _LOGGER.log(
self.entry.title, LOG_LEVEL_TO_LOGGER.get(msg.level, logging.DEBUG),
ANSI_ESCAPE_78BIT.sub(b"", log).decode("utf-8", "backslashreplace"), "%s: %s",
) self.entry.title,
line,
)
@callback @callback
def _async_get_equivalent_log_level(self) -> LogLevel: def _async_get_equivalent_log_level(self) -> LogLevel: