Skip to content

Latest commit

 

History

History
1185 lines (957 loc) · 38.9 KB

File metadata and controls

1185 lines (957 loc) · 38.9 KB

Dyson REST API Client Documentation

This document prov # Dyson REST API Client Documentation

This document provides comprehensive documentation for both the synchronous and asynchronous Dyson REST API clients.

Table of Contents

Quick Start

Synchronous Usage

from libdyson_rest import DysonClient

# Two-step authentication (recommended)
client = DysonClient("user@example.com", "your_password")
if not client.authenticate():  # Returns False - OTP needed
    otp = input("Enter OTP from email: ")
    client.complete_authentication(otp)

devices = client.get_devices()
for device in devices:
    print(f"Device: {device.name} ({device.serial})")

# Context manager usage
with DysonClient("user@example.com", "your_password") as client:
    if not client.authenticate():
        otp = input("Enter OTP from email: ")
        client.complete_authentication(otp)
    devices = client.get_devices()

Asynchronous Usage

import asyncio
from libdyson_rest import AsyncDysonClient


async def main():
    # Two-step authentication (recommended)
    client = AsyncDysonClient("user@example.com", "your_password")
    if not await client.authenticate():  # Returns False - OTP needed
        otp = input("Enter OTP from email: ")
        await client.complete_authentication(otp)

    devices = await client.get_devices()
    for device in devices:
        print(f"Device: {device.name} ({device.serial})")
    await client.close()

    # Context manager usage
    async with AsyncDysonClient("user@example.com", "your_password") as client:
        if not await client.authenticate():
            otp = input("Enter OTP from email: ")
            await client.complete_authentication(otp)
        devices = await client.get_devices()


asyncio.run(main())

Client Overview

The library provides two client implementations:

  • DysonClient: Synchronous client using requests library
  • AsyncDysonClient: Asynchronous client using httpx library

Both clients provide identical APIs except for the async/await syntax and context manager protocols.

Synchronous Client (DysonClient)

Constructor

class DysonClient:
    def __init__(
        self,
        email: str | None = None,
        password: str | None = None,
        auth_token: str | None = None,
        request_timeout: int = 30,
        user_agent: str = "android client"
    ) -> None

Parameters:

  • email (str | None): User's email address for authentication
  • password (str | None): User's password for authentication
  • auth_token (str | None): Pre-existing authentication token
  • request_timeout (int): Request timeout in seconds (default: 30)
  • user_agent (str): User agent string for requests (default: "android client")

Authentication Methods

authenticate()

def authenticate(self) -> bool

Initiates the authentication process. Returns True if authentication completes without OTP, False if OTP is required.

Returns: bool - True if authenticated, False if OTP required

Raises:

  • DysonAuthError: Authentication failed
  • DysonConnectionError: Network/connection issues

complete_authentication(otp_code: str)

def complete_authentication(self, otp_code: str) -> None

Completes authentication using the OTP code received via email.

Parameters:

  • otp_code (str): One-time password from email

Raises:

  • DysonAuthError: Invalid OTP or authentication failed
  • DysonConnectionError: Network/connection issues

login_challenge(email: str, password: str)

def login_challenge(self, email: str, password: str) -> LoginChallenge

Low-level method to initiate login challenge (used internally by authenticate()).

Parameters:

  • email (str): User's email address
  • password (str): User's password

Returns: LoginChallenge object containing challenge details

complete_login(challenge_id: str, otp_code: str, email: str | None = None, password: str | None = None)

def complete_login(
    self,
    challenge_id: str,
    otp_code: str,
    email: str | None = None,
    password: str | None = None,
) -> LoginInformation

Low-level method to complete login with OTP (used internally by complete_authentication()).

Parameters:

  • challenge_id (str): Challenge ID from login_challenge response
  • otp_code (str): One-time password from email
  • email (str | None): User's email (optional if set in constructor)
  • password (str | None): User's password (optional if set in constructor)

Returns: LoginInformation object containing auth token and account details

Device Management Methods

get_devices()

def get_devices(self) -> list[DysonDevice]

