jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 06:07:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add WiZ occupancy sensor support (#66231)

Browse Source
This commit is contained in:
J. Nick Koston 2022-02-14 07:25:15 -06:00 committed by GitHub
parent b2ee7cebc9
commit 00d7fdd274
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 242 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
homeassistant/components/wiz/__init__.py
View File

@ -4,13 +4,15 @@ from datetime import timedelta
import logging
from typing import Any
from pywizlight import wizlight
from pywizlight import PilotParser, wizlight
from pywizlight.bulb import PIR_SOURCE
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -19,6 +21,7 @@ from .const import (
DISCOVER_SCAN_TIMEOUT,
DISCOVERY_INTERVAL,
DOMAIN,
SIGNAL_WIZ_PIR,
WIZ_CONNECT_EXCEPTIONS,
WIZ_EXCEPTIONS,
)
@ -27,7 +30,7 @@ from .models import WizData
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.LIGHT, Platform.SWITCH]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.LIGHT, Platform.SWITCH]
REQUEST_REFRESH_DELAY = 0.35
@ -76,7 +79,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
),
)
await bulb.start_push(lambda _: coordinator.async_set_updated_data(None))
@callback
def _async_push_update(state: PilotParser) -> None:
"""Receive a push update."""
_LOGGER.debug("%s: Got push update: %s", bulb.mac, state.pilotResult)
coordinator.async_set_updated_data(None)
if state.get_source() == PIR_SOURCE:
async_dispatcher_send(hass, SIGNAL_WIZ_PIR.format(bulb.mac))
await bulb.start_push(_async_push_update)
bulb.set_discovery_callback(lambda bulb: async_trigger_discovery(hass, [bulb]))
await coordinator.async_config_entry_first_refresh()

81
homeassistant/components/wiz/binary_sensor.py Normal file
View File

