added watchdog rules

changelog.md

@@ -6,6 +6,7 @@
 - added rule ``habapp_rules.actors.heating.KnxHeating`` which can be used to set the target temperature of a KNX heating actor which only supports temperature offsets
 - added temperature difference item of ``habapp_rules.sensors.sun.SensorTemperatureDifference`` to ``filtered_signal_groups``
 - added rule ``habapp_rules.actors.power.CurrentSwitch`` which can be used to enable a switch item if current is above a threshold
+- added rule ``habapp_rules.system.watchdog.Watchdog`` which can be used to check if an item was updated in time
 
 ## Bugfix
 

habapp_rules/system/config/watchdog.py

@@ -0,0 +1,22 @@
+"""Config models for watchdog rules."""
+import HABApp.openhab.items
+import pydantic
+
+import habapp_rules.core.pydantic_base
+
+
+class WatchdogItems(habapp_rules.core.pydantic_base.ItemBase):
+	"""Items for watchdog rule."""
+	observed: HABApp.openhab.items.OpenhabItem = pydantic.Field(..., description="observed item")
+	warning: HABApp.openhab.items.SwitchItem = pydantic.Field(..., description="warning item, which will be set to ON if the observed item was not updated in the expected time")
+
+
+class WatchdogParameter(habapp_rules.core.pydantic_base.ParameterBase):
+	"""Parameter for watchdog rule."""
+	timeout: int = pydantic.Field(3600, description="timeout in seconds")
+
+
+class WatchdogConfig(habapp_rules.core.pydantic_base.ConfigBase):
+	"""Config for watchdog rule."""
+	items: WatchdogItems = pydantic.Field(..., description="items for watchdog rule")
+	parameter: WatchdogParameter = pydantic.Field(WatchdogParameter(), description="parameters for watchdog rule")

habapp_rules/system/watchdog.py

@@ -0,0 +1,44 @@
+"""Watchdog rules."""
+import HABApp
+
+import habapp_rules.core.helper
+from habapp_rules.system.config.watchdog import WatchdogConfig
+
+
+class Watchdog(HABApp.Rule):
+	"""Watchdog rule to check if the observed item was updated in time.
+
+	# Items:
+	Switch    Item_To_Observe   "Item which should be observed"
+	Switch    Warning           "Warning, item was not updated in time"
+
+	# Config:
+	config = habapp_rules.system.config.WatchdogConfig(
+		items=habapp_rules.system.config.WatchdogItems(
+			observed="Item_To_Observe",
+			warning="Warning")
+	)
+
+	# Rule init:
+	habapp_rules.system.watchdog.Watchdog(config)
+	"""
+
+	def __init__(self, config: WatchdogConfig):
+		"""Init watchdog rule.
+
+		:param config: Config for watchdog rule
+		"""
+		HABApp.Rule.__init__(self)
+		self._config = config
+
+		self._countdown = self.run.countdown(self._config.parameter.timeout, habapp_rules.core.helper.send_if_different, item=self._config.items.warning, value="ON")
+		self._countdown.reset()
+		self._config.items.observed.listen_event(self._cb_observed_state_updated, HABApp.openhab.events.ItemStateUpdatedEventFilter())
+
+	def _cb_observed_state_updated(self, event: HABApp.openhab.events.ItemStateUpdatedEvent) -> None:
+		"""Callback which is called if the observed item was updated.
+
+		:param event: event which triggered this callback
+		"""
+		habapp_rules.core.helper.send_if_different(self._config.items.warning, "OFF")
+		self._countdown.reset()

tests/system/watchdog.py

@@ -0,0 +1,55 @@
+"""Tests for Watchdog Rule."""
+import unittest.mock
+
+import HABApp.openhab.items
+
+import habapp_rules.system.watchdog
+import tests.helper.oh_item
+import tests.helper.test_case_base
+from habapp_rules.system.config.watchdog import WatchdogConfig, WatchdogItems, WatchdogParameter
+
+
+class TestWatchdog(tests.helper.test_case_base.TestCaseBase):
+	"""Tests for Watchdog Rule."""
+
+	def setUp(self) -> None:
+		"""Setup test case."""
+		tests.helper.test_case_base.TestCaseBase.setUp(self)
+
+		tests.helper.oh_item.add_mock_item(HABApp.openhab.items.NumberItem, "Unittest_Number", None)
+		tests.helper.oh_item.add_mock_item(HABApp.openhab.items.SwitchItem, "Unittest_Switch", None)
+		tests.helper.oh_item.add_mock_item(HABApp.openhab.items.SwitchItem, "Unittest_Number_Warning", None)
+		tests.helper.oh_item.add_mock_item(HABApp.openhab.items.SwitchItem, "Unittest_Switch_Warning", None)
+
+		self._watchdog_number = habapp_rules.system.watchdog.Watchdog(WatchdogConfig(
+			items=WatchdogItems(
+				observed="Unittest_Number",
+				warning="Unittest_Number_Warning"
+			)
+		))
+
+		self._watchdog_switch = habapp_rules.system.watchdog.Watchdog(WatchdogConfig(
+			items=WatchdogItems(
+				observed="Unittest_Switch",
+				warning="Unittest_Switch_Warning"
+			),
+			parameter=WatchdogParameter(
+				timeout=10
+			)
+		))
+
+	def test_cb_observed_state_updated(self):
+		"""Callback which is called if the observed item was updated."""
+
+		with unittest.mock.patch.object(self._watchdog_number, "_countdown") as number_countdown_mock, unittest.mock.patch.object(self._watchdog_switch, "_countdown") as switch_countdown_mock:
+			tests.helper.oh_item.item_state_event("Unittest_Number", 42)
+			number_countdown_mock.reset.assert_called_once()
+			switch_countdown_mock.reset.assert_not_called()
+			tests.helper.oh_item.assert_value("Unittest_Number_Warning", "OFF")
+			tests.helper.oh_item.assert_value("Unittest_Switch_Warning", None)
+
+			tests.helper.oh_item.item_state_event("Unittest_Switch", "OFF")
+			number_countdown_mock.reset.assert_called_once()
+			switch_countdown_mock.reset.assert_called_once()
+			tests.helper.oh_item.assert_value("Unittest_Number_Warning", "OFF")
+			tests.helper.oh_item.assert_value("Unittest_Switch_Warning", "OFF")