check_home_assistant_state.py

@@ -4,16 +4,15 @@
 
 import argparse
 import dataclasses
-import json
 import logging
 import re
 import sys
-from pathlib import Path
-from typing import Any
 from urllib.parse import urljoin
+from typing import Optional
+from pathlib import Path
 
-import nagiosplugin
 import requests
+import nagiosplugin
 
 _log = logging.getLogger("nagiosplugin")
 
@@ -61,6 +60,9 @@ class AttributeFilter:
         else:
             return cls(k, v, False)
 
+    def check(self, attributes: dict[str, str]) -> bool:
+        return self.negated ^ (attributes.get(self.attribute) == self.value)
+
 
 @dataclasses.dataclass
 class Entities(nagiosplugin.Resource):
@@ -68,18 +70,16 @@ class Entities(nagiosplugin.Resource):
     token: str
     device_class: str
     numeric: bool
-    area: str
-    label: str
     filters: list[AttributeFilter]
-    attribute: str | None
+    attribute: Optional[str]
     friendly_name: bool
     ignore_missing: bool
     include: list[str]
     exclude: list[str]
-    min: float | None = None
+    min: Optional[float] = None
-    max: float | None = None
+    max: Optional[float] = None
 
-    def hass_get(self, endpoint: str) -> Any:
+    def hass_get(self, endpoint: str) -> requests.Response:
         headers = {
             "Authorization": "Bearer " + self.token,
             "Content-Type": "application/json",
@@ -91,18 +91,6 @@ class Entities(nagiosplugin.Resource):
 
         return r.json()
 
-    def hass_post(self, endpoint: str, data: Any) -> Any:
-        headers = {
-            "Authorization": "Bearer " + self.token,
-            "Content-Type": "application/json",
-        }
-
-        r = requests.post(urljoin(self.url, endpoint), headers=headers, json=data)
-        if not r.ok:
-            raise Exception("Failed to query Home Assistant API: " + r.text)
-
-        return r.json()
-
     def check_api(self):
         message = self.hass_get("/api/").get("message")
         if message != "API running.":
@@ -148,47 +136,14 @@ class Entities(nagiosplugin.Resource):
         )
 
     def probe(self):
-        template_filter = f"states|selectattr('attributes.device_class', 'eq', '{ self.device_class }')"
+        response = self.hass_get("/api/states")
-        for f in self.filters:
-            op = "ne" if f.negated else "eq"
-            template_filter += (
-                f"|selectattr('attributes.{ f.attribute }', '{ op }', '{ f.value }')"
-            )
-
-        if self.area:
-            template_filter += (
-                f"|selectattr('entity_id', 'in', area_entities('{ self.area }'))"
-            )
-        if self.label:
-            template_filter += (
-                f"|selectattr('entity_id', 'in', label_entities('{ self.label }'))"
-            )
-
-        if self.exclude:
-            template_filter += (
-                f"|rejectattr('entity_id', 'in', { json.dumps(self.exclude) })"
-            )
-        if self.include:
-            template_filter = (
-                f"({template_filter}|list + "
-                f" states|selectattr('entity_id', 'in', { json.dumps(self.include) })|list"
-                ")|unique(attribute='entity_id')"
-            )
-        template = (
-            "{% set ns = namespace(out=[]) %}\n"
-            f"{{% for s in { template_filter } %}}\n"
-            "  {% set ns.out = ns.out + [{\n"
-            '    "entity_id": s.entity_id,\n'
-            '    "state": s.state,\n'
-            '    "attributes": s.attributes,\n'
-            "  }]\n"
-            "  %}\n"
-            "{% endfor %}\n"
-            "{{ ns.out|tojson }}"
-        )
-
-        response = self.hass_post("/api/template", {"template": template})
         for state in response:
+            if (
+                state["attributes"].get("device_class") == self.device_class
+                and all(filter.check(state["attributes"]) for filter in self.filters)
+                and (len(self.include) == 0 or state["entity_id"] in self.include)
+                and state["entity_id"] not in self.exclude
+            ):
                 yield from self._state_to_metric(state)
 
 
@@ -196,9 +151,9 @@ class RegexContext(nagiosplugin.Context):
     def __init__(
         self,
         name: str,
-        ok: str | None = None,
+        ok: str = None,
-        warning: str | None = None,
+        warning: str = None,
-        critical: str | None = None,
+        critical: str = None,
         fmt_metric=None,
         result_cls=nagiosplugin.Result,
     ):
@@ -252,7 +207,7 @@ class Icinga2ConfAction(argparse.Action):
         self,
         param_prefix: str,
         parser: argparse.ArgumentParser,
-        order: int | None = None,
+        order: Optional[int] = None,
     ):
         for action in parser._actions:
             if action.dest in [self.dest, "help", "version"] or isinstance(
@@ -374,8 +329,6 @@ def main():
         action="extend",
         help="explicitly include entities by id. Other entities will not be considered if this is specified. Listed entities must also match the specified device class",
     )
-    common_args.add_argument("--area", type=str, help="area to filter by")
-    common_args.add_argument("--label", type=str, help="label to filter by")
     common_args.add_argument(
         "-e",
         "--exclude",
@@ -452,8 +405,6 @@ def main():
         "device_class": args.device_class,
         "attribute": args.attribute,
         "filters": args.filter,
-        "area": args.area,
-        "label": args.label,
         "include": args.include,
         "exclude": args.exclude,
         "friendly_name": args.friendly,

pyproject.toml

@@ -1,6 +1 @@
-[tool.ruff]
+[tool.black]
-line-length = 88
-
-[tool.ruff.lint]
-select = ["E4", "E7", "E9", "F", "I", "C4", "UP", "PERF", "PL", "SIM"]
-