blob: 0b662b7ddd3a1cd543f63b8bb7577937624f9c12 (
plain) (
tree)
|
|
#!/usr/bin/env python3
''' Check on esphome-devices '''
import os
import sys
import json
from datetime import datetime
import aioesphomeapi
import asyncio
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()
noise_psk = os.environ['el_esphome_api_psk']
#@retry(aioesphomeapi.core.SocketAPIError, tries=10, delay=1, backoff=2)
async def main():
"""Connect to an ESPHome device and get details."""
log.info('function main()')
# Establish connection
api = aioesphomeapi.APIClient("airgradient-pro.hissig.org", 6053, password='', noise_psk=noise_psk)
log.info('Connecting')
await api.connect(login=True)
print('api.api_version')
print(api.api_version)
print()
# Show device details
log.debug('Getting device info')
device_info = await api.device_info()
print('device_info')
print(device_info)
print()
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()
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()
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]))
value = state.state
if 'accuracy_decimals' in mysensors[state.key]:
decimals = mysensors[state.key]['accuracy_decimals']
log.debug('Accuracy decimals: ' + str(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()
async def on_connect() -> None:
log.debug('function on_connect()')
try:
await api.subscribe_states(cb)
except APIConnectionError:
await api.disconnect()
async def on_disconnect() -> None:
log.debug('function on_disconnect()')
log.warning("Disconnected")
logic = aioesphomeapi.ReconnectLogic(
client=api,
on_connect=on_connect,
on_disconnect=on_disconnect,
zeroconf_instance=zeroconf.Zeroconf()
)
await logic.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.')
sys.exit(0)
|
