evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Motion blinds duplication reduction using entity baseclass (#99444)

Browse source
This commit is contained in:
starkillerOG 2023-09-02 21:51:58 +02:00 • committed by GitHub
parent 0b065b118b
commit 3c30ad1850
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 117 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -750,6 +750,7 @@ omit =
homeassistant/components/moehlenhoff_alpha2/sensor.py
homeassistant/components/motion_blinds/__init__.py
homeassistant/components/motion_blinds/cover.py
homeassistant/components/motion_blinds/entity.py
homeassistant/components/motion_blinds/sensor.py
homeassistant/components/mpd/media_player.py
homeassistant/components/mqtt_room/sensor.py

23
homeassistant/components/motion_blinds/__init__.py
View file

@ -5,13 +5,12 @@ import logging
from socket import timeout
from typing import TYPE_CHECKING, Any
from motionblinds import DEVICE_TYPES_WIFI, AsyncMotionMulticast, ParseException
from motionblinds import AsyncMotionMulticast, ParseException
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import (
@ -27,8 +26,6 @@ from .const import (
KEY_MULTICAST_LISTENER,
KEY_SETUP_LOCK,
KEY_UNSUB_STOP,
KEY_VERSION,
MANUFACTURER,
PLATFORMS,
UPDATE_INTERVAL,
UPDATE_INTERVAL_FAST,
@ -183,32 +180,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Fetch initial data so we have data when entities subscribe
await coordinator.async_config_entry_first_refresh()
if motion_gateway.firmware is not None:
version = f"{motion_gateway.firmware}, protocol: {motion_gateway.protocol}"
else:
version = f"Protocol: {motion_gateway.protocol}"
hass.data[DOMAIN][entry.entry_id] = {
KEY_GATEWAY: motion_gateway,
KEY_COORDINATOR: coordinator,
KEY_VERSION: version,
}
if TYPE_CHECKING:
assert entry.unique_id is not None
if motion_gateway.device_type not in DEVICE_TYPES_WIFI:
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, motion_gateway.mac)},
identifiers={(DOMAIN, motion_gateway.mac)},
manufacturer=MANUFACTURER,
name=entry.title,
model="Wi-Fi bridge",
sw_version=version,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))

1
homeassistant/components/motion_blinds/const.py
View file

@ -18,7 +18,6 @@ KEY_COORDINATOR = "coordinator"
KEY_MULTICAST_LISTENER = "multicast_listener"
KEY_SETUP_LOCK = "setup_lock"
KEY_UNSUB_STOP = "unsub_stop"
KEY_VERSION = "version"
ATTR_WIDTH = "width"
ATTR_ABSOLUTE_POSITION = "absolute_position"

56
homeassistant/components/motion_blinds/cover.py
View file

