john/room_control
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rich logging config in yaml file in package data

Browse Source
This commit is contained in:
John Lancaster
2024-05-07 23:38:42 -05:00
parent 7b67d062a0
commit 463a68a27a
8 changed files with 170 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/room_control/.rich_logging.yaml Normal file
View File

@@ -0,0 +1,26 @@
version: 1
disable_existing_loggers: false
formatters:
rich:
style: "{"
format: "[room]{room}[/] {message}"
# format: "{message}"
datefmt: '%H:%M:%S.%f'
rich_component:
style: "{"
format: "[room]{room}[/] [component]{component}[/] {message}"
# format: "{message}"
datefmt: '%H:%M:%S.%f'
handlers:
rich:
formatter: rich
'()': 'rich.logging.RichHandler'
markup: True
show_path: false
omit_repeated_times: false
rich_component:
formatter: rich_component
'()': 'rich.logging.RichHandler'
markup: True
show_path: false
omit_repeated_times: false

6
src/room_control/button.py
View File

@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, List
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from .console import setup_component_logging
from . import console
from .model import ButtonConfig
if TYPE_CHECKING:
@@ -20,9 +20,9 @@ class Button(Mqtt):
logger: Logger
async def initialize(self):
self.config = ButtonConfig(**self.args)
self.logger = setup_component_logging(self)
self.app: 'RoomController' = await self.get_app(self.args['app'])
self.logger = console.load_rich_config(self.app.name, type(self).__name__)
self.config = ButtonConfig(**self.args)
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
self.button = self.config.button

156
src/room_control/console.py
View File

@@ -1,9 +1,12 @@
import json
import logging
import logging.config
import re
from abc import ABC
from dataclasses import asdict
from importlib.resources import files
import yaml
from rich.console import Console
from rich.highlighter import RegexHighlighter
from rich.theme import Theme
@@ -42,6 +45,40 @@ class RCHighlighter(RegexHighlighter):
]
def load_rich_config(
room: str = None, component: str = None, level: str = 'INFO'
) -> logging.LoggerAdapter:
logger_name = f'Appdaemon.{room}'
if component is not None:
logger_name += f'.{component}'
with files('room_control').joinpath('.rich_logging.yaml').open('r') as f:
RICH_CFG = yaml.safe_load(f)
RICH_CFG['handlers']['rich']['console'] = console
RICH_CFG['handlers']['rich']['highlighter'] = RCHighlighter()
RICH_CFG['handlers']['rich_component']['console'] = console
RICH_CFG['handlers']['rich_component']['highlighter'] = RCHighlighter()
RICH_CFG['loggers'] = {
logger_name: {
'handlers': ['rich' if component is None else 'rich_component'],
'propagate': False,
'level': level,
}
}
extra = {'room': room}
if component is not None:
extra['component'] = component
logging.config.dictConfig(RICH_CFG)
logger = logging.getLogger(logger_name)
adapter = logging.LoggerAdapter(logger, extra)
return adapter
RICH_HANDLER_CFG = {
'()': 'rich.logging.RichHandler',
'markup': True,
@@ -61,19 +98,19 @@ class ContextSettingFilter(logging.Filter, ABC):
return record
class RoomFilter(logging.Filter):
"""Used to filter out messages that have a component field because they will have already been printed by their respective logger."""
def filter(self, record: logging.LogRecord) -> bool:
return getattr(record, 'component', None) is None
# @dataclass
# class RoomControllerFilter(ContextSettingFilter):
# room: str
# component: Optional[str] = None
class RoomFilter(logging.Filter):
"""Used to filter out messages that have a component field because they will have already been printed by their respective logger."""
def filter(self, record: logging.LogRecord) -> bool:
return not hasattr(record, 'component')
# class RoomControllerFormatter(logging.Formatter):
# def format(self, record: logging.LogRecord):
# return super().format(record)
@@ -87,6 +124,11 @@ class UnMarkupFilter(logging.Filter):
return record
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
return json.dumps(record.__dict__)
def room_logging_config(name: str):
return {
'version': 1,
@@ -106,6 +148,7 @@ def room_logging_config(name: str):
'format': '{asctime}.{msecs:03.0f} {levelname:8} {name}: {message}',
'datefmt': '%Y-%m-%d %H:%M:%S',
},
'json': {'()': 'room_control.console.JSONFormatter'},
},
'handlers': {
'rich_room': {
@@ -119,7 +162,14 @@ def room_logging_config(name: str):
'class': 'logging.handlers.RotatingFileHandler',
# 'class': 'logging.FileHandler',
'filename': f'/logs/{name}.log',
'mode': 'w',
'maxBytes': 1000000,
'backupCount': 3,
},
'json': {
'filters': ['unmarkup'],
'formatter': 'json',
'filename': f'/logs/{name}.jsonl',
'class': 'logging.handlers.RotatingFileHandler',
'maxBytes': 1000000,
'backupCount': 3,
},
@@ -128,54 +178,60 @@ def room_logging_config(name: str):
f'AppDaemon.{name}': {
'level': 'INFO',
'propagate': False,
'handlers': ['rich_room', 'file'],
'handlers': [
'rich_room',
'file',
'json',
],
},
},
}
def component_logging_config(parent_room: str, component: str):
logger_name = f'AppDaemon.{parent_room}.{component}'
# def component_logging_config(parent_room: str, component: str):
# logger_name = f'AppDaemon.{parent_room}.{component}'
LOG_CFG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'rich_component': {
'style': '{',
'format': '[room]{room}[/] [component]{component}[/] {message}',
'datefmt': '%H:%M:%S.%f',
},
},
'handlers': {
'rich_component': {
'formatter': 'rich_component',
**RICH_HANDLER_CFG,
},
},
'loggers': {
logger_name: {
# 'level': 'INFO',
'propagate': True,
'handlers': ['rich_component'],
}
},
}
return LOG_CFG
# cfg = load_rich_config()
# LOG_CFG = {
# 'version': 1,
# 'disable_existing_loggers': False,
# 'formatters': {
# 'rich_component': {
# 'style': '{',
# 'format': '[room]{room}[/] [component]{component}[/] {message}',
# 'datefmt': '%H:%M:%S.%f',
# },
# },
# 'handlers': {
# 'rich_component': {
# 'formatter': 'rich_component',
# **RICH_HANDLER_CFG,
# },
# },
# 'loggers': {
# logger_name: {
# # 'level': 'INFO',
# 'propagate': True,
# 'handlers': ['rich_component'],
# }
# },
# }
# return LOG_CFG
def setup_component_logging(self) -> logging.Logger:
"""Creates a logger for a subcomponent with a RichHandler"""
component = type(self).__name__
parent = self.args['app']
cfg_dict = component_logging_config(parent_room=parent, component=component)
logger_name = next(iter(cfg_dict['loggers']))
# def setup_component_logging(self) -> logging.Logger:
# """Creates a logger for a subcomponent with a RichHandler"""
# component = type(self).__name__
# parent = self.args['app']
# cfg_dict = component_logging_config(parent_room=parent, component=component)
# logger_name = next(iter(cfg_dict['loggers']))
try:
logging.config.dictConfig(cfg_dict)
except Exception:
console.print_exception()
else:
logger = logging.getLogger(logger_name)
logger = logging.LoggerAdapter(logger, {'room': parent, 'component': component})
return logger
# try:
# logging.config.dictConfig(cfg_dict)
# except Exception:
# console.print_exception()
# else:
# logger = logging.getLogger(logger_name)
# logger = logging.LoggerAdapter(logger, {'room': parent, 'component': component})
# return logger