Retrieves all Dyson devices associated with the account.

Returns: List of DysonDevice objects

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

get_device_by_serial(serial: str)

def get_device_by_serial(self, serial: str) -> DysonDevice | None

Retrieves a specific device by its serial number.

Parameters:

  • serial (str): Device serial number

Returns: DysonDevice object if found, None otherwise

get_device_credentials(device: DysonDevice)

def get_device_credentials(self, device: DysonDevice) -> dict[str, str]

Retrieves MQTT credentials for a specific device.

Parameters:

  • device (DysonDevice): Device object

Returns: Dictionary containing MQTT credentials (username, password, hostname)

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

Vis Nav Robot Vacuum Methods

get_clean_maps(serial_number: str, *, api_version: int, include_dust_map: bool = True)

def get_clean_maps(
    self, serial_number: str, *, api_version: int, include_dust_map: bool = True
) -> list[CleanRecord]

Retrieves recent cleaning run history for a Vis Nav robot vacuum.

Parameters:

  • serial_number (str): Device serial number
  • api_version (int): API version to use — 1 for Dyson 360 Vis Nav (H8T / product type 276B), 2 for Dyson 360 Spot+Clean and newer (VS6 / RB05 / product type 804A)
  • include_dust_map (bool): When True (default), fetches the aggregated dust-density map blob for each run

Returns: List of CleanRecord objects, newest first

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

get_persistent_map_metadata(serial_number: str, *, api_version: int)

def get_persistent_map_metadata(self, serial_number: str, *, api_version: int) -> list[PersistentMapMeta]

Retrieves the list of saved maps (zone names, IDs, and areas) for a Vis Nav.

Parameters:

  • serial_number (str): Device serial number
  • api_version (int): API version to use — 1 for Vis Nav (H8T), 2 for Spot+Clean and newer

Returns: List of PersistentMapMeta objects — one per stored map

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

get_persistent_map(serial_number: str, map_id: str, *, api_version: int)

def get_persistent_map(self, serial_number: str, map_id: str, *, api_version: int) -> PersistentMap

Retrieves the full map record for a single persistent map, including the floor-plan PNG.

Parameters:

  • serial_number (str): Device serial number
  • map_id (str): Persistent map ID (from get_persistent_map_metadata)
  • api_version (int): API version to use — 1 for Vis Nav (H8T), 2 for Spot+Clean and newer

Returns: PersistentMap with presentation image, display orientation, world offset, and zone definitions

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

get_recommended_cleans(serial_number: str)

def get_recommended_cleans(self, serial_number: str) -> list[RecommendedCleanMap]

Retrieves Dyson's per-zone clean recommendations ranked by accumulated dust load.

Parameters:

  • serial_number (str): Device serial number

Returns: List of RecommendedCleanMap objects (one per persistent map), each containing per-zone ZonePrediction entries with ZoneDustBreakdown data

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

set_zone_behaviour(serial_number, map_id, zone_id, strategy)

def set_zone_behaviour(
    self,
    serial_number: str,
    map_id: str,
    zone_id: str,
    strategy: CleaningStrategy | str,
) -> None

Sets the per-zone cleaning strategy. Equivalent to changing a zone's behaviour in the MyDyson app. The device applies the override on its next clean.

Note: The API path is /v1/app/{serial}/{mapId}/zones/{zoneId}/zone-behaviours — there is no persistent-maps/ segment in this path.

Parameters:

  • serial_number (str): Device serial number
  • map_id (str): Persistent map ID
  • zone_id (str): Zone ID to update
  • strategy (CleaningStrategy | str): Cleaning strategy — AUTO, QUICK, QUIET, or BOOST

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

EC Air Purifier Methods

get_daily_environment_data(serial_number: str, language: str = "en")

def get_daily_environment_data(
    self, serial_number: str, language: str = "en"
) -> DailyAirQualityData

Retrieves today's indoor air-quality history at 15-minute resolution.

Parameters:

  • serial_number (str): Device serial number
  • language (str): Language code for localised field values (default: "en")

Returns: DailyAirQualityData with sample series, start_time, and resolution_minutes

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