@ -0,0 +1,81 @@
"""WiZ integration binary sensor platform."""
from __future__ import annotations
from collections.abc import Callable
from pywizlight.bulb import PIR_SOURCE
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, SIGNAL_WIZ_PIR
from .entity import WizEntity
from .models import WizData
OCCUPANCY_UNIQUE_ID = "{}_occupancy"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the WiZ binary sensor platform."""
wiz_data: WizData = hass.data[DOMAIN][entry.entry_id]
mac = wiz_data.bulb.mac
if er.async_get(hass).async_get_entity_id(
Platform.BINARY_SENSOR, DOMAIN, OCCUPANCY_UNIQUE_ID.format(mac)
):
async_add_entities([WizOccupancyEntity(wiz_data, entry.title)])
return
cancel_dispatcher: Callable[[], None] | None = None
@callback
def _async_add_occupancy_sensor() -> None:
nonlocal cancel_dispatcher
assert cancel_dispatcher is not None
cancel_dispatcher()
cancel_dispatcher = None
async_add_entities([WizOccupancyEntity(wiz_data, entry.title)])
cancel_dispatcher = async_dispatcher_connect(
hass, SIGNAL_WIZ_PIR.format(mac), _async_add_occupancy_sensor
)
@callback
def _async_cancel_dispatcher() -> None:
nonlocal cancel_dispatcher
if cancel_dispatcher is not None:
cancel_dispatcher()
cancel_dispatcher = None
entry.async_on_unload(_async_cancel_dispatcher)
class WizOccupancyEntity(WizEntity, BinarySensorEntity):
"""Representation of WiZ Occupancy sensor."""
_attr_device_class = BinarySensorDeviceClass.OCCUPANCY
def __init__(self, wiz_data: WizData, name: str) -> None:
"""Initialize an WiZ device."""
super().__init__(wiz_data, name)
self._attr_unique_id = OCCUPANCY_UNIQUE_ID.format(self._device.mac)
self._attr_name = f"{name} Occupancy"
self._async_update_attrs()
@callback
def _async_update_attrs(self) -> None:
"""Handle updating _attr values."""
if self._device.state.get_source() == PIR_SOURCE:
self._attr_is_on = self._device.status

2
homeassistant/components/wiz/const.py
View File

@ -21,3 +21,5 @@ WIZ_EXCEPTIONS = (
ConnectionRefusedError,
)
WIZ_CONNECT_EXCEPTIONS = (WizLightNotKnownBulb, *WIZ_EXCEPTIONS)
SIGNAL_WIZ_PIR = "wiz_pir_{}"

18
homeassistant/components/wiz/entity.py
View File

@ -1,23 +1,24 @@
"""WiZ integration entities."""
from __future__ import annotations
from abc import abstractmethod
from typing import Any
from pywizlight.bulblibrary import BulbType
from homeassistant.core import callback
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity import DeviceInfo, ToggleEntity
from homeassistant.helpers.entity import DeviceInfo, Entity, ToggleEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .models import WizData
class WizToggleEntity(CoordinatorEntity, ToggleEntity):
"""Representation of WiZ toggle entity."""
class WizEntity(CoordinatorEntity, Entity):
"""Representation of WiZ entity."""
def __init__(self, wiz_data: WizData, name: str) -> None:
"""Initialize an WiZ device."""
"""Initialize a WiZ entity."""
super().__init__(wiz_data.coordinator)
self._device = wiz_data.bulb
bulb_type: BulbType = self._device.bulbtype
@ -41,6 +42,15 @@ class WizToggleEntity(CoordinatorEntity, ToggleEntity):
self._async_update_attrs()
super()._handle_coordinator_update()
@callback
@abstractmethod
def _async_update_attrs(self) -> None:
"""Handle updating _attr values."""
class WizToggleEntity(WizEntity, ToggleEntity):
"""Representation of WiZ toggle entity."""
@callback
def _async_update_attrs(self) -> None:
"""Handle updating _attr values."""

45
tests/components/wiz/__init__.py
View File

@ -13,6 +13,7 @@ from pywizlight.discovery import DiscoveredBulb
from homeassistant.components.wiz.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@ -110,6 +111,15 @@ FAKE_DIMMABLE_BULB = BulbType(
white_channels=1,
white_to_color_ratio=80,
)
FAKE_TURNABLE_BULB = BulbType(
bulb_type=BulbClass.TW,
name="ESP01_TW_03",
features=FEATURE_MAP[BulbClass.TW],
kelvin_range=KelvinRange(2700, 6500),
fw_version="1.0.0",
white_channels=1,
white_to_color_ratio=80,
)
FAKE_SOCKET = BulbType(
bulb_type=BulbClass.SOCKET,
name="ESP01_SOCKET_03",
@ -144,16 +154,18 @@ async def setup_integration(
def _mocked_wizlight(device, extended_white_range, bulb_type) -> wizlight:
bulb = MagicMock(auto_spec=wizlight)
bulb = MagicMock(auto_spec=wizlight, name="Mocked wizlight")
async def _save_setup_callback(callback: Callable) -> None:
bulb.data_receive_callback = callback
bulb.push_callback = callback
bulb.getBulbConfig = AsyncMock(return_value=device or FAKE_BULB_CONFIG)
bulb.getExtendedWhiteRange = AsyncMock(
return_value=extended_white_range or FAKE_EXTENDED_WHITE_RANGE
)
bulb.getMac = AsyncMock(return_value=FAKE_MAC)
bulb.turn_on = AsyncMock()
bulb.turn_off = AsyncMock()
bulb.updateState = AsyncMock(return_value=FAKE_STATE)
bulb.getSupportedScenes = AsyncMock(return_value=list(SCENES))
bulb.start_push = AsyncMock(side_effect=_save_setup_callback)
@ -169,8 +181,8 @@ def _mocked_wizlight(device, extended_white_range, bulb_type) -> wizlight:
def _patch_wizlight(device=None, extended_white_range=None, bulb_type=None):
@contextmanager
def _patcher():
bulb = _mocked_wizlight(device, extended_white_range, bulb_type)
with patch("homeassistant.components.wiz.wizlight", return_value=bulb,), patch(
bulb = device or _mocked_wizlight(device, extended_white_range, bulb_type)
with patch("homeassistant.components.wiz.wizlight", return_value=bulb), patch(
"homeassistant.components.wiz.config_flow.wizlight",
return_value=bulb,
):
@ -189,3 +201,28 @@ def _patch_discovery():
yield
return _patcher()
async def async_setup_integration(
hass, device=None, extended_white_range=None, bulb_type=None
):
"""Set up the integration with a mock device."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=FAKE_MAC,
data={CONF_HOST: FAKE_IP},
)
entry.add_to_hass(hass)
bulb = _mocked_wizlight(device, extended_white_range, bulb_type)
with _patch_discovery(), _patch_wizlight(device=bulb):
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
return bulb, entry
async def async_push_update(hass, device, params):
"""Push an update to the device."""
device.state = PilotParser(params)
device.status = params["state"]
device.push_callback(device.state)
await hass.async_block_till_done()

