#!/usr/bin/env python3
''' Check on esphome-devices '''
import os
import sys
from datetime import datetime
from aioesphomeapi import APIClient, ReconnectLogic, SensorState, APIConnectionError
import asyncio
import argparse
import logging
#import colorlog
import zeroconf
import common
sleepsec = 60
noise_psk = os.environ['el_esphome_api_psk']
async def main(args):
"""Connect to an ESPHome device and get details."""
log.debug('function main()')
# Establish connection
api = APIClient(
address=args.address,
port=6053,
password='',
noise_psk=noise_psk)
log.debug('Connecting')
try:
await api.connect(login=True)
except APIConnectionError as e:
log.error("esphome api connection error - %s", e)
sys.exit(1)
log.info('Connected. Api version: ' + str(api.api_version))
# Show device details
log.debug('Getting device info')
device = vars(await api.device_info())
log.info('Device name: ' + device['name'])
log.debug('Getting sensors')
rawsensors, _ = await api.list_entities_services()
sensors = {}
for sensor in rawsensors:
sensors[sensor.key] = vars(sensor)
if log.isEnabledFor(logging.DEBUG):
from pprint import pformat
log.debug('Sensors: \n' + pformat(sensors))
log.info('Disconnecting')
await api.disconnect()
def callback(state):
if type(state) == SensorState and state.missing_state == False:
log.debug('function callback(' + str(state) + ')')
sensor = sensors[state.key]
value = state.state
if 'accuracy_decimals' in sensor:
decimals = sensor['accuracy_decimals']
value = round(value, decimals) if decimals > 0 else round(value)
unit = sensor['unit_of_measurement'] if 'unit_of_measurement' in sensor else ''
device_class = sensor['device_class'] if 'device_class' in sensor else ''
log.info(sensor['name'] + ' ' + sensor['device_class'] + ' - ' + str(value) + str(unit))
common.statein(sensor['name'], value, device_class, unit, verbose=True)
async def on_connect() -> None:
log.debug('function on_connect()')
log.info('Connected to API')
try:
await api.subscribe_states(callback)
except APIConnectionError as e:
log.error("esphome api connection error - %s", e)
await api.disconnect()
except Exception as e:
log.error("catched exception - %s", e)
await api.disconnect()
async def on_disconnect() -> None:
log.debug('function on_disconnect()')
log.warning("Disconnected from API")
reconnect = ReconnectLogic(
client=api,
on_connect=on_connect,
on_disconnect=on_disconnect,
zeroconf_instance=zeroconf.Zeroconf()
)
await reconnect.start()
try:
while True:
try:
log.debug('Sleep for ' + str(sleepsec) + 's')
await asyncio.sleep(sleepsec)
except Exception as e:
log.error("catched exception - %s", e)
except KeyboardInterrupt:
log.error("Keyboard interrupt")
pass
except KeyboardInterrupt:
log.error("Keyboard interrupt")
await reconnect.stop()
finally:
log.info("Closing loop.")
if __name__ == "__main__":
# Logging
#handler = colorlog.StreamHandler()
#handler.setFormatter(colorlog.ColoredFormatter(
# "%(log_color)s%(levelname)s - %(message)s {%(filename)s:%(lineno)d}",
# log_colors={
# 'DEBUG': 'light_black',
# 'INFO': 'cyan',
# 'WARNING': 'yellow',
# 'ERROR': 'red',
# 'CRITICAL': 'red,bg_white'
# }))
#log = colorlog.getLogger(__name__)
#log.setLevel(logging.WARNING)
#log.addHandler(handler)
log = logging.getLogger(__name__)
log.setLevel(logging.WARNING)
logging.basicConfig(format="%(levelname)s - %(message)s {%(filename)s:%(lineno)d}")
parser = argparse.ArgumentParser(description="Connect to an esphome-device and access the native api")
parser.add_argument( "-a", "--address", help="Address of esp-device to connect to")
parser.add_argument( "-v", "--verbose", help="Set logging to debug mode", action="store_true")
args = parser.parse_args()
# Verbose logging?
if args.verbose:
log.setLevel(logging.DEBUG)
log.debug('asyncio.run(main(args))')
asyncio.run(main(args))
print('Bottom of script. Exiting.')
sys.exit(0)
