make _info_defs and _info_devs private

This commit is contained in:
Stefan Allius
2024-03-30 11:58:38 +01:00
parent e3fdeecf82
commit 71ec0570ac

View File

@@ -101,13 +101,13 @@ class Infos:
logging.info('Initialize proxy statistics') logging.info('Initialize proxy statistics')
# init proxy counter in the class.stat dictionary # init proxy counter in the class.stat dictionary
cls.stat['proxy'] = {} cls.stat['proxy'] = {}
for key in cls._info_defs: for key in cls.__info_defs:
name = cls._info_defs[key]['name'] name = cls.__info_defs[key]['name']
if name[0] == 'proxy': if name[0] == 'proxy':
cls.stat['proxy'][name[1]] = 0 cls.stat['proxy'][name[1]] = 0
# add values from the environment to the device definition table # add values from the environment to the device definition table
prxy = cls._info_devs['proxy'] prxy = cls.__info_devs['proxy']
prxy['sw'] = cls.version prxy['sw'] = cls.version
prxy['mdl'] = cls.app_name prxy['mdl'] = cls.app_name
@@ -115,7 +115,7 @@ class Infos:
self.db = {} self.db = {}
self.tracer = logging.getLogger('data') self.tracer = logging.getLogger('data')
_info_devs = { __info_devs = {
'proxy': {'singleton': True, 'name': 'Proxy', 'mf': 'Stefan Allius'}, # noqa: E501 'proxy': {'singleton': True, 'name': 'Proxy', 'mf': 'Stefan Allius'}, # noqa: E501
'controller': {'via': 'proxy', 'name': 'Controller', 'mdl': Register.CHIP_MODEL, 'mf': Register.CHIP_TYPE, 'sw': Register.COLLECTOR_FW_VERSION}, # noqa: E501 'controller': {'via': 'proxy', 'name': 'Controller', 'mdl': Register.CHIP_MODEL, 'mf': Register.CHIP_TYPE, 'sw': Register.COLLECTOR_FW_VERSION}, # noqa: E501
'inverter': {'via': 'controller', 'name': 'Micro Inverter', 'mdl': Register.EQUIPMENT_MODEL, 'mf': Register.MANUFACTURER, 'sw': Register.VERSION}, # noqa: E501 'inverter': {'via': 'controller', 'name': 'Micro Inverter', 'mdl': Register.EQUIPMENT_MODEL, 'mf': Register.MANUFACTURER, 'sw': Register.VERSION}, # noqa: E501
@@ -127,7 +127,7 @@ class Infos:
__comm_type_val_tpl = "{%set com_types = ['n/a','Wi-Fi', 'G4', 'G5', 'GPRS'] %}{{com_types[value_json['Communication_Type']|int(0)]|default(value_json['Communication_Type'])}}" # noqa: E501 __comm_type_val_tpl = "{%set com_types = ['n/a','Wi-Fi', 'G4', 'G5', 'GPRS'] %}{{com_types[value_json['Communication_Type']|int(0)]|default(value_json['Communication_Type'])}}" # noqa: E501
_info_defs = { __info_defs = {
# collector values used for device registration: # collector values used for device registration:
Register.COLLECTOR_FW_VERSION: {'name': ['collector', 'Collector_Fw_Version'], 'level': logging.INFO, 'unit': ''}, # noqa: E501 Register.COLLECTOR_FW_VERSION: {'name': ['collector', 'Collector_Fw_Version'], 'level': logging.INFO, 'unit': ''}, # noqa: E501
Register.CHIP_TYPE: {'name': ['collector', 'Chip_Type'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 Register.CHIP_TYPE: {'name': ['collector', 'Chip_Type'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501
@@ -216,19 +216,21 @@ class Infos:
@property @property
def info_devs(self) -> dict: def info_devs(self) -> dict:
return self._info_devs return self.__info_devs
@info_devs.setter
def info_devs(self, value: dict) -> None:
self._info_devs = value
@property @property
def info_defs(self) -> dict: def info_defs(self) -> dict:
return self._info_defs return self.__info_defs
'''
if __name__ == "app.src.messages":
@info_defs.setter @info_defs.setter
def info_defs(self, value: dict) -> None: def info_defs(self, value: dict) -> None:
self._info_defs = value self.__info_defs = value
@info_devs.setter
def info_devs(self, value: dict) -> None:
self.__info_devs = value
'''
def dev_value(self, idx: str | int) -> str | int | float | None: def dev_value(self, idx: str | int) -> str | int | float | None:
'''returns the stored device value from our database '''returns the stored device value from our database
@@ -240,8 +242,8 @@ class Infos:
''' '''
if type(idx) is str: if type(idx) is str:
return idx # return idx as a fixed value return idx # return idx as a fixed value
elif idx in self._info_defs: elif idx in self.info_defs:
row = self._info_defs[idx] row = self.info_defs[idx]
if 'singleton' in row and row['singleton']: if 'singleton' in row and row['singleton']:
dict = self.stat dict = self.stat
else: else:
@@ -255,7 +257,7 @@ class Infos:
dict = dict[key] dict = dict[key]
return dict # value of the reqeusted entry return dict # value of the reqeusted entry
return None # unknwon idx, not in __info_defs return None # unknwon idx, not in info_defs
def inc_counter(self, counter: str) -> None: def inc_counter(self, counter: str) -> None:
'''inc proxy statistic counter''' '''inc proxy statistic counter'''
@@ -281,7 +283,7 @@ class Infos:
''' '''
# iterate over RegisterMap.map and get the register values for entries # iterate over RegisterMap.map and get the register values for entries
# with Singleton=True, which means that this is a proxy register # with Singleton=True, which means that this is a proxy register
for reg in self._info_defs.keys(): for reg in self.info_defs.keys():
res = self.ha_conf(reg, ha_prfx, node_id, snr, True) # noqa: E501 res = self.ha_conf(reg, ha_prfx, node_id, snr, True) # noqa: E501
if res: if res:
yield res yield res
@@ -320,7 +322,7 @@ class Infos:
attr['val_tpl'] = '{{value_json' + f"['{row['name'][-1]}'] {ha['fmt']}" + '}}' # eg. 'val_tpl': "{{ value_json['Output_Power']|float }} # noqa: E501 attr['val_tpl'] = '{{value_json' + f"['{row['name'][-1]}'] {ha['fmt']}" + '}}' # eg. 'val_tpl': "{{ value_json['Output_Power']|float }} # noqa: E501
else: else:
self.inc_counter('Internal_Error') self.inc_counter('Internal_Error')
logging.error(f"Infos._info_defs: the row for {key} do" logging.error(f"Infos.info_defs: the row for {key} do"
" not have a 'val_tpl' nor a 'fmt' value") " not have a 'val_tpl' nor a 'fmt' value")
# add unit_of_meas only, if status_class isn't none. If # add unit_of_meas only, if status_class isn't none. If
# status_cla is None we want a number format and not line # status_cla is None we want a number format and not line
@@ -364,7 +366,7 @@ class Infos:
dev['via_device'] = f"{via}_{snr}" dev['via_device'] = f"{via}_{snr}"
else: else:
self.inc_counter('Internal_Error') self.inc_counter('Internal_Error')
logging.error(f"Infos._info_defs: the row for " logging.error(f"Infos.info_defs: the row for "
f"{key} has an invalid via value: " f"{key} has an invalid via value: "
f"{via}") f"{via}")
for key in ('mdl', 'mf', 'sw', 'hw'): # add optional for key in ('mdl', 'mf', 'sw', 'hw'): # add optional
@@ -385,13 +387,13 @@ class Infos:
attr['o'] = origin attr['o'] = origin
else: else:
self.inc_counter('Internal_Error') self.inc_counter('Internal_Error')
logging.error(f"Infos._info_defs: the row for {key} " logging.error(f"Infos.info_defs: the row for {key} "
"missing 'dev' value for ha register") "missing 'dev' value for ha register")
return json.dumps(attr), component, node_id, attr['uniq_id'] return json.dumps(attr), component, node_id, attr['uniq_id']
return None return None
def _key_obj(self, id) -> list: def _key_obj(self, id) -> list:
d = self._info_defs.get(id, {'name': None, 'level': logging.DEBUG, d = self.info_defs.get(id, {'name': None, 'level': logging.DEBUG,
'unit': ''}) 'unit': ''})
if 'ha' in d and 'must_incr' in d['ha']: if 'ha' in d and 'must_incr' in d['ha']:
must_incr = d['ha']['must_incr'] must_incr = d['ha']['must_incr']