83
tests/components/wiz/test_binary_sensor.py Normal file
View File

@ -0,0 +1,83 @@
"""Tests for WiZ binary_sensor platform."""
from homeassistant import config_entries
from homeassistant.components import wiz
from homeassistant.components.wiz.binary_sensor import OCCUPANCY_UNIQUE_ID
from homeassistant.const import CONF_HOST, STATE_OFF, STATE_ON, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from . import (
FAKE_IP,
FAKE_MAC,
_mocked_wizlight,
_patch_discovery,
_patch_wizlight,
async_push_update,
async_setup_integration,
)
from tests.common import MockConfigEntry
async def test_binary_sensor_created_from_push_updates(hass: HomeAssistant) -> None:
"""Test a binary sensor created from push updates."""
bulb, _ = await async_setup_integration(hass)
await async_push_update(hass, bulb, {"mac": FAKE_MAC, "src": "pir", "state": True})
entity_id = "binary_sensor.mock_title_occupancy"
entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == f"{FAKE_MAC}_occupancy"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
await async_push_update(hass, bulb, {"mac": FAKE_MAC, "src": "pir", "state": False})
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
async def test_binary_sensor_restored_from_registry(hass: HomeAssistant) -> None:
"""Test a binary sensor restored from registry with state unknown."""
entry = MockConfigEntry(
domain=wiz.DOMAIN,
unique_id=FAKE_MAC,
data={CONF_HOST: FAKE_IP},
)
entry.add_to_hass(hass)
bulb = _mocked_wizlight(None, None, None)
entity_registry = er.async_get(hass)
reg_ent = entity_registry.async_get_or_create(
Platform.BINARY_SENSOR, wiz.DOMAIN, OCCUPANCY_UNIQUE_ID.format(bulb.mac)
)
entity_id = reg_ent.entity_id
with _patch_discovery(), _patch_wizlight(device=bulb):
await async_setup_component(hass, wiz.DOMAIN, {wiz.DOMAIN: {}})
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_UNKNOWN
await async_push_update(hass, bulb, {"mac": FAKE_MAC, "src": "pir", "state": True})
assert entity_registry.async_get(entity_id).unique_id == f"{FAKE_MAC}_occupancy"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == config_entries.ConfigEntryState.NOT_LOADED
async def test_binary_sensor_never_created_no_error_on_unload(
hass: HomeAssistant,
) -> None:
"""Test a binary sensor does not error on unload."""
_, entry = await async_setup_integration(hass)
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == config_entries.ConfigEntryState.NOT_LOADED

22
tests/components/wiz/test_config_flow.py
View File

