145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
from dataclasses import dataclass, field
|
|
from datetime import timedelta
|
|
from typing import List
|
|
|
|
from appdaemon.adapi import ADAPI
|
|
from appdaemon.entity import Entity
|
|
|
|
from appdaemon.plugins.hass.hassapi import Hass
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerBase(Hass):
|
|
entities: List[Entity]
|
|
|
|
def initialize(self):
|
|
# assign fields
|
|
for arg, val in self.args.items():
|
|
if arg not in ['class', 'module']:
|
|
setattr(self, arg, val)
|
|
self.log(f'Set {arg} to {val}')
|
|
|
|
for entity in self.entities:
|
|
assert self.entity_exists(entity), f'{entity} does not exist'
|
|
|
|
self.entities = [self.get_entity(e) for e in self.entities]
|
|
|
|
self.log(f'Initialized controller for {[e.friendly_name for e in self.entities]}')
|
|
|
|
@dataclass(init=False)
|
|
class ControllerRoom(ControllerBase):
|
|
def initialize(self):
|
|
super().initialize()
|
|
self.register_service(f'{self.name}/activate', self.activate)
|
|
self.register_service(f'{self.name}/deactivate', self.deactivate)
|
|
|
|
def activate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
|
|
for entity in self.entities:
|
|
self.log(f'Turning on {entity.name}')
|
|
entity.turn_on()
|
|
|
|
def deactivate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
|
|
for entity in self.entities:
|
|
self.log(f'Turning off {entity.name}')
|
|
entity.turn_off()
|
|
|
|
@property
|
|
def state(self) -> bool:
|
|
return any([e.get_state() == 'on' for e in self.entities])
|
|
|
|
@dataclass(init=False)
|
|
class ControllerMotion(ControllerBase):
|
|
room: ControllerRoom
|
|
off_duration: timedelta
|
|
|
|
def initialize(self):
|
|
super().initialize()
|
|
# self.log('Motion Controller init')
|
|
|
|
# convert room to App
|
|
self.room: ControllerRoom = self.get_app(self.room)
|
|
|
|
# convert off_duration
|
|
try:
|
|
hours, minutes, seconds = map(int, self.args['off_duration'].split(':'))
|
|
self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
|
except Exception:
|
|
self.off_duration = timedelta()
|
|
|
|
self.sync_state()
|
|
|
|
self.listen_state(self.sync_state, [e.entity_id for e in self.entities])
|
|
|
|
@property
|
|
def current_state(self) -> bool:
|
|
return any(e.get_state() == 'on' for e in self.entities)
|
|
|
|
def sync_state(self,
|
|
entity=None,
|
|
attribute=None,
|
|
old=None,
|
|
new=None,
|
|
kwargs=None):
|
|
self.log(f'Syncing state, current state: {self.current_state}')
|
|
if self.current_state:
|
|
self.room.activate()
|
|
self.listen_motion_off()
|
|
else:
|
|
self.room.deactivate()
|
|
self.listen_motion_on()
|
|
|
|
def listen_motion_on(self):
|
|
self.listen_state(
|
|
callback=self.callback_motion_on,
|
|
entity_id=[e.entity_id for e in self.entities],
|
|
new='on',
|
|
oneshot=True
|
|
)
|
|
self.log(f'Waiting for motion on {[e.friendly_name for e in self.entities]} to turn on room {self.room.name}')
|
|
|
|
def listen_motion_off(self):
|
|
self.listen_state(
|
|
callback=self.callback_motion_off,
|
|
entity_id=[e.entity_id for e in self.entities],
|
|
new='off',
|
|
duration=self.off_duration.total_seconds(),
|
|
oneshot=True
|
|
)
|
|
self.log(f'Waiting for motion off {[e.friendly_name for e in self.entities]} for {self.off_duration}')
|
|
|
|
def callback_motion_on(self, entity, attribute, old, new, kwargs):
|
|
self.log(f'Motion detected on {self.friendly_name(entity)}')
|
|
self.room.activate()
|
|
|
|
def callback_motion_off(self, entity, attribute, old, new, kwargs):
|
|
self.log(f'Motion stopped on {self.friendly_name(entity)} for {self.off_duration}')
|
|
self.room.deactivate()
|
|
|
|
@dataclass(init=False)
|
|
class ControllerButton(Hass):
|
|
room: ControllerRoom
|
|
buttons: List[str]
|
|
|
|
def initialize(self):
|
|
self.buttons = self.args['buttons']
|
|
|
|
# convert room to App
|
|
self.room: ControllerRoom = self.get_app(self.args['room'])
|
|
|
|
for button in self.buttons:
|
|
self.listen_event(
|
|
self.callback_button,
|
|
event='deconz_event',
|
|
id=button,
|
|
)
|
|
self.log(f'Listening to presses on button ID={button}')
|
|
|
|
def callback_button(self, event_name, data, kwargs):
|
|
# single press
|
|
if data['event'] == 1002:
|
|
self.log(f"Single press: {data['id']}")
|
|
if self.room.state:
|
|
self.room.deactivate()
|
|
else:
|
|
self.room.activate()
|
|
|