ad-nix/apps/sleep.py

from datetime import datetime, timedelta
import json

from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from appdaemon.adbase import ADBase


class SleepTV(ADBase):
    handle: str = None

    def initialize(self):
        self.adapi = self.get_ad_api()
        self.adapi.set_log_level('DEBUG')
        self.sleep_time.listen_state(self.handle_sleep_time_change)

    @property
    def sleep_time(self) -> Entity:
        return self.adapi.get_entity(self.args['sleep_time'])
    
    @property
    def tv(self) -> Entity:
        return self.adapi.get_entity(self.args['tv'])

    def handle_sleep_time_change(self, entity: str, attribute: str, old: str, new: str, **kwargs):
        now = self.adapi.get_now()
        dt = datetime.strptime(new, '%H:%M:%S')
        dt = datetime.combine(now.date(), dt.time())

        if dt.time() < now.time():
            dt += timedelta(days=1)

        self.adapi.cancel_timer(self.handle, silent=True)
        self.handle = self.adapi.run_at(lambda **kwargs: self.tv.turn_off(), dt)


class SleepSetter(Hass, Mqtt):
    def initialize(self):
        assert self.entity_exists(entity_id=self.variable), f'{self.variable} does not exist'
        self.variable_entity.listen_state(self.handle_state)
        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 Exception:
                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:
            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('Deactivating apps')
        for app_name in self.args['off_apps']:
            try:
                self.get_app(app_name).deactivate(cause='sleep setter')
            except Exception:
                self.log(f'Failed to deactivate {app_name}')
                continue

        self.log('Turning off entities')
        for entity in self.args['off_entities']:
            try:
                self.turn_off(entity)
            except Exception:
                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 Exception:
                    return
                else:
                    self.log(f'Activated {app_name}')