@ -12,7 +12,6 @@ from homeassistant.const import CONF_HOST
from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_FORM
from . import (
FAKE_BULB_CONFIG,
FAKE_DIMMABLE_BULB,
FAKE_EXTENDED_WHITE_RANGE,
FAKE_IP,
@ -20,7 +19,6 @@ from . import (
FAKE_RGBW_BULB,
FAKE_RGBWW_BULB,
FAKE_SOCKET,
FAKE_SOCKET_CONFIG,
TEST_CONNECTION,
TEST_SYSTEM_INFO,
_patch_discovery,
@ -184,12 +182,11 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
@pytest.mark.parametrize(
"source, data, device, bulb_type, extended_white_range, name",
"source, data, bulb_type, extended_white_range, name",
[
(
config_entries.SOURCE_DHCP,
DHCP_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_DIMMABLE_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ Dimmable White ABCABC",
@ -197,7 +194,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_INTEGRATION_DISCOVERY,
INTEGRATION_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_DIMMABLE_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ Dimmable White ABCABC",
@ -205,7 +201,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_DHCP,
DHCP_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_RGBW_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ RGBW Tunable ABCABC",
@ -213,7 +208,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_INTEGRATION_DISCOVERY,
INTEGRATION_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_RGBW_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ RGBW Tunable ABCABC",
@ -221,7 +215,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_DHCP,
DHCP_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_RGBWW_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ RGBWW Tunable ABCABC",
@ -229,7 +222,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_INTEGRATION_DISCOVERY,
INTEGRATION_DISCOVERY,
FAKE_BULB_CONFIG,
FAKE_RGBWW_BULB,
FAKE_EXTENDED_WHITE_RANGE,
"WiZ RGBWW Tunable ABCABC",
@ -237,7 +229,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_DHCP,
DHCP_DISCOVERY,
FAKE_SOCKET_CONFIG,
FAKE_SOCKET,
None,
"WiZ Socket ABCABC",
@ -245,7 +236,6 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
(
config_entries.SOURCE_INTEGRATION_DISCOVERY,
INTEGRATION_DISCOVERY,
FAKE_SOCKET_CONFIG,
FAKE_SOCKET,
None,
"WiZ Socket ABCABC",
@ -253,11 +243,11 @@ async def test_discovered_by_dhcp_connection_fails(hass, source, data):
],
)
async def test_discovered_by_dhcp_or_integration_discovery(
hass, source, data, device, bulb_type, extended_white_range, name
hass, source, data, bulb_type, extended_white_range, name
):
"""Test we can configure when discovered from dhcp or discovery."""
with _patch_wizlight(
device=device, extended_white_range=extended_white_range, bulb_type=bulb_type
device=None, extended_white_range=extended_white_range, bulb_type=bulb_type
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}, data=data
@ -268,7 +258,7 @@ async def test_discovered_by_dhcp_or_integration_discovery(
assert result["step_id"] == "discovery_confirm"
with _patch_wizlight(
device=device, extended_white_range=extended_white_range, bulb_type=bulb_type
device=None, extended_white_range=extended_white_range, bulb_type=bulb_type
), patch(
"homeassistant.components.wiz.async_setup_entry",
return_value=True,
@ -423,7 +413,7 @@ async def test_setup_via_discovery_cannot_connect(hass):
async def test_discovery_with_firmware_update(hass):
"""Test we check the device again between first discovery and config entry creation."""
with _patch_wizlight(
device=FAKE_BULB_CONFIG,
device=None,
extended_white_range=FAKE_EXTENDED_WHITE_RANGE,
bulb_type=FAKE_RGBW_BULB,
):
@ -447,7 +437,7 @@ async def test_discovery_with_firmware_update(hass):
) as mock_setup_entry, patch(
"homeassistant.components.wiz.async_setup", return_value=True
) as mock_setup, _patch_wizlight(
device=FAKE_BULB_CONFIG,
device=None,
extended_white_range=FAKE_EXTENDED_WHITE_RANGE,
bulb_type=FAKE_RGBWW_BULB,
):