libdeye package

Submodules

libdeye.cli module

CLI tool for testing libdeye library during development.

async libdeye.cli.authenticate(session: ClientSession, username: str, password: str, auth_token: str | None = None) → DeyeCloudApi

Authenticate with Deye Cloud API.

async libdeye.cli.get_classic_mqtt_info(api: DeyeCloudApi) → None

Get and display Classic platform MQTT information.

async libdeye.cli.get_device_state(api: DeyeCloudApi, device_id: str) → None

Get the current state of a device.

async libdeye.cli.get_fog_mqtt_info(api: DeyeCloudApi) → None

Get and display Fog platform MQTT information.

async libdeye.cli.list_devices(api: DeyeCloudApi) → None

List all devices associated with the account.

async libdeye.cli.list_products(api: DeyeCloudApi) → None

List all available products from Deye.

libdeye.cli.load_env_file(env_file: str = '.env') → dict[str, str]

Load environment variables from a .env file.

libdeye.cli.main() → None

Main entry point for the CLI.

async libdeye.cli.monitor_device(api: DeyeCloudApi, device_id: str) → None

Monitor a device for state updates.

async libdeye.cli.print_auth_token(api: DeyeCloudApi) → None

Print the authentication token.

libdeye.cli.print_device_state(state: DeyeDeviceState) → None

Print the state of a device.

async libdeye.cli.refresh_token(api: DeyeCloudApi) → None

Force refresh the authentication token.

async libdeye.cli.run_cli(args: Namespace, username: str, password: str, auth_token: str | None, device_id: str | None) → None

Run the CLI with the given arguments.

async libdeye.cli.set_device_state(api: DeyeCloudApi, device_id: str, power: bool | None = None, mode: DeyeDeviceMode | None = None, fan_speed: DeyeFanSpeed | None = None, target_humidity: int | None = None, anion: bool | None = None, water_pump: bool | None = None, oscillating: bool | None = None, child_lock: bool | None = None) → None

Set the state of a device.

libdeye.cloud_api module

Deye Cloud API related stuffs

class libdeye.cloud_api.DeyeApiRequestFogPlatformDeviceCommands

Bases: TypedDict

Device Command for the Fog platform sent to the API

KeyLock: int

Mode: int

NegativeIon: int

Power: int

SetHumidity: int

SwingingWind: int

WaterPump: int

WindSpeed: int

class libdeye.cloud_api.DeyeApiResponseClassicPlatformMqttInfo

Bases: TypedDict

MQTT information for the Deye platform returned by the API

clientid: str

endpoint: str

loginname: str

mqtthost: str

mqttport: int

password: str

sslport: int

class libdeye.cloud_api.DeyeApiResponseDeviceInfo

Bases: TypedDict

Device information returned by the API

alias: str

device_id: str

device_name: str

deviceid: str

gatewaytype: int

is_combo: bool

mac: str

online: bool

payload: Any

picture_v3: str

platform: DeyeIotPlatform

product_icon: str

product_id: str

product_name: str

product_type: str

producttype_id: int

protocol_version: str

role: int

user_count: int

work_time: int

class libdeye.cloud_api.DeyeApiResponseEnvelope

Bases: TypedDict

Message envelope for all API responses

data: Any

meta: DeyeApiResponseEnvelopeMeta

class libdeye.cloud_api.DeyeApiResponseEnvelopeMeta

Bases: TypedDict

Meta information for the API message envelope

code: int

message: str

class libdeye.cloud_api.DeyeApiResponseFogPlatformDeviceProperties

Bases: TypedDict

Device properties for the Fog platform returned by the API

CompressorStatus: int

CurrentAmbientTemperature: int

CurrentCoilTemperature: int

CurrentEnvironmentalHumidity: int

CurrentExhaustTemperature: int

Demisting: int

EnvironmentalRating: int

Fan: int

KeyLock: int

Mode: int

NegativeIon: int

Power: int

ProtocolVersion: int

SetHumidity: int

SolenoidValve: int

SwingingWind: int

TimedOff: int

TimedOn: int

TimedShutdownHourSetting: int

TimedShutdownMinuteSettingTime: int

TimedShutdownTimeRemainingHours: int

TimedShutdownTimeRemainingMinutes: int

TimedStartupHoursSetTime: int

TimedStartupMinuteSettingTime: int

TimedStartupTimeRemainingHours: int

TimedStartupTimeRemainingMinutes: int

WaterPump: int

WaterTank: int

WindSpeed: int

fault: dict[str, int]

class libdeye.cloud_api.DeyeApiResponseFogPlatformMqttInfo

