86 lines
3 KiB
Python
86 lines
3 KiB
Python
|
import logging
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from homeassistant.components.lock import LockEntity, LockEntityDescription
|
||
|
|
from homeassistant.config_entries import ConfigEntry
|
||
|
|
from homeassistant.core import callback
|
||
|
|
from pyhon.parameter.base import HonParameter
|
||
|
|
from pyhon.parameter.range import HonParameterRange
|
||
|
|
|
||
|
|
from .const import DOMAIN
|
||
|
|
from .hon import HonEntity
|
||
|
|
|
||
|
|
_LOGGER = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
LOCKS: dict[str, tuple[LockEntityDescription, ...]] = {
|
||
|
|
"AP": (
|
||
|
|
LockEntityDescription(
|
||
|
|
key="lockStatus",
|
||
|
|
name="Lock Status",
|
||
|
|
translation_key="mode",
|
||
|
|
),
|
||
|
|
),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
|
||
|
|
entities = []
|
||
|
|
for device in hass.data[DOMAIN][entry.unique_id].appliances:
|
||
|
|
for description in LOCKS.get(device.appliance_type, []):
|
||
|
|
if (
|
||
|
|
f"settings.{description.key}" not in device.available_settings
|
||
|
|
or device.get(description.key) is None
|
||
|
|
):
|
||
|
|
continue
|
||
|
|
entity = HonLockEntity(hass, entry, device, description)
|
||
|
|
await entity.coordinator.async_config_entry_first_refresh()
|
||
|
|
entities.append(entity)
|
||
|
|
|
||
|
|
async_add_entities(entities)
|
||
|
|
|
||
|
|
|
||
|
|
class HonLockEntity(HonEntity, LockEntity):
|
||
|
|
entity_description: LockEntityDescription
|
||
|
|
|
||
|
|
@property
|
||
|
|
def is_locked(self) -> bool | None:
|
||
|
|
"""Return a boolean for the state of the lock."""
|
||
|
|
"""Return True if entity is on."""
|
||
|
|
return self._device.get(self.entity_description.key, 0) == 1
|
||
|
|
|
||
|
|
async def async_lock(self, **kwargs: Any) -> None:
|
||
|
|
"""Lock method."""
|
||
|
|
setting = self._device.settings[f"settings.{self.entity_description.key}"]
|
||
|
|
if type(setting) == HonParameter:
|
||
|
|
return
|
||
|
|
setting.value = setting.max if isinstance(setting, HonParameterRange) else 1
|
||
|
|
self.async_write_ha_state()
|
||
|
|
await self._device.commands["settings"].send()
|
||
|
|
await self.coordinator.async_refresh()
|
||
|
|
|
||
|
|
async def async_unlock(self, **kwargs: Any) -> None:
|
||
|
|
"""Unlock method."""
|
||
|
|
setting = self._device.settings[f"settings.{self.entity_description.key}"]
|
||
|
|
if type(setting) == HonParameter:
|
||
|
|
return
|
||
|
|
setting.value = setting.min if isinstance(setting, HonParameterRange) else 0
|
||
|
|
self.async_write_ha_state()
|
||
|
|
await self._device.commands["settings"].send()
|
||
|
|
await self.coordinator.async_refresh()
|
||
|
|
|
||
|
|
@property
|
||
|
|
def available(self) -> bool:
|
||
|
|
"""Return True if entity is available."""
|
||
|
|
return (
|
||
|
|
super().available
|
||
|
|
and int(self._device.get("remoteCtrValid", 1)) == 1
|
||
|
|
and self._device.get("attributes.lastConnEvent.category") != "DISCONNECTED"
|
||
|
|
)
|
||
|
|
|
||
|
|
@callback
|
||
|
|
def _handle_coordinator_update(self, update=True) -> None:
|
||
|
|
value = self._device.get(self.entity_description.key, 0)
|
||
|
|
self._attr_is_locked = self.is_locked
|
||
|
|
if update:
|
||
|
|
self.async_write_ha_state()
|