carbon-exporter/main.py

"""
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')