"""Support for Prometheus metrics export.""" from __future__ import annotations from collections import defaultdict from collections.abc import Callable, Sequence from dataclasses import astuple, dataclass import logging import string from typing import Any, cast from aiohttp import web import prometheus_client from prometheus_client.metrics import MetricWrapperBase import voluptuous as vol from homeassistant import core as hacore from homeassistant.components.alarm_control_panel import AlarmControlPanelState from homeassistant.components.climate import ( ATTR_CURRENT_TEMPERATURE, ATTR_FAN_MODE, ATTR_FAN_MODES, ATTR_HVAC_ACTION, ATTR_HVAC_MODES, ATTR_TARGET_TEMP_HIGH, ATTR_TARGET_TEMP_LOW, HVACAction, ) from homeassistant.components.cover import ( ATTR_CURRENT_POSITION, ATTR_CURRENT_TILT_POSITION, ) from homeassistant.components.fan import ( ATTR_DIRECTION, ATTR_OSCILLATING, ATTR_PERCENTAGE, ATTR_PRESET_MODE, ATTR_PRESET_MODES, DIRECTION_FORWARD, DIRECTION_REVERSE, ) from homeassistant.components.http import KEY_HASS, HomeAssistantView from homeassistant.components.humidifier import ATTR_AVAILABLE_MODES, ATTR_HUMIDITY from homeassistant.components.light import ATTR_BRIGHTNESS from homeassistant.components.sensor import SensorDeviceClass # Alias water_heater constants to avoid name clashes with similarly named climate constants from homeassistant.components.water_heater import ( ATTR_AWAY_MODE as WATER_HEATER_ATTR_AWAY_MODE, ATTR_CURRENT_TEMPERATURE as WATER_HEATER_ATTR_CURRENT_TEMPERATURE, ATTR_MAX_TEMP as WATER_HEATER_ATTR_MAX_TEMP, ATTR_MIN_TEMP as WATER_HEATER_ATTR_MIN_TEMP, ATTR_OPERATION_LIST as WATER_HEATER_ATTR_OPERATION_LIST, ATTR_OPERATION_MODE as WATER_HEATER_ATTR_OPERATION_MODE, ATTR_TARGET_TEMP_HIGH as WATER_HEATER_ATTR_TARGET_TEMP_HIGH, ATTR_TARGET_TEMP_LOW as WATER_HEATER_ATTR_TARGET_TEMP_LOW, ) from homeassistant.const import ( ATTR_BATTERY_LEVEL, ATTR_DEVICE_CLASS, ATTR_FRIENDLY_NAME, ATTR_MODE, ATTR_TEMPERATURE, ATTR_UNIT_OF_MEASUREMENT, CONTENT_TYPE_TEXT_PLAIN, EVENT_STATE_CHANGED, PERCENTAGE, STATE_CLOSED, STATE_CLOSING, STATE_ON, STATE_OPEN, STATE_OPENING, STATE_UNAVAILABLE, STATE_UNKNOWN, UnitOfTemperature, ) from homeassistant.core import Event, EventStateChangedData, HomeAssistant, State from homeassistant.helpers import ( area_registry as ar, config_validation as cv, device_registry as dr, entity_registry as er, entityfilter, floor_registry as fr, state as state_helper, ) from homeassistant.helpers.area_registry import ( EVENT_AREA_REGISTRY_UPDATED, AreaEntry, EventAreaRegistryUpdatedData, ) from homeassistant.helpers.device_registry import ( EVENT_DEVICE_REGISTRY_UPDATED, EventDeviceRegistryUpdatedData, ) from homeassistant.helpers.entity_registry import ( EVENT_ENTITY_REGISTRY_UPDATED, EventEntityRegistryUpdatedData, ) from homeassistant.helpers.entity_values import EntityValues from homeassistant.helpers.floor_registry import ( EVENT_FLOOR_REGISTRY_UPDATED, EventFloorRegistryUpdatedData, FloorEntry, ) from homeassistant.helpers.typing import ConfigType from homeassistant.util.dt import as_timestamp from homeassistant.util.unit_conversion import TemperatureConverter _LOGGER = logging.getLogger(__name__) API_ENDPOINT = "/api/prometheus" IGNORED_STATES = frozenset({STATE_UNAVAILABLE, STATE_UNKNOWN}) DOMAIN = "prometheus" CONF_FILTER = "filter" CONF_REQUIRES_AUTH = "requires_auth" CONF_PROM_NAMESPACE = "namespace" CONF_COMPONENT_CONFIG = "component_config" CONF_COMPONENT_CONFIG_GLOB = "component_config_glob" CONF_COMPONENT_CONFIG_DOMAIN = "component_config_domain" CONF_DEFAULT_METRIC = "default_metric" CONF_OVERRIDE_METRIC = "override_metric" COMPONENT_CONFIG_SCHEMA_ENTRY = vol.Schema( {vol.Optional(CONF_OVERRIDE_METRIC): cv.string} ) ALLOWED_METRIC_CHARS = set(string.ascii_letters + string.digits + "_:") DEFAULT_NAMESPACE = "homeassistant" CONFIG_SCHEMA = vol.Schema( { DOMAIN: vol.All( { vol.Optional(CONF_FILTER, default={}): entityfilter.FILTER_SCHEMA, vol.Optional(CONF_PROM_NAMESPACE, default=DEFAULT_NAMESPACE): cv.string, vol.Optional(CONF_REQUIRES_AUTH, default=True): cv.boolean, vol.Optional(CONF_DEFAULT_METRIC): cv.string, vol.Optional(CONF_OVERRIDE_METRIC): cv.string, vol.Optional(CONF_COMPONENT_CONFIG, default={}): vol.Schema( {cv.entity_id: COMPONENT_CONFIG_SCHEMA_ENTRY} ), vol.Optional(CONF_COMPONENT_CONFIG_GLOB, default={}): vol.Schema( {cv.string: COMPONENT_CONFIG_SCHEMA_ENTRY} ), vol.Optional(CONF_COMPONENT_CONFIG_DOMAIN, default={}): vol.Schema( {cv.string: COMPONENT_CONFIG_SCHEMA_ENTRY} ), } ) }, extra=vol.ALLOW_EXTRA, ) def setup(hass: HomeAssistant, config: ConfigType) -> bool: """Activate Prometheus component.""" hass.http.register_view(PrometheusView(config[DOMAIN][CONF_REQUIRES_AUTH])) conf: dict[str, Any] = config[DOMAIN] entity_filter: entityfilter.EntityFilter = conf[CONF_FILTER] namespace: str = conf[CONF_PROM_NAMESPACE] climate_units = hass.config.units.temperature_unit override_metric: str | None = conf.get(CONF_OVERRIDE_METRIC) default_metric: str | None = conf.get(CONF_DEFAULT_METRIC) component_config = EntityValues( conf[CONF_COMPONENT_CONFIG], conf[CONF_COMPONENT_CONFIG_DOMAIN], conf[CONF_COMPONENT_CONFIG_GLOB], ) area_registry = ar.async_get(hass) device_registry = dr.async_get(hass) entity_registry = er.async_get(hass) floor_registry = fr.async_get(hass) metrics = PrometheusMetrics( entity_filter, namespace, climate_units, component_config, override_metric, default_metric, area_registry, device_registry, entity_registry, floor_registry, ) hass.bus.listen(EVENT_STATE_CHANGED, metrics.handle_state_changed_event) hass.bus.listen( EVENT_ENTITY_REGISTRY_UPDATED, metrics.handle_entity_registry_updated, ) hass.bus.listen( EVENT_DEVICE_REGISTRY_UPDATED, metrics.handle_device_registry_updated, ) hass.bus.listen(EVENT_AREA_REGISTRY_UPDATED, metrics.handle_area_registry_updated) hass.bus.listen(EVENT_FLOOR_REGISTRY_UPDATED, metrics.handle_floor_registry_updated) for floor in floor_registry.async_list_floors(): metrics.handle_floor(floor) for area in area_registry.async_list_areas(): metrics.handle_area(area) for state in hass.states.all(): if entity_filter(state.entity_id): metrics.handle_state(state) return True @dataclass(frozen=True, slots=True) class MetricNameWithLabelValues: """Class to represent a metric with its label values. The prometheus client library doesn't easily allow us to get back the information we put into it. Specifically, it is very expensive to query which label values have been set for metrics. This class is used to hold a bit of data we need to efficiently remove labelsets from metrics. """ metric_name: str label_values: tuple[str, ...] class PrometheusMetrics: """Model all of the metrics which should be exposed to Prometheus.""" def __init__( self, entity_filter: entityfilter.EntityFilter, namespace: str, climate_units: UnitOfTemperature, component_config: EntityValues, override_metric: str | None, default_metric: str | None, area_registry: ar.AreaRegistry, device_registry: dr.DeviceRegistry, entity_registry: er.EntityRegistry, floor_registry: fr.FloorRegistry, ) -> None: """Initialize Prometheus Metrics.""" self._component_config = component_config self._override_metric = override_metric self._default_metric = default_metric self._filter = entity_filter self._sensor_metric_handlers: list[ Callable[[State, str | None], str | None] ] = [ self._sensor_override_component_metric, self._sensor_override_metric, self._sensor_timestamp_metric, self._sensor_attribute_metric, self._sensor_default_metric, self._sensor_fallback_metric, ] if namespace: self.metrics_prefix = f"{namespace}_" else: self.metrics_prefix = "" self._metrics: dict[str, MetricWrapperBase] = {} self._metrics_by_entity_id: dict[str, set[MetricNameWithLabelValues]] = ( defaultdict(set) ) self._climate_units = climate_units self._area_info_metrics: dict[str, MetricNameWithLabelValues] = {} self._floor_info_metrics: dict[str, MetricNameWithLabelValues] = {} self.area_registry = area_registry self.device_registry = device_registry self.entity_registry = entity_registry self.floor_registry = floor_registry def handle_state_changed_event(self, event: Event[EventStateChangedData]) -> None: """Handle new messages from the bus.""" if (state := event.data.get("new_state")) is None: return if not self._filter(state.entity_id): _LOGGER.debug("Filtered out entity %s", state.entity_id) return if ( old_state := event.data.get("old_state") ) is not None and old_state.attributes.get( ATTR_FRIENDLY_NAME ) != state.attributes.get(ATTR_FRIENDLY_NAME): self._remove_labelsets(old_state.entity_id) self.handle_state(state) def handle_state(self, state: State) -> None: """Add/update a state in Prometheus.""" entity_id = state.entity_id _LOGGER.debug("Handling state update for %s", entity_id) if not self._metrics_by_entity_id[state.entity_id]: area_id = self._find_area_id(state.entity_id) if area_id is not None: self._add_entity_info(state.entity_id, area_id) labels = self._labels(state) self._metric( "state_change", prometheus_client.Counter, "The number of state changes", labels, ).inc() self._metric( "entity_available", prometheus_client.Gauge, "Entity is available (not in the unavailable or unknown state)", labels, ).set(float(state.state not in IGNORED_STATES)) self._metric( "last_updated_time_seconds", prometheus_client.Gauge, "The last_updated timestamp", labels, ).set(state.last_updated.timestamp()) if state.state in IGNORED_STATES: self._remove_labelsets( entity_id, { "state_change", "entity_available", "last_updated_time_seconds", "entity_info", }, ) else: domain, _ = hacore.split_entity_id(entity_id) handler = f"_handle_{domain}" if hasattr(self, handler) and state.state: getattr(self, handler)(state) def handle_entity_registry_updated( self, event: Event[EventEntityRegistryUpdatedData] ) -> None: """Listen for deleted, disabled or renamed entities and remove them from the Prometheus Registry.""" if event.data["action"] in (None, "create"): return entity_id = event.data.get("entity_id") _LOGGER.debug("Handling entity update for %s", entity_id) metrics_entity_id: str | None = None if event.data["action"] == "remove": metrics_entity_id = entity_id elif event.data["action"] == "update": changes = event.data["changes"] if "entity_id" in changes: metrics_entity_id = changes["entity_id"] elif "disabled_by" in changes: metrics_entity_id = entity_id elif "area_id" in changes or "device_id" in changes: if entity_id is not None: self._remove_entity_info(entity_id) area_id = self._find_area_id(entity_id) if area_id is not None: self._add_entity_info(entity_id, area_id) if metrics_entity_id: self._remove_labelsets(metrics_entity_id) def handle_device_registry_updated( self, event: Event[EventDeviceRegistryUpdatedData] ) -> None: """Listen for changes of devices' area_id.""" if event.data["action"] != "update" or "area_id" not in event.data["changes"]: return device_id = event.data["device_id"] _LOGGER.debug("Handling device update for %s", device_id) device = self.device_registry.async_get(device_id) if device is None: return area_id = device.area_id for entity_id in ( entity.entity_id for entity in er.async_entries_for_device(self.entity_registry, device_id) if entity.area_id is None and entity.entity_id in self._metrics_by_entity_id ): self._remove_entity_info(entity_id) if area_id is not None: self._add_entity_info(entity_id, area_id) def handle_area_registry_updated( self, event: Event[EventAreaRegistryUpdatedData] ) -> None: """Listen for changes to areas.""" area_id = event.data.get("area_id") if area_id is None: return action = event.data["action"] _LOGGER.debug("Handling area update for %s (%s)", area_id, action) if action in {"update", "remove"}: metric = self._area_info_metrics.pop(area_id, None) if metric is not None: metric_name, label_values = astuple(metric) self._metrics[metric_name].remove(*label_values) if action in {"update", "create"}: area = self.area_registry.async_get_area(area_id) if area is not None: self.handle_area(area) def handle_area(self, area: AreaEntry) -> None: """Add/update an area in Prometheus.""" metric_name = "area_info" labels = { "area": area.id, "area_name": area.name, "floor": area.floor_id if area.floor_id is not None else "", } self._area_info_metrics[labels["area"]] = MetricNameWithLabelValues( metric_name, tuple(labels.values()) ) self._metric( metric_name, prometheus_client.Gauge, "Area information", labels, ).set(1.0) def handle_floor_registry_updated( self, event: Event[EventFloorRegistryUpdatedData] ) -> None: """Listen for changes to floors.""" floor_id = event.data.get("floor_id") if floor_id is None: return action = event.data["action"] _LOGGER.debug("Handling floor update for %s (%s)", floor_id, action) if action in {"update", "remove"}: metric = self._floor_info_metrics.pop(str(floor_id), None) if metric is not None: metric_name, label_values = astuple(metric) self._metrics[metric_name].remove(*label_values) if action in {"update", "create"}: floor = self.floor_registry.async_get_floor(str(floor_id)) if floor is not None: self.handle_floor(floor) def handle_floor(self, floor: FloorEntry) -> None: """Add/update a floor in Prometheus.""" metric_name = "floor_info" labels = { "floor": floor.floor_id, "floor_name": floor.name, "floor_level": str(floor.level) if floor.level is not None else "", } self._floor_info_metrics[labels["floor"]] = MetricNameWithLabelValues( metric_name, tuple(labels.values()) ) self._metric( metric_name, prometheus_client.Gauge, "Floor information", labels, ).set(1.0) def _remove_labelsets( self, entity_id: str, ignored_metric_names: set[str] | None = None, ) -> None: """Remove labelsets matching the given entity id from all non-ignored metrics.""" if ignored_metric_names is None: ignored_metric_names = set() metric_set = self._metrics_by_entity_id[entity_id] removed_metrics = set() for metric in metric_set: metric_name, label_values = astuple(metric) if metric_name in ignored_metric_names: continue _LOGGER.debug( "Removing labelset %s from %s for entity_id: %s", label_values, metric_name, entity_id, ) removed_metrics.add(metric) self._metrics[metric_name].remove(*label_values) metric_set -= removed_metrics if not metric_set: del self._metrics_by_entity_id[entity_id] def _handle_attributes(self, state: State) -> None: for key, value in state.attributes.items(): try: value = float(value) except (ValueError, TypeError): continue self._metric( f"{state.domain}_attr_{key.lower()}", prometheus_client.Gauge, f"{key} attribute of {state.domain} entity", self._labels(state), ).set(value) def _metric[_MetricBaseT: MetricWrapperBase]( self, metric_name: str, factory: type[_MetricBaseT], documentation: str, labels: dict[str, str], ) -> _MetricBaseT: try: metric = cast(_MetricBaseT, self._metrics[metric_name]) except KeyError: full_metric_name = self._sanitize_metric_name( f"{self.metrics_prefix}{metric_name}" ) self._metrics[metric_name] = factory( full_metric_name, documentation, labels.keys(), registry=prometheus_client.REGISTRY, ) metric = cast(_MetricBaseT, self._metrics[metric_name]) if "entity" in labels: self._metrics_by_entity_id[labels["entity"]].add( MetricNameWithLabelValues(metric_name, tuple(labels.values())) ) return metric.labels(**labels) @staticmethod def _sanitize_metric_name(metric: str) -> str: metric.replace("\u03bc", "\u00b5") return "".join( [c if c in ALLOWED_METRIC_CHARS else f"u{hex(ord(c))}" for c in metric] ) @staticmethod def state_as_number(state: State) -> float | None: """Return state as a float, or None if state cannot be converted.""" try: if state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.TIMESTAMP: value = as_timestamp(state.state) else: value = state_helper.state_as_number(state) except ValueError: _LOGGER.debug("Could not convert %s to float", state) value = None return value @staticmethod def _labels( state: State, extra_labels: dict[str, str] | None = None, ) -> dict[str, Any]: if extra_labels is None: extra_labels = {} labels = { "entity": state.entity_id, "domain": state.domain, "friendly_name": state.attributes.get(ATTR_FRIENDLY_NAME), } if not labels.keys().isdisjoint(extra_labels.keys()): conflicting_keys = labels.keys() & extra_labels.keys() raise ValueError( f"extra_labels contains conflicting keys: {conflicting_keys}" ) return labels | extra_labels def _remove_entity_info(self, entity_id: str) -> None: """Remove an entity-area-relation in Prometheus.""" self._remove_labelsets( entity_id, { metric_set.metric_name for metric_set in self._metrics_by_entity_id[entity_id] if metric_set.metric_name != "entity_info" }, ) def _add_entity_info(self, entity_id: str, area_id: str) -> None: """Add/update an entity-area-relation in Prometheus.""" self._metric( "entity_info", prometheus_client.Gauge, "The area of an entity", { "entity": entity_id, "area": area_id, }, ).set(1.0) def _find_area_id(self, entity_id: str) -> str | None: """Find area of entity or parent device.""" entity = self.entity_registry.async_get(entity_id) if entity is None: return None area_id = entity.area_id if area_id is None and entity.device_id is not None: device = self.device_registry.async_get(entity.device_id) if device is not None: area_id = device.area_id return area_id def _battery_metric(self, state: State) -> None: if (battery_level := state.attributes.get(ATTR_BATTERY_LEVEL)) is None: return try: value = float(battery_level) except ValueError: return self._metric( "battery_level_percent", prometheus_client.Gauge, "Battery level as a percentage of its capacity", self._labels(state), ).set(value) def _temperature_metric( self, state: State, attr: str, metric_name: str, metric_description: str ) -> None: if (temp := state.attributes.get(attr)) is None: return if self._climate_units == UnitOfTemperature.FAHRENHEIT: temp = TemperatureConverter.convert( temp, UnitOfTemperature.FAHRENHEIT, UnitOfTemperature.CELSIUS ) self._metric( metric_name, prometheus_client.Gauge, metric_description, self._labels(state), ).set(temp) def _bool_metric( self, state: State, attr: str, metric_name: str, metric_description: str, true_values: set[Any] | None = None, ) -> None: value = state.attributes.get(attr) if value is None: return result = bool(value) if true_values is None else value in true_values self._metric( metric_name, prometheus_client.Gauge, metric_description, self._labels(state), ).set(float(result)) def _float_metric( self, state: State, attr: str, metric_name: str, metric_description: str, ) -> None: value = state.attributes.get(attr) if value is None: return self._metric( metric_name, prometheus_client.Gauge, metric_description, self._labels(state), ).set(float(value)) def _enum_metric( self, state: State, current_value: Any | None, values: Sequence[str] | None, metric_name: str, metric_description: str, enum_label_name: str, ) -> None: if current_value is None or values is None: return for value in values: self._metric( metric_name, prometheus_client.Gauge, metric_description, self._labels(state, {enum_label_name: value}), ).set(float(value == current_value)) def _numeric_metric(self, state: State, domain: str, title: str) -> None: if (value := self.state_as_number(state)) is None: return if unit := self._unit_string(state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)): metric = self._metric( f"{domain}_state_{unit}", prometheus_client.Gauge, f"State of the {title} measured in {unit}", self._labels(state), ) else: metric = self._metric( f"{domain}_state", prometheus_client.Gauge, f"State of the {title}", self._labels(state), ) if ( state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfTemperature.FAHRENHEIT ): value = TemperatureConverter.convert( value, UnitOfTemperature.FAHRENHEIT, UnitOfTemperature.CELSIUS ) metric.set(value) def _handle_binary_sensor(self, state: State) -> None: self._numeric_metric(state, "binary_sensor", "binary boolean") def _handle_input_boolean(self, state: State) -> None: self._numeric_metric(state, "input_boolean", "input boolean") def _handle_input_number(self, state: State) -> None: self._numeric_metric(state, "input_number", "input number") def _handle_number(self, state: State) -> None: self._numeric_metric(state, "number", "number") def _handle_device_tracker(self, state: State) -> None: self._numeric_metric(state, "device_tracker", "device tracker") def _handle_person(self, state: State) -> None: self._numeric_metric(state, "person", "person") def _handle_lock(self, state: State) -> None: self._numeric_metric(state, "lock", "lock") def _handle_cover(self, state: State) -> None: self._enum_metric( state, state.state, [STATE_CLOSED, STATE_CLOSING, STATE_OPEN, STATE_OPENING], "cover_state", "State of the cover (0/1)", "state", ) self._float_metric( state, ATTR_CURRENT_POSITION, "cover_position", "Position of the cover (0-100)", ) self._float_metric( state, ATTR_CURRENT_TILT_POSITION, "cover_tilt_position", "Tilt Position of the cover (0-100)", ) def _handle_light(self, state: State) -> None: if (value := self.state_as_number(state)) is None: return brightness = state.attributes.get(ATTR_BRIGHTNESS) if state.state == STATE_ON and brightness is not None: value = float(brightness) / 255.0 value = value * 100 self._metric( "light_brightness_percent", prometheus_client.Gauge, "Light brightness percentage (0..100)", self._labels(state), ).set(value) def _handle_climate(self, state: State) -> None: self._temperature_metric( state, ATTR_TEMPERATURE, "climate_target_temperature_celsius", "Target temperature in degrees Celsius", ) self._temperature_metric( state, ATTR_TARGET_TEMP_HIGH, "climate_target_temperature_high_celsius", "Target high temperature in degrees Celsius", ) self._temperature_metric( state, ATTR_TARGET_TEMP_LOW, "climate_target_temperature_low_celsius", "Target low temperature in degrees Celsius", ) self._temperature_metric( state, ATTR_CURRENT_TEMPERATURE, "climate_current_temperature_celsius", "Current temperature in degrees Celsius", ) self._enum_metric( state, ( (attr := state.attributes.get(ATTR_HVAC_ACTION)) and getattr(attr, "value", attr) ), [action.value for action in HVACAction], "climate_action", "HVAC action", "action", ) self._enum_metric( state, state.state, state.attributes.get(ATTR_HVAC_MODES), "climate_mode", "HVAC mode", "mode", ) self._enum_metric( state, state.attributes.get(ATTR_PRESET_MODE), state.attributes.get(ATTR_PRESET_MODES), "climate_preset_mode", "Preset mode enum", "mode", ) self._enum_metric( state, state.attributes.get(ATTR_FAN_MODE), state.attributes.get(ATTR_FAN_MODES), "climate_fan_mode", "Fan mode enum", "mode", ) def _handle_humidifier(self, state: State) -> None: self._numeric_metric(state, "humidifier", "humidifier") self._float_metric( state, ATTR_HUMIDITY, "humidifier_target_humidity_percent", "Target Relative Humidity", ) self._enum_metric( state, state.attributes.get(ATTR_MODE), state.attributes.get(ATTR_AVAILABLE_MODES), "humidifier_mode", "Humidifier Mode", "mode", ) def _handle_water_heater(self, state: State) -> None: # Temperatures self._temperature_metric( state, ATTR_TEMPERATURE, "water_heater_temperature_celsius", "Target temperature in degrees Celsius", ) self._temperature_metric( state, WATER_HEATER_ATTR_CURRENT_TEMPERATURE, "water_heater_current_temperature_celsius", "Target temperature in degrees Celsius", ) self._temperature_metric( state, WATER_HEATER_ATTR_TARGET_TEMP_HIGH, "water_heater_target_temperature_high_celsius", "Target high temperature in degrees Celsius", ) self._temperature_metric( state, WATER_HEATER_ATTR_TARGET_TEMP_LOW, "water_heater_target_temperature_low_celsius", "Target low temperature in degrees Celsius", ) self._temperature_metric( state, WATER_HEATER_ATTR_MIN_TEMP, "water_heater_min_temperature_celsius", "Minimum allowed temperature in degrees Celsius", ) self._temperature_metric( state, WATER_HEATER_ATTR_MAX_TEMP, "water_heater_max_temperature_celsius", "Maximum allowed temperature in degrees Celsius", ) self._enum_metric( state, state.attributes.get(WATER_HEATER_ATTR_OPERATION_MODE) or state.state, state.attributes.get(WATER_HEATER_ATTR_OPERATION_LIST), "water_heater_operation_mode", "Water heater operation mode", "mode", ) # Away mode bool self._bool_metric( state, WATER_HEATER_ATTR_AWAY_MODE, "water_heater_away_mode", "Whether away mode is on (0/1)", {STATE_ON}, ) def _handle_switch(self, state: State) -> None: self._numeric_metric(state, "switch", "switch") self._handle_attributes(state) def _handle_fan(self, state: State) -> None: self._numeric_metric(state, "fan", "fan") self._float_metric( state, ATTR_PERCENTAGE, "fan_speed_percent", "Fan speed percent (0-100)" ) self._bool_metric( state, ATTR_OSCILLATING, "fan_is_oscillating", "Whether the fan is oscillating (0/1)", ) self._enum_metric( state, state.attributes.get(ATTR_PRESET_MODE), state.attributes.get(ATTR_PRESET_MODES), "fan_preset_mode", "Fan preset mode enum", "mode", ) fan_direction = state.attributes.get(ATTR_DIRECTION) if fan_direction in {DIRECTION_FORWARD, DIRECTION_REVERSE}: self._bool_metric( state, ATTR_DIRECTION, "fan_direction_reversed", "Fan direction reversed (bool)", {DIRECTION_REVERSE}, ) def _handle_zwave(self, state: State) -> None: self._battery_metric(state) def _handle_automation(self, state: State) -> None: self._metric( "automation_triggered_count", prometheus_client.Counter, "Count of times an automation has been triggered", self._labels(state), ).inc() def _handle_counter(self, state: State) -> None: if (value := self.state_as_number(state)) is None: return self._metric( "counter_value", prometheus_client.Gauge, "Value of counter entities", self._labels(state), ).set(value) def _handle_update(self, state: State) -> None: self._numeric_metric(state, "update", "update") def _handle_alarm_control_panel(self, state: State) -> None: self._enum_metric( state, state.state, [alarm_state.value for alarm_state in AlarmControlPanelState], "alarm_control_panel_state", "State of the alarm control panel (0/1)", "state", ) def _handle_sensor(self, state: State) -> None: unit = self._unit_string(state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)) for metric_handler in self._sensor_metric_handlers: metric = metric_handler(state, unit) if metric is not None: break if metric is not None and (value := self.state_as_number(state)) is not None: documentation = "State of the sensor" if unit: documentation = f"Sensor data measured in {unit}" if ( state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfTemperature.FAHRENHEIT ): value = TemperatureConverter.convert( value, UnitOfTemperature.FAHRENHEIT, UnitOfTemperature.CELSIUS ) self._metric( metric, prometheus_client.Gauge, documentation, self._labels(state), ).set(value) self._battery_metric(state) def _sensor_default_metric(self, state: State, unit: str | None) -> str | None: """Get default metric.""" return self._default_metric @staticmethod def _sensor_attribute_metric(state: State, unit: str | None) -> str | None: """Get metric based on device class attribute.""" metric = state.attributes.get(ATTR_DEVICE_CLASS) if metric is not None: return f"sensor_{metric}_{unit}" return None @staticmethod def _sensor_timestamp_metric(state: State, unit: str | None) -> str | None: """Get metric for timestamp sensors, which have no unit of measurement attribute.""" metric = state.attributes.get(ATTR_DEVICE_CLASS) if metric == SensorDeviceClass.TIMESTAMP: return f"sensor_{metric}_seconds" return None def _sensor_override_metric(self, state: State, unit: str | None) -> str | None: """Get metric from override in configuration.""" if self._override_metric: return self._override_metric return None def _sensor_override_component_metric( self, state: State, unit: str | None ) -> str | None: """Get metric from override in component configuration.""" return self._component_config.get(state.entity_id).get(CONF_OVERRIDE_METRIC) @staticmethod def _sensor_fallback_metric(state: State, unit: str | None) -> str | None: """Get metric from fallback logic for compatibility.""" if unit not in (None, ""): return f"sensor_unit_{unit}" return "sensor_state" @staticmethod def _unit_string(unit: str | None) -> str | None: """Get a formatted string of the unit.""" if unit is None: return None units = { UnitOfTemperature.CELSIUS: "celsius", UnitOfTemperature.FAHRENHEIT: "celsius", # F should go into C metric PERCENTAGE: "percent", } default = unit.replace("/", "_per_") # Unit conversion for CONCENTRATION_MICROGRAMS_PER_CUBIC_METER "μg/m³" # "μ" == "\u03bc" but the API uses "\u00b5" default = default.replace("\u03bc", "\u00b5") default = default.lower() return units.get(unit, default) class PrometheusView(HomeAssistantView): """Handle Prometheus requests.""" url = API_ENDPOINT name = "api:prometheus" def __init__(self, requires_auth: bool) -> None: """Initialize Prometheus view.""" self.requires_auth = requires_auth async def get(self, request: web.Request) -> web.Response: """Handle request for Prometheus metrics.""" _LOGGER.debug("Received Prometheus metrics request") hass = request.app[KEY_HASS] body = await hass.async_add_executor_job( prometheus_client.generate_latest, prometheus_client.REGISTRY ) return web.Response( body=body, content_type=CONTENT_TYPE_TEXT_PLAIN, )