evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Binary Sensor for WeMo Insight & Maker (#55000)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Eric Severance 2021-08-22 11:09:22 -07:00 • committed by GitHub
parent 2d34ebc506
commit e6ba3b41cb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 247 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/wemo/__init__.py
View file

@ -33,9 +33,9 @@ WEMO_MODEL_DISPATCH = {
"CoffeeMaker": [SWITCH_DOMAIN],
"Dimmer": [LIGHT_DOMAIN],
"Humidifier": [FAN_DOMAIN],
"Insight": [SENSOR_DOMAIN, SWITCH_DOMAIN],
"Insight": [BINARY_SENSOR_DOMAIN, SENSOR_DOMAIN, SWITCH_DOMAIN],
"LightSwitch": [SWITCH_DOMAIN],
"Maker": [SWITCH_DOMAIN],
"Maker": [BINARY_SENSOR_DOMAIN, SWITCH_DOMAIN],
"Motion": [BINARY_SENSOR_DOMAIN],
"OutdoorPlug": [SWITCH_DOMAIN],
"Sensor": [BINARY_SENSOR_DOMAIN],

31
homeassistant/components/wemo/binary_sensor.py
View file

@ -2,6 +2,8 @@
import asyncio
import logging
from pywemo import Insight, Maker
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -16,7 +18,12 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
async def _discovered_wemo(coordinator):
"""Handle a discovered Wemo device."""
async_add_entities([WemoBinarySensor(coordinator)])
if isinstance(coordinator.wemo, Insight):
async_add_entities([InsightBinarySensor(coordinator)])
elif isinstance(coordinator.wemo, Maker):
async_add_entities([MakerBinarySensor(coordinator)])
else:
async_add_entities([WemoBinarySensor(coordinator)])
async_dispatcher_connect(hass, f"{WEMO_DOMAIN}.binary_sensor", _discovered_wemo)
@ -35,3 +42,25 @@ class WemoBinarySensor(WemoEntity, BinarySensorEntity):
def is_on(self) -> bool:
"""Return true if the state is on. Standby is on."""
return self.wemo.get_state()
class MakerBinarySensor(WemoEntity, BinarySensorEntity):
"""Maker device's sensor port."""
_name_suffix = "Sensor"
@property
def is_on(self) -> bool:
"""Return true if the Maker's sensor is pulled low."""
return self.wemo.has_sensor and self.wemo.sensor_state == 0
class InsightBinarySensor(WemoBinarySensor):
"""Sensor representing the device connected to the Insight Switch."""
_name_suffix = "Device"
@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"

24
homeassistant/components/wemo/entity.py
View file

@ -19,6 +19,12 @@ _LOGGER = logging.getLogger(__name__)
class WemoEntity(CoordinatorEntity):
"""Common methods for Wemo entities."""
# Most pyWeMo devices are associated with a single Home Assistant entity. When
# that is not the case, name_suffix & unique_id_suffix can be used to provide
# names and unique ids for additional Home Assistant entities.
_name_suffix: str | None = None
_unique_id_suffix: str | None = None
def __init__(self, coordinator: DeviceCoordinator) -> None:
"""Initialize the WeMo device."""
super().__init__(coordinator)
@ -26,9 +32,17 @@ class WemoEntity(CoordinatorEntity):
self._device_info = coordinator.device_info
self._available = True
@property
def name_suffix(self):
"""Suffix to append to the WeMo device name."""
return self._name_suffix
@property
def name(self) -> str:
"""Return the name of the device if any."""
suffix = self.name_suffix
if suffix:
return f"{self.wemo.name} {suffix}"
return self.wemo.name
@property
@ -36,9 +50,19 @@ class WemoEntity(CoordinatorEntity):
"""Return true if the device is available."""
return super().available and self._available
@property
def unique_id_suffix(self):
"""Suffix to append to the WeMo device's unique ID."""
if self._unique_id_suffix is None and self.name_suffix is not None:
return self._name_suffix.lower()
return self._unique_id_suffix
@property
def unique_id(self) -> str:
"""Return the id of this WeMo device."""
suffix = self.unique_id_suffix
if suffix:
return f"{self.wemo.serialnumber}_{suffix}"
return self.wemo.serialnumber
@property

8
homeassistant/components/wemo/sensor.py
View file

@ -45,14 +45,14 @@ class InsightSensor(WemoEntity, SensorEntity):
"""Common base for WeMo Insight power sensors."""
@property
def name(self) -> str:
def name_suffix(self) -> str:
"""Return the name of the entity if any."""
return f"{super().name} {self.entity_description.name}"
return self.entity_description.name
@property
def unique_id(self) -> str:
def unique_id_suffix(self) -> str:
"""Return the id of this entity."""
return f"{super().unique_id}_{self.entity_description.key}"
return self.entity_description.key
@property
def available(self) -> str:

15
tests/components/wemo/conftest.py
View file

@ -68,8 +68,14 @@ def pywemo_device_fixture(pywemo_registry, pywemo_model):
yield 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):
async def async_wemo_entity_fixture(hass, pywemo_device, wemo_entity_suffix):
"""Fixture for a Wemo entity in hass."""
assert await async_setup_component(
hass,
@ -84,7 +90,8 @@ async def async_wemo_entity_fixture(hass, pywemo_device):
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
entity_entries = list(entity_registry.entities.values())
assert len(entity_entries) == 1
for entry in entity_registry.entities.values():
if entry.entity_id.endswith(wemo_entity_suffix):
return entry
yield entity_entries[0]
return None

225
tests/components/wemo/test_binary_sensor.py
View file

@ -6,71 +6,190 @@ from homeassistant.components.homeassistant import (
DOMAIN as HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
)
from homeassistant.components.wemo.binary_sensor import (
InsightBinarySensor,
MakerBinarySensor,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from . import entity_test_helpers
@pytest.fixture
def pywemo_model():
"""Pywemo Motion models use the binary_sensor platform."""
return "Motion"
class EntityTestHelpers:
"""Common state update helpers."""
async def test_async_update_locked_multiple_updates(
self, hass, pywemo_device, wemo_entity
):
"""Test that two hass async_update state updates do not proceed at the same time."""
await entity_test_helpers.test_async_update_locked_multiple_updates(
hass, pywemo_device, wemo_entity
)
async def test_async_update_locked_multiple_callbacks(
self, hass, pywemo_device, wemo_entity
):
"""Test that two device callback state updates do not proceed at the same time."""
await entity_test_helpers.test_async_update_locked_multiple_callbacks(
hass, pywemo_device, wemo_entity
)
async def test_async_update_locked_callback_and_update(
self, hass, pywemo_device, wemo_entity
):
"""Test that a callback and a state update request can't both happen at the same time.
When a state update is received via a callback from the device at the same time
as hass is calling `async_update`, verify that only one of the updates proceeds.
"""
await entity_test_helpers.test_async_update_locked_callback_and_update(
hass, pywemo_device, wemo_entity
)
# Tests that are in common among wemo platforms. These test methods will be run
# in the scope of this test module. They will run using the pywemo_model from
# this test module (Motion).
test_async_update_locked_multiple_updates = (
entity_test_helpers.test_async_update_locked_multiple_updates
)
test_async_update_locked_multiple_callbacks = (
entity_test_helpers.test_async_update_locked_multiple_callbacks
)
test_async_update_locked_callback_and_update = (
entity_test_helpers.test_async_update_locked_callback_and_update
)
class TestMotion(EntityTestHelpers):
"""Test for the pyWeMo Motion device."""
@pytest.fixture
def pywemo_model(self):
"""Pywemo Motion models use the binary_sensor platform."""
return "Motion"
async def test_binary_sensor_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_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
await hass.async_block_till_done()
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON
# Off state.
pywemo_device.get_state.return_value = 0
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
async def test_binary_sensor_update_entity(
self, hass, pywemo_registry, pywemo_device, wemo_entity
):
"""Verify that the binary_sensor performs state updates."""
await async_setup_component(hass, HA_DOMAIN, {})
# On state.
pywemo_device.get_state.return_value = 1
await hass.services.async_call(
HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: [wemo_entity.entity_id]},
blocking=True,
)
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON
# Off state.
pywemo_device.get_state.return_value = 0
await hass.services.async_call(
HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: [wemo_entity.entity_id]},
blocking=True,
)
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF
async def test_binary_sensor_registry_state_callback(
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_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
await hass.async_block_till_done()
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON
class TestMaker(EntityTestHelpers):
"""Test for the pyWeMo Maker device."""
# Off state.
pywemo_device.get_state.return_value = 0
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
@pytest.fixture
def pywemo_model(self):
"""Pywemo Motion models use the binary_sensor platform."""
return "Maker"
@pytest.fixture
def wemo_entity_suffix(self):
"""Select the MakerBinarySensor entity."""
return MakerBinarySensor._name_suffix.lower()
@pytest.fixture(name="pywemo_device")
def pywemo_device_fixture(self, pywemo_device):
"""Fixture for WeMoDevice instances."""
pywemo_device.maker_params = {
"hassensor": 1,
"sensorstate": 1,
"switchmode": 1,
"switchstate": 0,
}
pywemo_device.has_sensor = pywemo_device.maker_params["hassensor"]
pywemo_device.sensor_state = pywemo_device.maker_params["sensorstate"]
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.sensor_state = 0
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
# Off state.
pywemo_device.sensor_state = 1
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
async def test_binary_sensor_update_entity(
hass, pywemo_registry, pywemo_device, wemo_entity
):
"""Verify that the binary_sensor performs state updates."""
await async_setup_component(hass, HA_DOMAIN, {})
class TestInsight(EntityTestHelpers):
"""Test for the pyWeMo Insight device."""
# On state.
pywemo_device.get_state.return_value = 1
await hass.services.async_call(
HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: [wemo_entity.entity_id]},
blocking=True,
)
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON
@pytest.fixture
def pywemo_model(self):
"""Pywemo Motion models use the binary_sensor platform."""
return "Insight"
# Off state.
pywemo_device.get_state.return_value = 0
await hass.services.async_call(
HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: [wemo_entity.entity_id]},
blocking=True,
)
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF
@pytest.fixture
def wemo_entity_suffix(self):
"""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_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_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_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
await hass.async_block_till_done()
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF

36
tests/components/wemo/test_sensor.py
View file

@ -6,14 +6,10 @@ from homeassistant.components.homeassistant import (
DOMAIN as HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
)
from homeassistant.components.wemo import CONF_DISCOVERY, CONF_STATIC
from homeassistant.components.wemo.const import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from . import entity_test_helpers
from .conftest import MOCK_HOST, MOCK_PORT
@pytest.fixture
@ -44,35 +40,11 @@ class InsightTestTemplate:
EXPECTED_STATE_VALUE: str
INSIGHT_PARAM_NAME: str
@pytest.fixture(name="wemo_entity")
@pytest.fixture(name="wemo_entity_suffix")
@classmethod
async def async_wemo_entity_fixture(cls, hass, pywemo_device):
"""Fixture for a Wemo entity in hass."""
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
CONF_DISCOVERY: False,
CONF_STATIC: [f"{MOCK_HOST}:{MOCK_PORT}"],
},
},
)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
correct_entity = None
to_remove = []
for entry in entity_registry.entities.values():
if entry.entity_id.endswith(cls.ENTITY_ID_SUFFIX):
correct_entity = entry
else:
to_remove.append(entry.entity_id)
for removal in to_remove:
entity_registry.async_remove(removal)
assert len(entity_registry.entities) == 1
return correct_entity
def wemo_entity_suffix_fixture(cls):
"""Select the appropriate entity for the test."""
return cls.ENTITY_ID_SUFFIX
# Tests that are in common among wemo platforms. These test methods will be run
# in the scope of this test module. They will run using the pywemo_model from