Improve nuki typing (#86736)
* Use NukiCoordinator * Make NukiEntity generic * Remove unnecessary ABC
This commit is contained in:
parent
138a522d2e
commit
caa1ba7e13
4 changed files with 54 additions and 49 deletions
|
@ -1,7 +1,10 @@
|
||||||
"""The nuki component."""
|
"""The nuki component."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Generic, TypeVar
|
||||||
|
|
||||||
import async_timeout
|
import async_timeout
|
||||||
from pynuki import NukiBridge, NukiLock, NukiOpener
|
from pynuki import NukiBridge, NukiLock, NukiOpener
|
||||||
|
@ -31,6 +34,8 @@ from .const import (
|
||||||
)
|
)
|
||||||
from .helpers import parse_id
|
from .helpers import parse_id
|
||||||
|
|
||||||
|
_NukiDeviceT = TypeVar("_NukiDeviceT", bound=NukiDevice)
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
PLATFORMS = [Platform.BINARY_SENSOR, Platform.LOCK, Platform.SENSOR]
|
PLATFORMS = [Platform.BINARY_SENSOR, Platform.LOCK, Platform.SENSOR]
|
||||||
|
@ -109,38 +114,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
return unload_ok
|
return unload_ok
|
||||||
|
|
||||||
|
|
||||||
class NukiEntity(CoordinatorEntity[DataUpdateCoordinator[None]]):
|
class NukiCoordinator(DataUpdateCoordinator[None]):
|
||||||
"""An entity using CoordinatorEntity.
|
|
||||||
|
|
||||||
The CoordinatorEntity class provides:
|
|
||||||
should_poll
|
|
||||||
async_update
|
|
||||||
async_added_to_hass
|
|
||||||
available
|
|
||||||
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self, coordinator: DataUpdateCoordinator[None], nuki_device: NukiDevice
|
|
||||||
) -> None:
|
|
||||||
"""Pass coordinator to CoordinatorEntity."""
|
|
||||||
super().__init__(coordinator)
|
|
||||||
self._nuki_device = nuki_device
|
|
||||||
|
|
||||||
@property
|
|
||||||
def device_info(self):
|
|
||||||
"""Device info for Nuki entities."""
|
|
||||||
return {
|
|
||||||
"identifiers": {(DOMAIN, parse_id(self._nuki_device.nuki_id))},
|
|
||||||
"name": self._nuki_device.name,
|
|
||||||
"manufacturer": "Nuki Home Solutions GmbH",
|
|
||||||
"model": self._nuki_device.device_type_str.capitalize(),
|
|
||||||
"sw_version": self._nuki_device.firmware_version,
|
|
||||||
"via_device": (DOMAIN, self.coordinator.bridge_id),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
class NukiCoordinator(DataUpdateCoordinator):
|
|
||||||
"""Data Update Coordinator for the Nuki integration."""
|
"""Data Update Coordinator for the Nuki integration."""
|
||||||
|
|
||||||
def __init__(self, hass, bridge, locks, openers):
|
def __init__(self, hass, bridge, locks, openers):
|
||||||
|
@ -217,3 +191,32 @@ class NukiCoordinator(DataUpdateCoordinator):
|
||||||
break
|
break
|
||||||
|
|
||||||
return events
|
return events
|
||||||
|
|
||||||
|
|
||||||
|
class NukiEntity(CoordinatorEntity[NukiCoordinator], Generic[_NukiDeviceT]):
|
||||||
|
"""An entity using CoordinatorEntity.
|
||||||
|
|
||||||
|
The CoordinatorEntity class provides:
|
||||||
|
should_poll
|
||||||
|
async_update
|
||||||
|
async_added_to_hass
|
||||||
|
available
|
||||||
|
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, coordinator: NukiCoordinator, nuki_device: _NukiDeviceT) -> None:
|
||||||
|
"""Pass coordinator to CoordinatorEntity."""
|
||||||
|
super().__init__(coordinator)
|
||||||
|
self._nuki_device = nuki_device
|
||||||
|
|
||||||
|
@property
|
||||||
|
def device_info(self):
|
||||||
|
"""Device info for Nuki entities."""
|
||||||
|
return {
|
||||||
|
"identifiers": {(DOMAIN, parse_id(self._nuki_device.nuki_id))},
|
||||||
|
"name": self._nuki_device.name,
|
||||||
|
"manufacturer": "Nuki Home Solutions GmbH",
|
||||||
|
"model": self._nuki_device.device_type_str.capitalize(),
|
||||||
|
"sw_version": self._nuki_device.firmware_version,
|
||||||
|
"via_device": (DOMAIN, self.coordinator.bridge_id),
|
||||||
|
}
|
||||||
|
|
|
@ -1,6 +1,8 @@
|
||||||
"""Doorsensor Support for the Nuki Lock."""
|
"""Doorsensor Support for the Nuki Lock."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from pynuki.constants import STATE_DOORSENSOR_OPENED
|
from pynuki.constants import STATE_DOORSENSOR_OPENED
|
||||||
|
from pynuki.device import NukiDevice
|
||||||
|
|
||||||
from homeassistant.components.binary_sensor import (
|
from homeassistant.components.binary_sensor import (
|
||||||
BinarySensorDeviceClass,
|
BinarySensorDeviceClass,
|
||||||
|
@ -9,9 +11,8 @@ from homeassistant.components.binary_sensor import (
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
||||||
|
|
||||||
from . import NukiEntity
|
from . import NukiCoordinator, NukiEntity
|
||||||
from .const import ATTR_NUKI_ID, DATA_COORDINATOR, DATA_LOCKS, DOMAIN as NUKI_DOMAIN
|
from .const import ATTR_NUKI_ID, DATA_COORDINATOR, DATA_LOCKS, DOMAIN as NUKI_DOMAIN
|
||||||
|
|
||||||
|
|
||||||
|
@ -20,7 +21,7 @@ async def async_setup_entry(
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the Nuki lock binary sensor."""
|
"""Set up the Nuki lock binary sensor."""
|
||||||
data = hass.data[NUKI_DOMAIN][entry.entry_id]
|
data = hass.data[NUKI_DOMAIN][entry.entry_id]
|
||||||
coordinator: DataUpdateCoordinator[None] = data[DATA_COORDINATOR]
|
coordinator: NukiCoordinator = data[DATA_COORDINATOR]
|
||||||
|
|
||||||
entities = []
|
entities = []
|
||||||
|
|
||||||
|
@ -31,7 +32,7 @@ async def async_setup_entry(
|
||||||
async_add_entities(entities)
|
async_add_entities(entities)
|
||||||
|
|
||||||
|
|
||||||
class NukiDoorsensorEntity(NukiEntity, BinarySensorEntity):
|
class NukiDoorsensorEntity(NukiEntity[NukiDevice], BinarySensorEntity):
|
||||||
"""Representation of a Nuki Lock Doorsensor."""
|
"""Representation of a Nuki Lock Doorsensor."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
_attr_has_entity_name = True
|
||||||
|
|
|
@ -1,11 +1,12 @@
|
||||||
"""Nuki.io lock platform."""
|
"""Nuki.io lock platform."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from abc import ABC, abstractmethod
|
from abc import abstractmethod
|
||||||
from typing import Any
|
from typing import Any, TypeVar
|
||||||
|
|
||||||
from pynuki import NukiLock, NukiOpener
|
from pynuki import NukiLock, NukiOpener
|
||||||
from pynuki.constants import MODE_OPENER_CONTINUOUS
|
from pynuki.constants import MODE_OPENER_CONTINUOUS
|
||||||
|
from pynuki.device import NukiDevice
|
||||||
from requests.exceptions import RequestException
|
from requests.exceptions import RequestException
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
|
@ -14,9 +15,8 @@ from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
||||||
|
|
||||||
from . import NukiEntity
|
from . import NukiCoordinator, NukiEntity
|
||||||
from .const import (
|
from .const import (
|
||||||
ATTR_BATTERY_CRITICAL,
|
ATTR_BATTERY_CRITICAL,
|
||||||
ATTR_ENABLE,
|
ATTR_ENABLE,
|
||||||
|
@ -30,13 +30,15 @@ from .const import (
|
||||||
)
|
)
|
||||||
from .helpers import CannotConnect
|
from .helpers import CannotConnect
|
||||||
|
|
||||||
|
_NukiDeviceT = TypeVar("_NukiDeviceT", bound=NukiDevice)
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(
|
async def async_setup_entry(
|
||||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the Nuki lock platform."""
|
"""Set up the Nuki lock platform."""
|
||||||
data = hass.data[NUKI_DOMAIN][entry.entry_id]
|
data = hass.data[NUKI_DOMAIN][entry.entry_id]
|
||||||
coordinator: DataUpdateCoordinator[None] = data[DATA_COORDINATOR]
|
coordinator: NukiCoordinator = data[DATA_COORDINATOR]
|
||||||
|
|
||||||
entities: list[NukiDeviceEntity] = [
|
entities: list[NukiDeviceEntity] = [
|
||||||
NukiLockEntity(coordinator, lock) for lock in data[DATA_LOCKS]
|
NukiLockEntity(coordinator, lock) for lock in data[DATA_LOCKS]
|
||||||
|
@ -64,7 +66,7 @@ async def async_setup_entry(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class NukiDeviceEntity(NukiEntity, LockEntity, ABC):
|
class NukiDeviceEntity(NukiEntity[_NukiDeviceT], LockEntity):
|
||||||
"""Representation of a Nuki device."""
|
"""Representation of a Nuki device."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
_attr_has_entity_name = True
|
||||||
|
@ -101,11 +103,9 @@ class NukiDeviceEntity(NukiEntity, LockEntity, ABC):
|
||||||
"""Open the door latch."""
|
"""Open the door latch."""
|
||||||
|
|
||||||
|
|
||||||
class NukiLockEntity(NukiDeviceEntity):
|
class NukiLockEntity(NukiDeviceEntity[NukiLock]):
|
||||||
"""Representation of a Nuki lock."""
|
"""Representation of a Nuki lock."""
|
||||||
|
|
||||||
_nuki_device: NukiLock
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_locked(self) -> bool:
|
def is_locked(self) -> bool:
|
||||||
"""Return true if lock is locked."""
|
"""Return true if lock is locked."""
|
||||||
|
@ -144,11 +144,9 @@ class NukiLockEntity(NukiDeviceEntity):
|
||||||
raise CannotConnect from err
|
raise CannotConnect from err
|
||||||
|
|
||||||
|
|
||||||
class NukiOpenerEntity(NukiDeviceEntity):
|
class NukiOpenerEntity(NukiDeviceEntity[NukiOpener]):
|
||||||
"""Representation of a Nuki opener."""
|
"""Representation of a Nuki opener."""
|
||||||
|
|
||||||
_nuki_device: NukiOpener
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_locked(self) -> bool:
|
def is_locked(self) -> bool:
|
||||||
"""Return true if either ring-to-open or continuous mode is enabled."""
|
"""Return true if either ring-to-open or continuous mode is enabled."""
|
||||||
|
|
|
@ -1,4 +1,7 @@
|
||||||
"""Battery sensor for the Nuki Lock."""
|
"""Battery sensor for the Nuki Lock."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from pynuki.device import NukiDevice
|
||||||
|
|
||||||
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
|
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
@ -23,7 +26,7 @@ async def async_setup_entry(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class NukiBatterySensor(NukiEntity, SensorEntity):
|
class NukiBatterySensor(NukiEntity[NukiDevice], SensorEntity):
|
||||||
"""Representation of a Nuki Lock Battery sensor."""
|
"""Representation of a Nuki Lock Battery sensor."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
_attr_has_entity_name = True
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue