Add Surepetcare locks (#56396)

* Surepetcare, add lock

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Fix tests

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Surepetcare, lock name

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* surepetcare_id

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* typing

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Fix review comment

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Fix review comment

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Fix review comment

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* add more tests

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

* Fix review comment

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
This commit is contained in:
Daniel Hjelseth Høyer 2021-09-29 18:17:12 +02:00 committed by GitHub
parent 50fffe48f8
commit 60eb426451
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 179 additions and 4 deletions

View File

@ -40,7 +40,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["binary_sensor", "sensor"]
PLATFORMS = ["binary_sensor", "lock", "sensor"]
SCAN_INTERVAL = timedelta(minutes=3)
CONFIG_SCHEMA = vol.Schema(
@ -118,7 +118,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
vol.Required(ATTR_LOCK_STATE): vol.All(
cv.string,
vol.Lower,
vol.In(coordinator.lock_states.keys()),
vol.In(coordinator.lock_states_callbacks.keys()),
),
}
)
@ -171,7 +171,7 @@ class SurePetcareDataCoordinator(DataUpdateCoordinator):
api_timeout=SURE_API_TIMEOUT,
session=async_get_clientsession(hass),
)
self.lock_states = {
self.lock_states_callbacks = {
LockState.UNLOCKED.name.lower(): self.surepy.sac.unlock,
LockState.LOCKED_IN.name.lower(): self.surepy.sac.lock_in,
LockState.LOCKED_OUT.name.lower(): self.surepy.sac.lock_out,
@ -195,7 +195,7 @@ class SurePetcareDataCoordinator(DataUpdateCoordinator):
"""Call when setting the lock state."""
flap_id = call.data[ATTR_FLAP_ID]
state = call.data[ATTR_LOCK_STATE]
await self.lock_states[state](flap_id)
await self.lock_states_callbacks[state](flap_id)
await self.async_request_refresh()
def get_pets(self) -> dict[str, int]:

View File

@ -0,0 +1,98 @@
"""Support for Sure PetCare Flaps locks."""
from __future__ import annotations
import logging
from typing import Any
from surepy.entities import SurepyEntity
from surepy.enums import EntityType, LockState
from homeassistant.components.lock import STATE_LOCKED, STATE_UNLOCKED, LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SurePetcareDataCoordinator
from .const import DOMAIN
from .entity import SurePetcareEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Sure PetCare locks on a config entry."""
entities: list[SurePetcareLock] = []
coordinator: SurePetcareDataCoordinator = hass.data[DOMAIN][entry.entry_id]
for surepy_entity in coordinator.data.values():
if surepy_entity.type not in [
EntityType.CAT_FLAP,
EntityType.PET_FLAP,
]:
continue
for lock_state in (
LockState.LOCKED_IN,
LockState.LOCKED_OUT,
LockState.LOCKED_ALL,
):
entities.append(SurePetcareLock(surepy_entity.id, coordinator, lock_state))
async_add_entities(entities)
class SurePetcareLock(SurePetcareEntity, LockEntity):
"""A lock implementation for Sure Petcare Entities."""
coordinator: SurePetcareDataCoordinator
def __init__(
self,
surepetcare_id: int,
coordinator: SurePetcareDataCoordinator,
lock_state: LockState,
) -> None:
"""Initialize a Sure Petcare lock."""
self._lock_state = lock_state.name.lower()
self._available = False
super().__init__(surepetcare_id, coordinator)
self._attr_name = f"{self._device_name} {self._lock_state.replace('_', ' ')}"
self._attr_unique_id = f"{self._device_id}-{self._lock_state}"
@property
def available(self) -> bool:
"""Return true if entity is available."""
return self._available and super().available
@callback
def _update_attr(self, surepy_entity: SurepyEntity) -> None:
"""Update the state."""
status = surepy_entity.raw_data()["status"]
self._attr_is_locked = (
LockState(status["locking"]["mode"]).name.lower() == self._lock_state
)
self._available = bool(status.get("online"))
async def async_lock(self, **kwargs: Any) -> None:
"""Lock the lock."""
if self.state == STATE_LOCKED:
return
await self.coordinator.lock_states_callbacks[self._lock_state](self._id)
self._attr_is_locked = True
self.async_write_ha_state()
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the lock."""
if self.state == STATE_UNLOCKED:
return
await self.coordinator.surepy.sac.unlock(self._id)
self._attr_is_locked = False
self.async_write_ha_state()

View File

@ -38,6 +38,7 @@ MOCK_CAT_FLAP = {
"locking": {"mode": 0},
"learn_mode": 0,
"signal": {"device_rssi": 65, "hub_rssi": 64},
"online": True,
},
}
@ -52,6 +53,7 @@ MOCK_PET_FLAP = {
"locking": {"mode": 0},
"learn_mode": 0,
"signal": {"device_rssi": 70, "hub_rssi": 65},
"online": True,
},
}

