diff options
| -rw-r--r-- | scripts/esphomeapi.py | 238 |
1 files changed, 112 insertions, 126 deletions
diff --git a/scripts/esphomeapi.py b/scripts/esphomeapi.py index 0b662b7..b3c684b 100644 --- a/scripts/esphomeapi.py +++ b/scripts/esphomeapi.py @@ -3,173 +3,159 @@ import os import sys -import json from datetime import datetime -import aioesphomeapi +from aioesphomeapi import APIClient, ReconnectLogic, SensorState, APIConnectionError import asyncio +import argparse import logging import colorlog import zeroconf -#from retry import retry import common -# Logging -handler = colorlog.StreamHandler() -handler.setFormatter(colorlog.ColoredFormatter( - "%(log_color)s%(asctime)s - %(levelname)s - %(message)s", - log_colors={ - 'DEBUG': 'light_black', - 'INFO': 'cyan', - 'WARNING': 'yellow', - 'ERROR': 'red', - 'CRITICAL': 'red,bg_white' - })) -log = colorlog.getLogger('example') -log.setLevel(logging.DEBUG) -log.addHandler(handler) -log -log.critical("critical") -log.error("error") -log.warning("warning") -log.info("info") -log.debug("debug") - - - -#sys.exit() -# -#log = logging.Logger(__name__) -## Stream/console output -##log.handler = logging.StreamHandler(sys.stdout) -##log.handler.setLevel(logging.DEBUG) -##formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") -##log.handler.setFormatter(formatter) -# -#ch = logging.StreamHandler() -#ch.setLevel(logging.DEBUG) -#ch.setFormatter(CustomFormatter()) -# -#log.addHandler(ch) -# -# -##log.handler.setFormatter(CustomFormatter) -##log.addHandler(log.handler) -# -#log.critical("critical") -#log.error("error") -#log.warning("warning") -#log.info("info") -#log.debug("debug") -# -##sys.exit() - - +sleepsec = 60 noise_psk = os.environ['el_esphome_api_psk'] -#@retry(aioesphomeapi.core.SocketAPIError, tries=10, delay=1, backoff=2) -async def main(): +async def main(args): """Connect to an ESPHome device and get details.""" - log.info('function main()') + log.debug('function main()') # Establish connection - api = aioesphomeapi.APIClient("airgradient-pro.hissig.org", 6053, password='', noise_psk=noise_psk) + api = APIClient( + address=args.address, + port=6053, + password='', + noise_psk=noise_psk) - log.info('Connecting') - await api.connect(login=True) - print('api.api_version') - print(api.api_version) - print() + 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_info = await api.device_info() - print('device_info') - print(device_info) - print() + device = vars(await api.device_info()) + log.info('Device name: ' + device['name']) log.debug('Getting sensors') - sensors, _ = await api.list_entities_services() - sensor_by_keys = dict((sensor.key, sensor.name) for sensor in sensors) - - # List all entities of the device - print('sensors') - print(sensors) - print() - import pprint - pp = pprint.PrettyPrinter(indent=4) - pp.pprint(sensors) - print() - - print('LOOOL') - mysensors = {} - for sensor in sensors: - mysensors[sensor.key] = vars(sensor) - print(mysensors[sensor.key]) - print() - - print('mysensors') - print(mysensors) - print() + rawsensors, _ = await api.list_entities_services() + sensors = {} + for sensor in rawsensors: + sensors[sensor.key] = vars(sensor) - print('for senso in mysensors') - for senso in mysensors: - print('senso') - print(mysensors[senso]['object_id']) - - - - print('sensor_by_keys') - print(sensor_by_keys) - print() + if log.isEnabledFor(logging.DEBUG): + from pprint import pformat + log.debug('Sensors: \n' + pformat(sensors)) log.info('Disconnecting') await api.disconnect() - def cb(state): - if type(state) == aioesphomeapi.SensorState and state.missing_state == False: - log.debug('function cb(state)') - log.debug('state: ' + str(state)) - log.debug('sensor: ' + str(mysensors[state.key])) + 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 mysensors[state.key]: - decimals = mysensors[state.key]['accuracy_decimals'] - log.debug('Accuracy decimals: ' + str(decimals)) + if 'accuracy_decimals' in sensor: + decimals = sensor['accuracy_decimals'] value = round(value, decimals) if decimals > 0 else round(value) - if 'unit_of_measurement' in mysensors[state.key]: - value = str(value) + str(mysensors[state.key]['unit_of_measurement']) - print(mysensors[state.key]['name'] + ' - ' + str(value)) - print() + unit = sensor['unit_of_measurement'] if 'unit_of_measurement' in sensor else '' + device_class = sensor['device_class'] if 'device_class' in sensor else '' + + sql = """INSERT INTO states + (sensor_id, + value, + device_class, + unit, + time) + SELECT sensors.id, %s, %s, %s, %s + FROM sensors + WHERE sensors.name = %s;""" + values = (value, + device_class, + unit, + datetime.now(), + device['friendly_name']) + + log.info(sensor['name'] + ' ' + sensor['device_class'] + ' - ' + str(value) + str(unit)) + common.dbi(sql, values, verbose=True) async def on_connect() -> None: log.debug('function on_connect()') + log.info('Connected to API') try: - await api.subscribe_states(cb) - except APIConnectionError: + 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") + log.warning("Disconnected from API") - logic = aioesphomeapi.ReconnectLogic( + reconnect = ReconnectLogic( client=api, on_connect=on_connect, on_disconnect=on_disconnect, zeroconf_instance=zeroconf.Zeroconf() ) - await logic.start() + await reconnect.start() try: - log.debug('asyncio.Event().wait()') - await asyncio.Event().wait() # sleep - except: - log.debug('logic.stop()') - await logic.stop() - -log.info('asyncio.run(main())') -asyncio.run(main()) - -log.warning('Bottom of script. Exiting.') + 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.INFO) + #logging.basicConfig(format="%(name)s: %(levelname)s %(message)s") + + 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) |