Bases: TypedDict

MQTT information for the Fog platform returned by the API

clientid: str

expire: int

mqtt_host: str

password: str

ssl_port: str

topic: DeyeApiResponseFogPlatformMqttTopics

username: str

ws_port: str

class libdeye.cloud_api.DeyeApiResponseFogPlatformMqttTopics

Bases: TypedDict

MQTT topics returned by the API

all: list[str]

pub: list[str]

sub: list[str]

class libdeye.cloud_api.DeyeApiResponseProductDefinition

Bases: TypedDict

Product definition information returned by the API

brand: str

configType: int

config_guide: str

model: str

picture: str

picture_v3: str | None

pname: str

productid: str

status: int

class libdeye.cloud_api.DeyeApiResponseProductType

Bases: TypedDict

Product type information returned by the API

pdata: list[DeyeApiResponseProductDefinition]

ptype: str

ptypename: str

class libdeye.cloud_api.DeyeCloudApi(session: ClientSession, username: str, password: str, auth_token: str | None = None)

Bases: object

Interact with Deye Cloud APIs.

property auth_token: str | None

Get the auth token

async authenticate() → None

Authenticate by username/password and set the auth token.

async get_device_list() → list[DeyeApiResponseDeviceInfo]

Get all connected devices for current user

async get_deye_platform_mqtt_info() → DeyeApiResponseClassicPlatformMqttInfo

Get MQTT server info / credentials for current user (Deye platform)

async get_fog_platform_device_properties(device_id: str) → DeyeApiResponseFogPlatformDeviceProperties

Get properties for a device on the Fog platform

async get_fog_platform_mqtt_info() → DeyeApiResponseFogPlatformMqttInfo

Get MQTT server info / credentials for current user (Fog platform)

async get_product_list() → list[DeyeApiResponseProductType]

Get all available products

on_auth_token_refreshed: Callable[[str], None] | None

async poll_fog_platform_device_properties(device_id: str) → None

Poll properties for a device on the Fog platform

async refresh_token_if_near_expiry(force: bool = False) → None

Get a new auth token by calling /refreshToken if the current auth token is about to be expired. This will be automatically called for each API call.

Parameters:

force – If True, refresh the token regardless of expiry time.

async set_fog_platform_device_properties(device_id: str, params: object) → None

Set properties for a device on the Fog platform

user_id: str | None

exception libdeye.cloud_api.DeyeCloudApiCannotConnectError

Bases: Exception

Error to indicate we cannot connect.

exception libdeye.cloud_api.DeyeCloudApiInvalidAuthError

Bases: Exception

Error to indicate there is invalid auth.

class libdeye.cloud_api.DeyeIotPlatform(value)

Bases: IntEnum

IoT platform of devices

Classic = 1

Fog = 2

libdeye.cloud_api.ensure_valid_response_code(result: DeyeApiResponseEnvelope) → None

Raise errors if we don’t have a valid result[“meta”][“code”]

libdeye.const module

Constants for the Deye Cloud API.

class libdeye.const.DeyeDeviceMode(value)

Bases: IntEnum

All supported mode

AIR_PURIFIER_MODE = 2

AUTO_MODE = 3

CLOTHES_DRYER_MODE = 1

MANUAL_MODE = 0

SLEEP_MODE = 6

UNKNOWN_MODE = 4

UNKNOWN_MODE_2 = 5

class libdeye.const.DeyeFanSpeed(value)

Bases: IntEnum

All supported fan speed

FULL = 4

HIGH = 3

LOW = 1

MIDDLE = 2

STOPPED = 0

class libdeye.const.DeyeProductConfig

Bases: TypedDict

Feature config for a specific Deye product

anion: bool

fan_speed: list[DeyeFanSpeed]

max_target_humidity: int

min_target_humidity: int

mode: list[DeyeDeviceMode]

oscillating: bool

water_pump: bool

class libdeye.const.DeyeProductPartialConfig

Bases: TypedDict

Feature config for a specific Deye product (partial)

anion: bool

fan_speed: list[DeyeFanSpeed]

max_target_humidity: int

min_target_humidity: int

mode: list[DeyeDeviceMode]

oscillating: bool

water_pump: bool

libdeye.const.get_product_feature_config(product_id: str) → DeyeProductConfig

Get supported features of the product

libdeye.device_command module

Utilities for device command parsing

class libdeye.device_command.DeyeDeviceCommand(anion_switch: bool = False, water_pump_switch: bool = False, power_switch: bool = False, oscillating_switch: bool = False, child_lock_switch: bool = False, fan_speed: DeyeFanSpeed = DeyeFanSpeed.LOW, mode: DeyeDeviceMode = DeyeDeviceMode.MANUAL_MODE, target_humidity: int = 60)

