Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

replace manual opts with configargparser #125

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

replace manual opts with configargparser #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
12ebe68
replace manual opts with configargparser
morph027 Jun 26, 2023
6c513bf
add configargparse to requirements
morph027 Jun 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
252 changes: 138 additions & 114 deletions SunGather/sungather.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from SungrowClient import SungrowClient
from version import __version__

import configargparse
import importlib
import logging
import logging.handlers
Expand All @@ -11,138 +12,154 @@
import yaml
import time

def main():
configfilename = 'config.yaml'
registersfilename = 'registers-sungrow.yaml'
logfolder = ''

try:
opts, args = getopt.getopt(sys.argv[1:],"hc:r:l:v:", "runonce")
except getopt.GetoptError:
logging.debug(f'No options passed via command line')

for opt, arg in opts:
if opt == '-h':
print(f'\nSunGather {__version__}')
print(f'\nhttps://sungather.app')
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'-r registers-file.yaml : Specify registers file.')
print(f'-l /logs/ : Specify folder to store logs.')
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':
configfilename = arg
elif opt == '-r':
registersfilename = arg
elif opt == '-l':
logfolder = 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__}')
logging.info(f'Need Help? https://github.com/bohdan-s/SunGather')
logging.info(f'NEW HomeAssistant Add-on: https://github.com/bohdan-s/hassio-repository')
def arguments():
"""
command line arguments
"""
parser = configargparse.ArgumentParser(
prog="sungather",
auto_env_var_prefix="SUNGATHER_",
add_env_var_help=True,
)
parser.add("--version", action="version", version=__version__)
parser.add("--config", "-c", default="config.yaml", help="Specify config file.")
parser.add(
"--registers",
"-r",
default="registers-sungrow.yaml",
help="Specify registers file.",
)
parser.add("--logs", "-l", help="Specify folder to store logs.")
parser.add(
"--verbosity",
"-v",
type=int,
default=30,
choices=[10, 20, 30, 40],
help="Logging Level, 10 = Debug, 20 = Info, 30 = Warning, 40 = Error",
)
parser.add("--runonce", action="store_true", help="Run once then exit.")
return parser.parse_known_args()


def main():
"""
sungather command line tool
"""
args, extra_args = arguments()
logging.info(f"Starting SunGather {__version__}")
logging.info("Need Help? https://github.com/bohdan-s/SunGather")
logging.info(
"NEW HomeAssistant Add-on: https://github.com/bohdan-s/hassio-repository"
)

try:
configfile = yaml.safe_load(open(configfilename, encoding="utf-8"))
logging.info(f"Loaded config: {configfilename}")
configfile = yaml.safe_load(open(args.config, encoding="utf-8"))
logging.info(f"Loaded config: {args.config}")
except Exception as err:
logging.error(f"Failed: Loading config: {configfilename} \n\t\t\t {err}")
logging.error(f"Failed: Loading config: {args.config} \n\t\t\t {err}")
sys.exit(1)
if not configfile.get('inverter'):
logging.error(f"Failed Loading config, missing Inverter settings")
sys.exit(f"Failed Loading config, missing Inverter settings")
if not configfile.get("inverter"):
logging.error("Failed Loading config, missing Inverter settings")
sys.exit("Failed Loading config, missing Inverter settings")

try:
registersfile = yaml.safe_load(open(registersfilename, encoding="utf-8"))
logging.info(f"Loaded registers: {registersfilename}")
logging.info(f"Registers file version: {registersfile.get('version','UNKNOWN')}")
registersfile = yaml.safe_load(open(args.registers, encoding="utf-8"))
logging.info(f"Loaded registers: {args.registers}")
logging.info(
f"Registers file version: {registersfile.get('version','UNKNOWN')}"
)
except Exception as err:
logging.error(f"Failed: Loading registers: {registersfilename} {err}")
sys.exit(f"Failed: Loading registers: {registersfilename} {err}")
logging.error(f"Failed: Loading registers: {args.registers} {err}")
sys.exit(f"Failed: Loading registers: {args.registers} {err}")