get_scheduled_events(serial_number: str, product_type: str | None = None)

def get_scheduled_events(
    self, serial_number: str, product_type: str | None = None
) -> ScheduledEventsData

Retrieves the MyDyson-app automation schedule for a device.

Parameters:

  • serial_number (str): Device serial number
  • product_type (str | None): Device product-type code (e.g. "438K"). Required by the server to return the correct schedule schema; omit only if the product type is unknown.

Returns: ScheduledEventsData with schedule_enabled flag and list of ScheduledEvent objects

Raises:

  • DysonAuthError: Not authenticated or token expired
  • DysonAPIError: API request failed
  • DysonConnectionError: Network/connection issues

Context Manager Support

with DysonClient("user@example.com", "password") as client:
    if not client.authenticate():
        otp = input("Enter OTP: ")
        client.complete_authentication(otp)
    devices = client.get_devices()
# Client is automatically cleaned up

Vis Nav Robot Vacuum Additional Methods

get_clean_map_data(serial_number: str, clean_id: str)

def get_clean_map_data(self, serial_number: str, clean_id: str) -> dict[str, Any]

Retrieves detailed data for a single cleaning session (path, duration, area coverage, etc.).

Parameters:

  • serial_number (str): Device serial number
  • clean_id (str): Cleaning session ID (from get_clean_maps)

Returns: Raw dict with detailed session data

update_persistent_map(serial_number: str, map_id: str, name: str | None = None)

def update_persistent_map(self, serial_number: str, map_id: str, name: str | None = None) -> None

Updates an existing persistent map (e.g. rename it).

delete_persistent_map(serial_number: str, map_id: str)

def delete_persistent_map(self, serial_number: str, map_id: str) -> None

Permanently deletes a persistent map from the device.

update_map_metadata(serial_number: str, map_id: str, name: str | None = None)

def update_map_metadata(self, serial_number: str, map_id: str, name: str | None = None) -> None

Updates the persistent map metadata stored by the API (e.g. the display name).

get_clean_estimation(serial_number: str, map_id: str, zone_ids: list[str] | None = None)

def get_clean_estimation(
    self,
    serial_number: str,
    map_id: str,
    zone_ids: list[str] | None = None,
) -> dict[str, Any]

Requests a server-side estimation of cleaning time and area for a set of zones.

Parameters:

  • serial_number (str): Device serial number
  • map_id (str): Persistent map ID
  • zone_ids (list[str] | None): Zone IDs to estimate; None estimates all zones

Returns: Raw dict with estimated duration and coverage

get_restrictions(serial_number: str, map_id: str)

def get_restrictions(self, serial_number: str, map_id: str) -> dict[str, Any]

Retrieves the no-go and restricted-area definitions for a persistent map.

update_restrictions(serial_number: str, map_id: str, body: dict[str, Any])

def update_restrictions(self, serial_number: str, map_id: str, body: dict[str, Any]) -> None

Replaces the restriction definitions (no-go zones, keep-out areas) for a persistent map.

divide_zone(serial_number: str, map_id: str, body: dict[str, Any])

def divide_zone(self, serial_number: str, map_id: str, body: dict[str, Any]) -> None

Splits a single zone into two sub-zones along a specified boundary.

merge_zones(serial_number: str, map_id: str, body: dict[str, Any])

def merge_zones(self, serial_number: str, map_id: str, body: dict[str, Any]) -> None

Merges two or more zones into a single zone on a persistent map.

get_live_map_cleaning(serial_number: str)

def get_live_map_cleaning(self, serial_number: str) -> dict[str, Any]

Returns the robot's real-time position and cleaned footprint during an active cleaning session. Poll this endpoint while a clean is in progress to display a live map.

get_live_map_mapping(serial_number: str)

def get_live_map_mapping(self, serial_number: str) -> dict[str, Any]

Returns the robot's real-time position and discovered floor plan during an active mapping session.

set_scheduled_events(serial_number, enabled, events, product_type=None)

def set_scheduled_events(
    self,
    serial_number: str,
    enabled: bool,
    events: list[dict[str, Any]],
    product_type: str | None = None,
) -> None