@ -16,15 +16,9 @@ from homeassistant.components.cover import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_platform,
)
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTR_ABSOLUTE_POSITION,
@ -33,13 +27,12 @@ from .const import (
DOMAIN,
KEY_COORDINATOR,
KEY_GATEWAY,
KEY_VERSION,
MANUFACTURER,
SERVICE_SET_ABSOLUTE_POSITION,
UPDATE_DELAY_STOP,
UPDATE_INTERVAL_MOVING,
UPDATE_INTERVAL_MOVING_WIFI,
)
from .entity import MotionCoordinatorEntity
from .gateway import device_name
_LOGGER = logging.getLogger(__name__)
@ -96,7 +89,6 @@ async def async_setup_entry(
entities = []
motion_gateway = hass.data[DOMAIN][config_entry.entry_id][KEY_GATEWAY]
coordinator = hass.data[DOMAIN][config_entry.entry_id][KEY_COORDINATOR]
sw_version = hass.data[DOMAIN][config_entry.entry_id][KEY_VERSION]
for blind in motion_gateway.device_list.values():
if blind.type in POSITION_DEVICE_MAP:
@ -105,7 +97,6 @@ async def async_setup_entry(
coordinator,
blind,
POSITION_DEVICE_MAP[blind.type],
sw_version,
)
)
@ -115,7 +106,6 @@ async def async_setup_entry(
coordinator,
blind,
TILT_DEVICE_MAP[blind.type],
sw_version,
)
)
@ -125,7 +115,6 @@ async def async_setup_entry(
coordinator,
blind,
TILT_ONLY_DEVICE_MAP[blind.type],
sw_version,
)
)
@ -135,7 +124,6 @@ async def async_setup_entry(
coordinator,
blind,
TDBU_DEVICE_MAP[blind.type],
sw_version,
"Top",
)
)
@ -144,7 +132,6 @@ async def async_setup_entry(
coordinator,
blind,
TDBU_DEVICE_MAP[blind.type],
sw_version,
"Bottom",
)
)
@ -153,7 +140,6 @@ async def async_setup_entry(
coordinator,
blind,
TDBU_DEVICE_MAP[blind.type],
sw_version,
"Combined",
)
)
@ -168,7 +154,6 @@ async def async_setup_entry(
coordinator,
blind,
POSITION_DEVICE_MAP[BlindType.RollerBlind],
sw_version,
)
)
@ -182,44 +167,27 @@ async def async_setup_entry(
)
class MotionPositionDevice(CoordinatorEntity, CoverEntity):
class MotionPositionDevice(MotionCoordinatorEntity, CoverEntity):
"""Representation of a Motion Blind Device."""
_restore_tilt = False
def __init__(self, coordinator, blind, device_class, sw_version):
def __init__(self, coordinator, blind, device_class):
"""Initialize the blind."""
super().__init__(coordinator)
super().__init__(coordinator, blind)
self._blind = blind
self._api_lock = coordinator.api_lock
self._requesting_position: CALLBACK_TYPE | None = None
self._previous_positions = []
if blind.device_type in DEVICE_TYPES_WIFI:
self._update_interval_moving = UPDATE_INTERVAL_MOVING_WIFI
via_device = ()
connections = {(dr.CONNECTION_NETWORK_MAC, blind.mac)}
else:
self._update_interval_moving = UPDATE_INTERVAL_MOVING
via_device = (DOMAIN, blind._gateway.mac)
connections = {}
sw_version = None
name = device_name(blind)
self._attr_device_class = device_class
self._attr_name = name
self._attr_unique_id = blind.mac
self._attr_device_info = DeviceInfo(
connections=connections,
identifiers={(DOMAIN, blind.mac)},
manufacturer=MANUFACTURER,
model=blind.blind_type,
name=name,
via_device=via_device,
sw_version=sw_version,
hw_version=blind.wireless_name,
)
@property
def available(self) -> bool:
@ -249,16 +217,6 @@ class MotionPositionDevice(CoordinatorEntity, CoverEntity):
return None
return self._blind.position == 100
async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes and register signal handler."""
self._blind.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed."""
self._blind.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass()
async def async_scheduled_update_request(self, *_):
"""Request a state update from the blind at a scheduled point in time."""
# add the last position to the list and keep the list at max 2 items
@ -439,9 +397,9 @@ class MotionTiltOnlyDevice(MotionTiltDevice):
class MotionTDBUDevice(MotionPositionDevice):
"""Representation of a Motion Top Down Bottom Up blind Device."""
def __init__(self, coordinator, blind, device_class, sw_version, motor):
def __init__(self, coordinator, blind, device_class, motor):
"""Initialize the blind."""
super().__init__(coordinator, blind, device_class, sw_version)
super().__init__(coordinator, blind, device_class)
self._motor = motor
self._motor_key = motor[0]
self._attr_name = f"{device_name(blind)} {motor}"

94
homeassistant/components/motion_blinds/entity.py Normal file
View file

@ -0,0 +1,94 @@
"""Support for Motion Blinds using their WLAN API."""
from __future__ import annotations
from motionblinds import DEVICE_TYPES_GATEWAY, DEVICE_TYPES_WIFI, MotionGateway
from motionblinds.motion_blinds import MotionBlind
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DataUpdateCoordinatorMotionBlinds
from .const import (
ATTR_AVAILABLE,
DEFAULT_GATEWAY_NAME,
DOMAIN,
KEY_GATEWAY,
MANUFACTURER,
)
from .gateway import device_name
class MotionCoordinatorEntity(CoordinatorEntity[DataUpdateCoordinatorMotionBlinds]):
"""Representation of a Motion Blind entity."""
def __init__(
self,
coordinator: DataUpdateCoordinatorMotionBlinds,
blind: MotionGateway | MotionBlind,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._blind = blind
self._api_lock = coordinator.api_lock
if blind.device_type in DEVICE_TYPES_GATEWAY:
gateway = blind
else:
gateway = blind._gateway
if gateway.firmware is not None:
sw_version = f"{gateway.firmware}, protocol: {gateway.protocol}"
else:
sw_version = f"Protocol: {gateway.protocol}"
if blind.device_type in DEVICE_TYPES_GATEWAY:
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, blind.mac)},
identifiers={(DOMAIN, blind.mac)},
manufacturer=MANUFACTURER,
name=DEFAULT_GATEWAY_NAME,
model="Wi-Fi bridge",
sw_version=sw_version,
)
elif blind.device_type in DEVICE_TYPES_WIFI:
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, blind.mac)},
identifiers={(DOMAIN, blind.mac)},
manufacturer=MANUFACTURER,
model=blind.blind_type,
name=device_name(blind),
sw_version=sw_version,
hw_version=blind.wireless_name,
)
else:
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, blind.mac)},
manufacturer=MANUFACTURER,
model=blind.blind_type,
name=device_name(blind),
via_device=(DOMAIN, blind._gateway.mac),
hw_version=blind.wireless_name,
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
if self.coordinator.data is None:
return False
gateway_available = self.coordinator.data[KEY_GATEWAY][ATTR_AVAILABLE]
if not gateway_available or self._blind.device_type in DEVICE_TYPES_GATEWAY:
return gateway_available
return self.coordinator.data[self._blind.mac][ATTR_AVAILABLE]
async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes and register signal handler."""
self._blind.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed."""
self._blind.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass()

84
homeassistant/components/motion_blinds/sensor.py
View file

