airbeld-api-sdk 0.4.0

Async Python SDK for the Airbeld backend

Airbeld Python SDK

Async Python SDK for the Airbeld API, providing access to air quality devices and telemetry data.

If you want to contribute, read CONTRIBUTING and DEVELOPER.

---

Installation

# Using pip
pip install airbeld-api-sdk

# Using uv (recommended)
uv add airbeld-api-sdk

Examples

Running Examples

All examples require setting environment variables before running:

# Export required environment variables
export AIRBELD_API_BASE="https://api.airbeld.com"
export AIRBELD_API_TOKEN="your-jwt-token-here"

# Run an example
python examples/quickstart.py

Environment Variables:

• AIRBELD_API_BASE: API base URL (default: https://api.airbeld.com)
• AIRBELD_API_TOKEN: JWT access token for authentication

Quickstart Example

import asyncio
import os
from airbeld import AirbeldClient

async def main():
    # Initialize client (JWT token obtained outside SDK)
    base_url = os.environ.get("AIRBELD_API_BASE", "https://api.airbeld.com")
    token = os.environ["AIRBELD_API_TOKEN"]

    async with AirbeldClient(token=token, base_url=base_url) as client:
        # Get all devices
        devices = await client.async_get_devices()
        print(f"Found {len(devices)} devices")

        if devices:
            device = devices[0]
            print(f"Device: {device.name} ({device.status})")

            # Get latest telemetry readings (no date range)
            readings = await client.async_get_readings_by_date(
                device_id=device.id,
                sensor="temperature"  # Optional: filter to single sensor
            )

            # Print latest temperature reading
            if "temperature" in readings.sensors:
                latest_temp = readings.get_latest_value("temperature")
                print(f"Latest temperature: {latest_temp}°C")

if __name__ == "__main__":
    asyncio.run(main())

Note: JWT token acquisition happens outside the SDK. The SDK only handles API requests with a ready token.

Getting Latest Readings

You can fetch the latest sensor readings without specifying a date range:

import asyncio
import os
from airbeld import AirbeldClient

async def main():
    base_url = os.environ.get("AIRBELD_API_BASE", "https://api.airbeld.com")
    token = os.environ["AIRBELD_API_TOKEN"]

    async with AirbeldClient(token=token, base_url=base_url) as client:
        devices = await client.async_get_devices()

        if devices:
            device = devices[0]

            # Get latest readings without specifying start_date/end_date
            readings = await client.async_get_readings_by_date(device_id=device.id)

            # Display latest values
            for sensor_name, metric in readings.sensors.items():
                latest = readings.get_latest_value(sensor_name)
                print(f"{metric.display_name}: {latest} {metric.unit}")

if __name__ == "__main__":
    asyncio.run(main())

Getting Historical Readings

Fetch sensor readings for a specific date range with hourly or daily aggregation:

import asyncio
import os
from airbeld import AirbeldClient

async def main():
    base_url = os.environ.get("AIRBELD_API_BASE", "https://api.airbeld.com")
    token = os.environ["AIRBELD_API_TOKEN"]

    async with AirbeldClient(token=token, base_url=base_url) as client:
        devices = await client.async_get_devices()

        if devices:
            device = devices[0]

            # Get hourly readings for a specific date range
            readings = await client.async_get_readings_by_date(
                device_id=device.id,
                start_date="2025-09-19",  # Format: YYYY-MM-DD or 'today'
                end_date="2025-09-19",    # Same day = 24 hours of data
                period="hour",            # Aggregation: 'hour' or 'day'
                sensor="temperature"      # Optional: single sensor filter
            )

            # Access all temperature values
            if "temperature" in readings.sensors:
                temp_metric = readings.sensors["temperature"]
                print(f"Temperature readings: {len(temp_metric.values)} values")

                for reading in temp_metric.values:
                    print(f"  {reading.timestamp}: {reading.value}°C")

if __name__ == "__main__":
    asyncio.run(main())

API Reference

Client Methods

AirbeldClient(token, base_url, timeout)

Initialize the Airbeld API client.

Parameters:

• token (str, required): JWT access token for authentication
• base_url (str, optional): API base URL. Default: "https://api.airbeld.com"
• timeout (float, optional): Request timeout in seconds. Default: 10.0

Usage:

async with AirbeldClient(token=token, base_url=base_url) as client:
    # Use client...

---

async_get_devices()

Get list of all devices.

Parameters: None

Returns: list[DeviceSummary] - List of device objects

Raises:

• AuthError: Authentication failed (401/403)
• ApiError: Other API errors (4xx/5xx)
• NetworkError: Network connectivity issues

Example:

devices = await client.async_get_devices()
for device in devices:
    print(f"{device.name} - {device.status}")

---

async_get_readings_by_date(device_id, start_date, end_date, sensor, period)

Get telemetry readings for a device.

Parameters:

• device_id (int, required): Device ID
• start_date (str, optional): Start date in YYYY-MM-DD format or 'today'. If omitted, returns latest data.
• end_date (str, optional): End date in YYYY-MM-DD format or 'today'. If omitted, returns latest data.
• sensor (str, optional): Single sensor name to filter (e.g., "temperature", "pm2p5"). If omitted, returns all sensors.
• period (str, optional): Data aggregation period. Values: "hour" or "day".

Returns: Readings - Object containing sensor readings

Raises:

• AuthError: Authentication failed (401/403)
• ApiError: API errors including 404 (device not found), 413 (range too large)
• RateLimitError: Rate limit exceeded (429)
• NetworkError: Network connectivity issues

Examples:

# Get latest readings (no date range)
readings = await client.async_get_readings_by_date(device_id=123)

# Get historical readings with date range
readings = await client.async_get_readings_by_date(
    device_id=123,
    start_date="2025-09-19",
    end_date="2025-09-19",
    period="hour"
)

# Filter to single sensor
readings = await client.async_get_readings_by_date(
    device_id=123,
    sensor="temperature"
)

---

async_get_all_readings_by_date(start_date, end_date, sensor, period)

Get telemetry readings for all user devices.

Parameters:

• start_date (str, optional): Start date in YYYY-MM-DD format or 'today'. If omitted, returns latest data.
• end_date (str, optional): End date in YYYY-MM-DD format or 'today'. If omitted, returns latest data.
• sensor (str, optional): Single sensor name to filter (e.g., "temperature", "pm2p5"). If omitted, returns all sensors.
• period (str, optional): Data aggregation period. Values: "hour" or "day".

Returns: list[DeviceReadings] - List of device objects with metadata and sensor readings

Raises:

• AuthError: Authentication failed (401/403)
• ApiError: API errors including 413 (range too large)
• RateLimitError: Rate limit exceeded (429)
• NetworkError: Network connectivity issues

Examples:

# Get latest readings for all devices
devices = await client.async_get_all_readings_by_date()
for device in devices:
    print(f"Device: {device.display_name or device.name}")
    temp = device.get_latest_value("temperature")
    print(f"  Temperature: {temp}°C")

# Get historical readings with date range
devices = await client.async_get_all_readings_by_date(
    start_date="2025-10-14",
    end_date="2025-10-14",
    period="hour"
)

# Filter to single sensor
devices = await client.async_get_all_readings_by_date(
    sensor="pm2p5"
)

---

set_token(new_token)

Update the authorization token at runtime.

Parameters:

• new_token (str, required): New JWT access token

Returns: None

Example:

client.set_token(refreshed_token)

Authentication Functions

async_login(base_url, email, password, timeout)

Authenticate with email and password to obtain JWT tokens.

Parameters:

• base_url (str, required): API base URL
• email (str, required): User email address
• password (str, required): User password
• timeout (float, optional): Request timeout in seconds. Default: 10.0

Returns: TokenSet - Object containing access token, refresh token, expires_in, and token_type

Raises:

• AuthError: Invalid credentials (401)
• RateLimitError: Rate limit exceeded (429)
• ApiError: Other API errors (4xx/5xx)
• NetworkError: Network connectivity issues

Example:

from airbeld import async_login

token_set = await async_login(
    base_url="https://api.airbeld.com",
    email="[e-mail]",
    password="password"
)
print(f"Access token: {token_set.access_token}")
print(f"Expires in: {token_set.expires_in} seconds")

Data Models

DeviceSummary

Device information object.

Attributes:

• uid (str): Unique device identifier
• id (int): Device ID
• name (str): Device name
• display_name (str | None): Custom display name
• description (str): Device description
• type (str | None): Device type
• is_locked (bool): Whether device is locked
• status (str): Device status - "online" or "offline"
• sector (str | None): Sector name
• sector_id (int | None): Sector ID
• location (str | None): Location name
• location_id (int | None): Location ID
• timezone (str): IANA timezone

DeviceReadings

Device with telemetry readings (returned by async_get_all_readings_by_date).

Attributes:

• id (int): Device ID
• uid (str): Unique device identifier
• name (str): Device name
• display_name (str | None): Custom display name
• location (str | None): Location name
• sector (str | None): Sector name
• timezone (str): IANA timezone
• sensors (dict[str, TelemetryMetric]): Dictionary of sensor metrics by sensor name

Methods:

• get_latest_value(metric_name: str) -> float | None: Get the latest value for a specific sensor
• pm2_5 (property): Shortcut to access PM 2.5 metric (returns TelemetryMetric | None)

Readings

Container for sensor readings.

Attributes:

• sensors (dict[str, TelemetryMetric]): Dictionary of sensor metrics by sensor name

Methods:

• get_latest_value(metric_name: str) -> float | None: Get the latest value for a specific sensor
• pm2_5 (property): Shortcut to access PM 2.5 metric (returns TelemetryMetric | None)

TelemetryMetric

Individual sensor metric with metadata and values.

Attributes:

• name (str): Sensor name
• display_name (str | None): Display name
• unit (str): Measurement unit
• description (str | None): Sensor description
• values (list[TelemetryValue]): List of readings

TelemetryValue

Single sensor reading.

Attributes:

• timestamp (datetime): Reading timestamp
• value (float | None): Measured value

---

Authentication Options

The SDK supports two authentication paths depending on your use case:

Home Assistant Integration

For Home Assistant users, authentication is handled by the integration:

• Home Assistant performs OAuth2 flow automatically
• SDK receives a ready JWT token from the integration
• See docs/home-assistant-auth.md for details

Standalone Applications (CLI, Scripts, etc.)

For standalone applications, use the built-in authentication:

import asyncio
import os
from airbeld import async_login, AirbeldClient

async def main():
    # Authenticate with email/password
    token_set = await async_login(
        base_url="https://api.airbeld.com",
        email=os.environ["AIRBELD_USER_EMAIL"],
        password=os.environ["AIRBELD_USER_PASSWORD"]
    )
    
    # Create client with access token
    async with AirbeldClient(token=token_set.access_token) as client:
        # List devices
        devices = await client.async_get_devices()
        print(f"Found {len(devices)} devices:")
        
        for device in devices:
            print(f"  - {device.name} ({device.status})")

if __name__ == "__main__":
    asyncio.run(main())

⚠️ Security Warning: Never commit real credentials or tokens to version control. Use environment variables or secure secret storage systems.

Token Management

For applications requiring token refresh or runtime token updates:

# Update token at runtime
client.set_token(new_token)

# Or use refresh token (if implemented)
new_token_set = await async_login(...)
client.set_token(new_token_set.access_token)

---

Development

Installation

Using uv (recommended):

# Clone the repository
git clone https://github.com/Embio-Diagnostics/airbeld-api-sdk.git
cd airbeld-api-sdk

# Create virtual environment and install dependencies
uv sync

Using pip:

# Clone the repository
git clone https://github.com/Embio-Diagnostics/airbeld-api-sdk.git
cd airbeld-api-sdk

# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install package with dev dependencies
pip install -e ".[dev]"

Running Tests

# Using uv
uv run pytest

# Or with pip
pytest

Linting and Formatting

# Check code style
uv run ruff check .

# Format code
uv run ruff format .

# Or with pip
ruff check .
ruff format .

Type Checking

# Using uv
uv run mypy src

# Or with pip
mypy src

Building the Package

# Using uv
uv build

# Or with pip
python -m build

---

Resources

• Contributing: See CONTRIBUTING.md for contribution guidelines
• Changelog: See CHANGELOG.md for version history
• License: This project is licensed under the MIT License - see LICENSE file for details