Improve MyQ code quality through creation of MyQ entity (#54728)
parent
4f9c788216
commit
d3f7312834
|
@ -3,6 +3,13 @@ from datetime import timedelta
|
|||
import logging
|
||||
|
||||
import pymyq
|
||||
from pymyq.const import (
|
||||
DEVICE_STATE as MYQ_DEVICE_STATE,
|
||||
DEVICE_STATE_ONLINE as MYQ_DEVICE_STATE_ONLINE,
|
||||
KNOWN_MODELS,
|
||||
MANUFACTURER,
|
||||
)
|
||||
from pymyq.device import MyQDevice
|
||||
from pymyq.errors import InvalidCredentialsError, MyQError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
@ -10,7 +17,11 @@ from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
|||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
|
||||
from .const import DOMAIN, MYQ_COORDINATOR, MYQ_GATEWAY, PLATFORMS, UPDATE_INTERVAL
|
||||
|
||||
|
@ -63,3 +74,46 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
|
|||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
class MyQEntity(CoordinatorEntity):
|
||||
"""Base class for MyQ Entities."""
|
||||
|
||||
def __init__(self, coordinator: DataUpdateCoordinator, device: MyQDevice) -> None:
|
||||
"""Initialize class."""
|
||||
super().__init__(coordinator)
|
||||
self._device = device
|
||||
self._attr_unique_id = device.device_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name if any, name can change if user changes it within MyQ."""
|
||||
return self._device.name
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device_info of the device."""
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._device.device_id)},
|
||||
"name": self._device.name,
|
||||
"manufacturer": MANUFACTURER,
|
||||
"sw_version": self._device.firmware_version,
|
||||
}
|
||||
model = (
|
||||
KNOWN_MODELS.get(self._device.device_id[2:4])
|
||||
if self._device.device_id is not None
|
||||
else None
|
||||
)
|
||||
if model:
|
||||
device_info["model"] = model
|
||||
if self._device.parent_device_id:
|
||||
device_info["via_device"] = (DOMAIN, self._device.parent_device_id)
|
||||
return device_info
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if the device is online."""
|
||||
# Not all devices report online so assume True if its missing
|
||||
return super().available and self._device.device_json[MYQ_DEVICE_STATE].get(
|
||||
MYQ_DEVICE_STATE_ONLINE, True
|
||||
)
|
||||
|
|
|
@ -1,17 +1,10 @@
|
|||
"""Support for MyQ gateways."""
|
||||
from pymyq.const import (
|
||||
DEVICE_STATE as MYQ_DEVICE_STATE,
|
||||
DEVICE_STATE_ONLINE as MYQ_DEVICE_STATE_ONLINE,
|
||||
KNOWN_MODELS,
|
||||
MANUFACTURER,
|
||||
)
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASS_CONNECTIVITY,
|
||||
BinarySensorEntity,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import MyQEntity
|
||||
from .const import DOMAIN, MYQ_COORDINATOR, MYQ_GATEWAY
|
||||
|
||||
|
||||
|
@ -29,16 +22,11 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
async_add_entities(entities)
|
||||
|
||||
|
||||
class MyQBinarySensorEntity(CoordinatorEntity, BinarySensorEntity):
|
||||
class MyQBinarySensorEntity(MyQEntity, BinarySensorEntity):
|
||||
"""Representation of a MyQ gateway."""
|
||||
|
||||
_attr_device_class = DEVICE_CLASS_CONNECTIVITY
|
||||
|
||||
def __init__(self, coordinator, device):
|
||||
"""Initialize with API object, device id."""
|
||||
super().__init__(coordinator)
|
||||
self._device = device
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the garage door if any."""
|
||||
|
@ -47,35 +35,9 @@ class MyQBinarySensorEntity(CoordinatorEntity, BinarySensorEntity):
|
|||
@property
|
||||
def is_on(self):
|
||||
"""Return if the device is online."""
|
||||
if not self.coordinator.last_update_success:
|
||||
return False
|
||||
|
||||
# Not all devices report online so assume True if its missing
|
||||
return self._device.device_json[MYQ_DEVICE_STATE].get(
|
||||
MYQ_DEVICE_STATE_ONLINE, True
|
||||
)
|
||||
return super().available
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Entity is always available."""
|
||||
return True
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return a unique, Home Assistant friendly identifier for this entity."""
|
||||
return self._device.device_id
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device_info of the device."""
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._device.device_id)},
|
||||
"name": self.name,
|
||||
"manufacturer": MANUFACTURER,
|
||||
"sw_version": self._device.firmware_version,
|
||||
}
|
||||
model = KNOWN_MODELS.get(self._device.device_id[2:4])
|
||||
if model:
|
||||
device_info["model"] = model
|
||||
|
||||
return device_info
|
||||
|
|
|
@ -1,13 +1,7 @@
|
|||
"""Support for MyQ-Enabled Garage Doors."""
|
||||
import logging
|
||||
|
||||
from pymyq.const import (
|
||||
DEVICE_STATE as MYQ_DEVICE_STATE,
|
||||
DEVICE_STATE_ONLINE as MYQ_DEVICE_STATE_ONLINE,
|
||||
DEVICE_TYPE_GATE as MYQ_DEVICE_TYPE_GATE,
|
||||
KNOWN_MODELS,
|
||||
MANUFACTURER,
|
||||
)
|
||||
from pymyq.const import DEVICE_TYPE_GATE as MYQ_DEVICE_TYPE_GATE
|
||||
from pymyq.errors import MyQError
|
||||
|
||||
from homeassistant.components.cover import (
|
||||
|
@ -19,8 +13,8 @@ from homeassistant.components.cover import (
|
|||
)
|
||||
from homeassistant.const import STATE_CLOSED, STATE_CLOSING, STATE_OPEN, STATE_OPENING
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import MyQEntity
|
||||
from .const import DOMAIN, MYQ_COORDINATOR, MYQ_GATEWAY, MYQ_TO_HASS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
@ -33,16 +27,18 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
coordinator = data[MYQ_COORDINATOR]
|
||||
|
||||
async_add_entities(
|
||||
[MyQDevice(coordinator, device) for device in myq.covers.values()]
|
||||
[MyQCover(coordinator, device) for device in myq.covers.values()]
|
||||
)
|
||||
|
||||
|
||||
class MyQDevice(CoordinatorEntity, CoverEntity):
|
||||
class MyQCover(MyQEntity, CoverEntity):
|
||||
"""Representation of a MyQ cover."""
|
||||
|
||||
_attr_supported_features = SUPPORT_OPEN | SUPPORT_CLOSE
|
||||
|
||||
def __init__(self, coordinator, device):
|
||||
"""Initialize with API object, device id."""
|
||||
super().__init__(coordinator)
|
||||
super().__init__(coordinator, device)
|
||||
self._device = device
|
||||
if device.device_type == MYQ_DEVICE_TYPE_GATE:
|
||||
self._attr_device_class = DEVICE_CLASS_GATE
|
||||
|
@ -50,19 +46,6 @@ class MyQDevice(CoordinatorEntity, CoverEntity):
|
|||
self._attr_device_class = DEVICE_CLASS_GARAGE
|
||||
self._attr_unique_id = device.device_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the garage door if any."""
|
||||
return self._device.name
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if the device is online."""
|
||||
# Not all devices report online so assume True if its missing
|
||||
return super().available and self._device.device_json[MYQ_DEVICE_STATE].get(
|
||||
MYQ_DEVICE_STATE_ONLINE, True
|
||||
)
|
||||
|
||||
@property
|
||||
def is_closed(self):
|
||||
"""Return true if cover is closed, else False."""
|
||||
|
@ -83,11 +66,6 @@ class MyQDevice(CoordinatorEntity, CoverEntity):
|
|||
"""Return if the cover is opening or not."""
|
||||
return MYQ_TO_HASS.get(self._device.state) == STATE_OPENING
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Flag supported features."""
|
||||
return SUPPORT_OPEN | SUPPORT_CLOSE
|
||||
|
||||
async def async_close_cover(self, **kwargs):
|
||||
"""Issue close command to cover."""
|
||||
if self.is_closing or self.is_closed:
|
||||
|
@ -133,19 +111,3 @@ class MyQDevice(CoordinatorEntity, CoverEntity):
|
|||
|
||||
if not result:
|
||||
raise HomeAssistantError(f"Opening of cover {self._device.name} failed")
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device_info of the device."""
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._device.device_id)},
|
||||
"name": self._device.name,
|
||||
"manufacturer": MANUFACTURER,
|
||||
"sw_version": self._device.firmware_version,
|
||||
}
|
||||
model = KNOWN_MODELS.get(self._device.device_id[2:4])
|
||||
if model:
|
||||
device_info["model"] = model
|
||||
if self._device.parent_device_id:
|
||||
device_info["via_device"] = (DOMAIN, self._device.parent_device_id)
|
||||
return device_info
|
||||
|
|
|
@ -1,19 +1,13 @@
|
|||
"""Support for MyQ-Enabled lights."""
|
||||
import logging
|
||||
|
||||
from pymyq.const import (
|
||||
DEVICE_STATE as MYQ_DEVICE_STATE,
|
||||
DEVICE_STATE_ONLINE as MYQ_DEVICE_STATE_ONLINE,
|
||||
KNOWN_MODELS,
|
||||
MANUFACTURER,
|
||||
)
|
||||
from pymyq.errors import MyQError
|
||||
|
||||
from homeassistant.components.light import LightEntity
|
||||
from homeassistant.const import STATE_OFF, STATE_ON
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import MyQEntity
|
||||
from .const import DOMAIN, MYQ_COORDINATOR, MYQ_GATEWAY, MYQ_TO_HASS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
@ -30,29 +24,11 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
)
|
||||
|
||||
|
||||
class MyQLight(CoordinatorEntity, LightEntity):
|
||||
class MyQLight(MyQEntity, LightEntity):
|
||||
"""Representation of a MyQ light."""
|
||||
|
||||
_attr_supported_features = 0
|
||||
|
||||
def __init__(self, coordinator, device):
|
||||
"""Initialize with API object, device id."""
|
||||
super().__init__(coordinator)
|
||||
self._device = device
|
||||
self._attr_unique_id = device.device_id
|
||||
self._attr_name = device.name
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if the device is online."""
|
||||
if not super().available:
|
||||
return False
|
||||
|
||||
# Not all devices report online so assume True if its missing
|
||||
return self._device.device_json[MYQ_DEVICE_STATE].get(
|
||||
MYQ_DEVICE_STATE_ONLINE, True
|
||||
)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if the light is on, else False."""
|
||||
|
@ -92,24 +68,3 @@ class MyQLight(CoordinatorEntity, LightEntity):
|
|||
|
||||
# Write new state to HASS
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device_info of the device."""
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._device.device_id)},
|
||||
"name": self._device.name,
|
||||
"manufacturer": MANUFACTURER,
|
||||
"sw_version": self._device.firmware_version,
|
||||
}
|
||||
if model := KNOWN_MODELS.get(self._device.device_id[2:4]):
|
||||
device_info["model"] = model
|
||||
if self._device.parent_device_id:
|
||||
device_info["via_device"] = (DOMAIN, self._device.parent_device_id)
|
||||
return device_info
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Subscribe to updates."""
|
||||
self.async_on_remove(
|
||||
self.coordinator.async_add_listener(self.async_write_ha_state)
|
||||
)
|
||||
|
|
Loading…
Reference in New Issue