diff options
Diffstat (limited to 'scripts/esphomeapi.py')
| -rw-r--r-- | scripts/esphomeapi.py | 90 |
1 files changed, 54 insertions, 36 deletions
diff --git a/scripts/esphomeapi.py b/scripts/esphomeapi.py index 5fcbe37..390fbd5 100644 --- a/scripts/esphomeapi.py +++ b/scripts/esphomeapi.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -''' Check on esphome-devices ''' +""" Check on esphome-devices """ import argparse import asyncio @@ -9,40 +9,42 @@ import sys import common -#import colorlog +# import colorlog import zeroconf from aioesphomeapi import APIClient, APIConnectionError, ReconnectLogic, SensorState sleepsec = 60 -noise_psk = os.environ['el_esphome_api_psk'] +noise_psk = os.environ["el_esphome_api_psk"] + async def main(args): """Connect to an ESPHome device and get details.""" - log.debug('function main()') + log.debug("function main()") # Establish connection api = APIClient( address=args.address, port=6053, - password='', + password="", client_info="esphomeapi.py", - noise_psk=noise_psk) + noise_psk=noise_psk, + ) - log.debug('Connecting') + 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)) + log.info("Connected. Api version: " + str(api.api_version)) # Show device details - log.debug('Getting device info') + log.debug("Getting device info") device = vars(await api.device_info()) - log.info('Device name: ' + device['name']) + log.info("Device name: " + device["name"]) - log.debug('Getting sensors') + log.debug("Getting sensors") rawsensors, _ = await api.list_entities_services() sensors = {} for sensor in rawsensors: @@ -50,29 +52,41 @@ async def main(args): if log.isEnabledFor(logging.DEBUG): from pprint import pformat - log.debug('Sensors: \n' + pformat(sensors)) - log.info('Disconnecting') + log.debug("Sensors: \n" + pformat(sensors)) + + log.info("Disconnecting") await api.disconnect() def callback(state): if type(state) == SensorState and state.missing_state is False: - log.debug('function callback(' + str(state) + ')') + log.debug("function callback(" + str(state) + ")") sensor = sensors[state.key] value = state.state - if 'accuracy_decimals' in sensor: - decimals = sensor['accuracy_decimals'] + 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(device['friendly_name'], value, device_class, unit, verbose=True) + 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( + device["friendly_name"], value, device_class, unit, verbose=True + ) async def on_connect() -> None: - log.debug('function on_connect()') - log.info('Connected to API') + log.debug("function on_connect()") + log.info("Connected to API") try: await api.subscribe_states(callback) except APIConnectionError as e: @@ -83,7 +97,7 @@ async def main(args): await api.disconnect() async def on_disconnect() -> None: - log.debug('function on_disconnect()') + log.debug("function on_disconnect()") log.warning("Disconnected from API") await asyncio.sleep(sleepsec) @@ -91,14 +105,14 @@ async def main(args): client=api, on_connect=on_connect, on_disconnect=on_disconnect, - zeroconf_instance=zeroconf.Zeroconf() + zeroconf_instance=zeroconf.Zeroconf(), ) await reconnect.start() try: while True: try: - log.debug('Sleep for ' + str(sleepsec) + 's') + log.debug("Sleep for " + str(sleepsec) + "s") await asyncio.sleep(sleepsec) except Exception as e: log.error("catched exception - %s", e) @@ -114,8 +128,8 @@ async def main(args): if __name__ == "__main__": # Logging - #handler = colorlog.StreamHandler() - #handler.setFormatter(colorlog.ColoredFormatter( + # handler = colorlog.StreamHandler() + # handler.setFormatter(colorlog.ColoredFormatter( # "%(log_color)s%(levelname)s - %(message)s {%(filename)s:%(lineno)d}", # log_colors={ # 'DEBUG': 'light_black', @@ -124,26 +138,30 @@ if __name__ == "__main__": # 'ERROR': 'red', # 'CRITICAL': 'red,bg_white' # })) - #log = colorlog.getLogger(__name__) - #log.setLevel(logging.WARNING) - #log.addHandler(handler) + # 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") + 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))') + log.debug("asyncio.run(main(args))") asyncio.run(main(args)) -print('Bottom of script. Exiting.') +print("Bottom of script. Exiting.") sys.exit(0) |