@ -1,5 +1,5 @@
"""Support for Motion Blinds sensors."""
from motionblinds import DEVICE_TYPES_WIFI, BlindType
from motionblinds import DEVICE_TYPES_GATEWAY, DEVICE_TYPES_WIFI, BlindType
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
@ -9,16 +9,13 @@ from homeassistant.const import (
EntityCategory,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTR_AVAILABLE, DOMAIN, KEY_COORDINATOR, KEY_GATEWAY
from .const import DOMAIN, KEY_COORDINATOR, KEY_GATEWAY
from .entity import MotionCoordinatorEntity
from .gateway import device_name
ATTR_BATTERY_VOLTAGE = "battery_voltage"
TYPE_BLIND = "blind"
TYPE_GATEWAY = "gateway"
async def async_setup_entry(
@ -32,7 +29,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][config_entry.entry_id][KEY_COORDINATOR]
for blind in motion_gateway.device_list.values():
entities.append(MotionSignalStrengthSensor(coordinator, blind, TYPE_BLIND))
entities.append(MotionSignalStrengthSensor(coordinator, blind))
if blind.type == BlindType.TopDownBottomUp:
entities.append(MotionTDBUBatterySensor(coordinator, blind, "Bottom"))
entities.append(MotionTDBUBatterySensor(coordinator, blind, "Top"))
@ -42,14 +39,12 @@ async def async_setup_entry(
# Do not add signal sensor twice for direct WiFi blinds
if motion_gateway.device_type not in DEVICE_TYPES_WIFI:
entities.append(
MotionSignalStrengthSensor(coordinator, motion_gateway, TYPE_GATEWAY)
)
entities.append(MotionSignalStrengthSensor(coordinator, motion_gateway))
async_add_entities(entities)
class MotionBatterySensor(CoordinatorEntity, SensorEntity):
class MotionBatterySensor(MotionCoordinatorEntity, SensorEntity):
"""Representation of a Motion Battery Sensor."""
_attr_device_class = SensorDeviceClass.BATTERY
@ -57,24 +52,11 @@ class MotionBatterySensor(CoordinatorEntity, SensorEntity):
def __init__(self, coordinator, blind):
"""Initialize the Motion Battery Sensor."""
super().__init__(coordinator)
super().__init__(coordinator, blind)
self._blind = blind
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, blind.mac)})
self._attr_name = f"{device_name(blind)} battery"
self._attr_unique_id = f"{blind.mac}-battery"
@property
def available(self) -> bool:
"""Return True if entity is available."""
if self.coordinator.data is None:
return False
if not self.coordinator.data[KEY_GATEWAY][ATTR_AVAILABLE]:
return False
return self.coordinator.data[self._blind.mac][ATTR_AVAILABLE]
@property
def native_value(self):
"""Return the state of the sensor."""
@ -85,16 +67,6 @@ class MotionBatterySensor(CoordinatorEntity, SensorEntity):
"""Return device specific state attributes."""
return {ATTR_BATTERY_VOLTAGE: self._blind.battery_voltage}
async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes."""
self._blind.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed."""
self._blind.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass()
class MotionTDBUBatterySensor(MotionBatterySensor):
"""Representation of a Motion Battery Sensor for a Top Down Bottom Up blind."""
@ -125,7 +97,7 @@ class MotionTDBUBatterySensor(MotionBatterySensor):
return attributes
class MotionSignalStrengthSensor(CoordinatorEntity, SensorEntity):
class MotionSignalStrengthSensor(MotionCoordinatorEntity, SensorEntity):
"""Representation of a Motion Signal Strength Sensor."""
_attr_device_class = SensorDeviceClass.SIGNAL_STRENGTH
@ -133,47 +105,19 @@ class MotionSignalStrengthSensor(CoordinatorEntity, SensorEntity):
_attr_native_unit_of_measurement = SIGNAL_STRENGTH_DECIBELS_MILLIWATT
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, coordinator, device, device_type):
def __init__(self, coordinator, blind):
"""Initialize the Motion Signal Strength Sensor."""
super().__init__(coordinator)
super().__init__(coordinator, blind)
if device_type == TYPE_GATEWAY:
if blind.device_type in DEVICE_TYPES_GATEWAY:
name = "Motion gateway signal strength"
else:
name = f"{device_name(device)} signal strength"
name = f"{device_name(blind)} signal strength"
self._device = device
self._device_type = device_type
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device.mac)})
self._attr_unique_id = f"{device.mac}-RSSI"
self._attr_unique_id = f"{blind.mac}-RSSI"
self._attr_name = name
@property
def available(self) -> bool:
"""Return True if entity is available."""
if self.coordinator.data is None:
return False
gateway_available = self.coordinator.data[KEY_GATEWAY][ATTR_AVAILABLE]
if self._device_type == TYPE_GATEWAY:
return gateway_available
return (
gateway_available
and self.coordinator.data[self._device.mac][ATTR_AVAILABLE]
)
@property
def native_value(self):
"""Return the state of the sensor."""
return self._device.RSSI
async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes."""
self._device.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed."""
self._device.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass()
return self._blind.RSSI