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

more work

Browse Source
This commit is contained in:
John Lancaster
2024-03-10 17:33:22 -05:00
parent e90ad5a071
commit 541475d68b
5 changed files with 93 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
button.py
View File

@@ -15,7 +15,7 @@ class Button(Mqtt):
async def initialize(self):
setup_component_logging(self)
self.app: RoomController = await self.get_app(self.args['app'])
self.log(f'Connected to AD app [room]{self.app.name}[/]')
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
self.button = self.args['button']
self.setup_buttons(self.button)

10
door.py
View File

@@ -1,8 +1,14 @@
from appdaemon.plugins.hass.hassapi import Hass
from console import setup_component_logging
from room_control import RoomController
class Door(Hass):
async def initialize(self):
app: RoomController = await self.get_app(self.args['app'])
await self.listen_state(app.activate_all_off, entity_id=self.args['door'], new='on', cause='door open')
setup_component_logging(self)
self.app: RoomController = await self.get_app(self.args['app'])
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
await self.listen_state(self.app.activate_all_off, entity_id=self.args['door'], new='on', cause='door open')

38
model.py Normal file
View File

@@ -0,0 +1,38 @@
from typing import Dict
from pydantic import (
BaseModel,
ValidationError,
field_validator,
)
def validate_int(v):
if not len(bytes(v)) == 1:
raise ValidationError()
class State(BaseModel):
state: bool = True
brightness: int = None
color_temp: int = None
@field_validator('brightness')
@classmethod
def validate_brightness(cls, v: int) -> int:
assert 0 <= v <= 255
return v
@field_validator('color_temp')
@classmethod
def validate_color_temp(cls, v: int) -> int:
assert 200 <= v <= 600
return v
# Scene = RootModel[Dict[str, State]]
class ApplyScene(BaseModel):
entities: Dict[str, State]
transition: int = None

14
motion.py
View File

@@ -28,7 +28,7 @@ class Motion(Hass):
def initialize(self):
setup_component_logging(self)
self.app: RoomController = self.get_app(self.args['app'])
self.log(f'Connected to AD app [room]{self.app.name}[/]')
self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
assert self.entity_exists(self.args['sensor'])
assert self.entity_exists(self.args['ref_entity'])
@@ -37,8 +37,14 @@ class Motion(Hass):
entity_id=self.ref_entity.entity_id,
immediate=True, # avoids needing to sync the state
)
if self.sensor_state != self.ref_entity_state:
self.log(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}'})
# don't need to await these because they'll already get turned into a task by the utils.sync_wrapper decorator
self.listen_state(**base_kwargs, attribute='brightness', callback=self.callback_light_on)
self.listen_state(**base_kwargs, attribute='brightness', callback=self.callback_light_on,)
self.listen_state(**base_kwargs, new='off', callback=self.callback_light_off)
def listen_motion_on(self):
@@ -53,7 +59,7 @@ class Motion(Hass):
)
self.log(f'Waiting for motion on [friendly_name]{self.sensor.friendly_name}[/]')
if self.sensor_state:
self.log(f'{self.sensor.friendly_name} is already on', level='WARNING')
self.log(f'[friendly_name]{self.sensor.friendly_name}[/] is already on', level='WARNING')
def listen_motion_off(self, duration: timedelta):
"""Sets up the motion off callback to deactivate the room"""
@@ -69,7 +75,7 @@ class Motion(Hass):
self.log(f'Waiting for motion to stop on [friendly_name]{self.sensor.friendly_name}[/] for {duration}')
if not self.sensor_state:
self.log(f'{self.sensor.friendly_name} is currently off', level='WARNING')
self.log(f'[friendly_name]{self.sensor.friendly_name}[/] is currently off', level='WARNING')
def callback_light_on(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
"""Called when the light turns on"""

51
room_control.py
View File

@@ -11,8 +11,9 @@ from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from astral import SunDirection
from console import console, setup_handler
from rich.console import Console, ConsoleOptions, RenderResult, Group
from rich.table import Table, Column
from model import ApplyScene
from rich.console import Console, ConsoleOptions, RenderResult
from rich.table import Column, Table
def str_to_timedelta(input_str: str) -> datetime.timedelta:
@@ -65,11 +66,23 @@ class RoomState:
return cls(**json_input)
def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult:
table = Table('Entity ID', 'State')
table = Table(
Column('Entity ID', width=15),
Column('State'),
highlight=True,
padding=1,
collapse_padding=True,
)
for name, state in self.scene.items():
table.add_row(name, str(state))
table.add_row(name, ApplyScene(entites=self.scene).model_dump(exclude_none=True))
yield table
def scene_model(self) -> ApplyScene:
return ApplyScene(
entities=self.scene,
transition=0
).model_dump(exclude_none=True)
@dataclass
class RoomConfig:
@@ -101,10 +114,16 @@ class RoomConfig:
table = Table(
Column('Time', width=15),
Column('Scene'),
highlight=True, padding=1, collapse_padding=True,
highlight=True,
padding=1,
collapse_padding=True,
)
for state in self.states:
lines = [f'{name:20}{state["state"]} Brightness: {state["brightness"]:<4} Temp: {state["color_temp"]}' for name, state in state.scene.items()]
scene_json = ApplyScene(entities=state.scene).model_dump(exclude_none=True)
lines = [
f'{name:20}{state["state"]} Brightness: {state["brightness"]:<4} Temp: {state["color_temp"]}'
for name, state in scene_json['entities'].items()
]
table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines))
yield table
@@ -257,15 +276,20 @@ class RoomController(Hass, Mqtt):
self.log(f'Getting state for {now}', level='DEBUG')
state = self._room_config.current_state(now)
self.log(f'Current state: {state}', level='DEBUG')
if self.logger.isEnabledFor(logging.DEBUG):
self.log('Current state', level='DEBUG')
console.print(state)
return state
def current_scene(self, now: datetime.time = None) -> Dict[str, Dict[str, str | int]]:
state = self.current_state(now)
assert type(state).__name__ == 'RoomState' # needed this way instead of isinstance(...) for the reloading to work
# needed this way instead of isinstance(...) for the reloading to work
assert type(state).__name__ == 'RoomState'
if self.logger.isEnabledFor(logging.DEBUG):
self.log('Current scene:')
self.log('Current scene:', level='DEBUG')
console.print(state)
return state.scene
@@ -339,12 +363,9 @@ class RoomController(Hass, Mqtt):
self.log(f'Turned on scene: {scene}')
elif isinstance(scene, dict):
# makes setting the state to 'on' optional in the yaml definition
for entity, settings in scene.items():
if 'state' not in settings:
scene[entity]['state'] = 'on'
self.call_service('scene/apply', entities=scene, transition=0)
kwargs = ApplyScene(entities=scene, transition=0).model_dump(exclude_none=True)
self.log('Validated scene JSON', level='DEBUG')
self.call_service('scene/apply', **kwargs)
if self.logger.isEnabledFor(logging.INFO):
self.log('Applied scene:')
console.print(scene)