added log_level config per app

This commit is contained in:
John Lancaster
2024-07-27 20:48:57 -05:00
parent 7cde3c75d8
commit e4fcd757ce
5 changed files with 22 additions and 44 deletions

View File

@@ -14,7 +14,9 @@ class Button:
button_name: str button_name: str
def __post_init__(self): def __post_init__(self):
self.logger = console.load_rich_config(self.adapi.name, 'Button', 'DEBUG') self.logger = console.load_rich_config(
self.adapi.name, 'Button', level=self.adapi.args.get('log_level', 'INFO')
)
topic = f'zigbee2mqtt/{self.button_name}' topic = f'zigbee2mqtt/{self.button_name}'
self.adapi.listen_event( self.adapi.listen_event(
self.handle_button, self.handle_button,

View File

@@ -13,7 +13,9 @@ class Door:
entity_id: str entity_id: str
def __post_init__(self): def __post_init__(self):
self.logger = console.load_rich_config(self.adapi.name, 'Door', 'DEBUG') self.logger = console.load_rich_config(
self.adapi.name, 'Door', level=self.adapi.args.get('log_level', 'INFO')
)
self.adapi.listen_state( self.adapi.listen_state(
lambda *args, **kwargs: self.adapi.activate_all_off(cause='door open'), lambda *args, **kwargs: self.adapi.activate_all_off(cause='door open'),

View File

@@ -1,6 +1,6 @@
import datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Annotated, Dict, List, Optional, Self from typing import Annotated, Dict, List, Optional, Self, Union
import yaml import yaml
from astral import SunDirection from astral import SunDirection
@@ -61,12 +61,20 @@ class ControllerStateConfig(BaseModel):
return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True) return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True)
class MotionSensorConfig(BaseModel):
sensor: str
ref_entity: str
class RoomControllerConfig(BaseModel): class RoomControllerConfig(BaseModel):
states: List[ControllerStateConfig] = Field(default_factory=list) states: List[ControllerStateConfig] = Field(default_factory=list)
off_duration: Optional[OffDuration] = Field(default_factory=datetime.timedelta) off_duration: Optional[OffDuration] = Field(default_factory=datetime.timedelta)
sleep_state: Optional[ControllerStateConfig] = None sleep_state: Optional[ControllerStateConfig] = None
rich: Optional[str] = None rich: Optional[str] = None
manual_mode: Optional[str] = None manual_mode: Optional[str] = None
button: Optional[Union[str, List[str]]] = None
motion: Optional[MotionSensorConfig] = None
log_level: Optional[str] = None
@classmethod @classmethod
def from_yaml(cls: Self, yaml_path: Path) -> Self: def from_yaml(cls: Self, yaml_path: Path) -> Self:
@@ -95,38 +103,3 @@ class RoomControllerConfig(BaseModel):
] ]
table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines)) table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines))
yield table yield table
def sort_states(self):
"""Should only be called after all the times have been resolved"""
assert all(
isinstance(state.time, datetime.time) for state in self.states
), 'Times have not all been resolved yet'
self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
def current_state(self, now: datetime.time) -> ControllerStateConfig:
self.sort_states()
for state in self.states:
if state.time <= now:
return state
else:
return self.states[0]
def current_scene(self, now: datetime.time) -> Dict:
state = self.current_state(now)
return state.scene
def current_off_duration(self, now: datetime.time) -> datetime.timedelta:
state = self.current_state(now)
if state.off_duration is None:
if self.off_duration is None:
raise ValueError('Need an off duration')
else:
return self.off_duration
else:
return state.off_duration
class ButtonConfig(BaseModel):
app: str
button: str | List[str]
ref_entity: str

View File

@@ -32,7 +32,9 @@ class MotionSensor:
ref_entity_id: str ref_entity_id: str
def __post_init__(self): def __post_init__(self):
self.logger = console.load_rich_config(self.adapi.name, 'Motion', 'DEBUG') self.logger = console.load_rich_config(
self.adapi.name, 'Motion', level=self.adapi.args.get('log_level', 'INFO')
)
assert self.sensor_entity.exists() assert self.sensor_entity.exists()
assert self.ref_entity.exists() assert self.ref_entity.exists()

View File

@@ -56,8 +56,7 @@ class RoomController(Hass):
return self.get_entity(f'{self.name}.state', namespace='controller') return self.get_entity(f'{self.name}.state', namespace='controller')
def initialize(self): def initialize(self):
self.logger = console.load_rich_config(self.name) self.logger = console.load_rich_config(self.name, level=self.args.get('log_level', 'INFO'))
self.set_log_level('DEBUG')
self.refresh_state_times() self.refresh_state_times()
self.run_daily(callback=self.refresh_state_times, start='00:00:00') self.run_daily(callback=self.refresh_state_times, start='00:00:00')
@@ -82,13 +81,13 @@ class RoomController(Hass):
if button := self.args.get('button'): if button := self.args.get('button'):
if isinstance(button, str): if isinstance(button, str):
self.button = Button(self, button_name=button) Button(self, button_name=button)
elif isinstance(button, list) and all(isinstance(b, str) for b in button): elif isinstance(button, list) and all(isinstance(b, str) for b in button):
self.button = [Button(self, button_name=b) for b in button] for b in button:
Button(self, button_name=b)
if door := self.args.get('door'): if door := self.args.get('door'):
if isinstance(door, str): if isinstance(door, str):
self.log('door--')
self.door = Door(self, entity_id=door) self.door = Door(self, entity_id=door)
if motion := self.args.get('motion'): if motion := self.args.get('motion'):