sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lint code with flake8

Browse Source
This commit is contained in:
Stefan Allius
2023-11-02 22:31:30 +01:00
parent c772eeeb28
commit 2c4af0b7d8
7 changed files with 585 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
app/src/config.py
View File

@@ -1,78 +1,90 @@
'''Config module handles the proxy configuration in the config.toml file'''
import shutil, tomllib, logging
import shutil
import tomllib
import logging
from schema import Schema, And, Use, Optional
class Config():
'''Static class Config is reads and sanitize the config.
Read config.toml file and sanitize it with read().
'''Static class Config is reads and sanitize the config.
Read config.toml file and sanitize it with read().
Get named parts of the config with get()'''
config = {}
conf_schema = Schema({ 'tsun': {
'enabled': Use(bool),
'host': Use(str),
'port': And(Use(int), lambda n: 1024 <= n <= 65535)},
conf_schema = Schema({
'tsun': {
'enabled': Use(bool),
'host': Use(str),
'port': And(Use(int), lambda n: 1024 <= n <= 65535)
},
'mqtt': {
'host': Use(str),
'port': And(Use(int), lambda n: 1024 <= n <= 65535),
'user': And(Use(str), Use(lambda s: s if len(s) > 0 else None)),
'passwd': And(Use(str), Use(lambda s: s if len(s) > 0 else None))
},
'ha': {
'auto_conf_prefix': Use(str),
'discovery_prefix': Use(str),
'entity_prefix': Use(str),
'proxy_node_id': Use(str),
'proxy_unique_id': Use(str)
},
'inverters': {
'allow_all': Use(bool), And(Use(str), lambda s: len(s) == 16): {
Optional('node_id', default=""): And(Use(str),
Use(lambda s: s + '/'
if len(s) > 0 and
s[-1] != '/' else s)),
'mqtt': {
'host': Use(str),
'port': And(Use(int), lambda n: 1024 <= n <= 65535),
'user': And(Use(str), Use(lambda s: s if len(s) >0 else None)),
'passwd': And(Use(str), Use(lambda s: s if len(s) >0 else None))},
'ha': {
'auto_conf_prefix': Use(str),
'discovery_prefix': Use(str),
'entity_prefix': Use(str),
'proxy_node_id': Use(str),
'proxy_unique_id': Use(str)},
'inverters': {
'allow_all' : Use(bool),
And(Use(str), lambda s: len(s) == 16 ): {
Optional('node_id', default=""): And(Use(str),Use(lambda s: s +'/' if len(s)> 0 and s[-1] != '/' else s)),
Optional('suggested_area', default=""): Use(str)
}}
}, ignore_extra_keys=True)
Optional('suggested_area', default=""): Use(str)
}}
}, ignore_extra_keys=True
)
@classmethod
def read(cls) -> None:
'''Read config file, merge it with the default config and sanitize the result'''
'''Read config file, merge it with the default config
and sanitize the result'''
config = {}
logger = logging.getLogger('data')
try:
# make the default config transparaent by copying it in the config.example file
# make the default config transparaent by copying it
# in the config.example file
shutil.copy2("default_config.toml", "config/config.example.toml")
# read example config file as default configuration
with open("default_config.toml", "rb") as f:
def_config = tomllib.load(f)
# overwrite the default values, with values from the config.toml file
# overwrite the default values, with values from
# the config.toml file
with open("config/config.toml", "rb") as f:
usr_config = tomllib.load(f)
config['tsun'] = def_config['tsun'] | usr_config['tsun']
config['mqtt'] = def_config['mqtt'] | usr_config['mqtt']
config['ha'] = def_config['ha'] | usr_config['ha']
config['inverters'] = def_config['inverters'] | usr_config['inverters']
config['tsun'] = def_config['tsun'] | usr_config['tsun']
config['mqtt'] = def_config['mqtt'] | usr_config['mqtt']
config['ha'] = def_config['ha'] | usr_config['ha']
config['inverters'] = def_config['inverters'] | \
usr_config['inverters']
cls.config = cls.conf_schema.validate(config)
#logging.debug(f'Readed config: "{cls.config}" ')
# logging.debug(f'Readed config: "{cls.config}" ')
except Exception as error:
logger.error(f'Config.read: {error}')
cls.config = {}
@classmethod
def get(cls, member:str = None):
'''Get a named attribute from the proxy config. If member == None it returns the complete config dict'''
def get(cls, member: str = None):
'''Get a named attribute from the proxy config. If member ==
None it returns the complete config dict'''
if member:
return cls.config.get(member, {})
else:
return cls.config
return cls.config