config_inverter = {
"host": configfile['inverter'].get('host',None),
"port": configfile['inverter'].get('port',502),
"timeout": configfile['inverter'].get('timeout',10),
"retries": configfile['inverter'].get('retries',3),
"slave": configfile['inverter'].get('slave',0x01),
"scan_interval": configfile['inverter'].get('scan_interval',30),
"connection": configfile['inverter'].get('connection',"modbus"),
"model": configfile['inverter'].get('model',None),
"smart_meter": configfile['inverter'].get('smart_meter',False),
"use_local_time": configfile['inverter'].get('use_local_time',False),
"log_console": configfile['inverter'].get('log_console','WARNING'),
"log_file": configfile['inverter'].get('log_file','OFF'),
"level": configfile['inverter'].get('level',1)
"host": configfile["inverter"].get("host", None),
"port": configfile["inverter"].get("port", 502),
"timeout": configfile["inverter"].get("timeout", 10),
"retries": configfile["inverter"].get("retries", 3),
"slave": configfile["inverter"].get("slave", 0x01),
"scan_interval": configfile["inverter"].get("scan_interval", 30),
"connection": configfile["inverter"].get("connection", "modbus"),
"model": configfile["inverter"].get("model", None),
"smart_meter": configfile["inverter"].get("smart_meter", False),
"use_local_time": configfile["inverter"].get("use_local_time", False),
"log_console": configfile["inverter"].get("log_console", "WARNING"),
"log_file": configfile["inverter"].get("log_file", "OFF"),
"level": configfile["inverter"].get("level", 1),
}

if 'loglevel' in locals():
logger.handlers[0].setLevel(loglevel)
else:
logger.handlers[0].setLevel(config_inverter['log_console'])

if not config_inverter['log_file'] == "OFF":
if config_inverter['log_file'] == "DEBUG" or config_inverter['log_file'] == "INFO" or config_inverter['log_file'] == "WARNING" or config_inverter['log_file'] == "ERROR":
logfile = logfolder + "SunGather.log"
fh = logging.handlers.RotatingFileHandler(logfile, mode='w', encoding='utf-8', maxBytes=10485760, backupCount=10) # Log 10mb files, 10 x files = 100mb
logger.handlers[0].setLevel(args.verbosity)

if not config_inverter["log_file"] == "OFF":
if (
config_inverter["log_file"] == "DEBUG"
or config_inverter["log_file"] == "INFO"
or config_inverter["log_file"] == "WARNING"
or config_inverter["log_file"] == "ERROR"
):
logfile = args.logs + "SunGather.log"
fh = logging.handlers.RotatingFileHandler(
logfile, mode="w", encoding="utf-8", maxBytes=10485760, backupCount=10
) # Log 10mb files, 10 x files = 100mb
fh.formatter = logger.handlers[0].formatter
fh.setLevel(config_inverter['log_file'])
fh.setLevel(config_inverter["log_file"])
logger.addHandler(fh)
else:
logging.warning(f"log_file: Valid options are: DEBUG, INFO, WARNING, ERROR and OFF")
logging.warning(
"log_file: Valid options are: DEBUG, INFO, WARNING, ERROR and OFF"
)

logging.info(f"Logging to console set to: {logging.getLevelName(logger.handlers[0].level)}")
logging.info(
f"Logging to console set to: {logging.getLevelName(logger.handlers[0].level)}"
)
if logger.handlers.__len__() == 3:
logging.info(f"Logging to file set to: {logging.getLevelName(logger.handlers[2].level)}")

logging.debug(f'Inverter Config Loaded: {config_inverter}')
logging.info(
f"Logging to file set to: {logging.getLevelName(logger.handlers[2].level)}"
)

logging.debug(f"Inverter Config Loaded: {config_inverter}")

