import logging import math import time from typing import Dict, Any, Optional, List import httpx from config import settings from schemas.weather import CurrentWeatherMetrics, TimeSeriesPoint, TimeSeriesMetric, WeatherHistoryResponse logger = logging.getLogger("grabengasse_backend.prometheus") def calculate_dew_point(temp: Optional[float], humidity: Optional[float]) -> Optional[float]: """Calculate dew point (°C) using the Magnus-Tetens formula.""" if temp is None or humidity is None or humidity <= 0: return None a = 17.625 b = 243.04 try: alpha = ((a * temp) / (b + temp)) + math.log(humidity / 100.0) dew_point = (b * alpha) / (a - alpha) return round(dew_point, 1) except Exception: return None METRIC_QUERIES = { "temperature": 'homeassistant_sensor_temperature_celsius{entity=~"sensor.shelly_ws90_temperature|sensor.shelly_ws90_temperatur"}', "apparent_temperature": 'homeassistant_sensor_temperature_celsius{entity="sensor.shelly_ws90_apparent_temperature"}', "wind_chill": 'homeassistant_sensor_temperature_celsius{entity="sensor.shelly_ws90_wind_chill"}', "dew_point": 'homeassistant_sensor_temperature_celsius{entity="sensor.shelly_ws90_dew_point"}', "humidex": 'homeassistant_sensor_temperature_celsius{entity="sensor.shelly_ws90_humidex"}', "humidity": 'homeassistant_sensor_humidity_percent{entity=~"sensor.shelly_ws90_humidity|sensor.shelly_ws90_luftfeuchtigkeit"}', "pressure": 'homeassistant_sensor_atmospheric_pressure_hpa{entity=~"sensor.shelly_ws90_pressure|sensor.shelly_ws90_atmospharischer_druck"}', "rain_status": 'homeassistant_binary_sensor_state{entity=~"binary_sensor.shelly_ws90_rain_status|binary_sensor.shelly_ws90_rain"}', "wind_direction": 'homeassistant_sensor_unit_u0xb0{entity="sensor.shelly_ws90_wind_direction"} or on() vector(0)', "wind_speed": 'homeassistant_sensor_wind_speed_km_per_h{entity="sensor.shelly_ws90_wind_speed"} or on() vector(0)', "gust_speed": 'homeassistant_sensor_wind_speed_km_per_h{entity="sensor.shelly_ws90_gust_speed"} or on() vector(0)', "illuminance_lux": 'homeassistant_sensor_illuminance_lx{entity="sensor.shelly_ws90_illuminance"}', "uv_index": 'homeassistant_sensor_state{entity="sensor.shelly_ws90_uv_index"} or on() vector(0)', "battery_percent": 'homeassistant_sensor_battery_percent{entity=~"sensor.shelly_ws90_battery|sensor.shelly_ws90_batterie"}', "linkquality_lqi": 'homeassistant_sensor_unit_lqi{entity=~"sensor.shelly_ws90_linkquality|sensor.shelly_ws90_lqi"}' } METRIC_METADATA = { "temperature": {"label": "Temperature", "unit": "°C"}, "apparent_temperature": {"label": "Apparent Temperature", "unit": "°C"}, "dew_point": {"label": "Dew Point", "unit": "°C"}, "humidity": {"label": "Relative Humidity", "unit": "%"}, "pressure": {"label": "Atmospheric Pressure", "unit": "hPa"}, "wind_speed": {"label": "Wind Speed", "unit": "km/h"}, "gust_speed": {"label": "Gust Speed", "unit": "km/h"}, "illuminance_lux": {"label": "Solar Illuminance", "unit": "LUX"}, "uv_index": {"label": "UV Index", "unit": ""} } class PrometheusService: def __init__(self, base_url: str = settings.PROMETHEUS_URL): self.base_url = base_url.rstrip("/") async def _query_instant(self, client: httpx.AsyncClient, query: str) -> Optional[float]: try: url = f"{self.base_url}/api/v1/query" response = await client.get(url, params={"query": query}, timeout=5.0) if response.status_code == 200: data = response.json() results = data.get("data", {}).get("result", []) if results: val = results[0].get("value", [None, None])[1] return float(val) if val is not None else None except Exception as e: logger.warning(f"Error querying Prometheus instant '{query}': {e}") return None async def get_current_metrics(self) -> CurrentWeatherMetrics: async with httpx.AsyncClient() as client: values: Dict[str, Any] = {} for key, query in METRIC_QUERIES.items(): values[key] = await self._query_instant(client, query) rain_val = values.get("rain_status") rain_int = int(rain_val) if rain_val is not None else 0 rain_label = "Wet (Rain)" if rain_int == 1 else "Dry" dew_point_val = values.get("dew_point") if dew_point_val is None: dew_point_val = calculate_dew_point(values.get("temperature"), values.get("humidity")) return CurrentWeatherMetrics( temperature=values.get("temperature"), apparent_temperature=values.get("apparent_temperature"), wind_chill=values.get("wind_chill"), dew_point=dew_point_val, humidex=values.get("humidex"), humidity=values.get("humidity"), pressure=values.get("pressure"), rain_status=rain_int, rain_status_label=rain_label, wind_speed=values.get("wind_speed"), gust_speed=values.get("gust_speed"), wind_direction=values.get("wind_direction"), illuminance_lux=values.get("illuminance_lux"), uv_index=values.get("uv_index"), battery_percent=values.get("battery_percent"), linkquality_lqi=values.get("linkquality_lqi"), updated_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) ) async def get_history(self, hours: int = 24, step: int = 300) -> WeatherHistoryResponse: end_time = int(time.time()) start_time = end_time - (hours * 3600) results_map: Dict[str, TimeSeriesMetric] = {} async with httpx.AsyncClient() as client: for key, meta in METRIC_METADATA.items(): query = METRIC_QUERIES.get(key) if not query: continue try: url = f"{self.base_url}/api/v1/query_range" params = { "query": query, "start": start_time, "end": end_time, "step": step } response = await client.get(url, params=params, timeout=20.0) points: List[TimeSeriesPoint] = [] if response.status_code == 200: data = response.json() res = data.get("data", {}).get("result", []) if res: values_raw = res[0].get("values", []) for t, v in values_raw: try: points.append(TimeSeriesPoint(timestamp=int(t), value=round(float(v), 2))) except (ValueError, TypeError): pass results_map[key] = TimeSeriesMetric( name=key, label=meta["label"], unit=meta["unit"], data=points ) except Exception as e: logger.warning(f"Error fetching history for '{key}': {e}") results_map[key] = TimeSeriesMetric( name=key, label=meta["label"], unit=meta["unit"], data=[] ) # Compute dew point history if missing from Prometheus dew_metric = results_map.get("dew_point") temp_metric = results_map.get("temperature") humidity_metric = results_map.get("humidity") if (not dew_metric or len(dew_metric.data) == 0) and temp_metric and humidity_metric: hum_by_ts = {p.timestamp: p.value for p in humidity_metric.data} computed_dew_points: List[TimeSeriesPoint] = [] for p in temp_metric.data: h_val = hum_by_ts.get(p.timestamp) if h_val is not None: d_val = calculate_dew_point(p.value, h_val) if d_val is not None: computed_dew_points.append(TimeSeriesPoint(timestamp=p.timestamp, value=d_val)) results_map["dew_point"] = TimeSeriesMetric( name="dew_point", label="Dew Point", unit="°C", data=computed_dew_points ) return WeatherHistoryResponse( timeframe=f"{hours}h", metrics=results_map ) prometheus_service = PrometheusService()