124 lines
3.9 KiB
Python
Executable File
124 lines
3.9 KiB
Python
Executable File
import json
|
|
|
|
from appdaemon.entity import Entity
|
|
from appdaemon.plugins.hass.hassapi import Hass
|
|
from appdaemon.plugins.mqtt.mqttapi import Mqtt
|
|
|
|
|
|
class SleepSetter(Hass, Mqtt):
|
|
def initialize(self):
|
|
assert self.entity_exists(entity_id=self.variable), f'{self.variable} does not exist'
|
|
self.listen_state(callback=self.handle_state, entity_id=self.variable)
|
|
self.setup_buttons()
|
|
|
|
def setup_buttons(self):
|
|
if isinstance(self.button, list):
|
|
for button in self.button:
|
|
self.setup_button(button)
|
|
else:
|
|
self.setup_button(button)
|
|
|
|
def setup_button(self, name: str):
|
|
topic = f'zigbee2mqtt/{name}'
|
|
self.mqtt_subscribe(topic, namespace='mqtt')
|
|
self.listen_event(self.handle_button, "MQTT_MESSAGE", topic=topic, namespace='mqtt', button=name)
|
|
self.log(f'Subscribed: {topic}')
|
|
|
|
@property
|
|
def button(self) -> str:
|
|
return self.args['button']
|
|
|
|
@property
|
|
def scene(self) -> str:
|
|
res = self.args['scene']
|
|
if not res.startswith('scene.'):
|
|
res = f'scene.{res}'
|
|
return res
|
|
|
|
@property
|
|
def variable(self) -> str:
|
|
return self.args['variable']
|
|
|
|
@property
|
|
def variable_entity(self) -> Entity:
|
|
return self.get_entity(self.variable)
|
|
|
|
@property
|
|
def state(self) -> bool:
|
|
return self.variable_entity.get_state('state') == 'on'
|
|
|
|
@state.setter
|
|
def state(self, new: bool):
|
|
state = 'on' if bool(new) else 'off'
|
|
self.log(f'Setting {self.variable} to {state}')
|
|
return self.variable_entity.set_state(state=state)
|
|
|
|
@property
|
|
def sun_elevation(self) -> float:
|
|
state = self.get_state('sun.sun', 'elevation')
|
|
assert isinstance(state, float)
|
|
return state
|
|
|
|
def handle_state(self, entity, attribute, old, new, kwargs):
|
|
self.log(f'new state: {self.state}')
|
|
if self.state:
|
|
self.all_off()
|
|
try:
|
|
self.call_service('scene/turn_on', entity_id=self.scene)
|
|
except:
|
|
return
|
|
else:
|
|
self.log(f'Turned on scene: {self.scene}')
|
|
# self.turn_on(self.scene)
|
|
|
|
def handle_button(self, event_name, data, kwargs):
|
|
topic = data['topic']
|
|
# self.log(f'Button event for: {topic}')
|
|
try:
|
|
payload = json.loads(data['payload'])
|
|
action = payload['action']
|
|
except json.JSONDecodeError:
|
|
self.log(f'Error decoding JSON from {data["payload"]}', level='ERROR')
|
|
except KeyError as e:
|
|
return
|
|
else:
|
|
self.handle_action(action)
|
|
|
|
def handle_action(self, action: str):
|
|
if action == '':
|
|
return
|
|
|
|
if action == 'hold':
|
|
self.log(f' {action.upper()} '.center(50, '='))
|
|
self.state = True
|
|
elif action == 'double':
|
|
self.log(f' {action.upper()} '.center(50, '='))
|
|
self.state = not self.state
|
|
self.on_apps()
|
|
|
|
def all_off(self):
|
|
self.log(f'Deactivating apps')
|
|
for app_name in self.args['off_apps']:
|
|
try:
|
|
self.get_app(app_name).deactivate(cause='sleep setter')
|
|
except:
|
|
self.log(f'Failed to deactivate {app_name}')
|
|
continue
|
|
|
|
self.log(f'Turning off entities')
|
|
for entity in self.args['off_entities']:
|
|
try:
|
|
self.turn_off(entity)
|
|
except:
|
|
self.log(f'Failed to turn off {entity}')
|
|
continue
|
|
|
|
def on_apps(self):
|
|
if (on_apps := self.args.get('on_apps', None)) is not None:
|
|
for app_name in on_apps:
|
|
try:
|
|
self.get_app(app_name).activate(kwargs={'cause': 'sleep setter'})
|
|
except:
|
|
return
|
|
else:
|
|
self.log(f'Activated {app_name}') |