aiokwikset 0.6.2
Python interface for Kwikset Smart Locks
License: Apache-2.0
Requires Python: >=3.10
Keywords: kwikset, home automation, kwikset halo, kwikset smart lock, smart home, iot

# aiokwikset

[![PyPI version](https://badge.fury.io/py/aiokwikset.svg)](https://badge.fury.io/py/aiokwikset)
[![Python Versions](https://img.shields.io/pypi/pyversions/aiokwikset.svg)](https://pypi.org/project/aiokwikset/)
[![CI](https://github.com/explosivo22/aiokwikset/actions/workflows/ci.yml/badge.svg)](https://github.com/explosivo22/aiokwikset/actions/workflows/ci.yml)
[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
[![AsyncIO](https://img.shields.io/badge/asyncio-yes-green.svg)](https://docs.python.org/3/library/asyncio.html)

An asynchronous Python client library for the Kwikset Smart Lock API.

This library provides a fully async interface for interacting with [Kwikset Smart Locks](https://www.kwikset.com/products/electronic/electronic-smart-locks) (including Halo and Aura models) via the Kwikset cloud API.

## Table of Contents

- [Features](#features)
- [Requirements](#requirements)
- [Installation](#installation)
- [Quick Start](#quick-start)
- [Usage](#usage)
  - [Basic Authentication](#basic-authentication)
  - [Using Context Manager](#using-context-manager)
  - [MFA Authentication](#mfa-authentication)
  - [Session Restoration](#session-restoration)
  - [Device Operations](#device-operations)
  - [Real-Time Subscriptions](#real-time-subscriptions)
  - [Token Persistence](#token-persistence)
  - [Home Assistant Integration](#home-assistant-integration)
- [API Reference](#api-reference)
  - [API Class](#api-class)
  - [Device Class](#device-class)
  - [User Class](#user-class)
  - [SubscriptionManager Class](#subscriptionmanager-class)
  - [Exceptions](#exceptions)
- [Error Handling](#error-handling)
- [Contributing](#contributing)
- [License](#license)

## Features

- **Fully Asynchronous**: Built with `asyncio` and `aiohttp` for non-blocking I/O
- **Real-Time Subscriptions**: Push-based device updates via AWS AppSync WebSocket connections
- **Type Annotated**: Complete type hints for better IDE support and code quality
- **MFA Support**: Full support for SMS and Software Token (TOTP) multi-factor authentication
- **Context Manager**: Proper resource cleanup with async context manager support
- **Token Management**: Automatic token refresh with optional persistence callbacks
- **Session Restoration**: Restore sessions from saved tokens without re-entering password
- **Retry Logic**: Automatic retries with exponential backoff for transient failures
- **Auto-Reconnect**: WebSocket connections automatically reconnect with exponential backoff
- **Configurable Timeouts**: Customizable request timeouts (default: 30 seconds)
- **Home Assistant Ready**: Purpose-built exceptions and patterns for HA integrations
- **Secure**: AWS Cognito authentication with secure token handling
- **Extensible**: Clean architecture with modular endpoint handlers

### Supported Operations

| Category | Operations |
|----------|------------|
| **Authentication** | Login, MFA (SMS/TOTP), Token Refresh |
| **Lock Control** | Lock, Unlock |
| **Device Settings** | LED Status, Audio Status, Secure Screen Status |
| **Information** | User Info, Homes, Devices, Device Details |
| **Real-Time Events** | Device Status, Battery, Door Sensor, Lock Events, WiFi Signal, and more |

## Requirements

- **Python**: 3.10 or higher
- **Kwikset Account**: You must have a registered Kwikset account with at least one home configured
- **Kwikset App**: Initial setup requires the official Kwikset mobile app
  - [iOS App Store](https://apps.apple.com/us/app/kwikset/id1465996742)
  - [Google Play Store](https://play.google.com/store/apps/details?id=com.kwikset.blewifi)

## Installation

### Using pip

```bash
pip install aiokwikset
```

### Using Poetry

```bash
poetry add aiokwikset
```

### From Source

```bash
git clone https://github.com/explosivo22/aiokwikset.git
cd aiokwikset
pip install -e .
```

### Development Installation

```bash
pip install -e ".[dev]"
```

## Quick Start

```python
import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("your-email@example.com", "your-password")
        
        # Get all homes
        homes = await api.user.get_homes()
        
        # Get devices for the first home
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        for device in devices:
            print(f"Device: {device['devicename']} - {device['deviceid']}")

asyncio.run(main())
```

## Usage

### Basic Authentication

```python
import asyncio
from aiokwikset import API

async def main() -> None:
    # Create API instance
    api = API()
    
    try:
        # Authenticate with Kwikset cloud
        await api.async_login("your-email@example.com", "your-password")
        
        # Check authentication status
        if api.is_authenticated:
            print("Successfully authenticated!")
            
            # Get user information
            user_info = await api.user.get_info()
            print(f"Welcome, {user_info['firstname']} {user_info['lastname']}!")
            
    finally:
        # Clean up resources
        await api.async_close()

asyncio.run(main())
```

### Using Context Manager

The recommended way to use the API is with an async context manager, which ensures proper cleanup:

```python
import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("your-email@example.com", "your-password")
        
        homes = await api.user.get_homes()
        for home in homes:
            print(f"Home: {home['homename']} (ID: {home['homeid']})")
        
        # Resources are automatically cleaned up when exiting the context

asyncio.run(main())
```

### MFA Authentication

If your account has Multi-Factor Authentication enabled, handle the MFA challenge:

```python
import asyncio
from aiokwikset import API, MFAChallengeRequired

async def main() -> None:
    async with API() as api:
        try:
            await api.async_login("your-email@example.com", "your-password")
            
        except MFAChallengeRequired as mfa_error:
            # Determine MFA type
            if mfa_error.mfa_type == "SOFTWARE_TOKEN_MFA":
                print("Enter the code from your authenticator app:")
            else:  # SMS_MFA
                print("Enter the code sent to your phone:")
            
            mfa_code = input("MFA Code: ")
            
            # Complete MFA authentication
            await api.async_respond_to_mfa_challenge(
                mfa_code=mfa_code,
                mfa_type=mfa_error.mfa_type,
                mfa_tokens=mfa_error.mfa_tokens,
            )
            
            print("MFA authentication successful!")
        
        # Continue with normal API usage
        homes = await api.user.get_homes()
        print(f"Found {len(homes)} home(s)")

asyncio.run(main())
```

For more detailed MFA examples, see [MFA_USAGE.md](./MFA_USAGE.md).

### Session Restoration

Restore sessions from previously saved tokens without requiring the user to re-enter their password. This is ideal for Home Assistant integrations and long-running applications:

```python
import asyncio
from aiokwikset import API, TokenExpiredError

async def main() -> None:
    # Create API with stored tokens
    api = API(
        id_token="stored_id_token",
        access_token="stored_access_token",
        refresh_token="stored_refresh_token",
    )
    
    async with api:
        try:
            # Authenticate using stored tokens (validates and refreshes if needed)
            await api.async_authenticate_with_tokens(
                id_token="stored_id_token",
                access_token="stored_access_token",
                refresh_token="stored_refresh_token",
            )
            
            # Session restored - continue with API usage
            homes = await api.user.get_homes()
            print(f"Session restored! Found {len(homes)} home(s)")
            
        except TokenExpiredError:
            # Tokens are no longer valid - re-authentication required
            print("Stored tokens expired. Please login again.")
            await api.async_login("your-email@example.com", "your-password")

asyncio.run(main())
```

### Device Operations

#### Locking and Unlocking

```python
import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("your-email@example.com", "your-password")
        
        # Get user info (required for lock/unlock operations)
        user_info = await api.user.get_info()
        
        # Get homes and devices
        homes = await api.user.get_homes()
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        # Get detailed info for the first device
        device_info = await api.device.get_device_info(devices[0]["deviceid"])
        
        # Lock the device
        await api.device.lock_device(device_info, user_info)
        print("Device locked!")
        
        # Unlock the device
        await api.device.unlock_device(device_info, user_info)
        print("Device unlocked!")

asyncio.run(main())
```

#### Device Information and Settings

```python
import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("your-email@example.com", "your-password")
        
        # Get user info (needed for lock/unlock operations)
        user_info = await api.user.get_info()
        
        # Get all homes and devices
        homes = await api.user.get_homes()
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        for device in devices:
            # Get detailed device information
            device_info = await api.device.get_device_info(device["deviceid"])
            
            if device_info:
                print(f"Device: {device_info['devicename']}")
                print(f"  Model: {device_info.get('modelnumber', 'Unknown')}")
                print(f"  Battery: {device_info.get('batterypercentage', 'N/A')}%")
                print(f"  Lock Status: {device_info.get('doorstatus', 'Unknown')}")
                
                # Configure device settings
                await api.device.set_ledstatus(device_info, "true")      # Enable LED
                await api.device.set_audiostatus(device_info, "true")    # Enable audio
                await api.device.set_securescreenstatus(device_info, "false")  # Disable secure screen

asyncio.run(main())
```

### Real-Time Subscriptions

Receive instant push notifications when device state changes, instead of polling. The library connects to Kwikset's AWS AppSync WebSocket endpoints and delivers events via a callback.

#### Basic Subscription

```python
import asyncio
from aiokwikset import API


def handle_event(subscription_name: str, data: dict) -> None:
    """Called whenever a real-time event arrives."""
    print(f"[{subscription_name}] {data}")


async def main() -> None:
    async with API() as api:
        await api.async_login("your-email@example.com", "your-password")

        # Set up real-time events
        api.subscriptions.set_callback(handle_event)
        await api.subscriptions.async_subscribe_device("your-email@example.com")

        # Keep running to receive events (Ctrl+C to stop)
        try:
            while True:
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            pass


asyncio.run(main())
```

The `onManageDevice` subscription delivers all device changes in a single stream:

```python
# Example callback data:
# subscription_name = "onManageDevice"
# data = {
#     "deviceid": "abc123",
#     "devicestatus": "Locked",
#     "batterypercentage": "95",
#     "connectivitystatus": "online",
#     "doorposition": "closed",
#     "operationtype": "statusUpdate",
#     ...
# }
```

#### Available Subscriptions

Two AppSync endpoints are available — the **Core API** for broad device/user/home events, and the **Halo V API** for granular, per-attribute updates:

**Core API** (recommended for most use cases):

| Method | Event Name | Description |
|--------|-----------|-------------|
| `async_subscribe_device(email)` | `onManageDevice` | Lock state, battery, door position, tamper, connectivity, and all device settings |
| `async_subscribe_user(email)` | `onManageUser` | Account profile changes (name, phone, MFA status) |
| `async_subscribe_home(email)` | `onManageHome` | Home sharing, invitations, access level changes |

**Halo V API** (fine-grained updates for newer devices):

| Method | Event Name | Description |
|--------|-----------|-------------|
| `async_subscribe_hav_device_status(email)` | `onUpdateHavDeviceStatus` | Lock/unlock state changes |
| `async_subscribe_hav_device(email)` | `onUpdateHavDevice` | Full device metadata updates |
| `async_subscribe_hav_event(email)` | `onUpdateHavEvent` | Activity log (who locked/unlocked, garage events) |
| `async_subscribe_hav_battery_voltage(email)` | `onUpdateHavBatteryVoltage` | Battery voltage changes |
| `async_subscribe_hav_sensor(email)` | `onUpdateHavSensor` | Door sensor alignment and alarm flags |
| `async_subscribe_hav_led_status(email)` | `onUpdateHavLedStatus` | LED indicator changes |
| `async_subscribe_hav_percentage(email)` | `onUpdateHavPercentage` | Garage door open percentage |
| `async_subscribe_hav_vacation_mode(email)` | `onUpdateHavVacationMode` | Vacation mode toggle |
| `async_subscribe_hav_wifi_signal(email)` | `onUpdateHavWifiSignal` | WiFi signal strength |
| `async_subscribe_hav_cycles(email)` | `onUpdateHavCycles` | Lock mechanism cycle count |
| `async_subscribe_hav_auto_close_schedule(email)` | `onUpdateHavAutoCloseSchedule` | Auto-lock schedule changes |
| `async_subscribe_hav_temporary_key(email)` | `onUpdateHavTemporaryKey` | Guest/temporary access keys |
| `async_subscribe_hav_brightness_level(email)` | `onUpdateHavBrightnessLevel` | Keypad brightness changes |

#### Managing Subscriptions

Each `subscribe` method returns a subscription ID that can be used to unsubscribe:

```python
# Subscribe
sub_id = await api.subscriptions.async_subscribe_device(email)
hav_id = await api.subscriptions.async_subscribe_hav_device_status(email)

# Unsubscribe from a specific subscription
await api.subscriptions.async_unsubscribe(sub_id)

# Close all subscriptions and WebSocket connections
await api.subscriptions.async_close()
```

#### Key Behaviors

- **Lazy connections**: WebSocket connections are only opened when you first subscribe to an endpoint
- **Auto-reconnect**: If the connection drops, all subscriptions are automatically re-established with exponential backoff
- **Token refresh**: Credentials are automatically refreshed on reconnection
- **Session reuse**: When a `websession` is injected (e.g., from Home Assistant), it is reused for WebSocket connections
- **Cleanup**: Calling `api.async_close()` or exiting the `async with` block closes all WebSocket connections

### Token Persistence

For long-running applications or integrations, you can persist tokens and receive callbacks when they're refreshed:

```python
import asyncio
import json
from pathlib import Path
from aiokwikset import API

TOKEN_FILE = Path("tokens.json")

async def save_tokens(id_token: str, access_token: str, refresh_token: str) -> None:
    """Callback to save tokens when they're refreshed."""
    TOKEN_FILE.write_text(json.dumps({
        "id_token": id_token,
        "access_token": access_token,
        "refresh_token": refresh_token,
    }))

async def main() -> None:
    # Load existing tokens if available
    tokens = {}
    if TOKEN_FILE.exists():
        tokens = json.loads(TOKEN_FILE.read_text())
    
    # Create API with token persistence callback
    api = API(
        id_token=tokens.get("id_token"),
        access_token=tokens.get("access_token"),
        refresh_token=tokens.get("refresh_token"),
        token_update_callback=save_tokens,
    )
    
    async with api:
        if not api.is_authenticated:
            await api.async_login("your-email@example.com", "your-password")
        
        # Tokens will be automatically saved when refreshed
        homes = await api.user.get_homes()
        print(f"Found {len(homes)} home(s)")

asyncio.run(main())
```

### Home Assistant Integration

When using with Home Assistant, pass the shared aiohttp session and handle the specific exceptions for proper integration behavior:

```python
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from aiokwikset import API, TokenExpiredError, ConnectionError, MFAChallengeRequired

async def async_setup_entry(hass, entry):
    """Set up Kwikset from a config entry."""
    session = async_get_clientsession(hass)
    
    async def async_update_tokens(id_token, access_token, refresh_token):
        """Update config entry with new tokens."""
        hass.config_entries.async_update_entry(
            entry,
            data={
                **entry.data,
                "id_token": id_token,
                "access_token": access_token,
                "refresh_token": refresh_token,
            },
        )
    
    api = API(
        websession=session,  # Reuses HA's session (won't be closed by API)
        timeout=30,  # Configurable timeout
        token_update_callback=async_update_tokens,
    )
    
    try:
        # Restore session from stored tokens
        await api.async_authenticate_with_tokens(
            id_token=entry.data["id_token"],
            access_token=entry.data["access_token"],
            refresh_token=entry.data["refresh_token"],
        )
    except TokenExpiredError:
        # Tokens expired - trigger reauth flow
        raise ConfigEntryAuthFailed("Authentication tokens expired")
    except ConnectionError:
        # Network issue - retry later
        raise ConfigEntryNotReady("Cannot connect to Kwikset API")
    
    # Store API instance for use by platforms
    hass.data.setdefault("kwikset", {})[entry.entry_id] = api
    
    # Set up real-time event subscriptions
    email = entry.data["email"]
    
    def handle_realtime_event(name: str, data: dict) -> None:
        """Handle real-time device updates."""
        if name == "onManageDevice":
            device_id = data.get("deviceid")
            if device_id:
                async_dispatcher_send(hass, f"kwikset_update_{device_id}", data)
    
    api.subscriptions.set_callback(handle_realtime_event)
    await api.subscriptions.async_subscribe_device(email)
    
    # Set up platforms
    await hass.config_entries.async_forward_entry_setups(entry, ["lock", "sensor"])
    
    return True
```

#### Config Flow Example

```python
from aiokwikset import API, MFAChallengeRequired, Unauthenticated

class KwiksetConfigFlow(ConfigFlow, domain="kwikset"):
    """Handle a config flow for Kwikset."""
    
    async def async_step_user(self, user_input=None):
        """Handle the initial step."""
        errors = {}
        
        if user_input is not None:
            api = API()
            try:
                await api.async_login(
                    user_input["email"],
                    user_input["password"],
                )
                
                # Success - create entry with tokens
                return self.async_create_entry(
                    title=user_input["email"],
                    data={
                        "email": user_input["email"],
                        "id_token": api.id_token,
                        "access_token": api.access_token,
                        "refresh_token": api.refresh_token,
                    },
                )
                
            except MFAChallengeRequired as err:
                # Store MFA context for next step
                self._mfa_tokens = err.mfa_tokens
                self._mfa_type = err.mfa_type
                self._email = user_input["email"]
                return await self.async_step_mfa()
                
            except Unauthenticated:
                errors["base"] = "invalid_auth"
            finally:
                await api.async_close()
        
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required("email"): str,
                vol.Required("password"): str,
            }),
            errors=errors,
        )
```

## API Reference

### API Class

The main entry point for interacting with the Kwikset cloud API.

```python
API(
    websession: Optional[ClientSession] = None,
    user_pool_region: str = "us-east-1",
    username: Optional[str] = None,
    timeout: int = 30,
    id_token: Optional[str] = None,
    access_token: Optional[str] = None,
    refresh_token: Optional[str] = None,
    token_update_callback: Optional[Callable[[str, str, str], Awaitable[None]]] = None,
)
```

| Parameter | Type | Description |
|-----------|------|-------------|
| `websession` | `Optional[ClientSession]` | Existing aiohttp session to reuse (won't be closed by API) |
| `user_pool_region` | `str` | AWS Cognito region (default: `"us-east-1"`) |
| `username` | `Optional[str]` | Pre-set username (email) |
| `timeout` | `int` | Request timeout in seconds (default: `30`) |
| `id_token` | `Optional[str]` | Pre-existing ID token for session restoration |
| `access_token` | `Optional[str]` | Pre-existing access token |
| `refresh_token` | `Optional[str]` | Pre-existing refresh token |
| `token_update_callback` | `Optional[Callable]` | Async callback when tokens are refreshed |

#### Properties

| Property | Type | Description |
|----------|------|-------------|
| `is_authenticated` | `bool` | Whether the client is currently authenticated |
| `device` | `Device` | Device endpoint handler (available after authentication) |
| `user` | `User` | User endpoint handler (available after authentication) |
| `subscriptions` | `SubscriptionManager` | Real-time event subscription manager (lazy, available after authentication) |

#### Methods

| Method | Description |
|--------|-------------|
| `async_login(email, password)` | Authenticate with email and password |
| `async_authenticate_with_tokens(id_token, access_token, refresh_token)` | Restore session from saved tokens |
| `async_respond_to_mfa_challenge(mfa_code, mfa_type, mfa_tokens)` | Complete MFA authentication |
| `async_renew_access_token()` | Manually refresh access token |
| `async_close()` | Clean up resources and clear tokens |

### Device Class

Handler for device-related API endpoints.

| Method | Parameters | Description |
|--------|------------|-------------|
| `get_devices(home_id)` | `home_id: str` | Get all devices for a home |
| `get_device_info(device_id)` | `device_id: str` | Get detailed device information |
| `lock_device(device, user)` | `device: dict`, `user: dict` | Lock the device |
| `unlock_device(device, user)` | `device: dict`, `user: dict` | Unlock the device |
| `set_ledstatus(device, status)` | `device: dict`, `status: str` | Set LED status (`"true"`/`"false"`) |
| `set_audiostatus(device, status)` | `device: dict`, `status: str` | Set audio status (`"true"`/`"false"`) |
| `set_securescreenstatus(device, status)` | `device: dict`, `status: str` | Set secure screen status (`"true"`/`"false"`) |
| `set_led_enabled(device, enabled)` | `device: dict`, `enabled: bool` | Enable/disable LED (convenience method) |
| `set_audio_enabled(device, enabled)` | `device: dict`, `enabled: bool` | Enable/disable audio (convenience method) |
| `set_secure_screen_enabled(device, enabled)` | `device: dict`, `enabled: bool` | Enable/disable secure screen (convenience method) |

### User Class

Handler for user-related API endpoints.

| Method | Description |
|--------|-------------|
| `get_info()` | Get current user account information |
| `get_homes()` | Get all homes associated with the user |

### SubscriptionManager Class

Manages real-time WebSocket subscriptions to both AppSync endpoints. Accessed via `api.subscriptions`.

| Method | Parameters | Description |
|--------|------------|-------------|
| `set_callback(callback)` | `callback: Callable[[str, dict], None]` | Set the event handler for all subscriptions |
| `async_subscribe_device(email)` | `email: str` | Subscribe to Core device events (returns subscription ID) |
| `async_subscribe_user(email)` | `email: str` | Subscribe to Core user events |
| `async_subscribe_home(email)` | `email: str` | Subscribe to Core home events |
| `async_subscribe_hav_device_status(email)` | `email: str` | Subscribe to Halo V device status |
| `async_subscribe_hav_device(email)` | `email: str` | Subscribe to Halo V device metadata |
| `async_subscribe_hav_event(email)` | `email: str` | Subscribe to Halo V activity events |
| `async_subscribe_hav_battery_voltage(email)` | `email: str` | Subscribe to Halo V battery updates |
| `async_subscribe_hav_sensor(email)` | `email: str` | Subscribe to Halo V door sensor/alarms |
| `async_subscribe_hav_led_status(email)` | `email: str` | Subscribe to Halo V LED status |
| `async_subscribe_hav_percentage(email)` | `email: str` | Subscribe to Halo V garage percentage |
| `async_subscribe_hav_vacation_mode(email)` | `email: str` | Subscribe to Halo V vacation mode |
| `async_subscribe_hav_wifi_signal(email)` | `email: str` | Subscribe to Halo V WiFi signal |
| `async_subscribe_hav_cycles(email)` | `email: str` | Subscribe to Halo V lock cycles |
| `async_subscribe_hav_auto_close_schedule(email)` | `email: str` | Subscribe to Halo V auto-close schedule |
| `async_subscribe_hav_temporary_key(email)` | `email: str` | Subscribe to Halo V temporary keys |
| `async_subscribe_hav_brightness_level(email)` | `email: str` | Subscribe to Halo V brightness |
| `async_unsubscribe(sub_id)` | `sub_id: str` | Unsubscribe by subscription ID |
| `async_close()` | — | Close all WebSocket connections |

### Exceptions

All exceptions inherit from `KwiksetError`.

#### Authentication Exceptions

| Exception | Description |
|-----------|-------------|
| `AuthenticationError` | Base exception for authentication-related errors |
| `MFAChallengeRequired` | MFA verification is required (contains `mfa_type` and `mfa_tokens`) |
| `Unauthenticated` | Invalid credentials or missing authentication |
| `TokenExpiredError` | Tokens expired and cannot be refreshed (triggers HA reauth) |
| `UserNotFound` | User account does not exist |
| `UserNotConfirmed` | User has not confirmed their email |
| `UserExists` | User already exists (during registration) |
| `PasswordChangeRequired` | Password must be changed |

#### Request Exceptions

| Exception | Description |
|-----------|-------------|
| `RequestError` | HTTP request failed (non-retryable) |
| `ConnectionError` | Network connection failed after retries (HA marks unavailable) |

#### Device Exceptions

| Exception | Description |
|-----------|-------------|
| `DeviceError` | Base exception for device-related errors |
| `InvalidDeviceError` | Device data is invalid or missing required fields |
| `InvalidUserError` | User data is invalid or missing required fields |
| `InvalidActionError` | Invalid action requested |
| `InvalidStatusError` | Invalid status value provided |
| `DeviceCommandError` | Device command failed |

#### General Exceptions

| Exception | Description |
|-----------|-------------|
| `KwiksetError` | Base exception for all library errors |
| `UnknownError` | Unknown error occurred |

#### Subscription Exceptions

| Exception | Description |
|-----------|-------------|
| `SubscriptionError` | Base exception for subscription/WebSocket errors |
| `SubscriptionTimeout` | WebSocket connection or subscription acknowledgment timed out |

## Error Handling

Implement comprehensive error handling for production applications:

```python
import asyncio
from aiokwikset import (
    API,
    KwiksetError,
    MFAChallengeRequired,
    RequestError,
    ConnectionError,
    TokenExpiredError,
    Unauthenticated,
    UserNotFound,
)

async def main() -> None:
    async with API() as api:
        try:
            await api.async_login("email@example.com", "password")
            
        except MFAChallengeRequired as err:
            # Handle MFA challenge
            print(f"MFA required: {err.mfa_type}")
            
        except Unauthenticated:
            print("Invalid email or password")
            
        except UserNotFound:
            print("Account does not exist")
            
        except ConnectionError as err:
            # Network issues - device should be marked unavailable
            print(f"Connection failed (after retries): {err}")
            
        except TokenExpiredError:
            # Tokens expired - need to re-authenticate
            print("Session expired. Please login again.")
            
        except RequestError as err:
            print(f"Request error: {err}")
            
        except KwiksetError as err:
            print(f"Unexpected Kwikset error: {err}")

asyncio.run(main())
```

### Home Assistant Error Handling

For Home Assistant integrations, map exceptions to appropriate HA behaviors:

```python
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from aiokwikset import ConnectionError, TokenExpiredError, Unauthenticated

async def async_setup_entry(hass, entry):
    try:
        await api.async_authenticate_with_tokens(...)
    except (TokenExpiredError, Unauthenticated):
        # Trigger reauth flow in HA
        raise ConfigEntryAuthFailed("Re-authentication required")
    except ConnectionError:
        # HA will retry setup later
        raise ConfigEntryNotReady("Cannot connect to Kwikset API")
```

## Contributing

Contributions are welcome! This is a community-maintained library.

1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Make your changes
4. Run tests (`pytest`)
5. Commit your changes (`git commit -m 'Add amazing feature'`)
6. Push to the branch (`git push origin feature/amazing-feature`)
7. Open a Pull Request

### Development Setup

```bash
# Clone the repository
git clone https://github.com/explosivo22/aiokwikset.git
cd aiokwikset

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

# Run tests
pytest

# Run linting
ruff check .

# Run type checking
mypy aiokwikset
```

## License

This project is licensed under the Apache License 2.0 - see the [LICENSE](LICENSE) file for details.

---

**Disclaimer**: This library is not affiliated with or endorsed by Kwikset. Use at your own risk.