11
src/room_control/door.py
View File

@@ -1,18 +1,21 @@
from logging import Logger
from typing import TYPE_CHECKING
from appdaemon.plugins.hass.hassapi import Hass
from room_control import RoomController
from . import console
from .console import setup_component_logging
if TYPE_CHECKING:
from room_control import RoomController
class Door(Hass):
app: 'RoomController'
logger: Logger
async def initialize(self):
self.logger = setup_component_logging(self)
self.app: RoomController = await self.get_app(self.args['app'])
self.app: 'RoomController' = await self.get_app(self.args['app'])
self.logger = console.load_rich_config(room=self.app.name, component=type(self).__name__)
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
await self.listen_state(

16
src/room_control/model.py
View File

@@ -47,9 +47,7 @@ class ApplyKwargs(BaseModel):
class ControllerStateConfig(BaseModel):
time: Optional[str | datetime] = None
elevation: Optional[float] = None
direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = (
None
)
direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = None
off_duration: Optional[OffDuration] = None
scene: dict[str, State] | str
@@ -57,13 +55,9 @@ class ControllerStateConfig(BaseModel):
def check_args(cls, values):
time, elevation = values.get('time'), values.get('elevation')
if time is not None and elevation is not None:
raise PydanticCustomError(
'bad_time_spec', 'Only one of time or elevation can be set.'
)
raise PydanticCustomError('bad_time_spec', 'Only one of time or elevation can be set.')
elif elevation is not None and 'direction' not in values:
raise PydanticCustomError(
'no_sun_dir', 'Needs sun direction with elevation'
)
raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation')
return values
def to_apply_kwargs(self, **kwargs):
@@ -85,9 +79,7 @@ class RoomControllerConfig(BaseModel):
if app_cfg['class'] == 'RoomController':
return cls.model_validate(app_cfg)
def __rich_console__(
self, console: Console, options: ConsoleOptions
) -> RenderResult:
def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult:
table = Table(
Column('Time', width=15),
Column('Scene'),

44
src/room_control/motion.py
View File

@@ -5,9 +5,9 @@ from typing import TYPE_CHECKING, Literal, Optional
from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass
from pydantic import BaseModel, TypeAdapter
from pydantic import BaseModel, ValidationError
from .console import setup_component_logging
from . import console
if TYPE_CHECKING:
from room_control import RoomController
@@ -52,9 +52,8 @@ class Motion(Hass):
return self.sensor_state != self.ref_entity_state
def initialize(self):
self.logger = setup_component_logging(self)
self.app: 'RoomController' = self.get_app(self.args['app'])
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
self.logger = console.load_rich_config(self.app.name, type(self).__name__)
assert self.entity_exists(self.args['sensor'])
assert self.entity_exists(self.args['ref_entity'])
@@ -66,14 +65,11 @@ class Motion(Hass):
if self.state_mismatch:
self.log(
f'Sensor is {self.sensor_state} '
f'and light is {self.ref_entity_state}',
f'Sensor is {self.sensor_state} ' f'and light is {self.ref_entity_state}',
level='WARNING',
)
if self.sensor_state:
self.app.activate(
kwargs={'cause': f'Syncing state with {self.sensor.entity_id}'}
)
self.app.activate(kwargs={'cause': f'Syncing state with {self.sensor.entity_id}'})
# don't need to await these because they'll already get turned into a task by the utils.sync_wrapper decorator
self.listen_state(
@@ -83,16 +79,20 @@ class Motion(Hass):
)
self.listen_state(**base_kwargs, new='off', callback=self.callback_light_off)
if callbacks := self.callbacks():
for handle, entry in callbacks.items():
self.log(
f'Handle [yellow]{handle[:4]}[/]: {entry.function}', level='DEBUG'
)
for handle, cb in self.callbacks():
self.log(f'Handle [yellow]{handle[:4]}[/]: {cb.function}', level='DEBUG')
self.log(f'Initialized [bold green]{type(self).__name__}[/]')
def callbacks(self):
"""Returns a dictionary of validated CallbackEntry objects that are associated with this app"""
data = TypeAdapter(Callbacks).validate_python(self.get_callback_entries())
return data.get(self.name, {})
self_callbacks = self.get_callback_entries().get(self.name, {})
for handle, cb_dict in self_callbacks.items():
try:
yield handle, CallbackEntry.model_validate(cb_dict)
except ValidationError as e:
self.logger.error('Error parsing callback dictionary')
self.logger.error(e)
def listen_motion_on(self):
"""Sets up the motion on callback to activate the room"""
@@ -104,9 +104,7 @@ class Motion(Hass):
oneshot=True,
cause='motion on',
)
self.log(
f'Waiting for sensor motion on [friendly_name]{self.sensor.friendly_name}[/]'
)
self.log(f'Waiting for sensor motion on [friendly_name]{self.sensor.friendly_name}[/]')
if self.sensor_state:
self.log(
f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is already on',
@@ -134,18 +132,14 @@ class Motion(Hass):
level='WARNING',
)
def callback_light_on(
self, entity=None, attribute=None, old=None, new=None, kwargs=None
):
def callback_light_on(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
"""Called when the light turns on"""
if new is not None:
self.log(f'Detected {entity} turning on', level='DEBUG')
duration = self.app.off_duration()
self.listen_motion_off(duration)
def callback_light_off(
self, entity=None, attribute=None, old=None, new=None, kwargs=None
):
def callback_light_off(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
"""Called when the light turns off"""
self.log(f'Detected {entity} turning off', level='DEBUG')
self.listen_motion_on()

19
src/room_control/room_control.py
View File

@@ -1,4 +1,5 @@
import datetime
import json
import logging
import logging.config
from copy import deepcopy
@@ -8,7 +9,7 @@ from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from .console import console, room_logging_config
from . import console
from .model import ControllerStateConfig, RoomControllerConfig
logger = logging.getLogger(__name__)
@@ -33,14 +34,8 @@ class RoomController(Hass, Mqtt):
assert all(isinstance(s, ControllerStateConfig) for s in new), f'Invalid: {new}'
self._room_config.states = new
def configure_logging(self) -> dict:
logging.config.dictConfig(room_logging_config(self.name))
self.logger = logging.LoggerAdapter(
logging.getLogger(f'AppDaemon.{self.name}'), {'room': self.name}
)
def initialize(self):
self.configure_logging()
self.logger = console.load_rich_config(self.name)
self.app_entities = self.gather_app_entities()
# self.log(f'entities: {self.app_entities}')
self.refresh_state_times()
@@ -95,10 +90,6 @@ class RoomController(Hass, Mqtt):
assert isinstance(state.time, datetime.time), f'Invalid time: {state.time}'
# if self.logger.isEnabledFor(logging.DEBUG):
# # table = self._room_config.rich_table(self.name)
# console.print(self._room_config)
self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
# schedule the transitions
@@ -209,9 +200,7 @@ class RoomController(Hass, Mqtt):
elif isinstance(scene_kwargs, dict):
self.call_service('scene/apply', **scene_kwargs)
if self.logger.isEnabledFor(logging.INFO):
self.log('Applied scene:')
console.print(scene_kwargs['entities'])
self.log(f'Applied scene:\n{json.dumps(scene_kwargs, indent=2)}', level='DEBUG')
elif scene_kwargs is None:
self.log('No scene, ignoring...')