A Python 3, asyncio-based library to interact with the PurpleAir API
Project description
🟣 aiopurpleair: A Python3, asyncio-based library to interact with the PurpleAir API
aiopurpleair
is a Python3, asyncio-based library to interact with the
PurpleAir API.
Installation
pip install aiopurpleair
Python Versions
aiopurpleair
is currently supported on:
- Python 3.9
- Python 3.10
- Python 3.11
Usage
In-depth documentation on the API can be found here:
https://api.purpleair.com/#api-welcome. Unless otherwise noted, aiopurpleair
endeavors
to follow the API as closely as possible.
Checking an API Key
To check whether an API key is valid and what properties it has:
import asyncio
from aiopurpleair import API
async def main() -> None:
"""Run."""
api = API("<API KEY>")
response = await api.async_check_api_key()
# >>> response.api_key_type == ApiKeyType.READ
# >>> response.api_version == "V1.0.11-0.0.41"
# >>> response.timestamp_utc == datetime(2022, 10, 27, 18, 25, 41)
asyncio.run(main())
Getting Sensors
import asyncio
from aiopurpleair import API
async def main() -> None:
"""Run."""
api = API("<API_KEY>")
response = await api.sensors.async_get_sensors(["name"])
# >>> response.api_version == "V1.0.11-0.0.41"
# >>> response.data == {
# >>> 131075: SensorModel(sensor_index=131075, name=Mariners Bluff),
# >>> 131079: SensorModel(sensor_index=131079, name=BRSKBV-outside),
# >>> }
# >>> response.data_timestamp_utc == datetime(2022, 11, 3, 19, 25, 31)
# >>> response.fields == ["sensor_index", "name"]
# >>> response.firmware_default_version == "7.02"
# >>> response.max_age == 604800
# >>> response.timestamp_utc == datetime(2022, 11, 3, 19, 26, 29)
asyncio.run(main())
Method Parameters
fields
(required): The sensor data fields to includelocation_type
(optional): An LocationType to filter bymax_age
(optional): Filter results modified within these secondsmodified_since
(optional): Filter results modified since a UTC datetimeread_keys
(optional): Read keys for private sensorssensor_indices
(optional): Filter results by sensor index
Getting a Single Sensor
import asyncio
from aiopurpleair import API
async def main() -> None:
"""Run."""
api = API("<API_KEY>")
response = await api.sensors.async_get_sensor(131075)
# >>> response.api_version == "V1.0.11-0.0.41"
# >>> response.data_timestamp_utc == datetime(2022, 11, 5, 16, 36, 21)
# >>> response.sensor == SensorModel(sensor_index=131075, ...),
# >>> response.timestamp_utc == datetime(2022, 11, 5, 16, 37, 3)
asyncio.run(main())
Method Parameters
sensor_index
(required): The sensor index of the sensor to retrieve.fields
(optional): The sensor data fields to include.read_key
(optional): A read key for a private sensor.
Getting Nearby Sensors
This method returns a list of NearbySensorResult
objects that are within a bounding box
around a given latitude/longitude pair. The list is sorted from nearest to furthest
(i.e., the first index in the list is the closest to the latitude/longitude).
NearbySensorResult
objects have two properties:
sensor
: the correspondingSensorModel
objectdistance
: the calculated distance (in kilometers) between this sensor and the provided latitude/longitude
import asyncio
from aiopurpleair import API
async def main() -> None:
"""Run."""
api = API("<API_KEY>")
sensors = await api.sensors.async_get_nearby_sensors(
["name"], 51.5285582, -0.2416796, 10
)
# >>> [NearbySensorResult(...), NearbySensorResult(...)]
asyncio.run(main())
Method Parameters
fields
(required): The sensor data fields to includelatitude
(required): The latitude of the point to measure distance fromlongitude
(required): The longitude of the point to measure distance fromdistance
(required): The distance from the measured point to search (in kilometers)limit
(optional): Limit the results
Connection Pooling
By default, the library creates a new connection to the PurpleAir 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 aiopurpleair import API
async def main() -> None:
"""Run."""
async with ClientSession() as session:
api = await API("<API KEY>")
# Get to work...
asyncio.run(main())
Contributing
- 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.
- Write tests that cover your new functionality.
- Run tests and ensure 100% code coverage:
poetry run pytest --cov aiopurpleair tests
- Update
README.md
with any new documentation. - Add yourself to
AUTHORS.md
. - Submit a pull request!
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aiopurpleair-2022.12.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f882495c67d9e3f5adaa365b57fe625d184868fc701de8e8c8c6ebe447e24d7e |
|
MD5 | 14abd7a51a5a530e62cfbcf21c656b6b |
|
BLAKE2b-256 | edc01a2ce72297728a7206e3ba01ce26f4fcf12e111d57d75582271d6f8553e9 |