AppDaemon App: CoverController

cover_controler.yaml

cover_controller_living:
  module: cover_controller
  class: CoverController
  cover_entities:
    - cover.living
  open_criteria:
    - binary_sensor.home
    - binary_sensor.daylight
    - schedule.awake
  override_position:
    - binary_sensor.not_home
    - binary_sensor.daylight
  close_position: 40
  close_prevention:
    - binary_sensor.living_patiodoor

cover_controller.py

import appdaemon.plugins.hass.hassapi as hass

class CoverController(hass.Hass):
    def initialize(self):
        self.cover_entities = self.args["cover_entities"]
        self.open_criteria = self.args.get("open_criteria", [])
        self.close_prevention = self.args.get("close_prevention", [])
        self.override_position = self.args.get("override_position", [])
        self.positions = {
            "closed": self.args.get("close_position", None),
            "opened": self.args.get("open_position", None)
        }
        self.state_name = f"sensor.{self.name}"
        self.requested_state = None

        for entity in self.open_criteria + self.close_prevention:
            self.listen_state(self.states_changed, entity)

        for entity in self.cover_entities:
            self.listen_state(self.cover_changed, entity)

        self.output_state = "open" if self.is_open() else "close"
        self.update()

    def states_changed(self, entity, attribute, old, new, kwargs):
        self.update()

    async def cover_changed(self, entity, attribute, old, new, kwargs):
        if self.requested_state is None or new != self.requested_state or self.is_position_overriden():
            return

        position = self.positions.get(new)
        if position is not None:
            await self.sleep(5)
            self.call_service("cover/set_cover_position", entity_id=entity, position=position)

    def update(self):
        cover_state = "open" if self.is_open() else "close"

        if cover_state != self.output_state:
            self.output_state = cover_state
            action = "open_cover" if cover_state == "open" else "close_cover"
            self.call_service(f"cover/{action}", entity_id=self.cover_entities)
            self.requested_state = "opened" if cover_state == "open" else "closed"
        else:
            self.requested_state = None

    def is_open(self):
        if all(self.get_state(entity) == "on" for entity in self.open_criteria):
            return not any(self.get_state(entity) == "on" for entity in self.close_prevention)
        return False

    def is_position_overriden(self):
        return any(self.get_state(entity) == "on" for entity in self.override_position)

    @property
    def output_state(self):
        return self.get_state(self.state_name)

    @output_state.setter
    def output_state(self, value):
        self.set_state(self.state_name, state=value)