mirror of
https://github.com/Tediore/battery2mqtt.git
synced 2025-04-12 03:13:43 +02:00
98 lines
4.2 KiB
Python
98 lines
4.2 KiB
Python
import os
|
|
import json
|
|
from sys import int_info
|
|
from time import sleep
|
|
import paho.mqtt.client as mqtt
|
|
|
|
MQTT_HOST = os.getenv('MQTT_HOST')
|
|
MQTT_PORT = int(os.getenv('MQTT_PORT', 1883))
|
|
MQTT_USER = os.getenv('MQTT_USER')
|
|
MQTT_PASSWORD = os.getenv('MQTT_PASSWORD')
|
|
MQTT_QOS = int(os.getenv('MQTT_QOS', 1))
|
|
MQTT_TOPIC = os.getenv('MQTT_TOPIC', 'server')
|
|
INTERVAL = int(os.getenv('INTERVAL', 60))
|
|
MONITORED_CONDITIONS = os.environ.get('MONITORED_CONDITIONS','status,capacity,energy_now,energy_full,energy_full_design,power_now,voltage_now')
|
|
SHOW_UNITS = int(os.getenv('SHOW_UNITS', 1))
|
|
BATTERY_HEALTH = int(os.getenv('BATTERY_HEALTH', 1))
|
|
TIME_REMAINING = int(os.getenv('TIME_REMAINING', 1))
|
|
AC_ADAPTER = int(os.getenv('AC_ADAPTER', 0))
|
|
|
|
client = mqtt.Client("battery2mqtt")
|
|
client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
|
|
|
|
monitored_conditions = MONITORED_CONDITIONS.split(',')
|
|
|
|
path = "/sys/class/power_supply/"
|
|
|
|
dirs = os.listdir(path)
|
|
|
|
payload = {}
|
|
health_calc = {}
|
|
time_remaining = {}
|
|
|
|
while True:
|
|
for dir in dirs:
|
|
if AC_ADAPTER == 1:
|
|
if dir.startswith('AC'):
|
|
try:
|
|
with open(path + dir + '/online', 'r') as file:
|
|
payload['ac_adapter'] = 'online' if int(file.read()) == 1 else 'offline'
|
|
except:
|
|
pass
|
|
|
|
for name in monitored_conditions:
|
|
try:
|
|
with open(path + dir + '/' + name, 'r') as file:
|
|
if name in ['alarm', 'capacity', 'cycle_count', 'online', 'present']:
|
|
payload[name] = int(file.read().replace('\n',''))
|
|
elif name.startswith('voltage') or name.startswith('energy') or name.startswith('power'):
|
|
if name.startswith('voltage'):
|
|
unit = ' V' if SHOW_UNITS == 1 else ''
|
|
elif name.startswith('energy'):
|
|
unit = ' Wh' if SHOW_UNITS == 1 else ''
|
|
else:
|
|
unit = ' W' if SHOW_UNITS == 1 else ''
|
|
|
|
if SHOW_UNITS == 1:
|
|
payload[name] = str(round(float(file.read().replace('\n','')) / 1000000,2)) + unit
|
|
else:
|
|
payload[name] = round(float(file.read().replace('\n','')) / 1000000,2)
|
|
else:
|
|
payload[name] = file.read().replace('\n','')
|
|
except:
|
|
payload[name] = "condition not found"
|
|
|
|
if BATTERY_HEALTH == 1:
|
|
unit = ' %' if SHOW_UNITS == 1 else ''
|
|
try:
|
|
for name in ['energy_full_design', 'energy_full']:
|
|
with open(path + dir + '/' + name, 'r') as file:
|
|
health_calc[name] = int(file.read().replace('\n',''))
|
|
if SHOW_UNITS == 1:
|
|
payload['battery_health'] = str(round((health_calc['energy_full'] / health_calc['energy_full_design']) * 100,2)) + unit
|
|
else:
|
|
payload['battery_health'] = round((health_calc['energy_full'] / health_calc['energy_full_design']) * 100,2)
|
|
except:
|
|
pass
|
|
|
|
if TIME_REMAINING == 1:
|
|
unit = ' h' if SHOW_UNITS == 1 else ''
|
|
try:
|
|
for name in ['energy_now', 'power_now']:
|
|
with open(path + dir + '/' + name, 'r') as file:
|
|
time_remaining[name] = int(file.read().replace('\n',''))
|
|
if SHOW_UNITS == 1:
|
|
payload['time_remaining'] = str(round((time_remaining['energy_now'] / time_remaining['power_now']),2) if round((time_remaining['energy_now'] / time_remaining['power_now']),2) < 24 else '> 24') + unit
|
|
else:
|
|
payload['time_remaining'] = round((time_remaining['energy_now'] / time_remaining['power_now']),2) if round((time_remaining['energy_now'] / time_remaining['power_now']),2) < 24 else '> 24'
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
client.connect(MQTT_HOST)
|
|
client.publish("battery2mqtt/" + MQTT_TOPIC + '/' + dir, json.dumps(payload), qos=MQTT_QOS, retain=False)
|
|
except:
|
|
print('Message send failed.')
|
|
|
|
sleep(INTERVAL)
|