182 lines
8.7 KiB
Python
182 lines
8.7 KiB
Python
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()
|