Voltorb ⚡

A simple (unofficial) Electricity Maps API client

Overview

voltorb is a simple (unofficial) client to the Electricity Maps API.

Built on top of snug, it provides a pythonic interface to query and manipulate results from all of its available API endpoints and may generally be useful for explorative data analysis and other applications.

[!NOTE]

This package is mainly a personal playground project to play around with a couple of libraries and concepts I am less familiar with (snug, atts, cattrs), and to refresh myself on the latest best-practices in terms of python tooling and packaging ecosystem.

As such, please be careful if actually using this in production - the package API is still in very early development status!

Installation

You can install voltorb from the python package index:

pip install voltorb

or locally from a clone of this repo:

pip install -e .

Authentication

An API token is required to query most Electricity Maps API endpoints. If you do not already have one, head over to https://api-portal.electricitymaps.com/ to explore all available options.

Once you have a token, you can setup token authentication as your authentication method:

import voltorb

auth = voltorb.token_auth("MY-API-TOKEN")

Use this later as an authentication method when executing your API queries.

[!TIP] To avoid hard-coding secrets, the following recipe uses python-dotenv to load the token from a local .env file:

import os

import voltorb
from dotenv import load_dotenv

load_dotenv()

API_TOKEN = os.environ.get("API_TOKEN")
auth = voltorb.token_auth(API_TOKEN)

Quickstart

You can find queries for all Electricity Maps API routes under the voltorb.electricity_maps namespace:

import voltorb

zone = voltorb.ZoneKey("IT")
live_carbon_intensity = voltorb.electricity_maps.carbon_intensity.get_latest(zone)

and execute them against the API backend:

import voltorb

response = voltorb.execute(live_carbon_intensity, auth=auth)

CarbonIntensity(
    zone='IT',
    carbon_intensity=327,
    datetime=datetime.datetime(2024, 4, 28, 21, 0, tzinfo=datetime.timezone.utc),
    updated_at=datetime.datetime(2024, 4, 28, 21, 47, 8, 110000, tzinfo=datetime.timezone.utc),
    created_at=datetime.datetime(2024, 4, 25, 21, 48, 58, 845000, tzinfo=datetime.timezone.utc),
    emission_factor_type=<EmissionFactorType.LIFECYCLE: 'lifecycle'>,
    is_estimated=True,
    estimation_method=<EstimationMethod.TIME_SLICER_AVERAGE: 'TIME_SLICER_AVERAGE'>
)

Executing queries

Queries can be executed in different ways. We have already seen execute(), but execute_async() is also available to perform asynchronous calls.

Both of these functions take arguments which affect:

• which authentication credentials are used
• which HTTP client is used (voltorb only includes basic sync and async HTTP clients)

For example, to use different credentials (or authentication methods):

import voltorb

# use personal token for authentication
auth_personal = voltorb.token_auth("MY-PERSONAL-API_TOKEN")
voltorb.execute(query, auth=auth_personal)

# use commercial token for authentication
auth_commercial = voltorb.token_auth("MY-COMMERCIAL-API_TOKEN")
voltorb.execute(query, auth=auth_commercial)

# use no authentication (auth=None)
voltorb.execute(query)

And to use different backend HTTP clients (for example httpx):

import httpx
import voltorb

with httpx.Client() as client:
    response = voltorb.execute(query, auth=auth, client=client)

You can also perform asynchronous queries through the same mechanisms:

import aiohttp
import asyncio
import voltorb

async def main():

    response = await voltorb.execute_async(query, auth=auth)

    async with aiohttp.ClientSession() as client:
        response = await voltorb.execute_async(query, auth=auth, client=client)

asyncio.run(main())

[!TIP] To make it easier to call execute()/execute_async() repeatedly with specific arguments, the executor()/async_executor() shortcuts can be used.

For example, to always add authentication to all of your queries:

import voltorb

executor = voltorb.executor(auth=auth)

some_response = executor(some_query)
another_response = executor(another_query)

# we can still override arguments at execution time
unauthenticated_response = executor(yet_another_query, auth=None)