Replace MQTT output with direct InfluxDB writes
Polls now cover meter 1, battery 1 and storage control blocks, plus inverter event registers for off-grid detection (vendor event bit 0). Register layout validated by probing a SE10K-RWB48BFN4 with fw 0004.0025.0011; battery and storage floats are word-swapped on this device. Failed Influx writes spool to disk and replay on the next success (Influx reached over WAN). Modbus blocks that fail to read are skipped for that cycle; the flaky gateway must not kill the daemon. Battery lifetime counters at 0xE176/0xE17A hang the device and are never read. BREAKING CHANGE: config.yaml 'mqtt:' section replaced by 'influx:', Home Assistant discovery removed.
This commit is contained in:
7 files changed
+544
-216
No files matched your search
+51
-51
@@ -1,17 +1,54 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import base64
|
||||
import os
|
||||
import solaredge_modbus
|
||||
import sys
|
||||
import time
|
||||
import paho.mqtt.client as mqtt
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
import yaml
|
||||
from solaredge_modbus_mqtt import SolarData
|
||||
from solaredge_modbus_mqtt import MyInverter
|
||||
from pymodbus.exceptions import ConnectionException
|
||||
|
||||
from solaredge_modbus_mqtt import MyInverter
|
||||
from solaredge_modbus_mqtt import SolarData
|
||||
from solaredge_modbus_mqtt import to_line_protocol
|
||||
|
||||
SPOOL_FILE = 'spool.txt'
|
||||
SPOOL_MAX_LINES = 20000 # ~16h of backlog at a 3s poll interval
|
||||
|
||||
|
||||
def send_to_influx(lines, influx):
|
||||
url = '{0}/write?db={1}&precision=ns'.format(
|
||||
influx['url'].rstrip('/'),
|
||||
urllib.parse.quote(influx['db'], safe='')
|
||||
)
|
||||
request = urllib.request.Request(url, data='\n'.join(lines).encode('utf-8'), method='POST')
|
||||
if influx.get('user'):
|
||||
credentials = '{0}:{1}'.format(influx['user'], influx.get('pass') or '')
|
||||
token = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
|
||||
request.add_header('Authorization', 'Basic ' + token)
|
||||
urllib.request.urlopen(request, timeout=5)
|
||||
|
||||
|
||||
def publish(line, influx):
|
||||
'''Write one line to InfluxDB. On failure, spool it to disk and replay the
|
||||
spool together with the next successful write (capped at SPOOL_MAX_LINES).'''
|
||||
try:
|
||||
backlog = []
|
||||
if os.path.exists(SPOOL_FILE):
|
||||
with open(SPOOL_FILE, 'r') as f:
|
||||
backlog = f.read().splitlines()[-SPOOL_MAX_LINES:]
|
||||
send_to_influx(backlog + [line], influx)
|
||||
if backlog:
|
||||
os.remove(SPOOL_FILE)
|
||||
except Exception as err:
|
||||
print('InfluxDB write failed, spooling line: {0}'.format(err))
|
||||
with open(SPOOL_FILE, 'a') as f:
|
||||
f.write(line + '\n')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
argparser = argparse.ArgumentParser()
|
||||
argparser.add_argument('--config', type=str, default=os.environ.get('CONFIG_FILE', 'config.yaml'), help='Path to the config.yaml')
|
||||
@@ -21,15 +58,8 @@ if __name__ == '__main__':
|
||||
try:
|
||||
config = yaml.safe_load(stream)
|
||||
except yaml.YAMLError as exc:
|
||||
print(exc)
|
||||
|
||||
try:
|
||||
mqtt = mqtt.Client(client_id='', clean_session=True, userdata=None, protocol=mqtt.MQTTv311, transport='tcp')
|
||||
mqtt.username_pw_set(config['mqtt']['user'], config['mqtt']['pass'])
|
||||
mqtt.connect(config['mqtt']['host'], config['mqtt']['port'], 60)
|
||||
except OSError as err:
|
||||
print('Cannot connect to MQTT on {0}:{1}, error: {2}'.format(config['mqtt']['host'], config['mqtt']['port'], err))
|
||||
sys.exit(1)
|
||||
print(exc)
|
||||
sys.exit(1)
|
||||
|
||||
inverter = MyInverter(
|
||||
host=config['modbus']['host'],
|
||||
@@ -38,45 +68,15 @@ if __name__ == '__main__':
|
||||
unit=config['modbus']['unit']
|
||||
)
|
||||
|
||||
run = True
|
||||
publishedAutoDiscovery = False
|
||||
while run:
|
||||
while True:
|
||||
try:
|
||||
raw = inverter.read()
|
||||
solar_data = SolarData(raw)
|
||||
line = to_line_protocol(solar_data.data, config['influx'].get('measurement', 'solaredge'))
|
||||
if line:
|
||||
publish(line, config['influx'])
|
||||
except ConnectionException as err:
|
||||
print('Cannot connect to SolarEdge Modbus server on {0}:{1}, error: {2}'.format(config['modbus']['host'], config['modbus']['port'], err))
|
||||
sys.exit(2)
|
||||
print('Cannot connect to SolarEdge Modbus server on {0}:{1}, error: {2}'.format(
|
||||
config['modbus']['host'], config['modbus']['port'], err))
|
||||
|
||||
mqtt.loop()
|
||||
|
||||
if (publishedAutoDiscovery == False):
|
||||
for k, v in solar_data.data.items():
|
||||
if k in ['model', 'version', 'serialnumber']:
|
||||
# These are not numbers/measurements, and should not be autodiscovered as a sensor
|
||||
continue
|
||||
|
||||
message = {
|
||||
'name': 'SolarEdge ' + k.replace('_', ' ').title(),
|
||||
'unique_id': 'solaredge_' + k,
|
||||
'unit_of_measurement': SolarData.fields[k]['unit'] if 'unit' in SolarData.fields[k] else '',
|
||||
'state_topic': config['mqtt']['state_topic'] + '/' + k,
|
||||
'force_update': 'True',
|
||||
'device': {
|
||||
'identifiers': ['solaredge_inverter'],
|
||||
'name': 'SolarEdge Inverter'
|
||||
}
|
||||
}
|
||||
mqtt.publish(
|
||||
config['mqtt']['auto_discovery_topic'] + '/' + str(k) + '/config',
|
||||
json.dumps(message),
|
||||
qos=0,
|
||||
retain=True
|
||||
)
|
||||
|
||||
publishedAutoDiscovery = True
|
||||
|
||||
for k, v in solar_data.data.items():
|
||||
mqtt.publish(config['mqtt']['state_topic'] + '/' + k, payload=str(v), qos=0, retain=False)
|
||||
|
||||
time.sleep(config['poll_interval'])
|
||||
time.sleep(config['poll_interval'])
|
||||
Reference in new issue
Block a user