Replaces the scheduled automation event list for a device. Each event dict should contain enabled, days (list of int, 0=Mon), startTime (HH:MM), and settings.

Parameters:

  • serial_number (str): Device serial number
  • enabled (bool): Whether the overall schedule is active
  • events (list[dict]): Replacement list of scheduled events
  • product_type (str | None): Product-type code, e.g. "438K"

get_schedule_binary(serial_number: str)

def get_schedule_binary(self, serial_number: str) -> bytes

Downloads the device schedule as a raw binary blob for BLE programming.

Returns: bytes — raw schedule binary

EC Air Purifier Additional Methods

get_environment_history(serial_number: str)

def get_environment_history(self, serial_number: str) -> dict[str, Any]

Retrieves the multi-day indoor air-quality history for a device. Unlike get_daily_environment_data which returns today's data at 15-minute resolution, this endpoint provides a longer historical dataset across multiple days.

Returns: Raw dict with historical air-quality data

get_energy_insights(serial_number: str, year: int | None = None, month: int | None = None)

def get_energy_insights(
    self,
    serial_number: str,
    year: int | None = None,
    month: int | None = None,
) -> dict[str, Any]

Retrieves monthly energy consumption insights for an EC air purifier.

Parameters:

  • serial_number (str): Device serial number
  • year (int | None): Year (defaults to current year on the server)
  • month (int | None): Month 1–12 (defaults to current month on the server)

Returns: Raw dict with monthly energy/EC usage data

Device Management Additional Methods

get_timezone(serial_number: str)

def get_timezone(self, serial_number: str) -> str | None

Retrieves the IANA timezone configured for a device (e.g. "Europe/London").

set_timezone(serial_number: str, timezone: str)

def set_timezone(self, serial_number: str, timezone: str) -> None

Sets the IANA timezone for a device.

get_ota_info(serial_number: str)

def get_ota_info(self, serial_number: str) -> dict[str, Any]

Returns over-the-air firmware update information for a device, including available firmware versions and update status.

is_banned_machine(serial_number: str)

def is_banned_machine(self, serial_number: str) -> bool

Checks whether a device serial number appears on Dyson's banned-machine list. Banned devices cannot connect to cloud services.

Returns: True if the device is banned, False otherwise

get_feature_support()

def get_feature_support(self) -> dict[str, Any]

Returns Dyson's global server-controlled feature-flag object. The MyDyson app uses this to enable or disable app features at runtime without an app update.

get_voice_languages(serial_number: str)

def get_voice_languages(self, serial_number: str) -> list[str]

Returns the list of supported voice-command language codes for a device.

Returns: list[str] of IETF language tags, e.g. ["en-GB", "fr-FR"]

Product Support Methods

get_product_faults(serial_number: str)

def get_product_faults(self, serial_number: str) -> dict[str, Any]

Returns known product fault codes and their recommended remedies for a device. Useful for building in-app troubleshooting flows.

get_product_guide(serial_number: str)

def get_product_guide(self, serial_number: str) -> dict[str, Any]

Returns the product user guide content for a device as served by the Dyson API.

get_product_voice_commands(serial_number: str)

def get_product_voice_commands(self, serial_number: str) -> dict[str, Any]

Returns the voice command reference content for a device.

Push Notification Methods

register_push_token(application_id, token, platform, serial_numbers=None)

def register_push_token(
    self,
    application_id: str,
    token: str,
    platform: str,
    serial_numbers: list[str] | None = None,
) -> dict[str, Any]

Registers an APNs or FCM push notification token with the Dyson API.

Parameters:

  • application_id (str): Application/token identifier
  • token (str): Platform push notification token
  • platform (str): "ios" or "android"
  • serial_numbers (list[str] | None): Device serial numbers to associate

Returns: Raw dict with registration confirmation

get_notification_permissions(application_id: str, serial_number: str)

def get_notification_permissions(self, application_id: str, serial_number: str) -> dict[str, Any]

Retrieves the per-device push notification permission settings for an application.