if config_inverter.get('host'):
if config_inverter.get("host"):
inverter = SungrowClient.SungrowClient(config_inverter)
else:
logging.error(f"Error: host option in config is required")
logging.error("Error: host option in config is required")
sys.exit("Error: host option in config is required")

if not inverter.checkConnection():
logging.error(f"Error: Connection to inverter failed: {config_inverter.get('host')}:{config_inverter.get('port')}")
sys.exit(f"Error: Connection to inverter failed: {config_inverter.get('host')}:{config_inverter.get('port')}")
logging.error(
f"Error: Connection to inverter failed: {config_inverter.get('host')}:{config_inverter.get('port')}"
)
sys.exit(
f"Error: Connection to inverter failed: {config_inverter.get('host')}:{config_inverter.get('port')}"
)

inverter.configure_registers(registersfile)
if not inverter.inverter_config['connection'] == "http": inverter.close()

if not inverter.inverter_config["connection"] == "http":
inverter.close()

# Now we know the inverter is working, lets load the exports
exports = []
if configfile.get('exports'):
for export in configfile.get('exports'):
if configfile.get("exports"):
for export in configfile.get("exports"):
try:
if export.get('enabled', False):
export_load = importlib.import_module("exports." + export.get('name'))
if export.get("enabled", False):
export_load = importlib.import_module(
"SunGather.exports." + export.get("name")
)
logging.info(f"Loading Export: exports\{export.get('name')}")
exports.append(getattr(export_load, "export_" + export.get('name'))())
exports.append(
getattr(export_load, "export_" + export.get("name"))()
)
retval = exports[-1].configure(export, inverter)
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")
logging.error(
f"Failed loading export: {err}"
+ f"\n\t\t\t Please make sure {export.get('name')}.py exists in the exports folder"
)

scan_interval = config_inverter.get('scan_interval')
scan_interval = config_inverter.get("scan_interval")

# Core polling loop
while True:
Expand All @@ -153,40 +170,47 @@ def main():
# Scrape the inverter
success = inverter.scrape()

if(success):
if success:
for export in exports:
export.publish(inverter)
if not inverter.inverter_config['connection'] == "http": inverter.close()
if not inverter.inverter_config["connection"] == "http":
inverter.close()
else:
inverter.disconnect()
logging.warning(f"Data collection failed, skipped exporting data. Retying in {scan_interval} secs")
logging.warning(
f"Data collection failed, skipped exporting data. Retying in {scan_interval} secs"
)

loop_end = time.perf_counter()
process_time = round(loop_end - loop_start, 2)
logging.debug(f'Processing Time: {process_time} secs')
logging.debug(f"Processing Time: {process_time} secs")

if 'runonce' in locals():
if "runonce" in locals():
sys.exit(0)

# Sleep until the next scan
if scan_interval - process_time <= 1:
logging.warning(f"SunGather is taking {process_time} to process, which is longer than interval {scan_interval}, Please increase scan interval")
logging.warning(
f"SunGather is taking {process_time} to process, which is longer than interval {scan_interval}, Please increase scan interval"
)
time.sleep(process_time)
else:
logging.info(f'Next scrape in {int(scan_interval - process_time)} secs')
time.sleep(scan_interval - process_time)
logging.info(f"Next scrape in {int(scan_interval - process_time)} secs")
time.sleep(scan_interval - process_time)


logging.basicConfig(
format='%(asctime)s %(levelname)-8s %(message)s',
format="%(asctime)s %(levelname)-8s %(message)s",
level=logging.DEBUG,
datefmt='%Y-%m-%d %H:%M:%S')
datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger('')
logger = logging.getLogger("")
ch = logging.StreamHandler()
ch.setLevel(logging.WARNING)
logger.addHandler(ch)

if __name__== "__main__":
if __name__ == "__main__":
main()

sys.exit()
1 change: 1 addition & 0 deletions requirements.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
configarparse>=1.5.3
PyYAML>=6.0
requests>=2.26.0
paho-mqtt>=1.5.1
Expand Down