🌤 aioambient: An async library for Ambient Weather Personal Weather Stations
aioambient
is a Python3, asyncio-driven library that interfaces with both the REST and
Websocket APIs provided by Ambient Weather.
Installation
pip install aioambient
Python Versions
aioambient
is currently supported on:
- Python 3.10
- Python 3.11
- Python 3.12
API and Application Keys
Utilizing aioambient
requires both an Application Key and an API Key from Ambient
Weather. You can generate both from the Profile page in your
Ambient Weather Dashboard.
Usage
REST API
import asyncio
from datetime import date
from aiohttp import ClientSession
from aioambient import API
async def main() -> None:
"""Create the aiohttp session and run the example."""
api = API("<YOUR APPLICATION KEY>", "<YOUR API KEY>")
await api.get_devices()
await api.get_device_details("<DEVICE MAC ADDRESS>")
await api.get_device_details("<DEVICE MAC ADDRESS>", end_date=date(2019, 1, 16))
asyncio.run(main())
By default, the library creates a new connection to Ambient Weather 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 datetime import date
from aiohttp import ClientSession
from aioambient import API
async def main() -> None:
"""Create the aiohttp session and run the example."""
async with ClientSession() as session:
api = API("<YOUR APPLICATION KEY>", "<YOUR API KEY>")
await api.get_devices()
await api.get_device_details("<DEVICE MAC ADDRESS>")
await api.get_device_details("<DEVICE MAC ADDRESS>", end_date=date(2019, 1, 16))
asyncio.run(main())
Please be aware of Ambient Weather's
rate limiting policies.
Websocket API
import asyncio
from aiohttp import ClientSession
from aioambient import Websocket
async def main() -> None:
"""Create the aiohttp session and run the example."""
websocket = Websocket("<YOUR APPLICATION KEY>", "<YOUR API KEY>")
websocket = Websocket("YOUR APPLICATION KEY", ["<API KEY 1>", "<API KEY 2>"])
def connect_method():
"""Print a simple "hello" message."""
print("Client has connected to the websocket")
websocket.on_connect(connect_method)
async def connect_coroutine():
"""Waits for 3 seconds, then print a simple "hello" message."""
await asyncio.sleep(3)
print("Client has connected to the websocket")
websocket.async_on_connect(connect_coroutine)
def subscribed_method(data):
"""Print the data received upon subscribing."""
print(f"Subscription data received: {data}")
websocket.on_subscribed(subscribed_method)
async def subscribed_coroutine(data):
"""Waits for 3 seconds, then print the incoming data."""
await asyncio.sleep(3)
print(f"Subscription data received: {data}")
websocket.async_on_subscribed(subscribed_coroutine)
def data_method(data):
"""Print the data received."""
print(f"Data received: {data}")
websocket.on_data(data_method)
async def data_coroutine(data):
"""Wait for 3 seconds, then print the data received."""
await asyncio.sleep(3)
print(f"Data received: {data}")
websocket.async_on_data(data_coroutine)
def disconnect_method(data):
"""Print a simple "goodbye" message."""
print("Client has disconnected from the websocket")
websocket.on_disconnect(disconnect_method)
async def disconnect_coroutine(data):
"""Wait for 3 seconds, then print a simple "goodbye" message."""
await asyncio.sleep(3)
print("Client has disconnected from the websocket")
websocket.async_on_disconnect(disconnect_coroutine)
await websocket.connect()
await websocket.disconnect()
asyncio.run(main())
Open REST API
The official REST API and Websocket API require an API and application key to access
data for the devices you own. This API cannot be used if you do not own a personal
weather station.
However, there is a second, undocumented API that is used by the https://ambientweather.net
web application that does not require an API and application key. You can use the
OpenAPI
class to retrieve weather station data from this API:
import asyncio
from datetime import date
from aiohttp import ClientSession
from aioambient import OpenAPI
async def main() -> None:
"""Create the aiohttp session and run the example."""
api = OpenAPI()
await api.get_devices_by_location(32.5, -97.3, 3.0)
await api.get_device_details("<DEVICE MAC ADDRESS>")
asyncio.run(main())
Contributing
Thanks to all of our contributors so far!
- Check for open features/bugs or initiate a discussion on one.
- Fork the repository.
- (optional, but highly recommended) Create a virtual environment:
python3 -m venv .venv
- (optional, but highly recommended) Enter the virtual environment:
source ./.venv/bin/activate
- Install the dev environment:
script/setup
- Code your new feature or bug fix on a new branch.
- Write tests that cover your new functionality.
- Run tests and ensure 100% code coverage:
poetry run pytest --cov aioambient tests
- Update
README.md
with any new documentation. - Submit a pull request!