Library Design & Modules¶
This page aims to provide some details on the design and internals of this library. You might be interested in this if you want to improve this library, or if you are just looking to access some information that is not currently exposed.
Initialization¶
Use discover()
to perform udp-based broadcast discovery on the network.
This will return you a list of device instances based on the discovery replies.
If the device’s host is already known, you can use to construct a device instance with
connect()
.
The connect()
also enables support for connecting to new
KASA SMART protocol and TAPO devices directly using the parameter DeviceConfig
.
Simply serialize the config
property via to_dict()
and then deserialize it later with from_dict()
and then pass it into connect()
.
Update Cycle¶
When update()
is called,
the library constructs a query to send to the device based on supported modules.
Internally, each module defines query()
to describe what they want query during the update.
The returned data is cached internally to avoid I/O on property accesses. All properties defined both in the device class and in the module classes follow this principle.
While the properties are designed to provide a nice API to use for common use cases,
you may sometimes want to access the raw, cached data as returned by the device.
This can be done using the internal_state
property.
Modules¶
The functionality provided by all SmartDevice
instances is (mostly) done inside separate modules.
While the individual device-type specific classes provide an easy access for the most import features,
you can also access individual modules through kasa.SmartDevice.modules
.
You can get the list of supported modules for a given device instance using supported_modules
.
Note
If you only need some module-specific information,
you can call the wanted method on the module to avoid using update()
.
Protocols and Transports¶
The library supports two different TP-Link protocols, IOT
and SMART
.
IOT
is the original Kasa protocol and SMART
is the newer protocol supported by TAPO devices and newer KASA devices.
The original protocol has a target
, command
, args
interface whereas the new protocol uses a different set of
commands and has a method
, parameters
interface.
Confusingly TP-Link originally called the Kasa line “Kasa Smart” and hence this library used “Smart” in a lot of the
module and class names but actually they were built to work with the IOT
protocol.
In 2021 TP-Link started updating the underlying communication transport used by Kasa devices to make them more secure.
It switched from a TCP connection with static XOR type of encryption to a transport called KLAP
which communicates
over http and uses handshakes to negotiate a dynamic encryption cipher.
This automatic update was put on hold and only seemed to affect UK HS100 models.
In 2023 TP-Link started updating the underlying communication transport used by Tapo devices to make them more secure.
It switched from AES encryption via public key exchange to use KLAP
encryption and negotiation due to concerns
around impersonation with AES.
The encryption cipher is the same as for Kasa KLAP but the handshake seeds are slightly different.
Also in 2023 TP-Link started releasing newer Kasa branded devices using the SMART
protocol.
This appears to be driven by hardware version rather than firmware.
In order to support these different configurations the library migrated from a single protocol class TPLinkSmartHomeProtocol
to support pluggable transports and protocols.
The classes providing this functionality are:
API documentation for modules¶
Module for individual feature modules.
- class kasa.modules.AmbientLight(device: SmartDevice, module: str)[source]
Implements ambient light controls for the motion sensor.
- call(method, params=None)
Call the given method with the given parameters.
- async current_brightness() int [source]
Return current brightness.
Return value units.
- property data
Return the module specific raw data from the last update.
- property enabled: bool
Return True if the module is enabled.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- property presets: dict
Return device-defined presets for brightness setting.
- query()[source]
Request configuration.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- async set_brightness_limit(value: int)[source]
Set the limit when the motion sensor is inactive.
See presets for preset values. Custom values are also likely allowed.
- async set_enabled(state: bool)[source]
Enable/disable LAS.
- class kasa.modules.Antitheft(device: SmartDevice, module: str)[source]
Implementation of the antitheft module.
This shares the functionality among other rule-based modules.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- async delete_all_rules()
Delete all rules.
- async delete_rule(rule: Rule)
Delete the given rule.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- query()
Prepare the query for rules.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property rules: List[Rule]
Return the list of rules for the service.
- async set_enabled(state: bool)
Enable or disable the service.
- class kasa.modules.Cloud(device: SmartDevice, module: str)[source]
Module implementing support for cloud services.
- call(method, params=None)
Call the given method with the given parameters.
- connect(username: str, password: str)[source]
Login to the cloud using given information.
- property data
Return the module specific raw data from the last update.
- disconnect()[source]
Disconnect from the cloud.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- get_available_firmwares()[source]
Return list of available firmwares.
- property info: CloudInfo
Return information about the cloud connectivity.
- property is_supported: bool
Return whether the module is supported by the device.
- query()[source]
Request cloud connectivity info.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- set_server(url: str)[source]
Set the update server URL.
- class kasa.modules.Countdown(device: SmartDevice, module: str)[source]
Implementation of countdown module.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- async delete_all_rules()
Delete all rules.
- async delete_rule(rule: Rule)
Delete the given rule.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- query()
Prepare the query for rules.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property rules: List[Rule]
Return the list of rules for the service.
- async set_enabled(state: bool)
Enable or disable the service.
- class kasa.modules.Emeter(device: SmartDevice, module: str)[source]
Emeter module.
- call(method, params=None)
Call the given method with the given parameters.
- property daily_data
Return statistics on daily basis.
- property data
Return the module specific raw data from the last update.
- property emeter_this_month: Optional[float]
Return this month’s energy consumption in kWh.
- property emeter_today: Optional[float]
Return today’s energy consumption in kWh.
- async erase_stats()[source]
Erase all stats.
Uses different query than usage meter.
- property estimated_query_response_size
Estimated maximum query response size.
- async get_daystat(*, year=None, month=None, kwh=True) Dict [source]
Return daily stats for the given year & month.
The return value is a dictionary of {day: energy, …}.
- async get_monthstat(*, year=None, kwh=True) Dict [source]
Return monthly stats for the given year.
The return value is a dictionary of {month: energy, …}.
- async get_raw_daystat(*, year=None, month=None) Dict
Return raw daily stats for the given year & month.
- async get_raw_monthstat(*, year=None) Dict
Return raw monthly stats for the given year.
- async get_realtime()[source]
Return real-time statistics.
- property is_supported: bool
Return whether the module is supported by the device.
- property monthly_data
Return statistics on monthly basis.
- query()
Return the base query.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property realtime: EmeterStatus
Return current energy readings.
- property usage_this_month
Return usage in this month in minutes.
- property usage_today
Return today’s usage in minutes.
- class kasa.modules.Module(device: SmartDevice, module: str)[source]
Base class implemention for all modules.
The base classes should implement query to return the query they want to be executed during the regular update cycle.
- call(method, params=None)[source]
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- abstract query()[source]
Query to execute during the update cycle.
The inheriting modules implement this to include their wanted queries to the query that gets executed when Device.update() gets called.
- query_for_command(query, params=None)[source]
Create a request object for the given parameters.
- class kasa.modules.Motion(device: SmartDevice, module: str)[source]
Implements the motion detection (PIR) module.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- property enabled: bool
Return True if module is enabled.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property inactivity_timeout: int
Return inactivity timeout in milliseconds.
- property is_supported: bool
Return whether the module is supported by the device.
- query()[source]
Request PIR configuration.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property range: Range
Return motion detection range.
- async set_enabled(state: bool)[source]
Enable/disable PIR.
- async set_inactivity_timeout(timeout: int)[source]
Set inactivity timeout in milliseconds.
Note, that you need to delete the default “Smart Control” rule in the app to avoid reverting this back to 60 seconds after a period of time.
- async set_range(*, range: Optional[Range] = None, custom_range: Optional[int] = None)[source]
Set the range for the sensor.
- Parameters:
range – for using standard ranges
custom_range – range in decimeters, overrides the range parameter
- class kasa.modules.Rule(*, id: str, name: str, enable: bool, wday: List[int], repeat: bool, sact: Optional[Action] = None, stime_opt: TimeOption, smin: int, eact: Optional[Action] = None, etime_opt: TimeOption, emin: int, s_light: Optional[Dict] = None)[source]
Representation of a rule.
- Config
alias of
BaseConfig
- classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) Model
Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values
- copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) Model
Duplicate a model, optionally choose which fields to include, exclude and change.
- Parameters:
include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model
- Returns:
new model instance
- dict(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) DictStrAny
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
- eact: Optional[Action]
- emin: int
- enable: bool
- etime_opt: TimeOption
- classmethod from_orm(obj: Any) Model
- id: str
- json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) str
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
- name: str
- classmethod parse_file(path: Union[str, Path], *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) Model
- classmethod parse_obj(obj: Any) Model
- classmethod parse_raw(b: Union[str, bytes], *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) Model
- repeat: bool
- s_light: Optional[Dict]
- sact: Optional[Action]
- classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') DictStrAny
- classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) str
- smin: int
- stime_opt: TimeOption
- classmethod update_forward_refs(**localns: Any) None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
- classmethod validate(value: Any) Model
- wday: List[int]
- class kasa.modules.RuleModule(device: SmartDevice, module: str)[source]
Base class for rule-based modules, such as countdown and antitheft.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- async delete_all_rules()[source]
Delete all rules.
- async delete_rule(rule: Rule)[source]
Delete the given rule.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- query()[source]
Prepare the query for rules.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property rules: List[Rule]
Return the list of rules for the service.
- async set_enabled(state: bool)[source]
Enable or disable the service.
- class kasa.modules.Schedule(device: SmartDevice, module: str)[source]
Implements the scheduling interface.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- async delete_all_rules()
Delete all rules.
- async delete_rule(rule: Rule)
Delete the given rule.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- property is_supported: bool
Return whether the module is supported by the device.
- query()
Prepare the query for rules.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property rules: List[Rule]
Return the list of rules for the service.
- async set_enabled(state: bool)
Enable or disable the service.
- class kasa.modules.Time(device: SmartDevice, module: str)[source]
Implements the timezone settings.
- call(method, params=None)
Call the given method with the given parameters.
- property data
Return the module specific raw data from the last update.
- property estimated_query_response_size
Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response will be so that queries can be split should an estimated response be too large
- async get_time()[source]
Return current device time.
- async get_timezone()[source]
Request timezone information from the device.
- property is_supported: bool
Return whether the module is supported by the device.
- query()[source]
Request time and timezone.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property time: datetime
Return current device time.
- property timezone
Return current timezone.
- class kasa.modules.Usage(device: SmartDevice, module: str)[source]
Baseclass for emeter/usage interfaces.
- call(method, params=None)
Call the given method with the given parameters.
- property daily_data
Return statistics on daily basis.
- property data
Return the module specific raw data from the last update.
- async erase_stats()[source]
Erase all stats.
- property estimated_query_response_size
Estimated maximum query response size.
- async get_daystat(*, year=None, month=None) Dict [source]
Return daily stats for the given year & month.
The return value is a dictionary of {day: time, …}.
- async get_monthstat(*, year=None) Dict [source]
Return monthly stats for the given year.
The return value is a dictionary of {month: time, …}.
- async get_raw_daystat(*, year=None, month=None) Dict [source]
Return raw daily stats for the given year & month.
- async get_raw_monthstat(*, year=None) Dict [source]
Return raw monthly stats for the given year.
- property is_supported: bool
Return whether the module is supported by the device.
- property monthly_data
Return statistics on monthly basis.
- query()[source]
Return the base query.
- query_for_command(query, params=None)
Create a request object for the given parameters.
- property usage_this_month
Return usage in this month in minutes.
- property usage_today
Return today’s usage in minutes.
API documentation for protocols and transports¶
- class kasa.protocol.BaseProtocol(*, transport: BaseTransport)[source]¶
Base class for all TP-Link Smart Home communication.
- property config: DeviceConfig¶
Return the connection parameters the device is using.
- class kasa.iotprotocol.IotProtocol(*, transport: BaseTransport)[source]¶
Class for the legacy TPLink IOT KASA Protocol.
- BACKOFF_SECONDS_AFTER_TIMEOUT = 1¶
- property config: DeviceConfig¶
Return the connection parameters the device is using.
- class kasa.smartprotocol.SmartProtocol(*, transport: BaseTransport)[source]¶
Class for the new TPLink SMART protocol.
- BACKOFF_SECONDS_AFTER_TIMEOUT = 1¶
- DEFAULT_MULTI_REQUEST_BATCH_SIZE = 5¶
- property config: DeviceConfig¶
Return the connection parameters the device is using.
- class kasa.protocol.BaseTransport(*, config: DeviceConfig)[source]¶
Base class for all TP-Link protocol transports.
- DEFAULT_TIMEOUT = 5¶
- abstract property credentials_hash: str¶
The hashed credentials used by the transport.
- abstract property default_port: int¶
The default port for the transport.
- class kasa.xortransport.XorTransport(*, config: DeviceConfig)[source]¶
XorTransport class.
- BLOCK_SIZE = 4¶
- DEFAULT_PORT: int = 9999¶
- DEFAULT_TIMEOUT = 5¶
- close_without_wait() None [source]¶
Close the connection without waiting for the connection to close.
- property credentials_hash: str¶
The hashed credentials used by the transport.
- property default_port¶
Default port for the transport.
- class kasa.klaptransport.KlapTransport(*, config: DeviceConfig)[source]¶
Implementation of the KLAP encryption protocol.
KLAP is the name used in device discovery for TP-Link’s new encryption protocol, used by newer firmware versions.
- DEFAULT_PORT: int = 80¶
- DEFAULT_TIMEOUT = 5¶
- DISCOVERY_QUERY = {'system': {'get_sysinfo': None}}¶
- SESSION_COOKIE_NAME = 'TP_SESSIONID'¶
- TIMEOUT_COOKIE_NAME = 'TIMEOUT'¶
- property credentials_hash: str¶
The hashed credentials used by the transport.
- property default_port¶
Default port for the transport.
- static generate_auth_hash(creds: Credentials)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- static generate_owner_hash(creds: Credentials)[source]¶
Return the MD5 hash of the username in this object.
- static handshake1_seed_auth_hash(local_seed: bytes, remote_seed: bytes, auth_hash: bytes)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- static handshake2_seed_auth_hash(local_seed: bytes, remote_seed: bytes, auth_hash: bytes)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- async perform_handshake() Any [source]¶
Perform handshake1 and handshake2.
Sets the encryption_session if successful.
- class kasa.klaptransport.KlapTransportV2(*, config: DeviceConfig)[source]¶
Implementation of the KLAP encryption protocol with v2 hanshake hashes.
- DEFAULT_PORT: int = 80¶
- DEFAULT_TIMEOUT = 5¶
- DISCOVERY_QUERY = {'system': {'get_sysinfo': None}}¶
- SESSION_COOKIE_NAME = 'TP_SESSIONID'¶
- TIMEOUT_COOKIE_NAME = 'TIMEOUT'¶
- async close() None ¶
Close the http client and reset internal state.
- property credentials_hash: str¶
The hashed credentials used by the transport.
- property default_port¶
Default port for the transport.
- static generate_auth_hash(creds: Credentials)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- static generate_owner_hash(creds: Credentials)¶
Return the MD5 hash of the username in this object.
- static handshake1_seed_auth_hash(local_seed: bytes, remote_seed: bytes, auth_hash: bytes)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- static handshake2_seed_auth_hash(local_seed: bytes, remote_seed: bytes, auth_hash: bytes)[source]¶
Generate an md5 auth hash for the protocol on the supplied credentials.
- async perform_handshake() Any ¶
Perform handshake1 and handshake2.
Sets the encryption_session if successful.
- async perform_handshake1() Tuple[bytes, bytes, bytes] ¶
Perform handshake1.
- async perform_handshake2(local_seed, remote_seed, auth_hash) KlapEncryptionSession ¶
Perform handshake2.
- async reset() None ¶
Reset internal handshake state.
- async send(request: str)¶
Send the request.
- class kasa.aestransport.AesTransport(*, config: DeviceConfig)[source]¶
Implementation of the AES encryption protocol.
AES is the name used in device discovery for TP-Link’s TAPO encryption protocol, sometimes used by newer firmware versions on kasa devices.
- COMMON_HEADERS = {'Accept': 'application/json', 'Content-Type': 'application/json', 'requestByApp': 'true'}¶
- CONTENT_LENGTH = 'Content-Length'¶
- DEFAULT_PORT: int = 80¶
- DEFAULT_TIMEOUT = 5¶
- KEY_PAIR_CONTENT_LENGTH = 314¶
- SESSION_COOKIE_NAME = 'TP_SESSIONID'¶
- TIMEOUT_COOKIE_NAME = 'TIMEOUT'¶
- property credentials_hash: str¶
The hashed credentials used by the transport.
- property default_port: int¶
Default port for the transport.
- static hash_credentials(login_v2: bool, credentials: Credentials) Tuple[str, str] [source]¶
Hash the credentials.