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

added some rich logging

Browse Source
This commit is contained in:
John Lancaster
2024-03-03 14:54:11 -06:00
parent d8a9d3d83a
commit 215d47ea00
3 changed files with 97 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
button.py
View File

@@ -1,17 +1,21 @@
import json import json
from appdaemon.plugins.mqtt.mqttapi import Mqtt from appdaemon.plugins.mqtt.mqttapi import Mqtt
from console import setup_logging
from room_control import RoomController from room_control import RoomController
class Button(Mqtt): class Button(Mqtt):
async def initialize(self): async def initialize(self):
if self.args.get('rich', False):
setup_logging(self)
self.app: RoomController = await self.get_app(self.args['app']) self.app: RoomController = await self.get_app(self.args['app'])
self.setup_buttons(self.args['button']) self.setup_buttons(self.args['button'])
def setup_buttons(self, buttons): def setup_buttons(self, buttons):
if isinstance(buttons, list): if isinstance(buttons, list):
for button in buttons: for button in buttons:
self.setup_button(button) self.setup_button(button)
else: else:
self.setup_button(buttons) self.setup_button(buttons)
@@ -19,7 +23,7 @@ class Button(Mqtt):
def setup_button(self, name: str): def setup_button(self, name: str):
topic = f'zigbee2mqtt/{name}' topic = f'zigbee2mqtt/{name}'
self.mqtt_subscribe(topic, namespace='mqtt') self.mqtt_subscribe(topic, namespace='mqtt')
self.listen_event(self.handle_button, "MQTT_MESSAGE", topic=topic, namespace='mqtt', button=name) self.listen_event(self.handle_button, 'MQTT_MESSAGE', topic=topic, namespace='mqtt', button=name)
self.log(f'"{topic}" controls app {self.app.name}') self.log(f'"{topic}" controls app {self.app.name}')
def handle_button(self, event_name, data, kwargs): def handle_button(self, event_name, data, kwargs):
@@ -28,7 +32,7 @@ class Button(Mqtt):
action = payload['action'] action = payload['action']
except json.JSONDecodeError: except json.JSONDecodeError:
self.log(f'Error decoding JSON from {data["payload"]}', level='ERROR') self.log(f'Error decoding JSON from {data["payload"]}', level='ERROR')
except KeyError as e: except KeyError:
return return
else: else:
if action != '': if action != '':
@@ -38,12 +42,10 @@ class Button(Mqtt):
if action == 'single': if action == 'single':
self.log(f' {action.upper()} '.center(50, '=')) self.log(f' {action.upper()} '.center(50, '='))
state = self.get_state(self.args['ref_entity']) state = self.get_state(self.args['ref_entity'])
kwargs = { kwargs = {'kwargs': {'cause': f'button single click: toggle while {state}'}}
'kwargs': {'cause': f'button single click: toggle while {state}'}
}
if state == 'on': if state == 'on':
self.app.deactivate(**kwargs) self.app.deactivate(**kwargs)
else: else:
self.app.activate(**kwargs) self.app.activate(**kwargs)
else: else:
pass pass

20
console.py Normal file
View File

@@ -0,0 +1,20 @@
from appdaemon.adapi import ADAPI
from rich.console import Console
from rich.logging import RichHandler
console = Console(width=150)
handler = RichHandler(
console=console,
markup=True,
show_path=False,
log_time_format='%Y-%m-%d %I:%M:%S %p',
)
def setup_logging(self: ADAPI):
if not any(isinstance(h, RichHandler) for h in self.logger.handlers):
self.logger.propagate = False
self.logger.addHandler(handler)
self.log(f'Added rich handler for [bold green]{self.logger.name}[/]')
self.log(f'Formatter [bold green]{self.logger.handlers[0].formatter}[/]')

140
room_control.py
View File