update_notification_permissions(application_id, serial_number, permissions)

def update_notification_permissions(
    self,
    application_id: str,
    serial_number: str,
    permissions: dict[str, Any],
) -> None

Updates the push notification permission settings for a device and application.

Smart Home (NCP/NSP) Methods

get_registered_products()

def get_registered_products(self) -> dict[str, Any]

Returns smart-home products registered with Dyson's NCP (Notification/Control Platform).

register_ncp(body: dict[str, Any])

def register_ncp(self, body: dict[str, Any]) -> None

Registers a device with Dyson's NCP smart-home platform.

register_nsp(body: dict[str, Any])

def register_nsp(self, body: dict[str, Any]) -> None

Registers a device with Dyson's NSP (another smart-home integration platform).

Constructor

class AsyncDysonClient:
    def __init__(
        self,
        email: str | None = None,
        password: str | None = None,
        auth_token: str | None = None,
        request_timeout: int = 30,
        user_agent: str = "android client"
    ) -> None

Parameters: Same as DysonClient

Authentication Methods

authenticate()

async def authenticate(self) -> bool

Async version of authenticate method.

Returns: bool - True if authenticated, False if OTP required

complete_authentication(otp_code: str)

async def complete_authentication(self, otp_code: str) -> None

Async version of complete_authentication method.

Parameters:

  • otp_code (str): One-time password from email

login_challenge(email: str, password: str)

async def login_challenge(self, email: str, password: str) -> LoginChallenge

Async version of login_challenge method.

complete_login(challenge_id: str, otp_code: str, email: str | None = None, password: str | None = None)

async def complete_login(
    self,
    challenge_id: str,
    otp_code: str,
    email: str | None = None,
    password: str | None = None,
) -> LoginInformation

Async version of complete_login method.

Device Management Methods

get_devices()

async def get_devices(self) -> list[DysonDevice]

Async version of get_devices method.

get_device_by_serial(serial: str)

async def get_device_by_serial(self, serial: str) -> DysonDevice | None

Async version of get_device_by_serial method.

get_device_credentials(device: DysonDevice)

async def get_device_credentials(self, device: DysonDevice) -> dict[str, str]

Async version of get_device_credentials method.

Vis Nav Robot Vacuum Methods

get_clean_maps(serial_number: str, *, api_version: int, include_dust_map: bool = True)

async def get_clean_maps(
    self, serial_number: str, *, api_version: int, include_dust_map: bool = True
) -> list[CleanRecord]

Async version of get_clean_maps. See sync client for full parameter/return documentation.

get_persistent_map_metadata(serial_number: str, *, api_version: int)

async def get_persistent_map_metadata(self, serial_number: str, *, api_version: int) -> list[PersistentMapMeta]

Async version of get_persistent_map_metadata.

get_persistent_map(serial_number: str, map_id: str, *, api_version: int)

async def get_persistent_map(self, serial_number: str, map_id: str, *, api_version: int) -> PersistentMap

Async version of get_persistent_map.

get_recommended_cleans(serial_number: str)

async def get_recommended_cleans(self, serial_number: str) -> list[RecommendedCleanMap]

Async version of get_recommended_cleans.

set_zone_behaviour(serial_number, map_id, zone_id, strategy)

async def set_zone_behaviour(
    self,
    serial_number: str,
    map_id: str,
    zone_id: str,
    strategy: CleaningStrategy | str,
) -> None

Async version of set_zone_behaviour.

EC Air Purifier Methods

get_daily_environment_data(serial_number: str, language: str = "en")

async def get_daily_environment_data(
    self, serial_number: str, language: str = "en"
) -> DailyAirQualityData

Async version of get_daily_environment_data.

get_scheduled_events(serial_number: str, product_type: str | None = None)

async def get_scheduled_events(
    self, serial_number: str, product_type: str | None = None
) -> ScheduledEventsData

Async version of get_scheduled_events.

Resource Management

close()

async def close(self) -> None

Properly closes the HTTP client session. Always call this when done with the client, or use async context manager.

Async Context Manager Support

