aioleviton 0.2.1

Async Python client for the Leviton My Leviton cloud API

aioleviton

Async Python client for the Leviton My Leviton cloud API.

Supports LWHEM and DAU/LDATA Smart Load Centers with WebSocket real-time push and REST API fallback.

Features

• Pure asyncio with aiohttp -- no blocking calls
• Accepts an injected aiohttp.ClientSession for connection pooling
• WebSocket real-time push notifications with automatic subscription management
• Full REST API coverage: authentication, device discovery, breaker control, energy history
• Typed data models with PEP 561 py.typed marker
• Support for both hub types: LWHEM (IotWhem) and DAU/LDATA (ResidentialBreakerPanel)
• Two-factor authentication (2FA) support

Installation

pip install aioleviton

Quick Start

import aiohttp
from aioleviton import LevitonClient, LevitonWebSocket

async def main():
    async with aiohttp.ClientSession() as session:
        # Authenticate
        client = LevitonClient(session)
        auth = await client.login("[e-mail]", "password")

        # Discover devices
        permissions = await client.get_permissions()
        for perm in permissions:
            if perm.residential_account_id:
                residences = await client.get_residences(perm.residential_account_id)
                for residence in residences:
                    whems = await client.get_whems(residence.id)
                    panels = await client.get_panels(residence.id)

        # Get breakers for a LWHEM hub
        for whem in whems:
            breakers = await client.get_whem_breakers(whem.id)
            cts = await client.get_cts(whem.id)

        # Connect WebSocket for real-time updates
        ws = LevitonWebSocket(
            session=session,
            token=auth.token,
            user_id=auth.user_id,
            user=auth.user,
            token_created=auth.created,
            token_ttl=auth.ttl,
        )
        await ws.connect()

        # Subscribe to a hub (delivers all child breaker/CT updates)
        await ws.subscribe("IotWhem", whem.id)

        # Handle notifications
        ws.on_notification(lambda data: print("Update:", data))

Supported Devices

Device	API Model	Hub Type
LWHEM (Whole Home Energy Module)	IotWhem	Wi-Fi hub
DAU / LDATA (Data Acquisition Unit)	ResidentialBreakerPanel	Wi-Fi hub
Smart Breaker Gen 1 (trip only)	ResidentialBreaker	Child of LWHEM or DAU
Smart Breaker Gen 2 (on/off)	ResidentialBreaker	Child of LWHEM or DAU
Current Transformer (CT)	IotCt	Child of LWHEM only
LSBMA Add-on CT	ResidentialBreaker	Virtual composite

Breaker Control

# Trip a Gen 1 breaker (cannot turn back on remotely)
await client.trip_breaker(breaker_id)

# Turn on/off a Gen 2 breaker
await client.turn_on_breaker(breaker_id)
await client.turn_off_breaker(breaker_id)

# Blink LED on a breaker
await client.blink_led(breaker_id)

# Identify LED on a LWHEM hub
await client.identify_whem(whem_id)

Energy History

Energy history endpoints return consumption data for all devices in a residence. Data is keyed by hub ID, then by breaker position and CT channel.

# Daily energy (hourly data points)
day = await client.get_energy_for_day(
    residence_id=123456,
    start_day="2026-02-16",
    timezone="America/Los_Angeles",
)

# Weekly energy (daily data points for 7 days)
week = await client.get_energy_for_week(
    residence_id=123456,
    start_day="2026-02-17",
    timezone="America/Los_Angeles",
)

# Monthly energy (daily data points for billing month)
month = await client.get_energy_for_month(
    residence_id=123456,
    billing_day_in_month="2026-02-28",
    timezone="America/Los_Angeles",
)

# Yearly energy (monthly data points for 12 months)
year = await client.get_energy_for_year(
    residence_id=123456,
    billing_day_in_end_month="2026-02-16",
    timezone="America/Los_Angeles",
)

# Response structure:
# {
#   "<hub_id>": {
#     "residentialBreakers": {"<position>": [{x, timestamp, energyConsumption, totalCost, ...}]},
#     "iotCts": {"<channel>": [...]},
#     "totals": [...]
#   },
#   "totals": [...]  # residence-level totals
# }

Firmware Check

# Check for available firmware updates
firmware = await client.check_firmware(
    app_id="LWHEM",
    model="AZ",
    serial="1000_XXXX_XXXX",
    model_type="IotWhem",
)
# Returns list of firmware objects with version, fileUrl, signature, hash, size, notes
for fw in firmware:
    print(f"v{fw['version']}: {fw['fileUrl']}")

License

MIT