fpletz · February 23, 2024 00:47 · klattimer · Dec 21, 2021
diff --git a/esphome-print-async.py b/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()
diff --git a/esphome-print-simple.py b/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()
diff --git a/esphome-sensor-influxdb.py b/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()
	#!/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()