async with AsyncDysonClient("user@example.com", "password") as client:
    if not await client.authenticate():
        otp = input("Enter OTP: ")
        await client.complete_authentication(otp)
    devices = await client.get_devices()
# Client is automatically closed

Method Comparison Table

Feature Synchronous Asynchronous Notes
Constructor DysonClient() AsyncDysonClient() Same parameters
Authentication authenticate() await authenticate() Returns bool
Complete Auth complete_authentication() await complete_authentication() Takes OTP code
Get Devices get_devices() await get_devices() Returns device list
Get Device get_device_by_serial() await get_device_by_serial() Find by serial
Get Credentials get_device_credentials() await get_device_credentials() MQTT credentials
Clean Maps get_clean_maps() await get_clean_maps() Requires api_version=1 (Vis Nav) or api_version=2 (Spot+Clean)
Clean Map Data get_clean_map_data() await get_clean_map_data() Detailed session data (v2 only)
Map Metadata get_persistent_map_metadata() await get_persistent_map_metadata() Requires api_version=1 or api_version=2
Full Map get_persistent_map() await get_persistent_map() Requires api_version=1 or api_version=2
Update Map update_persistent_map() await update_persistent_map() Rename persistent map
Delete Map delete_persistent_map() await delete_persistent_map() Remove persistent map
Update Metadata update_map_metadata() await update_map_metadata() Update map metadata
Clean Estimate get_clean_estimation() await get_clean_estimation() Zone cleaning estimate
Restrictions get_restrictions() await get_restrictions() No-go zones
Update Restrictions update_restrictions() await update_restrictions() Set no-go zones
Divide Zone divide_zone() await divide_zone() Split a zone
Merge Zones merge_zones() await merge_zones() Combine zones
Live Map (Clean) get_live_map_cleaning() await get_live_map_cleaning() Real-time clean map
Live Map (Map) get_live_map_mapping() await get_live_map_mapping() Real-time mapping
Recommended get_recommended_cleans() await get_recommended_cleans() Dust predictions
Zone Strategy set_zone_behaviour() await set_zone_behaviour() Vis Nav zone config
Set Schedule set_scheduled_events() await set_scheduled_events() Replace schedule
Schedule Binary get_schedule_binary() await get_schedule_binary() BLE schedule blob
AQI History get_daily_environment_data() await get_daily_environment_data() EC purifier (today)
AQI Multi-Day get_environment_history() await get_environment_history() EC purifier history
Energy Insights get_energy_insights() await get_energy_insights() Monthly usage
Get Schedule get_scheduled_events() await get_scheduled_events() Automation schedule
Outdoor Env get_outdoor_environment_data() await get_outdoor_environment_data() Outdoor air quality
Timezone Get get_timezone() await get_timezone() IANA timezone
Timezone Set set_timezone() await set_timezone() IANA timezone
OTA Info get_ota_info() await get_ota_info() Firmware update info
Banned Check is_banned_machine() await is_banned_machine() Cloud ban check
Feature Flags get_feature_support() await get_feature_support() App feature flags
Voice Languages get_voice_languages() await get_voice_languages() Voice command langs
Product Faults get_product_faults() await get_product_faults() Fault codes
Product Guide get_product_guide() await get_product_guide() User guide content
Voice Commands get_product_voice_commands() await get_product_voice_commands() Voice cmd reference
Register Token register_push_token() await register_push_token() APNs/FCM token
Notif Perms get_notification_permissions() await get_notification_permissions() Per-device perms
Update Notif update_notification_permissions() await update_notification_permissions() Update perms
NCP Products get_registered_products() await get_registered_products() Smart home devices
Register NCP register_ncp() await register_ncp() NCP registration
Register NSP register_nsp() await register_nsp() NSP registration
Context Manager with client: async with client: Auto cleanup
Resource Cleanup Automatic await client.close() Manual or context manager

Authentication Flow

The library uses a two-step authentication process:

  1. Initial Authentication: Call authenticate() with email/password

    • If MFA is disabled: Returns True, authentication complete
    • If MFA is enabled: Returns False, OTP sent to email
  2. OTP Completion: If authenticate() returns False, call complete_authentication(otp)

    • Provide the OTP code received via email
    • Authentication completes successfully