@@ -1,30 +1,31 @@
import datetime
from copy import deepcopy from copy import deepcopy
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, time, timedelta
from pathlib import Path from pathlib import Path
from typing import Dict, List from typing import Dict, List
import appdaemon.utils as utils
import yaml import yaml
from appdaemon.entity import Entity from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt from appdaemon.plugins.mqtt.mqttapi import Mqtt
from astral import SunDirection from astral import SunDirection
from console import setup_logging
def str_to_timedelta(input_str: str) -> timedelta: def str_to_timedelta(input_str: str) -> datetime.timedelta:
try: try:
hours, minutes, seconds = map(int, input_str.split(':')) hours, minutes, seconds = map(int, input_str.split(':'))
return timedelta(hours=hours, minutes=minutes, seconds=seconds) return datetime.timedelta(hours=hours, minutes=minutes, seconds=seconds)
except Exception: except Exception:
return timedelta() return datetime.timedelta()
@dataclass @dataclass
class RoomState: class RoomState:
scene: Dict[str, Dict[str, str | int]] scene: Dict[str, Dict[str, str | int]]
off_duration: timedelta = None off_duration: datetime.timedelta = None
time: time = None time: datetime.time = None
time_fmt: List[str] = field(default_factory=lambda : ['%H:%M:%S', '%I:%M:%S %p'], repr=False) time_fmt: List[str] = field(default_factory=lambda: ['%H:%M:%S', '%I:%M:%S %p'], repr=False)
elevation: int | float = None elevation: int | float = None
direction: SunDirection = None direction: SunDirection = None
@@ -32,14 +33,14 @@ class RoomState:
if isinstance(self.time, str): if isinstance(self.time, str):
for fmt in self.time_fmt: for fmt in self.time_fmt:
try: try:
self.time = datetime.strptime(self.time, fmt).time() self.time = datetime.datetime.strptime(self.time, fmt).time()
except: except Exception:
continue continue
else: else:
break break
if self.elevation is not None: if self.elevation is not None:
assert self.direction is not None, f'Elevation setting requires a direction' assert self.direction is not None, 'Elevation setting requires a direction'
if self.direction.lower() == 'setting': if self.direction.lower() == 'setting':
self.direction = SunDirection.SETTING self.direction = SunDirection.SETTING
elif self.direction.lower() == 'rising': elif self.direction.lower() == 'rising':
@@ -61,7 +62,7 @@ class RoomState:
@dataclass @dataclass
class RoomConfig: class RoomConfig:
states: List[RoomState] states: List[RoomState]
off_duration: timedelta = None off_duration: datetime.timedelta = None
def __post_init__(self): def __post_init__(self):
if isinstance(self.off_duration, str): if isinstance(self.off_duration, str):
@@ -74,28 +75,24 @@ class RoomConfig:
else: else:
kwargs = {} kwargs = {}
self = cls( self = cls(states=[RoomState.from_json(s) for s in app_cfg['states']], **kwargs)
states=[RoomState.from_json(s) for s in app_cfg['states']],
**kwargs
)
return self return self
@classmethod @classmethod
def from_yaml(cls, yaml_path: Path, app_name: str): def from_yaml(cls, yaml_path: Path, app_name: str):
with yaml_path.open('r') as f: with yaml_path.open('r') as f:
cfg: Dict = yaml.load(f, Loader=yaml.SafeLoader)[app_name] cfg: Dict = yaml.load(f, Loader=yaml.SafeLoader)[app_name]
return cls.from_app_config(cfg) return cls.from_app_config(cfg)
def sort_states(self): def sort_states(self):
"""Should only be called after all the times have been resolved """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'
assert all(isinstance(state.time, 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) self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
def current_state(self, now: time) -> RoomState: def current_state(self, now: datetime.time) -> RoomState:
time_fmt = "%I:%M:%S %p" # time_fmt = '%I:%M:%S %p'
print(now.strftime(time_fmt)) # print(now.strftime(time_fmt))
self.sort_states() self.sort_states()
for state in self.states: for state in self.states:
@@ -104,16 +101,16 @@ class RoomConfig:
else: else:
# self.log(f'Defaulting to first state') # self.log(f'Defaulting to first state')
return self.states[0] return self.states[0]
def current_scene(self, now: time) -> Dict: def current_scene(self, now: datetime.time) -> Dict:
state = self.current_state(now) state = self.current_state(now)
return state.scene return state.scene
def current_off_duration(self, now: time) -> timedelta: def current_off_duration(self, now: datetime.time) -> datetime.timedelta:
state = self.current_state(now) state = self.current_state(now)
if state.off_duration is None: if state.off_duration is None:
if self.off_duration is None: if self.off_duration is None:
raise ValueError(f'Need an off duration') raise ValueError('Need an off duration')
else: else:
return self.off_duration return self.off_duration
else: else:
@@ -133,26 +130,31 @@ class RoomController(Hass, Mqtt):
@property @property
def states(self) -> List[RoomState]: def states(self) -> List[RoomState]:
return self._room_config.states return self._room_config.states
@states.setter @states.setter
def states(self, new: List[RoomState]): def states(self, new: List[RoomState]):
assert all(isinstance(s, RoomState) for s in new), f'Invalid: {new}' assert all(isinstance(s, RoomState) for s in new), f'Invalid: {new}'
self._room_config.states = new self._room_config.states = new
def initialize(self): def initialize(self):
if self.args.get('rich', False):
setup_logging(self)
self.app_entities = self.gather_app_entities() self.app_entities = self.gather_app_entities()
# self.log(f'entities: {self.app_entities}') # self.log(f'entities: {self.app_entities}')
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')
def gather_app_entities(self) -> List[str]: def gather_app_entities(self) -> List[str]:
"""Returns a list of all the entities involved in any of the states """Returns a list of all the entities involved in any of the states"""
"""
def generator(): def generator():
for settings in deepcopy(self.args['states']): for settings in deepcopy(self.args['states']):
if (scene := settings.get('scene')): if scene := settings.get('scene'):
if isinstance(scene, str): if isinstance(scene, str):
assert scene.startswith('scene.'), f"Scene definition must start with 'scene.' for app {self.name}" assert scene.startswith(
'scene.'
), f"Scene definition must start with 'scene.' for app {self.name}"
entity: Entity = self.get_entity(scene) entity: Entity = self.get_entity(scene)
entity_state = entity.get_state('all') entity_state = entity.get_state('all')
attributes = entity_state['attributes'] attributes = entity_state['attributes']
@@ -170,7 +172,7 @@ class RoomController(Hass, Mqtt):
def refresh_state_times(self, *args, **kwargs): def refresh_state_times(self, *args, **kwargs):
"""Resets the `self.states` attribute to a newly parsed version of the states. """Resets the `self.states` attribute to a newly parsed version of the states.
Parsed states have an absolute time for the current day. Parsed states have an absolute time for the current day.
""" """
# re-parse the state strings into times for the current day # re-parse the state strings into times for the current day
self._room_config = RoomConfig.from_app_config(self.args) self._room_config = RoomConfig.from_app_config(self.args)
@@ -179,14 +181,13 @@ class RoomController(Hass, Mqtt):
for state in self._room_config.states: for state in self._room_config.states:
if state.time is None and state.elevation is not None: if state.time is None and state.elevation is not None:
state.time = self.AD.sched.location.time_at_elevation( state.time = self.AD.sched.location.time_at_elevation(
elevation=state.elevation, elevation=state.elevation, direction=state.direction
direction=state.direction ).time()
).time()
elif isinstance(state.time, str): elif isinstance(state.time, str):
state.time = self.parse_time(state.time) state.time = self.parse_time(state.time)
assert isinstance(state.time, time), f'Invalid time: {state.time}' assert isinstance(state.time, datetime.time), f'Invalid time: {state.time}'
for state in self.states: for state in self.states:
self.log(f'State: {state.time.strftime("%I:%M:%S %p")} {state.scene}') self.log(f'State: {state.time.strftime("%I:%M:%S %p")} {state.scene}')
@@ -194,24 +195,20 @@ class RoomController(Hass, Mqtt):
# schedule the transitions # schedule the transitions
for state in self.states[::-1]: for state in self.states[::-1]:
# t: time = state['time'] # t: datetime.time = state['time']
t: time = state.time t: datetime.time = state.time
try: try:
self.run_at( self.run_at(callback=self.activate_any_on, start=t.strftime('%H:%M:%S'), cause='scheduled transition')
callback=self.activate_any_on,
start=t.strftime('%H:%M:%S'),
cause='scheduled transition'
)
except ValueError: except ValueError:
# happens when the callback time is in the past # happens when the callback time is in the past
pass pass
except Exception as e: except Exception as e:
self.log(f'Failed with {type(e)}: {e}') self.log(f'Failed with {type(e)}: {e}')
def current_state(self, now: time = None) -> RoomState: def current_state(self, now: datetime.time = None) -> RoomState:
if self.sleep_bool(): if self.sleep_bool():
self.log(f'sleep: active') self.log('sleep: active')
if (state := self.args.get('sleep_state')): if state := self.args.get('sleep_state'):
return RoomState.from_json(state) return RoomState.from_json(state)
else: else:
return RoomState(scene={}) return RoomState(scene={})
@@ -224,21 +221,22 @@ class RoomController(Hass, Mqtt):
return state return state
def current_scene(self, now: time = None) -> Dict[str, Dict[str, str | int]]: def current_scene(self, now: datetime.time = None) -> Dict[str, Dict[str, str | int]]:
state = self.current_state(now) state = self.current_state(now)
assert isinstance(state, RoomState) # print(f'{type(state).__name__}')
self.log(f'Current scene: {state}') # assert isinstance(state, RoomState), f'Invalid state: {type(state).__name__}'
assert type(state).__name__ == 'RoomState' # needed for the reloading to work
# self.log(f'Current scene: {state}')
self.log('Current scene:')
self.log(state)
return state.scene return state.scene
def app_entity_states(self) -> Dict[str, str]: def app_entity_states(self) -> Dict[str, str]:
states = { states = {entity: self.get_state(entity) for entity in self.app_entities}
entity: self.get_state(entity)
for entity in self.app_entities
}
return states return states
def all_off(self) -> bool: def all_off(self) -> bool:
""""All off" is the logic opposite of "any on" """ "All off" is the logic opposite of "any on"
Returns: Returns:
bool: Whether all the lights associated with the app are off bool: Whether all the lights associated with the app are off
@@ -247,7 +245,7 @@ class RoomController(Hass, Mqtt):
return all(state != 'on' for entity, state in states.items()) return all(state != 'on' for entity, state in states.items())
def any_on(self) -> bool: def any_on(self) -> bool:
""""Any on" is the logic opposite of "all off" """ "Any on" is the logic opposite of "all off"
Returns: Returns:
bool: Whether any of the lights associated with the app are on bool: Whether any of the lights associated with the app are on
@@ -256,7 +254,7 @@ class RoomController(Hass, Mqtt):
return any(state == 'on' for entity, state in states.items()) return any(state == 'on' for entity, state in states.items())
def sleep_bool(self) -> bool: def sleep_bool(self) -> bool:
if (sleep_var := self.args.get('sleep')): if sleep_var := self.args.get('sleep'):
return self.get_state(sleep_var) == 'on' return self.get_state(sleep_var) == 'on'
else: else:
return False return False
@@ -271,7 +269,7 @@ class RoomController(Hass, Mqtt):
# else: # else:
# raise ValueError('Sleep variable is undefined') # raise ValueError('Sleep variable is undefined')
def off_duration(self, now: time = None) -> timedelta: def off_duration(self, now: datetime.time = None) -> datetime.timedelta:
"""Determines the time that the motion sensor has to be clear before deactivating """Determines the time that the motion sensor has to be clear before deactivating
Priority: Priority:
@@ -284,17 +282,17 @@ class RoomController(Hass, Mqtt):
sleep_mode_active = self.sleep_bool() sleep_mode_active = self.sleep_bool()
if sleep_mode_active: if sleep_mode_active:
self.log(f'Sleeping mode active: {sleep_mode_active}') self.log(f'Sleeping mode active: {sleep_mode_active}')
return timedelta() return datetime.timedelta()
else: else:
now = now or self.get_now().time() now = now or self.get_now().time()
return self._room_config.current_off_duration(now) return self._room_config.current_off_duration(now)
def activate(self, entity = None, attribute = None, old = None, new = None, kwargs = None): def activate(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
if kwargs is not None: if kwargs is not None:
cause = kwargs.get('cause', 'unknown') cause = kwargs.get('cause', 'unknown')
else: else:
cause = 'unknown' cause = 'unknown'
self.log(f'Activating: {cause}') self.log(f'Activating: {cause}')
scene = self.current_scene() scene = self.current_scene()
@@ -312,27 +310,25 @@ class RoomController(Hass, Mqtt):
self.log(f'Applied scene: {scene}') self.log(f'Applied scene: {scene}')
elif scene is None: elif scene is None:
self.log(f'No scene, ignoring...') self.log('No scene, ignoring...')
# Need to act as if the light had just turned off to reset the motion (and maybe other things?) # Need to act as if the light had just turned off to reset the motion (and maybe other things?)
# self.callback_light_off() # self.callback_light_off()
else: else:
self.log(f'ERROR: unknown scene: {scene}') self.log(f'ERROR: unknown scene: {scene}')
def activate_all_off(self, *args, **kwargs): def activate_all_off(self, *args, **kwargs):
"""Activate if all of the entities are off. Args and kwargs are passed directly to self.activate() """Activate if all of the entities are off. Args and kwargs are passed directly to self.activate()"""
"""
if self.all_off(): if self.all_off():
self.activate(*args, **kwargs) self.activate(*args, **kwargs)
else: else:
self.log(f'Skipped activating - everything is not off') self.log('Skipped activating - everything is not off')
def activate_any_on(self, *args, **kwargs): def activate_any_on(self, *args, **kwargs):
"""Activate if any of the entities are on. Args and kwargs are passed directly to self.activate() """Activate if any of the entities are on. Args and kwargs are passed directly to self.activate()"""
"""
if self.any_on(): if self.any_on():
self.activate(*args, **kwargs) self.activate(*args, **kwargs)
else: else:
self.log(f'Skipped activating - everything is off') self.log('Skipped activating - everything is off')
def toggle_activate(self, *args, **kwargs): def toggle_activate(self, *args, **kwargs):
if self.any_on(): if self.any_on():
@@ -340,7 +336,7 @@ class RoomController(Hass, Mqtt):
else: else:
self.activate(*args, **kwargs) self.activate(*args, **kwargs)
def deactivate(self, entity = None, attribute = None, old = None, new = None, kwargs = None): def deactivate(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
cause = kwargs.get('cause', 'unknown') cause = kwargs.get('cause', 'unknown')
self.log(f'Deactivating: {cause}') self.log(f'Deactivating: {cause}')
for e in self.app_entities: for e in self.app_entities: