160 lines
3.2 KiB
Python
160 lines
3.2 KiB
Python
"""
|
|
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
|
|
|
|
|
|
class FuelMix(TypedDict):
|
|
"""
|
|
Generational Fuel Mix
|
|
|
|
Args:
|
|
TypedDict ([type]): the particular fuel's usage
|
|
"""
|
|
|
|
fuel: str
|
|
perc: float
|
|
|
|
|
|
class Intensity(TypedDict):
|
|
"""
|
|
Intensity forecast
|
|
|
|
Args:
|
|
TypedDict ([type]): the intensity forecast
|
|
"""
|
|
|
|
forecast: int
|
|
index: str
|
|
|
|
|
|
class PostcodeData(TypedDict):
|
|
"""
|
|
Data for a postcode
|
|
|
|
Args:
|
|
TypedDict ([type]): Describes the date range, intensity and fuel mix
|
|
"""
|
|
|
|
_from: str
|
|
to: str
|
|
intensity: Intensity
|
|
generationmix: list[FuelMix]
|
|
|
|
|
|
class RegionData(TypedDict):
|
|
"""
|
|
Data for a region
|
|
|
|
Args:
|
|
TypedDict ([type]): describes the region and its data
|
|
"""
|
|
|
|
regionid: int
|
|
dnoregion: str
|
|
shortname: str
|
|
postcode: str
|
|
data: list[PostcodeData]
|
|
|
|
|
|
class CarbonData(TypedDict):
|
|
"""
|
|
Overall data for a request
|
|
|
|
Args:
|
|
TypedDict ([type]): describes the returned data
|
|
"""
|
|
|
|
data: list[RegionData]
|
|
|
|
|
|
def fetch_carbon_data(postcode: str) -> CarbonData:
|
|
"""
|
|
Fetch current data for the carbon intensity of the provided region
|
|
"""
|
|
|
|
obs_data = requests.get(
|
|
f'https://api.carbonintensity.org.uk/regional/postcode/{postcode}'
|
|
)
|
|
|
|
carbon_data: CarbonData = json.loads(obs_data.content)
|
|
return carbon_data
|
|
|
|
|
|
@app.route('/metrics')
|
|
def metrics() -> Response:
|
|
"""
|
|
Output Prometheus-style metrics
|
|
"""
|
|
postcode = os.environ.get('CARBON_POSTCODE') or ''
|
|
|
|
latest_data = fetch_carbon_data(postcode)['data'][0]
|
|
|
|
ret_data: list[PromData] = []
|
|
for generation in latest_data['data'][0]['generationmix']:
|
|
ret_data.append(
|
|
{
|
|
'key': 'sensor_carbon_generation_perc',
|
|
'labels': {
|
|
'fuel': generation['fuel'],
|
|
'region': latest_data['dnoregion'],
|
|
},
|
|
'type': 'gauge',
|
|
'value': float(generation['perc'])
|
|
}
|
|
)
|
|
|
|
ret_data.append(
|
|
{
|
|
'key': 'sensor_carbon_intensity_forecast',
|
|
'labels': {
|
|
'region': latest_data['dnoregion'],
|
|
},
|
|
'type': 'gauge',
|
|
'value': float(latest_data['data'][0]['intensity']['forecast']),
|
|
}
|
|
)
|
|
|
|
ret_strs = []
|
|
for item in ret_data:
|
|
ret_strs.append(f'# HELP {item["key"]} Carbon 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')
|