diff --git a/README.md b/README.md index e571461d..f2e7db84 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, L530EA, L535E, L630 - **Light Strips**: L900-10, L900-5, L920-5, L930-5 - **Cameras**: C100, C101, C110, C210, C220, C225, C325WB, C460, C520WS, C720, TC40, TC65, TC70 diff --git a/SUPPORTED.md b/SUPPORTED.md index 1d38ee0e..a3d9c336 100644 --- a/SUPPORTED.md +++ b/SUPPORTED.md @@ -254,6 +254,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 03df6d11..b898f38e 100644 --- a/kasa/smart/modules/energy.py +++ b/kasa/smart/modules/energy.py @@ -5,10 +5,15 @@ from __future__ import annotations from typing import Any, NoReturn from ...emeterstatus import EmeterStatus -from ...exceptions import DeviceError, KasaException +from ...exceptions import DeviceError, KasaException, SmartErrorCode from ...interfaces.energy import Energy as EnergyInterface from ..smartmodule import SmartModule, raise_if_update_error +_OPTIONAL_METHOD_ERRORS = { + SmartErrorCode.PARAMS_ERROR, + SmartErrorCode.UNKNOWN_METHOD_ERROR, +} + class Energy(SmartModule, EnergyInterface): """Implementation of energy monitoring module.""" @@ -18,6 +23,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,17 +61,8 @@ class Energy(SmartModule, EnergyInterface): 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 := self._get_current_power_mw(data)) 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") - ) is not None: - self._current_consumption = power else: self._current_consumption = None @@ -63,7 +81,7 @@ class Energy(SmartModule, EnergyInterface): 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 @@ -76,10 +94,14 @@ class Energy(SmartModule, EnergyInterface): """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, } ) @@ -88,19 +110,50 @@ class Energy(SmartModule, EnergyInterface): @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 as ex: + if ex.error_code not in _OPTIONAL_METHOD_ERRORS: + raise + 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 as ex: + if ex.error_code not in _OPTIONAL_METHOD_ERRORS: + raise + 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/device_fixtures.py b/tests/device_fixtures.py index e8d131de..d928f542 100644 --- a/tests/device_fixtures.py +++ b/tests/device_fixtures.py @@ -118,6 +118,7 @@ SWITCHES_SMART = { "S500D", "S505", "S505D", + "S515D", "TS15", } SWITCHES = {*SWITCHES_IOT, *SWITCHES_SMART} @@ -126,7 +127,7 @@ STRIPS_SMART = {"P300", "P304M", "TP25", "EP40M", "P210M", "P306", "P316M"} 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, @@ -150,7 +151,7 @@ THERMOSTATS_SMART = {"KE100"} 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 00000000..c36f0c25 --- /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 96da5ea8..18a6ccc4 100644 --- a/tests/smart/modules/test_energy.py +++ b/tests/smart/modules/test_energy.py @@ -10,16 +10,83 @@ from kasa.exceptions import SmartErrorCode 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 + +s515d_smart = parametrize( + "s515d smart", + 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"}, +) -@has_emeter_smart -async def test_supported(dev: SmartDevice) -> None: +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 + + +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) 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: @@ -35,11 +102,8 @@ async def test_get_energy_usage_error( """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}.") - - version = dev._components["energy_monitoring"] + energy_module = _get_energy_module(dev) + version = energy_module.supported_version expected_raise = does_not_raise() if version > 1 else pytest.raises(DeviceError) if version > 1: @@ -54,13 +118,10 @@ 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 ed := resp.get("get_emeter_data"): - ed["power_mw"] = 2002 - if cp := resp.get("get_current_power"): - 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,28 +143,284 @@ 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() - resp = copy.deepcopy(last_update) - if cp := resp.get("get_current_power"): - cp["current_power"] = 2.002 - resp["get_energy_usage"] = SmartErrorCode.JSON_DECODE_FAIL_ERROR - - # Remove get_emeter_data from the response and from the device which will - # remember it otherwise. - resp.pop("get_emeter_data", None) - dev._last_update.pop("get_emeter_data", None) +@has_emeter_smart +@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) + resp = _copy_last_update(dev, *remove_keys) + resp.update(response_updates) with patch.object(dev.protocol, "query", return_value=resp): await dev.update() - with expected_raise: + assert energy_module.current_consumption == expected_power + assert energy_module.status.power == expected_power + + +@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() + + if expect_energy_usage_removed: assert "get_energy_usage" not in energy_module.data - assert energy_module.current_consumption == expected_current_consumption + assert energy_module.current_consumption == expected_power + assert energy_module.status.power == expected_power - # message should only be logged once - assert msg not in caplog.text + +@has_emeter_smart +@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.UNKNOWN_METHOD_ERROR + ), + "get_energy_usage": _energy_usage(current_power=2002), + }, + ["get_emeter_data", "get_energy_usage"], + 2.002, + id="falls_back_after_unknown_method", + ), + 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: + """get_status should use the same source precedence as cached status.""" + energy_module = _get_v2_energy_module(dev) + calls: list[str] = [] + + with patch.object(dev.protocol, "query", side_effect=_mock_query(responses, calls)): + status = await energy_module.get_status() + + assert calls == expected_calls + assert status.power == expected_power + + +@has_emeter_smart +@pytest.mark.parametrize( + ("responses", "expected_calls", "expected_match"), + [ + pytest.param( + { + "get_emeter_data": _device_error( + "get_emeter_data", SmartErrorCode.JSON_DECODE_FAIL_ERROR + ), + }, + ["get_emeter_data"], + "get_emeter_data", + id="get_emeter_data", + ), + 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.JSON_DECODE_FAIL_ERROR + ), + }, + ["get_emeter_data", "get_energy_usage", "get_current_power"], + "get_current_power", + id="get_current_power", + ), + ], +) +async def test_v2_get_status_raises_unexpected_optional_method_errors( + dev: SmartDevice, + responses: dict[str, object], + expected_calls: list[str], + expected_match: str, +) -> None: + """Only unsupported optional method errors should fall back silently.""" + energy_module = _get_v2_energy_module(dev) + calls: list[str] = [] + + with ( + patch.object(dev.protocol, "query", side_effect=_mock_query(responses, calls)), + pytest.raises(DeviceError, match=expected_match), + ): + await energy_module.get_status() + + assert calls == expected_calls + + +@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 + + +@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)) + raise DeviceError(method, error_code=SmartErrorCode.JSON_DECODE_FAIL_ERROR) + + with ( + patch.object(dev.protocol, "query", side_effect=query), + pytest.raises(DeviceError, match="get_energy_usage"), + ): + await energy_module.get_status() + + +@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 + assert energy_module.current_consumption == 0.0 + assert energy_module.status.power == 0.0