Async Apps in AppDaemon Example

test.py

import appdaemon.plugins.hass.hassapi as hass
import asyncio


class TestAllAsync(hass.Hass):

    async def terminate(self):
        self.log('i will die in 3 seconds')
        await asyncio.sleep(3)
        self.log('i am dead')

    async def initialize(self):
        self.log('i will init in 3 seconds')
        await asyncio.sleep(3)
        await self.run_in(self.report_cb, 1)
        self.log('i am init')

    async def report_cb(self, kwargs):
        self.log('report')
        await self.fire_event('TESTALLASYNC')
        await self.run_in(self.report_cb, 1)


class TestAsync(hass.Hass):

    def initialize(self):
        self.log('init')
        self.listen_event(self.async_event_cb)
        self.run_coroutine(self.wait(3), callback=self.coro_cb)

    async def wait(self, delay):
        self.log('start wait {}'.format(delay))
        ret = await asyncio.sleep(delay, result="slept {}s".format(delay))
        self.log('end wait {}'.format(delay))
        await self.fire_event('TESTNONASYNC FROMASYNC')
        return ret

    async def async_event_cb(self, event_name, data, kwargs):
        self.log('async event: {}'.format(event_name))

    def coro_cb(self, res, kwargs):
        self.log(res)
        self.fire_event('TESTNONASYNC FROMNONASYNC')
        self.run_coroutine(self.wait(3), callback=self.coro_cb)


class TestNonAsync(hass.Hass):
    def initialize(self):
        self.log('init')
        self.run_in(self.run_in_cb, 1)

    def run_in_cb(self, kwargs):
        self.log('run in cb')
        self.run_in(self.run_in_cb, 1)
        self.fire_event('TESTNONASYNC')