From c66d8e74ca1cf522fb71bbce56e1874c475d54b8 Mon Sep 17 00:00:00 2001 From: ZeliardM Date: Fri, 27 Feb 2026 23:47:33 -0500 Subject: [PATCH 1/4] Add support for V2 Energy Monitoring for S515D --- README.md | 2 +- SUPPORTED.md | 2 + kasa/smart/modules/energy.py | 2 +- tests/device_fixtures.py | 5 +- tests/fixtures/smart/S515D(US)_1.6_1.0.4.json | 688 ++++++++++++++++++ tests/smart/modules/test_energy.py | 40 +- 6 files changed, 733 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/smart/S515D(US)_1.6_1.0.4.json diff --git a/README.md b/README.md index 6b3117295..8d1796e71 100644 --- a/README.md +++ b/README.md @@ -198,7 +198,7 @@ The following devices have been tested and confirmed as working. If your device - **Plugs**: P100, P105, P110, P110M, P115, P125M, P135, TP10, TP15 - **Power Strips**: P210M, P300, P304M, P306, P316M, TP25 -- **Wall Switches**: S210, S220, S500, S500D, S505, S505D, TS15 +- **Wall Switches**: S210, S220, S500, S500D, S505, S505D, S515D, TS15 - **Bulbs**: L430C, L430P, L510B, L510E, L530B, L530E, L535E, L630 - **Light Strips**: L900-10, L900-5, L920-5, L930-5 - **Cameras**: C100, C110, C210, C220, C225, C325WB, C460, C520WS, C720, TC40, TC65, TC70 diff --git a/SUPPORTED.md b/SUPPORTED.md index 79260b174..e05e7b78f 100644 --- a/SUPPORTED.md +++ b/SUPPORTED.md @@ -252,6 +252,8 @@ All Tapo devices require authentication.
Hub-Connected Devices may work acros - Hardware: 1.0 (US) / Firmware: 1.0.2 - **S505D** - Hardware: 1.0 (US) / Firmware: 1.1.0 +- **S515D** + - Hardware: 1.6 (US) / Firmware: 1.0.4 - **TS15** - Hardware: 1.0 (US) / Firmware: 1.2.2 diff --git a/kasa/smart/modules/energy.py b/kasa/smart/modules/energy.py index 03df6d11c..83a237cae 100644 --- a/kasa/smart/modules/energy.py +++ b/kasa/smart/modules/energy.py @@ -63,7 +63,7 @@ def query(self) -> dict: def optional_response_keys(self) -> list[str]: """Return optional response keys for the module.""" if self.supported_version > 1: - return ["get_energy_usage"] + return ["get_energy_usage", "get_current_power"] return [] @property diff --git a/tests/device_fixtures.py b/tests/device_fixtures.py index 992cbfd97..d3ea3b0a8 100644 --- a/tests/device_fixtures.py +++ b/tests/device_fixtures.py @@ -111,6 +111,7 @@ "S500D", "S505", "S505D", + "S515D", "TS15", } SWITCHES = {*SWITCHES_IOT, *SWITCHES_SMART} @@ -119,7 +120,7 @@ STRIPS = {*STRIPS_IOT, *STRIPS_SMART} DIMMERS_IOT = {"ES20M", "HS220", "KS220", "KS220M", "KS230", "KP405"} -DIMMERS_SMART = {"HS220", "KS225", "S500D", "P135"} +DIMMERS_SMART = {"HS220", "KS225", "S500D", "S505D", "S515D", "P135"} DIMMERS = { *DIMMERS_IOT, *DIMMERS_SMART, @@ -143,7 +144,7 @@ VACUUMS_SMART = {"RV20"} WITH_EMETER_IOT = {"EP25", "HS110", "HS300", "KP115", "KP125", *BULBS_IOT} -WITH_EMETER_SMART = {"P110", "P110M", "P115", "KP125M", "EP25", "P304M"} +WITH_EMETER_SMART = {"P110", "P110M", "P115", "KP125M", "EP25", "P304M", "S515D"} WITH_EMETER = {*WITH_EMETER_IOT, *WITH_EMETER_SMART} DIMMABLE = {*BULBS, *DIMMERS} diff --git a/tests/fixtures/smart/S515D(US)_1.6_1.0.4.json b/tests/fixtures/smart/S515D(US)_1.6_1.0.4.json new file mode 100644 index 000000000..c36f0c25f --- /dev/null +++ b/tests/fixtures/smart/S515D(US)_1.6_1.0.4.json @@ -0,0 +1,688 @@ +{ + "component_nego": { + "component_list": [ + { + "id": "device", + "ver_code": 2 + }, + { + "id": "firmware", + "ver_code": 2 + }, + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "time", + "ver_code": 1 + }, + { + "id": "wireless", + "ver_code": 1 + }, + { + "id": "schedule", + "ver_code": 2 + }, + { + "id": "countdown", + "ver_code": 2 + }, + { + "id": "antitheft", + "ver_code": 1 + }, + { + "id": "account", + "ver_code": 1 + }, + { + "id": "synchronize", + "ver_code": 1 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "led", + "ver_code": 1 + }, + { + "id": "cloud_connect", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "device_local_time", + "ver_code": 1 + }, + { + "id": "default_states", + "ver_code": 1 + }, + { + "id": "brightness", + "ver_code": 1 + }, + { + "id": "preset", + "ver_code": 1 + }, + { + "id": "on_off_gradually", + "ver_code": 2 + }, + { + "id": "dimmer_calibration", + "ver_code": 1 + }, + { + "id": "localSmart", + "ver_code": 1 + }, + { + "id": "matter", + "ver_code": 2 + }, + { + "id": "dimmer_custom_action", + "ver_code": 1 + }, + { + "id": "energy_monitoring", + "ver_code": 2 + }, + { + "id": "current_protection", + "ver_code": 1 + }, + { + "id": "satellite_check", + "ver_code": 1 + } + ] + }, + "discovery_result": { + "error_code": 0, + "result": { + "device_id": "00000000000000000000000000000000", + "device_model": "S515D(US)", + "device_type": "SMART.TAPOSWITCH", + "factory_default": false, + "ip": "127.0.0.123", + "is_support_iot_cloud": true, + "mac": "CC-BA-BD-00-00-00", + "mgt_encrypt_schm": { + "encrypt_type": "KLAP", + "http_port": 80, + "is_support_https": false, + "lv": 2 + }, + "obd_src": "tplink", + "owner": "00000000000000000000000000000000", + "protocol_version": 1 + } + }, + "get_antitheft_rules": { + "antitheft_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_auto_update_info": { + "enable": true, + "random_range": 120, + "time": 180 + }, + "get_connect_cloud_state": { + "status": 0 + }, + "get_countdown_rules": { + "countdown_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_device_info": { + "avatar": "switch", + "brightness": 34, + "dc_state": 0, + "default_states": { + "re_power_type": "always_off", + "re_power_type_capability": [ + "last_states", + "always_on", + "always_off" + ], + "type": "last_states" + }, + "device_id": "0000000000000000000000000000000000000000", + "device_on": false, + "fw_id": "00000000000000000000000000000000", + "fw_ver": "1.0.4 Build 240805 Rel.204647", + "has_set_location_info": false, + "hw_id": "00000000000000000000000000000000", + "hw_ver": "1.6", + "ip": "127.0.0.123", + "lang": "en_US", + "latitude": 0, + "longitude": 0, + "mac": "CC-BA-BD-00-00-00", + "model": "S515D", + "nickname": "I01BU0tFRF9OQU1FIw==", + "oem_id": "00000000000000000000000000000000", + "on_time": 0, + "overcurrent_status": "normal", + "region": "America/Los_Angeles", + "rssi": -49, + "signal_level": 3, + "specs": "", + "ssid": "I01BU0tFRF9TU0lEIw==", + "time_diff": -480, + "type": "SMART.TAPOSWITCH" + }, + "get_device_time": { + "region": "America/Los_Angeles", + "time_diff": -480, + "timestamp": 1771710913 + }, + "get_device_usage": { + "power_usage": { + "past30": 44, + "past7": 31, + "today": 1 + }, + "saved_power": { + "past30": 266, + "past7": 179, + "today": 0 + }, + "time_usage": { + "past30": 310, + "past7": 210, + "today": 1 + } + }, + "get_electricity_price_config": { + "constant_price": 0, + "time_of_use_config": { + "summer": { + "midpeak": 0, + "offpeak": 0, + "onpeak": 0, + "period": [ + 0, + 0, + 0, + 0 + ], + "weekday_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ], + "weekend_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ] + }, + "winter": { + "midpeak": 0, + "offpeak": 0, + "onpeak": 0, + "period": [ + 0, + 0, + 0, + 0 + ], + "weekday_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ], + "weekend_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ] + } + }, + "type": "constant" + }, + "get_emeter_data": { + "current_ma": 0, + "energy_wh": 44, + "power_mw": 0, + "voltage_mv": 117232 + }, + "get_emeter_vgain_igain": { + "igain": 3698, + "vgain": 122959 + }, + "get_energy_usage": { + "current_power": 0, + "electricity_charge": [ + 0, + 0, + 0 + ], + "local_time": "2026-02-21 13:55:15", + "month_energy": 44, + "month_runtime": 310, + "today_energy": 1, + "today_runtime": 1 + }, + "get_fw_download_state": { + "auto_upgrade": false, + "download_progress": 0, + "reboot_time": 5, + "status": 0, + "upgrade_time": 5 + }, + "get_inherit_info": null, + "get_latest_fw": { + "fw_size": 0, + "fw_ver": "1.0.4 Build 240805 Rel.204647", + "hw_id": "", + "need_to_upgrade": false, + "oem_id": "", + "release_date": "", + "release_note": "", + "type": 0 + }, + "get_led_info": { + "bri_config": { + "bri_type": "overall", + "overall_bri": 50 + }, + "led_rule": "always", + "led_status": true, + "night_mode": { + "end_time": 420, + "night_mode_type": "custom", + "start_time": 1320 + } + }, + "get_matter_setup_info": { + "setup_code": "00000000000", + "setup_payload": "00:0000000000000000000" + }, + "get_next_event": {}, + "get_on_off_gradually_info": { + "off_state": { + "duration": 1, + "enable": true, + "max_duration": 60 + }, + "on_state": { + "duration": 1, + "enable": true, + "max_duration": 60 + } + }, + "get_preset_rules": { + "brightness": [ + 100, + 75, + 50, + 25, + 1 + ] + }, + "get_schedule_rules": { + "enable": false, + "rule_list": [], + "schedule_rule_max_count": 32, + "start_index": 0, + "sum": 0 + }, + "get_wireless_scan_info": { + "ap_list": [ + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 3, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 2, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 2, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 2, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 2, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 0, + "key_type": "none", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + }, + { + "bssid": "000000000000", + "channel": 0, + "cipher_type": 2, + "key_type": "wpa2_psk", + "signal_level": 1, + "ssid": "I01BU0tFRF9TU0lEIw==" + } + ], + "start_index": 0, + "sum": 26, + "wep_supported": false + }, + "qs_component_nego": { + "component_list": [ + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "ble_whole_setup", + "ver_code": 1 + }, + { + "id": "matter", + "ver_code": 2 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "firmware", + "ver_code": 2 + }, + { + "id": "satellite_check", + "ver_code": 1 + } + ], + "extra_info": { + "device_model": "S515D", + "device_type": "SMART.TAPOSWITCH", + "is_klap": true + } + } +} diff --git a/tests/smart/modules/test_energy.py b/tests/smart/modules/test_energy.py index 7b31d74bf..cd700f232 100644 --- a/tests/smart/modules/test_energy.py +++ b/tests/smart/modules/test_energy.py @@ -59,7 +59,7 @@ async def test_get_energy_usage_error( if ed := resp.get("get_emeter_data"): ed["power_mw"] = 2002 - if cp := resp.get("get_current_power"): + if (cp := resp.get("get_current_power")) and isinstance(cp, dict): cp["current_power"] = 2.002 resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR @@ -88,8 +88,11 @@ async def test_get_energy_usage_error( caplog.clear() resp = copy.deepcopy(last_update) - if cp := resp.get("get_current_power"): + if (cp := resp.get("get_current_power")) and isinstance(cp, dict): cp["current_power"] = 2.002 + expected_current_consumption = 2.002 + else: + expected_current_consumption = None resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR # Remove get_emeter_data from the response and from the device which will @@ -107,3 +110,36 @@ async def test_get_energy_usage_error( # message should only be logged once assert msg not in caplog.text + + +@pytest.mark.xdist_group(name="caplog") +@has_emeter_smart +async def test_missing_get_current_power_for_v2_fixture( + dev: SmartDevice, + caplog: pytest.LogCaptureFixture, +): + """Energy v2 devices should tolerate get_current_power query errors.""" + caplog.set_level(logging.DEBUG) + + energy_module = dev.modules.get(Module.Energy) + if not energy_module: + pytest.skip(f"Energy module not supported for {dev}.") + + assert isinstance(energy_module, SmartEnergyModule) + if energy_module.supported_version <= 1: + pytest.skip("Only applicable for energy v2+ fixtures.") + + resp = copy.deepcopy(dev._last_update) + resp["get_current_power"] = SmartErrorCode.PARAMS_ERROR + + with patch.object(dev.protocol, "query", return_value=resp): + await dev.update() + + assert energy_module.disabled is False + assert energy_module._last_update_error is None + assert "get_current_power" not in energy_module.data + + # Some fixtures may have already logged this key removal during initial update, + # so duplicate log messages can be suppressed. + if "Removed key get_current_power" in caplog.text: + assert "PARAMS_ERROR" in caplog.text From 443baf33fe8f14e2b17524a271d5af56d53a2ccc Mon Sep 17 00:00:00 2001 From: ZeliardM Date: Sat, 28 Feb 2026 12:47:34 -0500 Subject: [PATCH 2/4] Update energy handling for new and old devices --- kasa/smart/modules/energy.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/kasa/smart/modules/energy.py b/kasa/smart/modules/energy.py index 83a237cae..93ccbbd89 100644 --- a/kasa/smart/modules/energy.py +++ b/kasa/smart/modules/energy.py @@ -34,17 +34,14 @@ async def _post_update_hook(self) -> None: self._supported | EnergyInterface.ModuleFeature.VOLTAGE_CURRENT ) - if (power := self._energy.get("current_power")) is not None or ( - power := data.get("get_emeter_data", {}).get("power_mw") - ) is not None: + if (power := data.get("get_emeter_data", {}).get("power_mw")) is not None: self._current_consumption = power / 1_000 - # Fallback if get_energy_usage does not provide current_power, - # which can happen on some newer devices (e.g. P304M). - # This may not be valid scenario as it pre-dates trying get_emeter_data elif ( - power := self.data.get("get_current_power", {}).get("current_power") + power := data.get("get_current_power", {}).get("current_power") ) is not None: self._current_consumption = power + elif (power := self._energy.get("current_power")) is not None: + self._current_consumption = power / 1_000 else: self._current_consumption = None From 80fb27a9fbdae943aef3f3fed5474c1bf44958f9 Mon Sep 17 00:00:00 2001 From: ZeliardM Date: Wed, 8 Apr 2026 16:57:30 -0400 Subject: [PATCH 3/4] Address review: clean up energy module and rewrite tests - Extract _get_current_power_mw() helper with documented precedence chain - Remove all isinstance checks from tests per review feedback - Rewrite tests into focused, independent functions with controlled responses - Add -> None return annotations to all test functions --- kasa/smart/modules/energy.py | 85 ++++++++++++---- tests/smart/modules/test_energy.py | 150 +++++++++++++++++++---------- 2 files changed, 166 insertions(+), 69 deletions(-) diff --git a/kasa/smart/modules/energy.py b/kasa/smart/modules/energy.py index 93ccbbd89..2a80ad963 100644 --- a/kasa/smart/modules/energy.py +++ b/kasa/smart/modules/energy.py @@ -18,6 +18,28 @@ class Energy(SmartModule, EnergyInterface): _energy: dict[str, Any] _current_consumption: float | None + def _get_current_power_mw( + self, data: dict[str, Any], energy: dict[str, Any] | None = None + ) -> float | None: + """Return the best available current power reading in milliwatts.""" + energy = self._energy if energy is None else energy + + if (power := data.get("get_emeter_data", {}).get("power_mw")) is not None: + return power + + # Prefer the higher precision milliwatt readings from the energy usage + # payload. get_current_power is only a lower precision fallback used by + # devices such as P304M whose get_energy_usage omits current_power. + if (power := energy.get("current_power")) is not None: + return power + + if ( + power := data.get("get_current_power", {}).get("current_power") + ) is not None: + return power * 1_000 + + return None + async def _post_update_hook(self) -> None: try: data = self.data @@ -34,13 +56,7 @@ async def _post_update_hook(self) -> None: self._supported | EnergyInterface.ModuleFeature.VOLTAGE_CURRENT ) - if (power := data.get("get_emeter_data", {}).get("power_mw")) is not None: - self._current_consumption = power / 1_000 - elif ( - power := data.get("get_current_power", {}).get("current_power") - ) is not None: - self._current_consumption = power - elif (power := self._energy.get("current_power")) is not None: + if (power := self._get_current_power_mw(data)) is not None: self._current_consumption = power / 1_000 else: self._current_consumption = None @@ -73,10 +89,14 @@ def energy(self) -> dict: """Return get_energy_usage results.""" return self._energy - def _get_status_from_energy(self, energy: dict) -> EmeterStatus: + def _get_status_from_energy( + self, energy: dict[str, Any], power_mw: float | None = None + ) -> EmeterStatus: return EmeterStatus( { - "power_mw": energy.get("current_power", 0), + "power_mw": ( + power_mw if power_mw is not None else energy.get("current_power") + ), "total": energy.get("today_energy", 0) / 1_000, } ) @@ -85,19 +105,48 @@ def _get_status_from_energy(self, energy: dict) -> EmeterStatus: @raise_if_update_error def status(self) -> EmeterStatus: """Get the emeter status.""" - if "get_emeter_data" in self.data: - return EmeterStatus(self.data["get_emeter_data"]) - else: - return self._get_status_from_energy(self.energy) + data = self.data + if "get_emeter_data" in data: + return EmeterStatus(data["get_emeter_data"]) + + return self._get_status_from_energy( + self.energy, self._get_current_power_mw(data) + ) async def get_status(self) -> EmeterStatus: """Return real-time statistics.""" - if "get_emeter_data" in self.data: - res = await self.call("get_emeter_data") - return EmeterStatus(res["get_emeter_data"]) - else: + if self.supported_version > 1: + try: + res = await self.call("get_emeter_data") + except DeviceError: + pass + else: + return EmeterStatus(res["get_emeter_data"]) + + energy: dict[str, Any] = {} + try: res = await self.call("get_energy_usage") - return self._get_status_from_energy(res["get_energy_usage"]) + except DeviceError: + if self.supported_version <= 1: + raise + else: + energy = res["get_energy_usage"] + if energy.get("current_power") is not None: + return self._get_status_from_energy(energy) + + current_power: dict[str, Any] = {} + if self.supported_version > 1: + try: + res = await self.call("get_current_power") + except DeviceError: + pass + else: + current_power = res["get_current_power"] + + return self._get_status_from_energy( + energy, + self._get_current_power_mw({"get_current_power": current_power}, energy), + ) @property def consumption_this_month(self) -> float | None: diff --git a/tests/smart/modules/test_energy.py b/tests/smart/modules/test_energy.py index cd700f232..41e23a5d4 100644 --- a/tests/smart/modules/test_energy.py +++ b/tests/smart/modules/test_energy.py @@ -10,16 +10,47 @@ from kasa.interfaces.energy import Energy from kasa.smart import SmartDevice from kasa.smart.modules import Energy as SmartEnergyModule -from tests.conftest import has_emeter_smart +from ...device_fixtures import has_emeter_smart, parametrize -@has_emeter_smart -async def test_supported(dev: SmartDevice): +s515d_smart = parametrize( + "s515d smart", + model_filter={"S515D(US)_1.6_1.0.4"}, + protocol_filter={"SMART"}, +) + + +def _get_energy_module(dev: SmartDevice) -> SmartEnergyModule: energy_module = dev.modules.get(Module.Energy) if not energy_module: pytest.skip(f"Energy module not supported for {dev}.") assert isinstance(energy_module, SmartEnergyModule) + return energy_module + + +def _get_v2_energy_module(dev: SmartDevice) -> SmartEnergyModule: + energy_module = _get_energy_module(dev) + if energy_module.supported_version <= 1: + pytest.skip("Only applicable for energy v2+ fixtures.") + + return energy_module + + +def _energy_usage(current_power: int | None = None) -> dict[str, int]: + energy_usage = { + "month_energy": 0, + "today_energy": 0, + } + if current_power is not None: + energy_usage["current_power"] = current_power + + return energy_usage + + +@has_emeter_smart +async def test_supported(dev: SmartDevice) -> None: + energy_module = _get_energy_module(dev) assert energy_module.supports(Energy.ModuleFeature.CONSUMPTION_TOTAL) is False assert energy_module.supports(Energy.ModuleFeature.PERIODIC_STATS) is False if energy_module.supported_version < 2: @@ -31,14 +62,11 @@ async def test_supported(dev: SmartDevice): @has_emeter_smart async def test_get_energy_usage_error( dev: SmartDevice, caplog: pytest.LogCaptureFixture -): +) -> None: """Test errors on get_energy_usage.""" caplog.set_level(logging.DEBUG) - energy_module = dev.modules.get(Module.Energy) - if not energy_module: - pytest.skip(f"Energy module not supported for {dev}.") - + energy_module = _get_energy_module(dev) version = dev._components["energy_monitoring"] expected_raise = does_not_raise() if version > 1 else pytest.raises(DeviceError) @@ -57,10 +85,8 @@ async def test_get_energy_usage_error( last_update = copy.deepcopy(dev._last_update) resp = copy.deepcopy(last_update) - if ed := resp.get("get_emeter_data"): - ed["power_mw"] = 2002 - if (cp := resp.get("get_current_power")) and isinstance(cp, dict): - cp["current_power"] = 2.002 + if version > 1: + resp["get_emeter_data"] = {"power_mw": 2002} resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR # version 1 only has get_energy_usage so module should raise an error if @@ -82,64 +108,86 @@ async def test_get_energy_usage_error( if version > 1: assert msg in caplog.text - # Now test with no get_emeter_data - # This may not be valid scenario but we have a fallback to get_current_power - # just in case that should be tested. - caplog.clear() + +@has_emeter_smart +async def test_v2_current_power_source_precedence(dev: SmartDevice) -> None: + """Prefer higher precision current power sources for v2 devices.""" + energy_module = _get_v2_energy_module(dev) + last_update = copy.deepcopy(dev._last_update) + resp = copy.deepcopy(last_update) + resp["get_emeter_data"] = {"power_mw": 3003} + resp["get_energy_usage"] = _energy_usage(current_power=2002) + resp["get_current_power"] = {"current_power": 1.001} - if (cp := resp.get("get_current_power")) and isinstance(cp, dict): - cp["current_power"] = 2.002 - expected_current_consumption = 2.002 - else: - expected_current_consumption = None - resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR + with patch.object(dev.protocol, "query", return_value=resp): + await dev.update() - # Remove get_emeter_data from the response and from the device which will - # remember it otherwise. + assert energy_module.current_consumption == 3.003 + assert energy_module.status.power == 3.003 + + resp = copy.deepcopy(last_update) resp.pop("get_emeter_data", None) dev._last_update.pop("get_emeter_data", None) + resp["get_energy_usage"] = _energy_usage(current_power=2002) + resp["get_current_power"] = {"current_power": 1.001} with patch.object(dev.protocol, "query", return_value=resp): await dev.update() - with expected_raise: - assert "get_energy_usage" not in energy_module.data - - assert energy_module.current_consumption == expected_current_consumption - - # message should only be logged once - assert msg not in caplog.text + assert energy_module.current_consumption == 2.002 + assert energy_module.status.power == 2.002 -@pytest.mark.xdist_group(name="caplog") @has_emeter_smart -async def test_missing_get_current_power_for_v2_fixture( +async def test_get_energy_usage_error_falls_back_to_get_current_power( dev: SmartDevice, - caplog: pytest.LogCaptureFixture, -): - """Energy v2 devices should tolerate get_current_power query errors.""" - caplog.set_level(logging.DEBUG) - - energy_module = dev.modules.get(Module.Energy) - if not energy_module: - pytest.skip(f"Energy module not supported for {dev}.") - - assert isinstance(energy_module, SmartEnergyModule) - if energy_module.supported_version <= 1: - pytest.skip("Only applicable for energy v2+ fixtures.") +) -> None: + """Use get_current_power when it is the only remaining power source.""" + energy_module = _get_v2_energy_module(dev) resp = copy.deepcopy(dev._last_update) - resp["get_current_power"] = SmartErrorCode.PARAMS_ERROR + resp.pop("get_emeter_data", None) + dev._last_update.pop("get_emeter_data", None) + resp["get_current_power"] = {"current_power": 2.002} + resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR with patch.object(dev.protocol, "query", return_value=resp): await dev.update() + assert "get_energy_usage" not in energy_module.data + assert energy_module.current_consumption == 2.002 + assert energy_module.status.power == 2.002 + + +@has_emeter_smart +async def test_v2_get_status_falls_back_to_get_current_power(dev: SmartDevice) -> None: + """get_status should use the same fallback sources as the cached status.""" + energy_module = _get_v2_energy_module(dev) + + async def query(request: dict[str, dict | None]) -> dict[str, dict]: + method = next(iter(request)) + if method == "get_emeter_data": + raise DeviceError(method, error_code=SmartErrorCode.PARAMS_ERROR) + if method == "get_energy_usage": + return {"get_energy_usage": _energy_usage()} + if method == "get_current_power": + return {"get_current_power": {"current_power": 2.002}} + raise AssertionError(f"Unexpected request: {request}") + + with patch.object(dev.protocol, "query", side_effect=query): + status = await energy_module.get_status() + + assert status.power == 2.002 + + +@s515d_smart +async def test_s515d_missing_get_current_power_is_optional(dev: SmartDevice) -> None: + """S515D exposes current power without a working get_current_power query.""" + energy_module = _get_v2_energy_module(dev) + assert energy_module.disabled is False assert energy_module._last_update_error is None assert "get_current_power" not in energy_module.data - - # Some fixtures may have already logged this key removal during initial update, - # so duplicate log messages can be suppressed. - if "Removed key get_current_power" in caplog.text: - assert "PARAMS_ERROR" in caplog.text + assert energy_module.current_consumption == 0.0 + assert energy_module.status.power == 0.0 From fd044d2358541cbf6d2d430177e24379ed177042 Mon Sep 17 00:00:00 2001 From: ZeliardM Date: Thu, 9 Apr 2026 17:10:33 -0400 Subject: [PATCH 4/4] Update test coverages --- tests/smart/modules/test_energy.py | 265 ++++++++++++++++++++++++----- 1 file changed, 219 insertions(+), 46 deletions(-) diff --git a/tests/smart/modules/test_energy.py b/tests/smart/modules/test_energy.py index 41e23a5d4..390a49b88 100644 --- a/tests/smart/modules/test_energy.py +++ b/tests/smart/modules/test_energy.py @@ -18,6 +18,11 @@ model_filter={"S515D(US)_1.6_1.0.4"}, protocol_filter={"SMART"}, ) +p110_v1_smart = parametrize( + "p110 v1 smart", + model_filter={"P110(EU)_1.0_1.0.7"}, + protocol_filter={"SMART"}, +) def _get_energy_module(dev: SmartDevice) -> SmartEnergyModule: @@ -48,6 +53,37 @@ def _energy_usage(current_power: int | None = None) -> dict[str, int]: return energy_usage +def _copy_last_update(dev: SmartDevice, *remove_keys: str) -> dict[str, object]: + """Copy the cached update response, removing keys from the cache if needed.""" + response = copy.deepcopy(dev._last_update) + for key in remove_keys: + response.pop(key, None) + dev._last_update.pop(key, None) + + return response + + +def _device_error(method: str, error_code: SmartErrorCode) -> DeviceError: + return DeviceError(method, error_code=error_code) + + +def _mock_query(responses: dict[str, object], calls: list[str]): + async def query(request: dict[str, dict | None]) -> dict[str, object]: + method = next(iter(request)) + calls.append(method) + + if method not in responses: + raise AssertionError(f"Unexpected request: {request}") + + response = responses[method] + if isinstance(response, DeviceError): + raise response + + return {method: response} + + return query + + @has_emeter_smart async def test_supported(dev: SmartDevice) -> None: energy_module = _get_energy_module(dev) @@ -67,7 +103,7 @@ async def test_get_energy_usage_error( caplog.set_level(logging.DEBUG) energy_module = _get_energy_module(dev) - version = dev._components["energy_monitoring"] + version = energy_module.supported_version expected_raise = does_not_raise() if version > 1 else pytest.raises(DeviceError) if version > 1: @@ -82,8 +118,7 @@ async def test_get_energy_usage_error( assert energy_module.consumption_today is not None assert energy_module.consumption_this_month is not None - last_update = copy.deepcopy(dev._last_update) - resp = copy.deepcopy(last_update) + resp = _copy_last_update(dev) if version > 1: resp["get_emeter_data"] = {"power_mw": 2002} @@ -110,75 +145,213 @@ async def test_get_energy_usage_error( @has_emeter_smart -async def test_v2_current_power_source_precedence(dev: SmartDevice) -> None: +@pytest.mark.parametrize( + ("remove_keys", "response_updates", "expected_power"), + [ + pytest.param( + (), + { + "get_emeter_data": {"power_mw": 3003}, + "get_energy_usage": _energy_usage(current_power=2002), + "get_current_power": {"current_power": 1.001}, + }, + 3.003, + id="prefers_get_emeter_data", + ), + pytest.param( + ("get_emeter_data",), + { + "get_energy_usage": _energy_usage(current_power=2002), + "get_current_power": {"current_power": 1.001}, + }, + 2.002, + id="falls_back_to_get_energy_usage", + ), + ], +) +async def test_v2_current_power_source_precedence( + dev: SmartDevice, + remove_keys: tuple[str, ...], + response_updates: dict[str, object], + expected_power: float, +) -> None: """Prefer higher precision current power sources for v2 devices.""" energy_module = _get_v2_energy_module(dev) - last_update = copy.deepcopy(dev._last_update) - - resp = copy.deepcopy(last_update) - resp["get_emeter_data"] = {"power_mw": 3003} - resp["get_energy_usage"] = _energy_usage(current_power=2002) - resp["get_current_power"] = {"current_power": 1.001} + resp = _copy_last_update(dev, *remove_keys) + resp.update(response_updates) with patch.object(dev.protocol, "query", return_value=resp): await dev.update() - assert energy_module.current_consumption == 3.003 - assert energy_module.status.power == 3.003 + assert energy_module.current_consumption == expected_power + assert energy_module.status.power == expected_power + - resp = copy.deepcopy(last_update) - resp.pop("get_emeter_data", None) - dev._last_update.pop("get_emeter_data", None) - resp["get_energy_usage"] = _energy_usage(current_power=2002) - resp["get_current_power"] = {"current_power": 1.001} +@has_emeter_smart +@pytest.mark.parametrize( + ("response_updates", "expected_power", "expect_energy_usage_removed"), + [ + pytest.param( + { + "get_current_power": {"current_power": 2.002}, + "get_energy_usage": SmartErrorCode.JSON_DECODE_FAIL_ERROR, + }, + 2.002, + True, + id="falls_back_to_get_current_power", + ), + pytest.param( + { + "get_energy_usage": _energy_usage(), + "get_current_power": SmartErrorCode.PARAMS_ERROR, + }, + None, + False, + id="returns_none_without_current_power", + ), + ], +) +async def test_v2_current_power_fallbacks( + dev: SmartDevice, + response_updates: dict[str, object], + expected_power: float | None, + expect_energy_usage_removed: bool, +) -> None: + """Test fallback behavior when higher precision power sources are unavailable.""" + energy_module = _get_v2_energy_module(dev) + resp = _copy_last_update(dev, "get_emeter_data") + resp.update(response_updates) with patch.object(dev.protocol, "query", return_value=resp): await dev.update() - assert energy_module.current_consumption == 2.002 - assert energy_module.status.power == 2.002 + if expect_energy_usage_removed: + assert "get_energy_usage" not in energy_module.data + + assert energy_module.current_consumption == expected_power + assert energy_module.status.power == expected_power @has_emeter_smart -async def test_get_energy_usage_error_falls_back_to_get_current_power( +@pytest.mark.parametrize( + ("responses", "expected_calls", "expected_power"), + [ + pytest.param( + { + "get_emeter_data": { + "current_ma": 25, + "energy_wh": 321, + "power_mw": 3003, + "voltage_mv": 120456, + } + }, + ["get_emeter_data"], + 3.003, + id="prefers_get_emeter_data", + ), + pytest.param( + { + "get_emeter_data": _device_error( + "get_emeter_data", SmartErrorCode.PARAMS_ERROR + ), + "get_energy_usage": _energy_usage(current_power=2002), + }, + ["get_emeter_data", "get_energy_usage"], + 2.002, + id="falls_back_to_get_energy_usage", + ), + pytest.param( + { + "get_emeter_data": _device_error( + "get_emeter_data", SmartErrorCode.PARAMS_ERROR + ), + "get_energy_usage": _energy_usage(), + "get_current_power": {"current_power": 2.002}, + }, + ["get_emeter_data", "get_energy_usage", "get_current_power"], + 2.002, + id="falls_back_to_get_current_power", + ), + pytest.param( + { + "get_emeter_data": _device_error( + "get_emeter_data", SmartErrorCode.PARAMS_ERROR + ), + "get_energy_usage": _device_error( + "get_energy_usage", SmartErrorCode.JSON_DECODE_FAIL_ERROR + ), + "get_current_power": {"current_power": 2.002}, + }, + ["get_emeter_data", "get_energy_usage", "get_current_power"], + 2.002, + id="continues_after_get_energy_usage_error", + ), + pytest.param( + { + "get_emeter_data": _device_error( + "get_emeter_data", SmartErrorCode.PARAMS_ERROR + ), + "get_energy_usage": _energy_usage(), + "get_current_power": _device_error( + "get_current_power", SmartErrorCode.PARAMS_ERROR + ), + }, + ["get_emeter_data", "get_energy_usage", "get_current_power"], + None, + id="returns_none_when_all_sources_fail", + ), + ], +) +async def test_v2_get_status_current_power_sources( dev: SmartDevice, + responses: dict[str, object], + expected_calls: list[str], + expected_power: float | None, ) -> None: - """Use get_current_power when it is the only remaining power source.""" + """get_status should use the same source precedence as cached status.""" energy_module = _get_v2_energy_module(dev) + calls: list[str] = [] - resp = copy.deepcopy(dev._last_update) - resp.pop("get_emeter_data", None) - dev._last_update.pop("get_emeter_data", None) - resp["get_current_power"] = {"current_power": 2.002} - resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR + with patch.object(dev.protocol, "query", side_effect=_mock_query(responses, calls)): + status = await energy_module.get_status() - with patch.object(dev.protocol, "query", return_value=resp): - await dev.update() + assert calls == expected_calls + assert status.power == expected_power - assert "get_energy_usage" not in energy_module.data - assert energy_module.current_consumption == 2.002 - assert energy_module.status.power == 2.002 +@p110_v1_smart +async def test_v1_get_status_uses_energy_usage_only(dev: SmartDevice) -> None: + """V1 devices should return energy usage data without querying current power.""" + energy_module = _get_energy_module(dev) + calls: list[str] = [] + + with patch.object( + dev.protocol, + "query", + side_effect=_mock_query({"get_energy_usage": _energy_usage()}, calls), + ): + status = await energy_module.get_status() + + assert calls == ["get_energy_usage"] + assert status.power is None -@has_emeter_smart -async def test_v2_get_status_falls_back_to_get_current_power(dev: SmartDevice) -> None: - """get_status should use the same fallback sources as the cached status.""" - energy_module = _get_v2_energy_module(dev) + +@p110_v1_smart +async def test_v1_get_status_raises_on_get_energy_usage_error( + dev: SmartDevice, +) -> None: + """V1 devices should still raise when get_energy_usage fails.""" + energy_module = _get_energy_module(dev) async def query(request: dict[str, dict | None]) -> dict[str, dict]: method = next(iter(request)) - if method == "get_emeter_data": - raise DeviceError(method, error_code=SmartErrorCode.PARAMS_ERROR) - if method == "get_energy_usage": - return {"get_energy_usage": _energy_usage()} - if method == "get_current_power": - return {"get_current_power": {"current_power": 2.002}} - raise AssertionError(f"Unexpected request: {request}") - - with patch.object(dev.protocol, "query", side_effect=query): - status = await energy_module.get_status() + raise DeviceError(method, error_code=SmartErrorCode.JSON_DECODE_FAIL_ERROR) - assert status.power == 2.002 + with ( + patch.object(dev.protocol, "query", side_effect=query), + pytest.raises(DeviceError, match="get_energy_usage"), + ): + await energy_module.get_status() @s515d_smart