Example Flow

# Step 1: Initial authentication
client = DysonClient("user@example.com", "password")
if client.authenticate():
    print("Authentication complete!")
else:
    # Step 2: OTP required
    otp = input("Enter OTP from email: ")
    client.complete_authentication(otp)
    print("Authentication complete with OTP!")

# Now authenticated - can make API calls
devices = client.get_devices()

Legacy Single-Step Authentication

# Deprecated: Single-step with OTP (if known)
client = DysonClient("user@example.com", "password")
login_info = client.complete_login(challenge_id, "123456")

Error Handling

The library defines custom exceptions for different error scenarios:

Exception Hierarchy

DysonError (base)
├── DysonAuthError (authentication failures)
├── DysonAPIError (API response errors)
└── DysonConnectionError (network/connection issues)

Exception Details

DysonError

Base exception class for all Dyson-related errors.

DysonAuthError

Raised when authentication fails:

  • Invalid credentials
  • Invalid OTP code
  • Token expired
  • Account locked

DysonAPIError

Raised when API requests fail:

  • Invalid API response format
  • Server errors (5xx)
  • Rate limiting
  • Invalid device serial

DysonConnectionError

Raised when network issues occur:

  • Connection timeout
  • DNS resolution failure
  • SSL/TLS errors
  • Network unreachable

Error Handling Example

from libdyson_rest import (
    DysonClient,
    DysonAuthError,
    DysonAPIError,
    DysonConnectionError,
)

try:
    client = DysonClient("user@example.com", "password")
    if not client.authenticate():
        otp = input("Enter OTP: ")
        client.complete_authentication(otp)

    devices = client.get_devices()

except DysonAuthError as e:
    print(f"Authentication failed: {e}")
except DysonAPIError as e:
    print(f"API error: {e}")
except DysonConnectionError as e:
    print(f"Connection error: {e}")

Data Models

DysonDevice

Represents a Dyson device with the following attributes:

@dataclass
class DysonDevice:
    serial: str  # Device serial number
    name: str  # User-assigned device name
    product_type: str  # Product identifier (e.g., "520")
    version: str  # Firmware version
    auto_update: bool  # Auto-update enabled
    new_version_available: bool  # Firmware update available
    category: str  # Device category (e.g., "purifier")

    # Optional attributes (may be None)
    local_credentials: dict[str, str] | None  # Local MQTT credentials
    connection_type: str | None  # Connection type
    mqtt_server: str | None  # MQTT server hostname

LoginChallenge

Represents a login challenge response:

@dataclass
class LoginChallenge:
    challenge_id: str  # Challenge identifier for OTP completion
    user_id: str  # User account identifier

LoginInformation

Represents completed login information:

@dataclass
class LoginInformation:
    token: str  # Authentication token
    account: dict  # Account details
    challenge_id: str  # Challenge identifier used

Usage Examples with Data Models

# Working with devices
devices = client.get_devices()
for device in devices:
    print(f"Device: {device.name}")
    print(f"Serial: {device.serial}")
    print(f"Type: {device.product_type}")
    print(f"Category: {device.category}")
    
    if device.new_version_available:
        print("⚠️  Firmware update available")
    
    # Get MQTT credentials for device
    credentials = client.get_device_credentials(device)
    print(f"MQTT Host: {credentials['hostname']}")

# Find specific device
device = client.get_device_by_serial("ABC-DEF-123")
if device:
    print(f"Found: {device.name}")
else:
    print("Device not found")

Vis Nav Robot Vacuum Models

CleanRecord

A single cleaning run, including timeline, dust map, and cleaning programme.

@dataclass
class CleanRecord:
    start_time: datetime | None
    end_time: datetime | None
    timeline: list[CleanTimelineEvent]
    dust_map: DustMapData | None  # Aggregated dust-density grid
    clean_map_position: CleanMapPosition | None  # World origin of dust map
    cleaning_programme: CleaningProgramme | None  # Zone-clean config used
    footprint: CleanedFootprint | None  # Cleaned area + floor-plan crop
    raw: dict  # Full raw API response

    @property
    def is_zone_clean(self) -> bool: ...

