""" Present observational weather data to Prometheus """ import json import os from typing import TypedDict import requests from flask import Flask, Response app = Flask(__name__) class PromData(TypedDict): """ Describes the data being returned to Prometheus Args: TypedDict ([type]): data to return to Prometheus """ key: str labels: dict[str, str] type: str value: float def fetch_weather_data(location: str, apikey: str) -> dict[str, dict[str, float]]: """ Fetch current data from the Met Office for the provided postcode """ obs_data = requests.get(f"http://api.weatherapi.com/v1/current.json?key={apikey}&q={location}&aqi=yes", timeout=30) return dict(json.loads(obs_data.content)) @app.route("/metrics") def metrics() -> Response: """ Output Prometheus-style metrics """ apikey = os.environ.get("WEATHER_APIKEY") location = os.environ.get("WEATHER_LOCATION") if not location: return Response( f'There was a problem finding the weather for location: "{location}"', status=500, ) latest_data = fetch_weather_data(location, str(apikey))["current"] ret_data: list[PromData] = [ { "key": "sensor_weather_outdoor_temperature_celsius", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["temp_c"]), }, { "key": "sensor_weather_outdoor_humidity_percent", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["humidity"]), }, { "key": "sensor_weather_outdoor_uv", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["uv"]), }, { "key": "sensor_weather_outdoor_cloud_percent", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["cloud"]), }, { "key": "sensor_weather_outdoor_precip", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["precip_mm"]), }, { "key": "sensor_weather_outdoor_wind_speed", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["wind_kph"]), }, { "key": "sensor_weather_outdoor_wind_gust", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["gust_kph"]), }, { "key": "sensor_weather_outdoor_wind_direction", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["wind_degree"]), }, { "key": "sensor_weather_outdoor_temp_feel_c", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["feelslike_c"]), }, { "key": "sensor_weather_outdoor_pressure", "labels": { "location": location, }, "type": "gauge", "value": float(latest_data["pressure_mb"]), }, ] ret_strs = [] for item in ret_data: ret_strs.append(f'# HELP {item["key"]} Weather metric') ret_strs.append(f'# TYPE {item["key"]} {item["type"]}') ret_strs.append( str(item["key"]) + "{" + " ".join([f'{key}="{val}"' for key, val in item["labels"].items()]) + "} " + str( item["value"], ) ) resp = Response("\n".join(ret_strs)) resp.headers["Content-type"] = "text/plain" return resp if __name__ == "__main__": app.run(host="0.0.0.0")