aioesphomeapi minimal examples

esphome-print-async.py

#!/usr/bin/env python3

import aioesphomeapi
import asyncio

async def main():
    loop = asyncio.get_running_loop()
    cli = aioesphomeapi.APIClient(loop, "foobarhostname", 6053, "foobarpassword")

    await cli.connect(login=True)

    def cb(state):
        #if type(state) == aioesphomeapi.BinarySensorState:
        print(state)

    await cli.subscribe_states(cb)

    l = await cli.list_entities_services()
    print(l)

loop = asyncio.get_event_loop()
try:
    asyncio.ensure_future(main())
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    loop.close()

esphome-print-simple.py

#!/usr/bin/env python3

import aioesphomeapi
import asyncio

loop = asyncio.new_event_loop()

cli = aioesphomeapi.APIClient(loop, "foobarhostname", 6053, "foobarpassword")

loop.run_until_complete(cli.connect(login=True))

def cb(state):
    print(state)

loop.run_until_complete(cli.subscribe_states(cb))

loop.run_forever()

esphome-sensor-influxdb.py

#!/usr/bin/env python3

import aioesphomeapi
import asyncio

from influxdb import InfluxDBClient
from influxdb import SeriesHelper

influxclient = InfluxDBClient('influxhost', 8086, database="dbname")

class EspSeries(SeriesHelper):
    class Meta:
        client = influxclient
        series_name = 'events.stats.{sensor}'
        fields = [ 'value' ]
        tags = [ 'sensor' ]
        bulk_size = 1
        autocommit = True

async def main():
    loop = asyncio.get_running_loop()
    cli = aioesphomeapi.APIClient(loop, "foobarhostname", 6053, "foobarpassword")

    await cli.connect(login=True)

    sensors, services = await cli.list_entities_services()
    sensor_by_keys = dict((sensor.key, sensor.name) for sensor in sensors)

    def cb(state):
        if type(state) == aioesphomeapi.SensorState:
            print(state)
            print(sensor_by_keys[state.key])
            print(state.state)
            EspSeries(sensor=sensor_by_keys[state.key], value=state.state)

    await cli.subscribe_states(cb)


loop = asyncio.get_event_loop()
try:
    asyncio.ensure_future(main())
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    loop.close()