mirror of
https://github.com/nikdoof/homeassistant-prometheus-query.git
synced 2025-12-17 12:39:22 +00:00
4fbe9c5660
Fix lfasci/homeassistant-prometheus-query#5 Goal is to make a first pass to allow to be exposed in the energy dashboard. This requires usage of HA version 2021.9 (which is beta at time of writing) if we want to allow total_increase state_class
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
import logging, time
|
|
import requests
|
|
import json
|
|
import voluptuous as vol
|
|
import homeassistant.helpers.config_validation as cv
|
|
from datetime import timedelta
|
|
from homeassistant.helpers.config_validation import PLATFORM_SCHEMA
|
|
from homeassistant.helpers.entity import Entity
|
|
from homeassistant.util import Throttle
|
|
from homeassistant.const import (
|
|
CONF_NAME,
|
|
CONF_UNIT_OF_MEASUREMENT,
|
|
STATE_UNKNOWN,
|
|
)
|
|
from homeassistant.components.sensor import (
|
|
DEVICE_CLASSES_SCHEMA,
|
|
STATE_CLASSES_SCHEMA,
|
|
)
|
|
|
|
from prometheus_client import Summary
|
|
|
|
CONF_PROMETHEUS_URL = 'prometheus_url'
|
|
CONF_PROMETHEUS_QUERY = 'prometheus_query'
|
|
CONF_STATE_CLASS = 'state_class'
|
|
SCAN_INTERVAL = timedelta(seconds=600)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
|
vol.Required(CONF_PROMETHEUS_URL): cv.string,
|
|
vol.Required(CONF_PROMETHEUS_QUERY): cv.string,
|
|
vol.Required(CONF_NAME): cv.string,
|
|
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
|
|
vol.Optional(CONF_STATE_CLASS): STATE_CLASSES_SCHEMA,
|
|
})
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None):
|
|
"""Set up the sensor platform."""
|
|
prom_data = {
|
|
'url': str(config.get(CONF_PROMETHEUS_URL)) + "/api/v1/query",
|
|
'query': str(config.get(CONF_PROMETHEUS_QUERY)),
|
|
'name': str(config.get(CONF_NAME)),
|
|
'unit': str(config.get(CONF_UNIT_OF_MEASUREMENT)),
|
|
'state_class': str(config.get(CONF_STATE_CLASS)),
|
|
}
|
|
add_entities([PrometheusQuery(prom_data)], True)
|
|
|
|
|
|
class PrometheusQuery(Entity):
|
|
"""Representation of a Sensor."""
|
|
def __init__(self, prom_data):
|
|
"""Initialize the sensor."""
|
|
self._url = prom_data["url"]
|
|
self._query = prom_data["query"]
|
|
self._name = prom_data["name"]
|
|
self._state = None
|
|
self._unit_of_measurement = prom_data["unit"]
|
|
self._state_class = prom_data["state_class"]
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the sensor."""
|
|
return self._name
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the sensor."""
|
|
return self._state
|
|
|
|
@property
|
|
def unit_of_measurement(self):
|
|
"""Return the unit of measurement."""
|
|
return self._unit_of_measurement
|
|
|
|
@property
|
|
def state_class(self):
|
|
"""Return the state_class of the sensor"""
|
|
return self._state_class
|
|
|
|
def update(self):
|
|
"""Fetch new state data for the sensor.
|
|
This is the only method that should fetch new data for Home Assistant.
|
|
"""
|
|
try:
|
|
response = requests.get(self._url, params={'query': self._query})
|
|
if (response):
|
|
results = response.json()['data']['result']
|
|
self._state = results[0]['value'][1]
|
|
else:
|
|
self._state = STATE_UNKNOWN
|
|
except URLCallError:
|
|
_LOGGER.error("Error when retrieving update data")
|