"""Configuration for connecting directly to a device without discovery. If you are connecting to a newer KASA or TAPO device you can get the device via discovery or connect directly with :class:`DeviceConfig`. Discovery returns a list of discovered devices: >>> from kasa import Discover, Device >>> device = await Discover.discover_single( >>> "127.0.0.3", >>> username="user@example.com", >>> password="great_password", >>> ) >>> print(device.alias) # Alias is None because update() has not been called None >>> config_dict = device.config.to_dict() >>> # DeviceConfig.to_dict() can be used to store for later >>> print(config_dict) {'host': '127.0.0.3', 'timeout': 5, 'credentials': {'username': 'user@example.com', \ 'password': 'great_password'}, 'connection_type'\ : {'device_family': 'SMART.TAPOBULB', 'encryption_type': 'KLAP', 'login_version': 2, \ 'https': False}, 'uses_http': True} >>> later_device = await Device.connect(config=Device.Config.from_dict(config_dict)) >>> print(later_device.alias) # Alias is available as connect() calls update() Living Room Bulb """ from __future__ import annotations import logging from dataclasses import dataclass, field, replace from enum import Enum from typing import TYPE_CHECKING, TypedDict from aiohttp import ClientSession from mashumaro import field_options, pass_through from mashumaro.config import BaseConfig from .credentials import Credentials from .exceptions import KasaException from .json import DataClassJSONMixin if TYPE_CHECKING: from aiohttp import ClientSession _LOGGER = logging.getLogger(__name__) class KeyPairDict(TypedDict): """Class to represent a public/private key pair.""" private: str public: str class DeviceEncryptionType(Enum): """Encrypt type enum.""" Klap = "KLAP" Aes = "AES" Xor = "XOR" class DeviceFamily(Enum): """Encrypt type enum.""" IotSmartPlugSwitch = "IOT.SMARTPLUGSWITCH" IotSmartBulb = "IOT.SMARTBULB" IotIpCamera = "IOT.IPCAMERA" SmartKasaPlug = "SMART.KASAPLUG" SmartKasaSwitch = "SMART.KASASWITCH" SmartTapoPlug = "SMART.TAPOPLUG" SmartTapoBulb = "SMART.TAPOBULB" SmartTapoSwitch = "SMART.TAPOSWITCH" SmartTapoHub = "SMART.TAPOHUB" SmartKasaHub = "SMART.KASAHUB" SmartIpCamera = "SMART.IPCAMERA" SmartTapoRobovac = "SMART.TAPOROBOVAC" class _DeviceConfigBaseMixin(DataClassJSONMixin): """Base class for serialization mixin.""" class Config(BaseConfig): """Serialization config.""" omit_none = True @dataclass class DeviceConnectionParameters(_DeviceConfigBaseMixin): """Class to hold the the parameters determining connection type.""" device_family: DeviceFamily encryption_type: DeviceEncryptionType login_version: int | None = None https: bool = False @staticmethod def from_values( device_family: str, encryption_type: str, login_version: int | None = None, https: bool | None = None, ) -> DeviceConnectionParameters: """Return connection parameters from string values.""" try: if https is None: https = False return DeviceConnectionParameters( DeviceFamily(device_family), DeviceEncryptionType(encryption_type), login_version, https, ) except (ValueError, TypeError) as ex: raise KasaException( f"Invalid connection parameters for {device_family}." + f"{encryption_type}.{login_version}" ) from ex @dataclass class DeviceConfig(_DeviceConfigBaseMixin): """Class to represent paramaters that determine how to connect to devices.""" DEFAULT_TIMEOUT = 5 #: IP address or hostname host: str #: Timeout for querying the device timeout: int | None = DEFAULT_TIMEOUT #: Override the default 9999 port to support port forwarding port_override: int | None = None #: Credentials for devices requiring authentication credentials: Credentials | None = None #: Credentials hash for devices requiring authentication. #: If credentials are also supplied they take precendence over credentials_hash. #: Credentials hash can be retrieved from :attr:`Device.credentials_hash` credentials_hash: str | None = None #: The protocol specific type of connection. Defaults to the legacy type. batch_size: int | None = None #: The batch size for protoools supporting multiple request batches. connection_type: DeviceConnectionParameters = field( default_factory=lambda: DeviceConnectionParameters( DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Xor ) ) #: True if the device uses http. Consumers should retrieve rather than set this #: in order to determine whether they should pass a custom http client if desired. uses_http: bool = False #: Set a custom http_client for the device to use. http_client: ClientSession | None = field( default=None, compare=False, metadata=field_options(serialize="omit", deserialize=pass_through), ) aes_keys: KeyPairDict | None = None def __post_init__(self) -> None: if self.connection_type is None: self.connection_type = DeviceConnectionParameters( DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Xor ) def to_dict_control_credentials( self, *, credentials_hash: str | None = None, exclude_credentials: bool = False, ) -> dict[str, dict[str, str]]: """Convert deviceconfig to dict controlling how to serialize credentials. If credentials_hash is provided credentials will be None. If credentials_hash is '' credentials_hash and credentials will be None. exclude credentials controls whether to include credentials. The defaults are the same as calling to_dict(). """ if credentials_hash is None: if not exclude_credentials: return self.to_dict() else: return replace(self, credentials=None).to_dict() return replace( self, credentials_hash=credentials_hash if credentials_hash else None, credentials=None, ).to_dict()