Bases: object

A class to store the command to control the device

to_bytes() → bytes

Get binary representation of this command

to_json() → dict[str, int]

Get JSON representation of this command

to_json_diff(baseline: DeyeDeviceCommand | DeyeDeviceState) → dict[str, int]

Get JSON with only properties that differ from the baseline.

class libdeye.device_command.DeyeDeviceCommandFlag(value)

Bases: IntFlag

Bit flags used in the command

ANION_SWITCH = 64

CHILD_LOCK_SWITCH = 4

OSCILLATING_SWITCH = 2

POWEROFF_SWITCH = 8

POWERON_SWITCH = 16

POWER_SWITCH = 1

WATER_PUMP_SWITCH = 32

libdeye.device_state module

Utilities for device state parsing

class libdeye.device_state.DeyeDeviceState(state: str | DeyeApiResponseFogPlatformDeviceProperties)

Bases: object

A class to store the device state.

copy() → DeyeDeviceState

Return an independent copy of this state.

to_command() → DeyeDeviceCommand

Convert to a command that can be used to let the device get into this state

class libdeye.device_state.DeyeDeviceStateFlag(value)

Bases: IntFlag

Bit flags used in the state string

ANION_SWITCH = 1

CHILD_LOCK_SWITCH = 1024

DEFROSTING_STATE = 8192

ELECTROMAGNETIC_STATE = 4

ENVIRONMENT_DEGREE = 16

FAN_RUNNING_STATE = 32768

OSCILLATING_SWITCH = 512

POWEROFF_SWITCH = 2048

POWERON_SWITCH = 4096

POWER_SWITCH = 256

PRESS_STATE = 8

WATER_PUMP_SWITCH = 2

WATER_TANK_FULL_STATE = 16384

libdeye.mqtt_client module

MQTT related stuffs

class libdeye.mqtt_client.BaseDeyeMqttClient(cloud_api: DeyeCloudApi, tls_context: SSLContext | None = None)

Bases: ABC

Base class for MQTT clients connected to Deye MQTT servers.

async connect() → None

Connect the MQTT client to the server.

disconnect() → None

Disconnect the MQTT client to the server.

abstractmethod async publish_command(product_id: str, device_id: str, command: DeyeDeviceCommand, properties: dict[str, int] | None = None) → None

Publish commands to a device

abstractmethod async query_device_state(product_id: str, device_id: str) → DeyeDeviceState

Query the latest device state.

abstractmethod subscribe_availability_change(product_id: str, device_id: str, callback: Callable[[bool], None]) → Callable[[], None]

Subscribe to availability changes of specified device.

abstractmethod subscribe_state_change(product_id: str, device_id: str, callback: Callable[[DeyeDeviceState], None]) → Callable[[], None]

Subscribe to state changes of specified device.

class libdeye.mqtt_client.DeyeClassicMqttClient(cloud_api: DeyeCloudApi, tls_context: SSLContext | None = None)

Bases: BaseDeyeMqttClient

MQTT client for the Classic platform.

async publish_command(product_id: str, device_id: str, command: DeyeDeviceCommand | bytes, properties: dict[str, int] | None = None) → None

Publish commands to a device

async query_device_state(product_id: str, device_id: str) → DeyeDeviceState

Query the latest device state.

subscribe_availability_change(product_id: str, device_id: str, callback: Callable[[bool], None]) → Callable[[], None]

Subscribe to availability changes of specified device.

subscribe_state_change(product_id: str, device_id: str, callback: Callable[[DeyeDeviceState], None]) → Callable[[], None]

Subscribe to state changes of specified device.

class libdeye.mqtt_client.DeyeFogMqttClient(cloud_api: DeyeCloudApi, tls_context: SSLContext | None = None)

Bases: BaseDeyeMqttClient

MQTT client for the Fog platform.

async publish_command(product_id: str, device_id: str, command: DeyeDeviceCommand, properties: dict[str, int] | None = None) → None

For Fog platform, commands are not published via MQTT. Instead, use the cloud API to send commands.

async query_device_state(product_id: str, device_id: str) → DeyeDeviceState

Query the latest device state.

subscribe_availability_change(product_id: str, device_id: str, callback: Callable[[bool], None]) → Callable[[], None]

Subscribe to availability changes of specified device.

subscribe_state_change(product_id: str, device_id: str, callback: Callable[[DeyeDeviceState], None]) → Callable[[], None]

Subscribe to state changes of specified device.

Module contents