Use properties of wemo Insight device (#72316)
parent
c4893f96f6
commit
cf5e21a996
|
@ -2,7 +2,7 @@
|
|||
import asyncio
|
||||
from typing import cast
|
||||
|
||||
from pywemo import Insight, Maker
|
||||
from pywemo import Insight, Maker, StandbyState
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
@ -64,4 +64,5 @@ class InsightBinarySensor(WemoBinarySensor):
|
|||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return true device connected to the Insight Switch is on."""
|
||||
return super().is_on and self.wemo.insight_params["state"] == "1"
|
||||
# Note: wemo.get_standby_state is a @property.
|
||||
return super().is_on and self.wemo.get_standby_state == StandbyState.ON
|
||||
|
|
|
@ -5,7 +5,7 @@ import asyncio
|
|||
from datetime import datetime, timedelta
|
||||
from typing import Any, cast
|
||||
|
||||
from pywemo import CoffeeMaker, Insight, Maker
|
||||
from pywemo import CoffeeMaker, Insight, Maker, StandbyState
|
||||
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
@ -13,7 +13,6 @@ from homeassistant.const import STATE_OFF, STATE_ON, STATE_STANDBY, STATE_UNKNOW
|
|||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.util import convert
|
||||
|
||||
from .const import DOMAIN as WEMO_DOMAIN
|
||||
from .entity import WemoBinaryStateEntity
|
||||
|
@ -22,19 +21,18 @@ from .wemo_device import DeviceCoordinator
|
|||
SCAN_INTERVAL = timedelta(seconds=10)
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
# The WEMO_ constants below come from pywemo itself
|
||||
ATTR_COFFEMAKER_MODE = "coffeemaker_mode"
|
||||
ATTR_CURRENT_STATE_DETAIL = "state_detail"
|
||||
ATTR_ON_LATEST_TIME = "on_latest_time"
|
||||
ATTR_ON_TODAY_TIME = "on_today_time"
|
||||
ATTR_ON_TOTAL_TIME = "on_total_time"
|
||||
ATTR_POWER_THRESHOLD = "power_threshold_w"
|
||||
ATTR_SENSOR_STATE = "sensor_state"
|
||||
ATTR_SWITCH_MODE = "switch_mode"
|
||||
ATTR_CURRENT_STATE_DETAIL = "state_detail"
|
||||
ATTR_COFFEMAKER_MODE = "coffeemaker_mode"
|
||||
|
||||
MAKER_SWITCH_MOMENTARY = "momentary"
|
||||
MAKER_SWITCH_TOGGLE = "toggle"
|
||||
|
||||
WEMO_ON = 1
|
||||
WEMO_OFF = 0
|
||||
WEMO_STANDBY = 8
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
|
@ -83,20 +81,10 @@ class WemoSwitch(WemoBinaryStateEntity, SwitchEntity):
|
|||
attr[ATTR_CURRENT_STATE_DETAIL] = self.detail_state
|
||||
|
||||
if isinstance(self.wemo, Insight):
|
||||
attr["on_latest_time"] = WemoSwitch.as_uptime(
|
||||
self.wemo.insight_params.get("onfor", 0)
|
||||
)
|
||||
attr["on_today_time"] = WemoSwitch.as_uptime(
|
||||
self.wemo.insight_params.get("ontoday", 0)
|
||||
)
|
||||
attr["on_total_time"] = WemoSwitch.as_uptime(
|
||||
self.wemo.insight_params.get("ontotal", 0)
|
||||
)
|
||||
threshold = convert(
|
||||
self.wemo.insight_params.get("powerthreshold"), float, 0.0
|
||||
)
|
||||
assert isinstance(threshold, float)
|
||||
attr["power_threshold_w"] = threshold / 1000.0
|
||||
attr[ATTR_ON_LATEST_TIME] = self.as_uptime(self.wemo.on_for)
|
||||
attr[ATTR_ON_TODAY_TIME] = self.as_uptime(self.wemo.today_on_time)
|
||||
attr[ATTR_ON_TOTAL_TIME] = self.as_uptime(self.wemo.total_on_time)
|
||||
attr[ATTR_POWER_THRESHOLD] = self.wemo.threshold_power_watts
|
||||
|
||||
if isinstance(self.wemo, CoffeeMaker):
|
||||
attr[ATTR_COFFEMAKER_MODE] = self.wemo.mode
|
||||
|
@ -117,12 +105,13 @@ class WemoSwitch(WemoBinaryStateEntity, SwitchEntity):
|
|||
if isinstance(self.wemo, CoffeeMaker):
|
||||
return cast(str, self.wemo.mode_string)
|
||||
if isinstance(self.wemo, Insight):
|
||||
standby_state = int(self.wemo.insight_params.get("state", 0))
|
||||
if standby_state == WEMO_ON:
|
||||
# Note: wemo.get_standby_state is a @property.
|
||||
standby_state = self.wemo.get_standby_state
|
||||
if standby_state == StandbyState.ON:
|
||||
return STATE_ON
|
||||
if standby_state == WEMO_OFF:
|
||||
if standby_state == StandbyState.OFF:
|
||||
return STATE_OFF
|
||||
if standby_state == WEMO_STANDBY:
|
||||
if standby_state == StandbyState.STANDBY:
|
||||
return STATE_STANDBY
|
||||
return STATE_UNKNOWN
|
||||
assert False # Unreachable code statement.
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
"""Fixtures for pywemo."""
|
||||
import asyncio
|
||||
import contextlib
|
||||
from unittest.mock import create_autospec, patch
|
||||
|
||||
import pytest
|
||||
|
@ -52,9 +53,9 @@ def pywemo_discovery_responder_fixture():
|
|||
yield
|
||||
|
||||
|
||||
@pytest.fixture(name="pywemo_device")
|
||||
def pywemo_device_fixture(pywemo_registry, pywemo_model):
|
||||
"""Fixture for WeMoDevice instances."""
|
||||
@contextlib.contextmanager
|
||||
def create_pywemo_device(pywemo_registry, pywemo_model):
|
||||
"""Create a WeMoDevice instance."""
|
||||
cls = getattr(pywemo, pywemo_model)
|
||||
device = create_autospec(cls, instance=True)
|
||||
device.host = MOCK_HOST
|
||||
|
@ -83,15 +84,21 @@ def pywemo_device_fixture(pywemo_registry, pywemo_model):
|
|||
yield device
|
||||
|
||||
|
||||
@pytest.fixture(name="pywemo_device")
|
||||
def pywemo_device_fixture(pywemo_registry, pywemo_model):
|
||||
"""Fixture for WeMoDevice instances."""
|
||||
with create_pywemo_device(pywemo_registry, pywemo_model) as pywemo_device:
|
||||
yield pywemo_device
|
||||
|
||||
|
||||
@pytest.fixture(name="wemo_entity_suffix")
|
||||
def wemo_entity_suffix_fixture():
|
||||
"""Fixture to select a specific entity for wemo_entity."""
|
||||
return ""
|
||||
|
||||
|
||||
@pytest.fixture(name="wemo_entity")
|
||||
async def async_wemo_entity_fixture(hass, pywemo_device, wemo_entity_suffix):
|
||||
"""Fixture for a Wemo entity in hass."""
|
||||
async def async_create_wemo_entity(hass, pywemo_device, wemo_entity_suffix):
|
||||
"""Create a hass entity for a wemo device."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
DOMAIN,
|
||||
|
@ -106,7 +113,13 @@ async def async_wemo_entity_fixture(hass, pywemo_device, wemo_entity_suffix):
|
|||
|
||||
entity_registry = er.async_get(hass)
|
||||
for entry in entity_registry.entities.values():
|
||||
if entry.entity_id.endswith(wemo_entity_suffix):
|
||||
if entry.entity_id.endswith(wemo_entity_suffix or pywemo_device.name.lower()):
|
||||
return entry
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture(name="wemo_entity")
|
||||
async def async_wemo_entity_fixture(hass, pywemo_device, wemo_entity_suffix):
|
||||
"""Fixture for a Wemo entity in hass."""
|
||||
return await async_create_wemo_entity(hass, pywemo_device, wemo_entity_suffix)
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Tests for the Wemo binary_sensor entity."""
|
||||
|
||||
import pytest
|
||||
from pywemo import StandbyState
|
||||
|
||||
from homeassistant.components.homeassistant import (
|
||||
DOMAIN as HA_DOMAIN,
|
||||
|
@ -123,41 +124,27 @@ class TestInsight(EntityTestHelpers):
|
|||
"""Select the InsightBinarySensor entity."""
|
||||
return InsightBinarySensor._name_suffix.lower()
|
||||
|
||||
@pytest.fixture(name="pywemo_device")
|
||||
def pywemo_device_fixture(self, pywemo_device):
|
||||
"""Fixture for WeMoDevice instances."""
|
||||
pywemo_device.insight_params = {
|
||||
"currentpower": 1.0,
|
||||
"todaymw": 200000000.0,
|
||||
"state": "0",
|
||||
"onfor": 0,
|
||||
"ontoday": 0,
|
||||
"ontotal": 0,
|
||||
"powerthreshold": 0,
|
||||
}
|
||||
yield pywemo_device
|
||||
|
||||
async def test_registry_state_callback(
|
||||
self, hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
"""Verify that the binary_sensor receives state updates from the registry."""
|
||||
# On state.
|
||||
pywemo_device.get_state.return_value = 1
|
||||
pywemo_device.insight_params["state"] = "1"
|
||||
pywemo_device.get_standby_state = StandbyState.ON
|
||||
pywemo_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON
|
||||
|
||||
# Standby (Off) state.
|
||||
pywemo_device.get_state.return_value = 1
|
||||
pywemo_device.insight_params["state"] = "8"
|
||||
pywemo_device.get_standby_state = StandbyState.STANDBY
|
||||
pywemo_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF
|
||||
|
||||
# Off state.
|
||||
pywemo_device.get_state.return_value = 0
|
||||
pywemo_device.insight_params["state"] = "1"
|
||||
pywemo_device.get_standby_state = StandbyState.OFF
|
||||
pywemo_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF
|
||||
|
|
|
@ -12,21 +12,6 @@ def pywemo_model():
|
|||
return "Insight"
|
||||
|
||||
|
||||
@pytest.fixture(name="pywemo_device")
|
||||
def pywemo_device_fixture(pywemo_device):
|
||||
"""Fixture for WeMoDevice instances."""
|
||||
pywemo_device.insight_params = {
|
||||
"currentpower": 1.0,
|
||||
"todaymw": 200000000.0,
|
||||
"state": 0,
|
||||
"onfor": 0,
|
||||
"ontoday": 0,
|
||||
"ontotal": 0,
|
||||
"powerthreshold": 0,
|
||||
}
|
||||
yield pywemo_device
|
||||
|
||||
|
||||
class InsightTestTemplate(EntityTestHelpers):
|
||||
"""Base class for testing WeMo Insight Sensors."""
|
||||
|
||||
|
|
|
@ -1,17 +1,35 @@
|
|||
"""Tests for the Wemo switch entity."""
|
||||
|
||||
import pytest
|
||||
from pywemo.exceptions import ActionException
|
||||
import pywemo
|
||||
|
||||
from homeassistant.components.homeassistant import (
|
||||
DOMAIN as HA_DOMAIN,
|
||||
SERVICE_UPDATE_ENTITY,
|
||||
)
|
||||
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
|
||||
from homeassistant.components.wemo.switch import (
|
||||
ATTR_CURRENT_STATE_DETAIL,
|
||||
ATTR_ON_LATEST_TIME,
|
||||
ATTR_ON_TODAY_TIME,
|
||||
ATTR_ON_TOTAL_TIME,
|
||||
ATTR_POWER_THRESHOLD,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_STANDBY,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import entity_test_helpers
|
||||
from .conftest import (
|
||||
MOCK_INSIGHT_STATE_THRESHOLD_POWER,
|
||||
async_create_wemo_entity,
|
||||
create_pywemo_device,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
@ -80,7 +98,7 @@ async def test_available_after_update(
|
|||
hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
"""Test the avaliability when an On call fails and after an update."""
|
||||
pywemo_device.on.side_effect = ActionException
|
||||
pywemo_device.on.side_effect = pywemo.exceptions.ActionException
|
||||
pywemo_device.get_state.return_value = 1
|
||||
await entity_test_helpers.test_avaliable_after_update(
|
||||
hass, pywemo_registry, pywemo_device, wemo_entity, SWITCH_DOMAIN
|
||||
|
@ -90,3 +108,42 @@ async def test_available_after_update(
|
|||
async def test_turn_off_state(hass, wemo_entity):
|
||||
"""Test that the device state is updated after turning off."""
|
||||
await entity_test_helpers.test_turn_off_state(hass, wemo_entity, SWITCH_DOMAIN)
|
||||
|
||||
|
||||
async def test_insight_state_attributes(hass, pywemo_registry):
|
||||
"""Verify the switch attributes are set for the Insight device."""
|
||||
await async_setup_component(hass, HA_DOMAIN, {})
|
||||
with create_pywemo_device(pywemo_registry, "Insight") as insight:
|
||||
wemo_entity = await async_create_wemo_entity(hass, insight, "")
|
||||
attributes = hass.states.get(wemo_entity.entity_id).attributes
|
||||
assert attributes[ATTR_ON_LATEST_TIME] == "00d 00h 20m 34s"
|
||||
assert attributes[ATTR_ON_TODAY_TIME] == "00d 01h 34m 38s"
|
||||
assert attributes[ATTR_ON_TOTAL_TIME] == "00d 02h 30m 12s"
|
||||
assert attributes[ATTR_POWER_THRESHOLD] == MOCK_INSIGHT_STATE_THRESHOLD_POWER
|
||||
assert attributes[ATTR_CURRENT_STATE_DETAIL] == STATE_OFF
|
||||
|
||||
async def async_update():
|
||||
await hass.services.async_call(
|
||||
HA_DOMAIN,
|
||||
SERVICE_UPDATE_ENTITY,
|
||||
{ATTR_ENTITY_ID: [wemo_entity.entity_id]},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Test 'ON' state detail value.
|
||||
insight.get_standby_state = pywemo.StandbyState.ON
|
||||
await async_update()
|
||||
attributes = hass.states.get(wemo_entity.entity_id).attributes
|
||||
assert attributes[ATTR_CURRENT_STATE_DETAIL] == STATE_ON
|
||||
|
||||
# Test 'STANDBY' state detail value.
|
||||
insight.get_standby_state = pywemo.StandbyState.STANDBY
|
||||
await async_update()
|
||||
attributes = hass.states.get(wemo_entity.entity_id).attributes
|
||||
assert attributes[ATTR_CURRENT_STATE_DETAIL] == STATE_STANDBY
|
||||
|
||||
# Test 'UNKNOWN' state detail value.
|
||||
insight.get_standby_state = None
|
||||
await async_update()
|
||||
attributes = hass.states.get(wemo_entity.entity_id).attributes
|
||||
assert attributes[ATTR_CURRENT_STATE_DETAIL] == STATE_UNKNOWN
|
||||
|
|
Loading…
Reference in New Issue