Main code cleanup to improve reconnection in even of failure, added InfluxDB export

This commit is contained in:
bohdan-s
2022-01-07 16:53:03 +11:00
parent 94e82c9c11
commit e849c733e2
5 changed files with 287 additions and 222 deletions
+42
View File
@@ -0,0 +1,42 @@
from typing import Sequence
import influxdb_client
import logging
from influxdb_client.client.write_api import SYNCHRONOUS
class export_influxdb(object):
def __init__(self):
self.client = None
self.write_api = None
self.bucket = None
self.measurements = []
# Configure InfluxDB
def configure(self, config, config_inverter):
if not config.get('token') and config.get('org') and config.get('bucket') and config.get('measurements'):
logging.warning(f"InfluxDB: Please check configuration")
return False
self.client = influxdb_client.InfluxDBClient(
url=config.get('url', 'http://localhost:8086'),
token=config.get('token'),
org=config.get('org')
)
self.bucket=config.get('bucket')
for measurement in config.get('measurements'):
self.measurements.append(measurement)
self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
logging.info(f"Configured InfluxDB Client: {self.client.url}")
def publish(self, inverter):
sequence = []
for measurement in self.measurements:
sequence.append(f"{measurement.get('point')},inverter={inverter.get('device_type_code', 'unknown').replace('.','').replace('-','')} {measurement.get('register')}={inverter.get(measurement.get('register'))}")
logging.info(f'InfluxDB Write: {sequence}')
try:
self.write_api.write(self.bucket, self.client.org, sequence)
except Exception as err:
logging.error(err)
logging.info("Published to InfluxDB")
+1 -1
View File
@@ -29,7 +29,7 @@ class export_mqtt(object):
for ha_topic in config.get('ha_topics'):
self.ha_topics.append(ha_topic)
logging.info("Configured MQTT Client")
logging.info(f"Configured MQTT Client: {config.get('host')}:{config.get('port', 1883)}")
def publish(self, inverter):
global mqtt_client
+242 -220
View File
@@ -15,113 +15,138 @@ import yaml
import time
import os
config = None
registers = None
inverter = None
client = None
logging.basicConfig(
format='%(asctime)s %(levelname)-8s %(message)s',
level=30,
level=20,
datefmt='%Y-%m-%d %H:%M:%S')
try:
opts, args = getopt.getopt(sys.argv[1:],"hc:v:", "runonce")
except getopt.GetoptError:
logging.debug(f'No options passed via command line')
def connect_inverter():
client_payload = {
"host": config['inverter'].get('host', '127.0.0.1'),
"port": config['inverter'].get('port', 502),
"timeout": config['inverter'].get('timeout', 10),
"retries": config['inverter'].get('retries', 3),
"RetryOnEmpty": config['inverter'].get('RetryOnEmpty', False),
}
if config['inverter'].get("connection") == "http":
client_payload['port'] = config['inverter'].get('port', '8082')
new_client = SungrowModbusWebClient.SungrowModbusWebClient(**client_payload)
elif config['inverter'].get("connection") == "sungrow":
new_client = SungrowModbusTcpClient.SungrowModbusTcpClient(**client_payload)
elif config['inverter'].get("connection") == "modbus":
new_client = ModbusTcpClient(**client_payload)
else:
logging.warning(f'inverter > connection not specified, defaulting to ModbusTcpClient')
new_client = ModbusTcpClient(**client_payload)
logging.info("Connection: " + str(new_client))
configfile = 'config.yaml'
if not config['inverter'].get("connection") == "http":
new_client.connect()
new_client.close()
for opt, arg in opts:
if opt == '-h':
print(f'\nSunGather {__version__}')
print(f'usage: python3 sungather.py [options]')
print(f'\nCommandling arguments override any config file settings')
print(f'Options and arguments:')
print(f'-c config.yaml : Specify config file.')
print(f'-v 30 : Logging Level, 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error')
print(f'--runonce : Run once then exit')
print(f'-h : print this help message and exit (also --help)')
print(f'\nExample:')
print(f'python3 sungather.py -c /full/path/config.yaml\n')
sys.exit()
elif opt == '-c':
configfile = arg
elif opt == '-v':
if arg.isnumeric():
if int(arg) >= 0 and int(arg) <= 50:
loglevel = int(arg)
else:
logging.error(f"Valid verbose options: 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error")
sys.exit(2)
else:
logging.error(f"Valid verbose options: 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error")
sys.exit(2)
elif opt == '--runonce':
runonce = True
return new_client
logging.info(f'Starting SunGather {__version__}')
def scrape_inverter():
global inverter
try:
config = yaml.safe_load(open(configfile))
logging.info(f"Loaded config: {configfile}")
except Exception as err:
logging.error(f"Failed: Loading config: {configfile} \n\t\t\t {err}")
sys.exit(1)
try:
registers = yaml.safe_load(open('registers.yaml'))
logging.info(f"Loaded registers: {os.getcwd()}/registers.yaml")
except Exception as err:
logging.error(f"Failed: Loading registers: {os.getcwd()}/registers.yaml {err}")
sys.exit(1)
if not config.get('inverter'):
logging.error(f"Failed Loading config, missing Inverter settings")
sys.exit(1)
if 'loglevel' in locals():
logging.getLogger().setLevel(loglevel)
else:
logging.getLogger().setLevel(config['inverter'].get('logging',30))
client_payload = {
"host": config['inverter'].get('host', '127.0.0.1'),
"port": config['inverter'].get('port', 502),
"timeout": config['inverter'].get('timeout', 10),
"retries": config['inverter'].get('retries', 3),
"RetryOnEmpty": config['inverter'].get('RetryOnEmpty', False),
}
inverter = {}
exports = []
if config.get('exports'):
for export in config.get('exports'):
try:
if export.get('enabled', True):
export_load = importlib.import_module("exports." + export.get('name'))
logging.info(f"Loaded Export: exports\{export.get('name')}")
exports.append(getattr(export_load, "export_" + export.get('name'))())
exports[-1].configure(export, config['inverter'])
logging.info(f"Configured export: {export.get('name')}")
except Exception as err:
logging.error(f"Failed loading export: {err}" +
f"\n\t\t\t Please make sure {export.get('name')}.py exists in the exports folder")
if config['inverter'].get("connection") == "http":
client_payload['port'] = config['inverter'].get('port', '8082')
client = SungrowModbusWebClient.SungrowModbusWebClient(**client_payload)
elif config['inverter'].get("connection") == "sungrow":
client = SungrowModbusTcpClient.SungrowModbusTcpClient(**client_payload)
elif config['inverter'].get("connection") == "modbus":
client = ModbusTcpClient(**client_payload)
else:
logging.warning(f'inverter > connection not specified, defaulting to ModbusTcpClient')
client = ModbusTcpClient(**client_payload)
logging.info("Connection: " + str(client))
if not config['inverter'].get("connection") == "http":
""" Connect to the inverter and scrape the metrics """
client.connect()
client.close()
for scan in registers['scan']:
if scan.get('read'):
for subscan in registers['scan'][0]['read']:
if subscan.get('level',3) <= config['inverter'].get('level',1) or config['inverter'].get('level',1) == 3:
if not subscan.get('hybrid', False):
logging.debug(f'Scanning: read, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("read", int(subscan.get('start')), int(subscan.get('range'))):
return False
elif subscan.get('hybrid', False) and config['inverter'].get('hybrid',False):
logging.debug(f'Scanning: read, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("read", int(subscan.get('start')), int(subscan.get('range'))):
return False
if scan.get('hold'):
for subscan in registers['scan'][1]['hold']:
if not subscan.get('hybrid', False):
logging.debug(f'Scanning: hold, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("hold", int(subscan.get('start')), int(subscan.get('range'))):
return False
elif subscan.get('hybrid', False) and config['inverter'].get('hybrid',False):
logging.debug(f'Scanning: hold, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("hold", int(subscan.get('start')), int(subscan.get('range'))):
return False
# Create a registers for Power imported and exported to/from Grid
if config['inverter'].get('level',1) >= 1:
try:
inverter["export_to_grid"] = 0
inverter["import_from_grid"] = 0
power = inverter.get('meter_power', inverter.get('export_power', 0))
if power < 0:
inverter["export_to_grid"] = abs(power)
elif power >= 0:
inverter["import_from_grid"] = power
except Exception:
pass
try: # If inverter is returning no data for load_power, we can calculate it manually
if config['inverter'].get('manual_load', False):
inverter["load_power"] = int(inverter.get('total_active_power')) + int(inverter.get('meter_power'))
except Exception:
pass
# See if the inverter is running, This is added to inverters so can be read via MQTT etc...
# It is also used below, as some registers hold the last value on 'stop' so we need to set to 0
# to help with graphing.
try:
if inverter.get('start_stop'):
if inverter.get('start_stop', False) == 'Start' and inverter.get('work_state_1', False) == 'Run':
inverter["is_running"] = True
else:
if inverter.get('start_stop') == 'Stop':
inverter["is_running"] = False
except Exception:
pass
if config['inverter'].get('use_local_time',False):
inverter["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logging.debug(f'Using Local Computer Time: {inverter.get("timestamp")}')
del inverter["year"]
del inverter["month"]
del inverter["day"]
del inverter["hour"]
del inverter["minute"]
del inverter["second"]
else:
try:
inverter["timestamp"] = "%s-%s-%s %s:%02d:%02d" % (
inverter["year"],
inverter["month"],
inverter["day"],
inverter["hour"],
inverter["minute"],
inverter["second"],
)
logging.debug(f'Using Inverter Time: {inverter.get("timestamp")}')
del inverter["year"]
del inverter["month"]
del inverter["day"]
del inverter["hour"]
del inverter["minute"]
del inverter["second"]
except Exception:
pass
client.close()
return True
def load_registers(register_type, start, count=100):
global inverter
try:
logging.debug(f'load_registers: {register_type}, {start}:{count}')
if register_type == "read":
@@ -132,19 +157,19 @@ def load_registers(register_type, start, count=100):
raise RuntimeError(f"Unsupported register type: {type}")
except Exception as err:
logging.warning(f'No data returned for {register_type}, {start}:{count}\n\t\t\t\t{str(err)}')
return
return False
if rr.isError():
logging.warning(f"Modbus connection failed: {rr}")
return
return False
if not hasattr(rr, 'registers'):
logging.warning("No registers returned")
return
return False
if len(rr.registers) != count:
logging.warning(f"Mismatched number of registers read {len(rr.registers)} != {count}")
return
return False
for num in range(0, count):
run = int(start) + num + 1
@@ -238,135 +263,132 @@ def load_registers(register_type, start, count=100):
inverter[register_name] = register_value
else:
inverter[register_name] = register_value
return True
# Inverter Model Scanning
if config['inverter'].get('model'):
inverter['device_type_code'] = config['inverter'].get('model')
logging.info(f'Bypassing Model Detection, Using config: {inverter.get("device_type_code")}')
else:
if load_registers("read", 4999, 1):
if isinstance(inverter.get('device_type_code'),int):
logging.warning(f"Unknown Type Code Detected: {inverter.get('device_type_code')}")
inverter['device_type_code'] = 'unknown'
logging.info(f"Detected Model: {inverter.get('device_type_code')}")
else:
inverter['device_type_code'] = 'unknown'
logging.info(f'Model detection failed, please set model in config.py')
def main():
global config
global registers
global inverter
global client
# Core monitoring loop
def scrape_inverter():
""" Connect to the inverter and scrape the metrics """
client.connect()
configfile = 'config.yaml'
for scan in registers['scan']:
if scan.get('read'):
for subscan in registers['scan'][0]['read']:
if subscan.get('level',3) <= config['inverter'].get('level',1) or config['inverter'].get('level',1) == 3:
if not subscan.get('hybrid', False):
logging.debug(f'Scanning: read, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("read", int(subscan.get('start')), int(subscan.get('range'))):
return False
elif subscan.get('hybrid', False) and config['inverter'].get('hybrid',False):
logging.debug(f'Scanning: read, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("read", int(subscan.get('start')), int(subscan.get('range'))):
return False
if scan.get('hold'):
for subscan in registers['scan'][1]['hold']:
if not subscan.get('hybrid', False):
logging.debug(f'Scanning: hold, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("hold", int(subscan.get('start')), int(subscan.get('range'))):
return False
elif subscan.get('hybrid', False) and config['inverter'].get('hybrid',False):
logging.debug(f'Scanning: hold, {subscan.get("start")}:{subscan.get("range")}')
if not load_registers("hold", int(subscan.get('start')), int(subscan.get('range'))):
return False
# Create a registers for Power imported and exported to/from Grid
if config['inverter'].get('level',1) >= 1:
try:
inverter["export_to_grid"] = 0
inverter["import_from_grid"] = 0
power = inverter.get('meter_power', inverter.get('export_power', 0))
if power < 0:
inverter["export_to_grid"] = abs(power)
elif power >= 0:
inverter["import_from_grid"] = power
except Exception:
pass
try: # If inverter is returning no data for load_power, we can calculate it manually
if config['inverter'].get('manual_load', False):
inverter["load_power"] = int(inverter.get('total_active_power')) + int(inverter.get('meter_power'))
except Exception:
pass
# See if the inverter is running, This is added to inverters so can be read via MQTT etc...
# It is also used below, as some registers hold the last value on 'stop' so we need to set to 0
# to help with graphing.
try:
if inverter.get('start_stop'):
if inverter.get('start_stop', False) == 'Start' and inverter.get('work_state_1', False) == 'Run':
inverter["is_running"] = True
else:
if inverter.get('start_stop') == 'Stop':
inverter["is_running"] = False
except Exception:
pass
opts, args = getopt.getopt(sys.argv[1:],"hc:v:", "runonce")
except getopt.GetoptError:
logging.debug(f'No options passed via command line')
if config['inverter'].get('use_local_time',False):
inverter["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logging.debug(f'Using Local Computer Time: {inverter.get("timestamp")}')
del inverter["year"]
del inverter["month"]
del inverter["day"]
del inverter["hour"]
del inverter["minute"]
del inverter["second"]
for opt, arg in opts:
if opt == '-h':
print(f'\nSunGather {__version__}')
print(f'usage: python3 sungather.py [options]')
print(f'\nCommandling arguments override any config file settings')
print(f'Options and arguments:')
print(f'-c config.yaml : Specify config file.')
print(f'-v 30 : Logging Level, 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error')
print(f'--runonce : Run once then exit')
print(f'-h : print this help message and exit (also --help)')
print(f'\nExample:')
print(f'python3 sungather.py -c /full/path/config.yaml\n')
sys.exit()
elif opt == '-c':
configfile = arg
elif opt == '-v':
if arg.isnumeric():
if int(arg) >= 0 and int(arg) <= 50:
loglevel = int(arg)
else:
logging.error(f"Valid verbose options: 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error")
sys.exit(2)
else:
logging.error(f"Valid verbose options: 10 = Debug, 20 = Info, 30 = Warning (default), 40 = Error")
sys.exit(2)
elif opt == '--runonce':
runonce = True
logging.info(f'Starting SunGather {__version__}')
try:
config = yaml.safe_load(open(configfile))
logging.info(f"Loaded config: {configfile}")
except Exception as err:
logging.error(f"Failed: Loading config: {configfile} \n\t\t\t {err}")
sys.exit(1)
if not config.get('inverter'):
logging.error(f"Failed Loading config, missing Inverter settings")
sys.exit(1)
try:
registers = yaml.safe_load(open('registers.yaml'))
logging.info(f"Loaded registers: {os.getcwd()}/registers.yaml")
except Exception as err:
logging.error(f"Failed: Loading registers: {os.getcwd()}/registers.yaml {err}")
sys.exit(1)
if 'loglevel' in locals():
logging.getLogger().setLevel(loglevel)
else:
try:
inverter["timestamp"] = "%s-%s-%s %s:%02d:%02d" % (
inverter["year"],
inverter["month"],
inverter["day"],
inverter["hour"],
inverter["minute"],
inverter["second"],
)
logging.debug(f'Using Inverter Time: {inverter.get("timestamp")}')
del inverter["year"]
del inverter["month"]
del inverter["day"]
del inverter["hour"]
del inverter["minute"]
del inverter["second"]
except Exception:
pass
logging.getLogger().setLevel(config['inverter'].get('logging',30))
client.close()
return True
exports = []
if config.get('exports'):
for export in config.get('exports'):
try:
if export.get('enabled', True):
export_load = importlib.import_module("exports." + export.get('name'))
logging.info(f"Loaded Export: exports\{export.get('name')}")
exports.append(getattr(export_load, "export_" + export.get('name'))())
exports[-1].configure(export, config['inverter'])
logging.info(f"Configured export: {export.get('name')}")
except Exception as err:
logging.error(f"Failed loading export: {err}" +
f"\n\t\t\t Please make sure {export.get('name')}.py exists in the exports folder")
while True:
# Clear previous inverter values
tmp_model = inverter.get('device_type_code')
client = connect_inverter()
inverter = {}
inverter['device_type_code'] = tmp_model
# Scrape the inverter
success = scrape_inverter()
for export in exports:
t = Thread(target=export.publish, args=(inverter,))
t.start()
# if args.one_shot:
# logging.info("Exiting due to --one-shot")
# break
if not 'runonce' in locals():
# Sleep until the next scan
time.sleep(config['inverter'].get('scan_interval', 30))
# Inverter Model Scanning
model = 'unknown'
if config['inverter'].get('model'):
model = config['inverter'].get('model')
logging.info(f'Bypassing Model Detection, Using config: {model}')
else:
sys.exit(0)
if load_registers("read", 4999, 1):
if isinstance(inverter.get('device_type_code'),int):
logging.warning(f"Unknown Type Code Detected: {inverter.get('device_type_code')}")
else:
model = inverter.get('device_type_code')
logging.info(f"Detected Model: {model}")
else:
logging.info(f'Model detection failed, please set model in config.py')
# Core monitoring loop
while True:
# Clear previous inverter values, keep the model
inverter = {}
inverter['device_type_code'] = model
if not client:
client = connect_inverter()
time.sleep(3) # Wait 3 secs for connection, resvoles issues with modbus not reconnecting on failure
# Scrape the inverter
success = scrape_inverter()
if(success):
for export in exports:
t = Thread(target=export.publish, args=(inverter,))
t.start()
else:
client.close()
client = None
logging.warning(f"Data collection failed, skipped exporting data. Retying in {config['inverter'].get('scan_interval', 30)}")
if not 'runonce' in locals():
# Sleep until the next scan
time.sleep(config['inverter'].get('scan_interval', 30))
else:
sys.exit(0)
if __name__== "__main__":
main()
+1 -1
View File
@@ -1 +1 @@
__version__ = '0.1.2'
__version__ = '0.1.3'
+1
View File
@@ -4,3 +4,4 @@ paho-mqtt>=1.5.1
pymodbus>=2.4.0
SungrowModbusTcpClient>=0.1.5
SungrowModbusWebClient>=0.2.6
influxdb-client=>=1.24.0