path: root/scripts/esphomeapi.py

#!/usr/bin/env python3
""" Check on esphome-devices """

import argparse
import asyncio
import logging
import os
import sys

import common

# import colorlog
import zeroconf
from aioesphomeapi import APIClient, APIConnectionError, ReconnectLogic, SensorState

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="",
        client_info="esphomeapi.py",
        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 is 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(
                device["friendly_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")
        await asyncio.sleep(sleepsec)

    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)