Compare commits

..

No commits in common. "fa21f7e87942693c066856d0532e5bfb7d39620a" and "8d048bab85b804338e00b83c51c83a3b8611be80" have entirely different histories.

2 changed files with 23 additions and 77 deletions

View File

@ -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
max: float | None = None
min: Optional[float] = 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 }')"
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})
response = self.hass_get("/api/states")
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,
warning: str | None = None,
critical: str | None = None,
ok: str = None,
warning: str = 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,

View File

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