CleaningStrategy (Enum)

Per-zone cleaning intensity used with set_zone_behaviour():

Value Description
AUTO Dyson-selected intensity
QUICK Fast single pass
QUIET Low-noise mode
BOOST Intensive multi-pass

PersistentMapMeta

Metadata about a stored floor map, including its zones.

@dataclass
class PersistentMapMeta:
    id: str
    name: str | None
    zones: list[ZoneMeta]

    def zone_by_id(self, zone_id: str) -> ZoneMeta | None: ...
    def zone_by_name(self, name: str) -> ZoneMeta | None: ...

ZoneMeta

@dataclass
class ZoneMeta:
    id: str
    name: str | None
    icon: str | None
    area: float | None  # Zone area in m²

PersistentMap

Full map record including presentation image and zone definitions.

@dataclass
class PersistentMap:
    id: str
    offset_x: float | None  # World-mm X origin
    offset_y: float | None  # World-mm Y origin
    display_orientation: int  # Degrees rotation for display
    presentation_map_data: str | None  # Base64-encoded floor-plan PNG
    zones_definition: dict | None
    zones: list[ZoneMeta]

RecommendedCleanMap

@dataclass
class RecommendedCleanMap:
    persistent_map_id: str
    zone_predictions: list[ZonePrediction]

    def sorted_by_dust(self) -> list[ZonePrediction]: ...

ZonePrediction

@dataclass
class ZonePrediction:
    zone_id: str
    dust: ZoneDustBreakdown

ZoneDustBreakdown

Per-particle-class dust load (milligrams).

@dataclass
class ZoneDustBreakdown:
    extra_fine: float
    fine: float
    medium: float
    large: float
    other: float
    total: float  # Sum of all classes
    raw: list  # Original API array

DustMapData

Aggregated dust-density grid returned with each clean record.

@dataclass
class DustMapData:
    width: int
    height: int
    resolution: float  # mm per pixel
    dust_data: list[list[float]]  # 2-D grid, values 0–1 (divided by scaleFactor)

EC Air Purifier Models

DailyAirQualityData

@dataclass
class DailyAirQualityData:
    start_time: datetime | None
    resolution_minutes: int  # Typically 15
    samples: list[float | None]  # AQI values; None = no reading

    @property
    def latest_sample(self) -> float | None: ...
    @property
    def min_sample(self) -> float | None: ...
    @property
    def max_sample(self) -> float | None: ...

ScheduledEventsData

@dataclass
class ScheduledEventsData:
    schedule_enabled: bool
    events: list[ScheduledEvent]

    @property
    def active_events(self) -> list[ScheduledEvent]: ...

ScheduledEvent

@dataclass
class ScheduledEvent:
    enabled: bool
    days: list[int]  # 0 = Monday … 6 = Sunday
    start_time: str | None  # "HH:MM" local time
    raw: dict

Migration from v0.6.x

If upgrading from version 0.6.x, note the following changes:

Authentication Changes

  • Old (v0.6.x): Single authenticate(otp) method
  • New (v0.7.0): Two-step authenticate()complete_authentication(otp)
# Old v0.6.x approach
client = DysonClient("user@example.com", "password")
client.authenticate("123456")  # OTP required upfront

# New v0.7.0 approach
client = DysonClient("user@example.com", "password")
if not client.authenticate():  # Try without OTP first
    otp = input("Enter OTP: ")
    client.complete_authentication(otp)

Async Client Changes

  • New in v0.7.0: Full async support with AsyncDysonClient
  • SSL blocking issues resolved with lazy HTTP client initialization
  • Proper async context manager support

API Endpoint Updates

  • Updated to use latest Dyson API endpoints (/v3/manifest)
  • Better error handling and validation
  • Enhanced device metadata support

This documentation covers the complete API surface for both synchronous and asynchronous clients. For additional examples and troubleshooting guides, see the examples/ directory in the repository.

asyncio.run(main())