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

added RoomState and RoomConfig dataclasses

Browse Source
This commit is contained in:
John Lancaster
2024-01-02 22:56:58 -06:00
parent dda9d2b501
commit 12ac6c4cc4
1 changed files with 111 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
room_control.py
View File

@@ -1,13 +1,79 @@
import asyncio
from copy import deepcopy
from dataclasses import dataclass, field
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Dict, List
import appdaemon.utils as utils
import astral
import yaml
from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from astral import SunDirection
@dataclass
class RoomState:
scene: Dict[str, Dict[str, str | int]]
off_duration: timedelta = field(default_factory=timedelta)
time: time = None
time_fmt: List[str] = field(default_factory=lambda : ['%H:%M:%S', '%I:%M:%S %p'], repr=False)
elevation: int | float = None
direction: SunDirection = None
def __post_init__(self):
if isinstance(self.time, str):
for fmt in self.time_fmt:
try:
self.time = datetime.strptime(self.time, fmt).time()
except:
continue
else:
break
if self.elevation is not None:
assert self.direction is not None, f'Elevation setting requires a direction'
if self.direction.lower() == 'setting':
self.direction = SunDirection.SETTING
elif self.direction.lower() == 'rising':
self.direction = SunDirection.RISING
else:
raise ValueError(f'Invalid sun direction: {self.direction}')
if isinstance(self.elevation, str):
self.elevation = float(self.elevation)
if isinstance(self.off_duration, str):
try:
hours, minutes, seconds = map(int, self.off_duration.split(':'))
self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
except Exception:
self.off_duration = timedelta()
@classmethod
def from_json(cls, json_input):
return cls(**json_input)
@dataclass
class RoomConfig:
states: List[RoomState]
off_duration: timedelta = field(default_factory=timedelta)
@classmethod
def from_app_config(cls, app_cfg: Dict[str, Dict]):
if 'class' in app_cfg:
app_cfg.pop('class')
if 'module' in app_cfg:
app_cfg.pop('module')
return cls(states=[RoomState.from_json(s) for s in app_cfg['states']])
@classmethod
def from_yaml(cls, yaml_path: Path, app_name: str):
with yaml_path.open('r') as f:
cfg: Dict = yaml.load(f, Loader=yaml.SafeLoader)[app_name]
return cls.from_app_config(cfg)
class RoomController(Hass, Mqtt):
@@ -20,6 +86,15 @@ class RoomController(Hass, Mqtt):
- When the light comes on, check if it's attributes match what they should, given the time.
"""
@property
def states(self) -> List[RoomState]:
return self._room_config.states
@states.setter
def states(self, new: List[RoomState]):
assert all(isinstance(s, RoomState) for s in new), f'Invalid: {new}'
self._room_config.states = new
async def initialize(self):
self.app_entities = await self.gather_app_entities()
# self.log(f'entities: {self.app_entities}')
@@ -58,51 +133,42 @@ class RoomController(Hass, Mqtt):
Parsed states have an absolute time for a certain day.
"""
# re-parse the state strings into times for the current day
self.states = await self.parse_states()
self._room_config = RoomConfig.from_app_config(self.args)
self.log(f'{len(self._room_config.states)} states in the RoomConfig')
for state in self._room_config.states:
if state.time is None and state.elevation is not None:
state.time = self.AD.sched.location.time_at_elevation(
elevation=state.elevation,
direction=state.direction
).time()
elif isinstance(state.time, str):
state.time = await self.parse_time(state.time)
assert isinstance(state.time, time), f'Invalid time: {state.time}'
for state in self.states:
self.log(f'State: {state.time.strftime("%I:%M:%S %p")} {str(state.scene)[:50]}...')
self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
# schedule the transitions
for state in self.states:
t: time = state['time']
for state in self.states[::-1]:
# t: time = state['time']
t: time = state.time
try:
await self.run_at(callback=self.activate_any_on, start=t.strftime('%H:%M:%S'), cause='scheduled transition')
await self.run_at(
callback=self.activate_any_on,
start=t.strftime('%H:%M:%S'),
cause='scheduled transition'
)
except ValueError:
# happens when the callback time is in the past
pass
except Exception as e:
self.log(f'Failed with {type(e)}: {e}')
else:
self.log(f'Scheduled transition at: {t.strftime("%I:%M:%S %p")}')
async def parse_states(self):
async def gen():
for state in deepcopy(self.args['states']):
if (time := state.get('time')):
state['time'] = await self.parse_time(time)
elif isinstance((elevation := state.get('elevation')), (int, float)):
assert 'direction' in state, f'State needs a direction if it has an elevation'
if state['direction'] == 'rising':
dir = astral.SunDirection.RISING
elif state['direction'] == 'setting':
dir = astral.SunDirection.SETTING
else:
raise ValueError(f'Invalid sun direction: {state["direction"]}')
state['time'] = self.AD.sched.location.time_at_elevation(
elevation=elevation, direction=dir
).time()
else:
raise ValueError(f'Missing time')
yield state
states = [s async for s in gen()]
states = sorted(states, key=lambda s: s['time'], reverse=True)
return states
async def current_state(self, time: time = None):
async def current_state(self, time: time = None) -> RoomState:
if (await self.sleep_bool()):
self.log(f'sleep: active')
if (state := self.args.get('sleep_state')):
@@ -113,11 +179,11 @@ class RoomController(Hass, Mqtt):
# now: datetime = await self.get_now()
# self.log(f'Getting state for datetime: {now.strftime("%I:%M:%S %p")}')
time = time or (await self.get_now()).time()
time_fmt = "%I:%M %p"
time_fmt = "%I:%M:%S %p"
self.log(f'Getting state before: {time.strftime(time_fmt)}')
for state in self.states:
time_str = state["time"].strftime(time_fmt)
if state['time'] <= time:
time_str = state.time.strftime(time_fmt)
if state.time <= time:
self.log(f'Selected state from {time_str}')
return state
else:
@@ -126,9 +192,9 @@ class RoomController(Hass, Mqtt):
self.log(f'Defaulting to first state')
return self.states[0]
async def current_scene(self, time: time = None):
async def current_scene(self, time: time = None) -> Dict[str, Dict[str, str | int]]:
if (state := (await self.current_state(time=time))) is not None:
return state['scene']
return state.scene
async def app_entity_states(self) -> Dict[str, str]:
states = {
@@ -182,15 +248,9 @@ class RoomController(Hass, Mqtt):
"""
current_state = await self.current_state()
duration_str = current_state.get(
'off_duration',
self.args.get('off_duration', '00:00:00')
)
try:
hours, minutes, seconds = map(int, duration_str.split(':'))
return timedelta(hours=hours, minutes=minutes, seconds=seconds)
except Exception:
if isinstance(current_state.off_duration, timedelta):
return current_state.off_duration
else:
return timedelta()
@utils.sync_wrapper