Added warning to information collector
This commit is contained in:
1 parent
c34dc7fdd3
commit
956ade4d03
3 files changed
+82
No files matched your search
@@ -18,6 +18,10 @@ informations_collector:
|
|||||||
reset_error_count_button: input_button.reset_error_count
|
reset_error_count_button: input_button.reset_error_count
|
||||||
output_sensor: sensor.ad_errors
|
output_sensor: sensor.ad_errors
|
||||||
|
|
||||||
|
warning_collector:
|
||||||
|
reset_warning_count_button: input_button.reset_error_count
|
||||||
|
output_sensor: sensor.ad_warnings
|
||||||
|
|
||||||
min_max_temp_sensors:
|
min_max_temp_sensors:
|
||||||
temperature_sensors:
|
temperature_sensors:
|
||||||
- sensor.ext_front_temperature
|
- sensor.ext_front_temperature
|
||||||
|
|||||||
@@ -51,6 +51,24 @@ class InformationsCollector(SmartObject):
|
|||||||
else:
|
else:
|
||||||
self.log_error("No output_sensor defined for error_collector")
|
self.log_error("No output_sensor defined for error_collector")
|
||||||
|
|
||||||
|
if 'warning_collector' in self.args:
|
||||||
|
if 'output_sensor' in self.args['warning_collector']:
|
||||||
|
sensor_name = self.args['warning_collector']['output_sensor']
|
||||||
|
self.warning_list = []
|
||||||
|
self.warning_log_handle = self.listen_log(self.on_log_warning, "WARNING", log="main_log")
|
||||||
|
self.warning_count_sensor = self.create_entity(
|
||||||
|
sensor_name, state=0,
|
||||||
|
attributes={"last_warning": "", "warning_list": []},
|
||||||
|
icon="mdi:alert-outline"
|
||||||
|
)
|
||||||
|
if 'reset_warning_count_button' in self.args['warning_collector']:
|
||||||
|
self.listen_state(
|
||||||
|
self.on_reset_warning_count,
|
||||||
|
self.args['warning_collector']['reset_warning_count_button']
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self.log_error("No output_sensor defined for warning_collector")
|
||||||
|
|
||||||
if 'min_max_temp_sensors' in self.args:
|
if 'min_max_temp_sensors' in self.args:
|
||||||
for sensor in self.args['min_max_temp_sensors']['temperature_sensors']:
|
for sensor in self.args['min_max_temp_sensors']['temperature_sensors']:
|
||||||
self.listen_state(self.on_temperature_change, sensor)
|
self.listen_state(self.on_temperature_change, sensor)
|
||||||
@@ -218,6 +236,21 @@ class InformationsCollector(SmartObject):
|
|||||||
self.error_list.clear()
|
self.error_list.clear()
|
||||||
self.error_count_sensor.set_state(state=0, attributes={"last_error": "", "error_list": self.error_list})
|
self.error_count_sensor.set_state(state=0, attributes={"last_error": "", "error_list": self.error_list})
|
||||||
|
|
||||||
|
def on_log_warning(self, name, ts, level, type, message, kwargs):
|
||||||
|
warning_count = int(self.warning_count_sensor.get_state()) + 1
|
||||||
|
self.warning_list.append(message)
|
||||||
|
self.warning_count_sensor.set_state(
|
||||||
|
state=warning_count,
|
||||||
|
attributes={"last_warning": message, "warning_list": self.warning_list}
|
||||||
|
)
|
||||||
|
|
||||||
|
def on_reset_warning_count(self, entity, attribute, old, new, kwargs):
|
||||||
|
self.warning_list.clear()
|
||||||
|
self.warning_count_sensor.set_state(
|
||||||
|
state=0,
|
||||||
|
attributes={"last_warning": "", "warning_list": self.warning_list}
|
||||||
|
)
|
||||||
|
|
||||||
def _on_group_condition_changed(self, prev_result,result):
|
def _on_group_condition_changed(self, prev_result,result):
|
||||||
self.update_group_collectors()
|
self.update_group_collectors()
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
|
sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
|
||||||
|
str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
|
||||||
|
|
||||||
|
from informationscollector import InformationsCollector
|
||||||
|
|
||||||
|
|
||||||
|
class WarningCollectorTests(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.app = object.__new__(InformationsCollector)
|
||||||
|
self.app.warning_list = []
|
||||||
|
self.app.warning_count_sensor = Mock()
|
||||||
|
self.app.warning_count_sensor.get_state.return_value = '2'
|
||||||
|
|
||||||
|
def test_warning_is_counted_and_exposed_in_attributes(self):
|
||||||
|
self.app.on_log_warning(
|
||||||
|
'example', None, 'WARNING', None, 'Something happened', {}
|
||||||
|
)
|
||||||
|
|
||||||
|
self.app.warning_count_sensor.set_state.assert_called_once_with(
|
||||||
|
state=3,
|
||||||
|
attributes={
|
||||||
|
'last_warning': 'Something happened',
|
||||||
|
'warning_list': ['Something happened'],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_warning_reset_clears_count_and_attributes(self):
|
||||||
|
self.app.warning_list = ['Old warning']
|
||||||
|
|
||||||
|
self.app.on_reset_warning_count(None, None, None, None, {})
|
||||||
|
|
||||||
|
self.assertEqual(self.app.warning_list, [])
|
||||||
|
self.app.warning_count_sensor.set_state.assert_called_once_with(
|
||||||
|
state=0,
|
||||||
|
attributes={'last_warning': '', 'warning_list': []},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
Reference in new issue
Block a user