View File

@ -0,0 +1,75 @@
"""The tests for the Sure Petcare lock platform."""
from homeassistant.components.surepetcare.const import DOMAIN
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from . import HOUSEHOLD_ID, MOCK_CAT_FLAP, MOCK_CONFIG, MOCK_PET_FLAP
EXPECTED_ENTITY_IDS = {
"lock.cat_flap_locked_in": f"{HOUSEHOLD_ID}-{MOCK_CAT_FLAP['id']}-locked_in",
"lock.cat_flap_locked_out": f"{HOUSEHOLD_ID}-{MOCK_CAT_FLAP['id']}-locked_out",
"lock.cat_flap_locked_all": f"{HOUSEHOLD_ID}-{MOCK_CAT_FLAP['id']}-locked_all",
"lock.pet_flap_locked_in": f"{HOUSEHOLD_ID}-{MOCK_PET_FLAP['id']}-locked_in",
"lock.pet_flap_locked_out": f"{HOUSEHOLD_ID}-{MOCK_PET_FLAP['id']}-locked_out",
"lock.pet_flap_locked_all": f"{HOUSEHOLD_ID}-{MOCK_PET_FLAP['id']}-locked_all",
}
async def test_locks(hass, surepetcare) -> None:
"""Test the generation of unique ids."""
assert await async_setup_component(hass, DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
state_entity_ids = hass.states.async_entity_ids()
for entity_id, unique_id in EXPECTED_ENTITY_IDS.items():
surepetcare.reset_mock()
assert entity_id in state_entity_ids
state = hass.states.get(entity_id)
assert state
assert state.state == "unlocked"
entity = entity_registry.async_get(entity_id)
assert entity.unique_id == unique_id
await hass.services.async_call(
"lock", "unlock", {"entity_id": entity_id}, blocking=True
)
state = hass.states.get(entity_id)
assert state.state == "unlocked"
# already unlocked
assert surepetcare.unlock.call_count == 0
await hass.services.async_call(
"lock", "lock", {"entity_id": entity_id}, blocking=True
)
state = hass.states.get(entity_id)
assert state.state == "locked"
if "locked_in" in entity_id:
assert surepetcare.lock_in.call_count == 1
elif "locked_out" in entity_id:
assert surepetcare.lock_out.call_count == 1
elif "locked_all" in entity_id:
assert surepetcare.lock.call_count == 1
# lock again should not trigger another request
await hass.services.async_call(
"lock", "lock", {"entity_id": entity_id}, blocking=True
)
state = hass.states.get(entity_id)
assert state.state == "locked"
if "locked_in" in entity_id:
assert surepetcare.lock_in.call_count == 1
elif "locked_out" in entity_id:
assert surepetcare.lock_out.call_count == 1
elif "locked_all" in entity_id:
assert surepetcare.lock.call_count == 1
await hass.services.async_call(
"lock", "unlock", {"entity_id": entity_id}, blocking=True
)
state = hass.states.get(entity_id)
assert state.state == "unlocked"
assert surepetcare.unlock.call_count == 1