diff --git a/main.py b/main.py index 50a786b..557a34f 100644 --- a/main.py +++ b/main.py @@ -4,15 +4,93 @@ Present observational weather data to Prometheus import json import os -from typing import Dict, List +from typing import TypedDict -import requests # type: ignore -from flask import Flask, Response # type: ignore +import requests +from flask import Flask, Response app = Flask(__name__) -def fetch_carbon_data(postcode: str) -> Dict: +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 """ @@ -21,23 +99,24 @@ def fetch_carbon_data(postcode: str) -> Dict: f'https://api.carbonintensity.org.uk/regional/postcode/{postcode}' ) - return json.loads(obs_data.content) + carbon_data: CarbonData = json.loads(obs_data.content) + return carbon_data @app.route('/metrics') -def metrics(): +def metrics() -> Response: """ Output Prometheus-style metrics """ - postcode = os.environ.get('CARBON_POSTCODE') + postcode = os.environ.get('CARBON_POSTCODE') or '' latest_data = fetch_carbon_data(postcode)['data'][0] - ret_data: List = list() + ret_data: list[PromData] = [] for generation in latest_data['data'][0]['generationmix']: ret_data.append( { - 'key': f'sensor_carbon_generation_perc', + 'key': 'sensor_carbon_generation_perc', 'labels': { 'fuel': generation['fuel'], 'region': latest_data['dnoregion'], @@ -58,7 +137,7 @@ def metrics(): } ) - ret_strs = list() + 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"]}')