""" Present observational weather data to Prometheus """ import json import time import os import requests from flask import Flask, Response app = Flask(__name__) def fetch_weather_data(location: int, apikey: str) -> dict: """ Fetch current data from the Met Office for the provided postcode """ obs_data = requests.get( f'http://api.weatherapi.com/v1/current.json?' f'key={apikey}&' f'q={location}&' f'aqi=yes' ) return json.loads(obs_data.content) @app.route('/metrics') def metrics(): """ Output Prometheus-style metrics """ apikey = os.environ.get('WEATHER_APIKEY') location = os.environ.get('WEATHER_LOCATION') latest_data = fetch_weather_data(location, apikey)['current'] ret_data = [ { '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_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']), }, ] ret_strs = list() 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( 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')