Add AC input watchdog that notifies when commanded ON but grid power is not flowing

solixauto/rules.py

@@ -181,6 +181,42 @@ class BatteryFloor:
         )
 
 
+class AcInputWatchdog:
+    def __init__(self, raw):
+        if not isinstance(raw, dict):
+            raise ProfileError("safety.ac_input_watchdog must be a mapping")
+
+        self.enabled = bool(raw.get("enabled", False))
+        self.field = str(raw.get("field") or "ac_input_power")
+
+        try:
+            self.min_watts = float(raw.get("min_watts", 50))
+        except (TypeError, ValueError):
+            raise ProfileError(
+                "safety.ac_input_watchdog.min_watts must be a number"
+            ) from None
+
+        self.dwell = parse_duration(
+            raw.get("for", 300), "safety.ac_input_watchdog.for"
+        )
+        self.notify_recovered = bool(raw.get("notify_recovered", True))
+        self.notify_template = raw.get("notify_template") or (
+            "{clock} {source_name}: {target_name} has been commanded ON but "
+            "{field} has stayed under {min_watts}W for {duration}. Check that "
+            "grid power is actually reaching the plug."
+        )
+        self.recovered_template = raw.get("recovered_template") or (
+            "{clock} {source_name}: {field} is back above {min_watts}W. "
+            "{target_name} is charging normally again."
+        )
+
+    def describe(self):
+        return (
+            f"{self.field} < {self.min_watts:g}W while ON for "
+            f"{format_duration(self.dwell)}"
+        )
+
+
 class NotificationSettings:
     def __init__(self, raw):
         if not isinstance(raw, dict):
@@ -381,6 +417,9 @@ class PowerProfile:
         floor = safety.get("battery_floor")
         self.battery_floor = BatteryFloor(floor) if floor else None
 
+        watchdog = safety.get("ac_input_watchdog")
+        self.ac_input_watchdog = AcInputWatchdog(watchdog) if watchdog else None
+
         limits = raw.get("limits") or {}
         self.min_gap = parse_duration(
             limits.get("min_seconds_between_actions", DEFAULT_MIN_GAP),
@@ -506,6 +545,14 @@ def validate(profile):
             "field on this device"
         )
 
+    if profile.ac_input_watchdog is not None:
+        if profile.ac_input_watchdog.field not in available:
+            problems.append(
+                f"safety.ac_input_watchdog.field "
+                f"{profile.ac_input_watchdog.field!r} is not a field on this "
+                "device"
+            )
+
     on_rules = [r for r in profile.active_rules() if r.desired_state() is True]
     off_rules = [r for r in profile.active_rules() if r.desired_state() is False]
     if not on_rules:
@@ -551,6 +598,9 @@ def validate(profile):
             profile.notifications.switch_failed_template,
             profile.notifications.switch_unconfirmed_template,
         ]
+        if profile.ac_input_watchdog is not None:
+            templates_to_check.append(profile.ac_input_watchdog.notify_template)
+            templates_to_check.append(profile.ac_input_watchdog.recovered_template)
         for rule in profile.active_rules():
             if rule.notify_template:
                 templates_to_check.append(rule.notify_template)
@@ -559,7 +609,8 @@ def validate(profile):
             "profile", "rule", "action", "action_word", "source_name", "source_model",
             "source_serial", "target_name", "target_model", "target_host",
             "target_channel", "condition", "time", "clock", "reason", "event",
-            "last_seen", "outage", "attempted", "actual",
+            "last_seen", "outage", "attempted", "actual", "field", "min_watts",
+            "duration",
         }
         for template in templates_to_check:
             for field in notify_module.template_fields(template):