aiowatttime is a Python 3, asyncio-friendly library for interacting with WattTime emissions data.

• Python Versions
• Installation
• Usage
• Contributing

aiowatttime is currently supported on:

• Python 3.10
• Python 3.11
• Python 3.12

pip install aiowatttime

Simply clone this repo and run the included interactive script:

$ script/register

Note that WattTime offers three plans: Visitors, Analyst, and Pro. The type you use will determine which elements of this library are available to use. You can read more details here.

The Client is the primary method of interacting with the API:

import asyncio

from aiowatttime import Client


async def main() -> None:
    client = await Client.login("<USERNAME>", "<PASSWORD>")
    # ...


asyncio.run(main())

By default, the library creates a new connection to the API with each coroutine. If you are calling a large number of coroutines (or merely want to squeeze out every second of runtime savings possible), an aiohttp ClientSession can be used for connection pooling:

import asyncio

from aiohttp import ClientSession

from aiowatttime import Client


async def main() -> None:
    async with ClientSession() as session:
        client = await Client.login("<USERNAME>", "<PASSWORD>", session=session)
        # ...


asyncio.run(main())

await client.async_request_password_reset()

It may be useful to first get the "grid region" (i.e., geographical info) for the area you care about:

await client.emissions.async_get_grid_region(
    "<LATITUDE>", "<LONGITUDE>", "<SIGNAL_TYPE>"
)
# >>> { "region": "PSCO", "region_full_name": "Public Service Co of Colorado", "signal_type": "co2_moer" }

Getting emissions data will require the region abbreviation (PSCO in the example above).

await client.emissions.async_get_realtime_emissions("<REGION>")
# >>>
{"data": [...]}

from datetime import datetime

await client.emissions.async_get_forecasted_emissions(
    "<REGION>", "<SIGNAL_TYPE>", datetime(2021, 1, 1), datetime(2021, 2, 1)
)
# >>> { "data": [ ... ] }

await client.emissions.async_get_forecasted_emissions(
    "<REGION>", "<SIGNAL_TYPE>", datetime(2021, 1, 1), datetime(2021, 2, 1)
)
# >>> [ { "point_time": "2019-02-21T00:15:00.000Z", "value": 844, ... } ]

By default, aiowatttime will handle expired access tokens for you. When a token expires, the library will attempt the following sequence 3 times:

• Request a new token
• Pause for 1 second (to be respectful of the API rate limiting)
• Execute the original request again

Both the number of retries and the delay between retries can be configured when instantiating a client:

import asyncio

from aiohttp import ClientSession

from aiowatttime import Client


async def main() -> None:
    async with ClientSession() as session:
        client = await Client.async_login(
            "user",
            "password",
            session=session,
            # Make 7 retry attempts:
            request_retries=7,
            # Delay 4 seconds between attempts:
            request_retry_delay=4,
        )


asyncio.run(main())

As always, an invalid username/password combination will immediately throw an exception.

By default, aiowatttime provides its own logger. If you should wish to use your own, you can pass it to the client during instantiation:

import asyncio
import logging

from aiohttp import ClientSession

from aiowatttime import Client

CUSTOM_LOGGER = logging.getLogger("my_custom_logger")


async def main() -> None:
    async with ClientSession() as session:
        client = await Client.async_login(
            "user",
            "password",
            session=session,
            logger=logger,
        )


asyncio.run(main())

Thanks to all of our contributors so far!

1. Check for open features/bugs or initiate a discussion on one.
2. Fork the repository.
3. (optional, but highly recommended) Create a virtual environment: python3 -m venv .venv
4. (optional, but highly recommended) Enter the virtual environment: source ./.venv/bin/activate
5. Install the dev environment: script/setup
6. Code your new feature or bug fix on a new branch.
7. Write tests that cover your new functionality.
8. Run tests and ensure 100% code coverage: poetry run pytest --cov aiowatttime tests
9. Update README.md with any new documentation.
10. Submit a pull request!