github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-07-27 18:58:57 +02:00
Code Releases Activity

Improve devolo Home Control code quality (#40480)

Browse Source 
Co-authored-by: Markus Bong <2Fake1987@gmail.com>
This commit is contained in:
Guido Schmitz 2020-09-24 17:57:52 +02:00 committed by GitHub
parent dc30f0e00c
commit 0b11559031
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 100 additions and 257 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
homeassistant/components/devolo_home_control/binary_sensor.py
View File

@ -67,50 +67,35 @@ class DevoloBinaryDeviceEntity(DevoloDeviceEntity, BinarySensorEntity):
element_uid
)
self._device_class = DEVICE_CLASS_MAPPING.get(
self._binary_sensor_property.sub_type
or self._binary_sensor_property.sensor_type
)
name = device_instance.item_name
if self._device_class is None:
if device_instance.binary_sensor_property.get(element_uid).sub_type != "":
name += f" {device_instance.binary_sensor_property.get(element_uid).sub_type}"
else:
name += f" {device_instance.binary_sensor_property.get(element_uid).sensor_type}"
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
name=name,
sync=self._sync,
)
self._state = self._binary_sensor_property.state
self._device_class = DEVICE_CLASS_MAPPING.get(
self._binary_sensor_property.sub_type
or self._binary_sensor_property.sensor_type
)
self._subscriber = None
if self._device_class is None:
if device_instance.binary_sensor_property.get(element_uid).sub_type != "":
self._name += f" {device_instance.binary_sensor_property.get(element_uid).sub_type}"
else:
self._name += f" {device_instance.binary_sensor_property.get(element_uid).sensor_type}"
self._value = self._binary_sensor_property.state
@property
def is_on(self):
"""Return the state."""
return self._state
return self._value
@property
def device_class(self):
"""Return device class."""
return self._device_class
def _sync(self, message=None):
"""Update the binary sensor state."""
if message[0].startswith("devolo.BinarySensor"):
self._state = self._device_instance.binary_sensor_property[message[0]].state
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("No valid message received: %s", message)
self.schedule_update_ha_state()
class DevoloRemoteControl(DevoloDeviceEntity, BinarySensorEntity):
"""Representation of a remote control within devolo Home Control."""
@ -120,26 +105,22 @@ class DevoloRemoteControl(DevoloDeviceEntity, BinarySensorEntity):
self._remote_control_property = device_instance.remote_control_property.get(
element_uid
)
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=f"{element_uid}_{key}",
name=device_instance.item_name,
sync=self._sync,
)
self._key = key
self._state = False
self._subscriber = None
@property
def is_on(self):
"""Return the state."""
return self._state
def _sync(self, message=None):
def _sync(self, message):
"""Update the binary sensor state."""
if (
message[0] == self._remote_control_property.element_uid

32
homeassistant/components/devolo_home_control/climate.py
View File

@ -14,7 +14,7 @@ from homeassistant.const import PRECISION_HALVES
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
from .devolo_device import DevoloDeviceEntity
from .devolo_multi_level_switch import DevoloMultiLevelSwitchDeviceEntity
_LOGGER = logging.getLogger(__name__)
@ -42,29 +42,13 @@ async def async_setup_entry(
async_add_entities(entities, False)
class DevoloClimateDeviceEntity(DevoloDeviceEntity, ClimateEntity):
class DevoloClimateDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, ClimateEntity):
"""Representation of a climate/thermostat device within devolo Home Control."""
def __init__(self, homecontrol, device_instance, element_uid):
"""Initialize a devolo climate/thermostat device."""
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
name=device_instance.item_name,
sync=self._sync,
)
self._multi_level_switch_property = (
device_instance.multi_level_switch_property.get(element_uid)
)
self._temperature = self._multi_level_switch_property.value
@property
def current_temperature(self) -> Optional[float]:
"""Return the current temperature."""
return self._temperature
return self._value
@property
def hvac_mode(self) -> str:
@ -104,13 +88,3 @@ class DevoloClimateDeviceEntity(DevoloDeviceEntity, ClimateEntity):
def set_temperature(self, **kwargs):
"""Set new target temperature."""
self._multi_level_switch_property.set(kwargs[ATTR_TEMPERATURE])
def _sync(self, message=None):
"""Update the climate entity triggered by web socket connection."""
if message[0] == self._unique_id:
self._temperature = message[1]
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("Not valid message received: %s", message)
self.schedule_update_ha_state()

34
homeassistant/components/devolo_home_control/cover.py
View File

@ -12,7 +12,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
from .devolo_device import DevoloDeviceEntity
from .devolo_multi_level_switch import DevoloMultiLevelSwitchDeviceEntity
_LOGGER = logging.getLogger(__name__)
@ -37,29 +37,13 @@ async def async_setup_entry(
async_add_entities(entities, False)
class DevoloCoverDeviceEntity(DevoloDeviceEntity, CoverEntity):
class DevoloCoverDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, CoverEntity):
"""Representation of a cover device within devolo Home Control."""
def __init__(self, homecontrol, device_instance, element_uid):
"""Initialize a devolo blinds device."""
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
name=device_instance.item_name,
sync=self._sync,
)
self._multi_level_switch_property = (
device_instance.multi_level_switch_property.get(element_uid)
)
self._position = self._multi_level_switch_property.value
@property
def current_cover_position(self):
"""Return the current position. 0 is closed. 100 is open."""
return self._position
return self._value
@property
def device_class(self):
@ -69,7 +53,7 @@ class DevoloCoverDeviceEntity(DevoloDeviceEntity, CoverEntity):
@property
def is_closed(self):
"""Return if the blind is closed or not."""
return not bool(self._position)
return not bool(self._value)
@property
def supported_features(self):
@ -87,13 +71,3 @@ class DevoloCoverDeviceEntity(DevoloDeviceEntity, CoverEntity):
def set_cover_position(self, **kwargs):
"""Set the blind to the given position."""
self._multi_level_switch_property.set(kwargs["position"])
def _sync(self, message=None):
"""Update the binary sensor state."""
if message[0] == self._unique_id:
self._position = message[1]
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("Not valid message received: %s", message)
self.schedule_update_ha_state()

21
homeassistant/components/devolo_home_control/devolo_device.py
View File

@ -10,14 +10,17 @@ _LOGGER = logging.getLogger(__name__)
class DevoloDeviceEntity(Entity):
"""Representation of a sensor within devolo Home Control."""
"""Abstract representation of a device within devolo Home Control."""
def __init__(self, homecontrol, device_instance, element_uid, name, sync):
def __init__(self, homecontrol, device_instance, element_uid):
"""Initialize a devolo device entity."""
self._device_instance = device_instance
self._name = name
self._unique_id = element_uid
self._homecontrol = homecontrol
self._name = device_instance.item_name
self._device_class = None
self._value = None
self._unit = None
# This is not doing I/O. It fetches an internal state of the API
self._available = device_instance.is_online()
@ -27,7 +30,7 @@ class DevoloDeviceEntity(Entity):
self._model = device_instance.name
self.subscriber = None
self.sync_callback = sync
self.sync_callback = self._sync
async def async_added_to_hass(self) -> None:
"""Call when entity is added to hass."""
@ -73,3 +76,13 @@ class DevoloDeviceEntity(Entity):
def available(self) -> bool:
"""Return the online state."""
return self._available
def _sync(self, message):
"""Update the binary sensor state."""
if message[0] == self._unique_id:
self._value = message[1]
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("No valid message received: %s", message)
self.schedule_update_ha_state()

12
homeassistant/components/devolo_home_control/devolo_multi_level_switch.py
View File

@ -15,21 +15,9 @@ class DevoloMultiLevelSwitchDeviceEntity(DevoloDeviceEntity):
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
name=device_instance.item_name,
sync=self._sync,
)
self._multi_level_switch_property = device_instance.multi_level_switch_property[
element_uid
]
self._value = self._multi_level_switch_property.value
def _sync(self, message):
"""Update the multi level switch state."""
if message[0] == self._multi_level_switch_property.element_uid:
self._value = message[1]
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("No valid message received: %s", message)
self.schedule_update_ha_state()

126
homeassistant/components/devolo_home_control/sensor.py
View File

@ -33,7 +33,7 @@ async def async_setup_entry(
for device in hass.data[DOMAIN]["homecontrol"].multi_level_sensor_devices:
for multi_level_sensor in device.multi_level_sensor_property:
entities.append(
DevoloMultiLevelDeviceEntity(
DevoloGenericMultiLevelDeviceEntity(
homecontrol=hass.data[DOMAIN]["homecontrol"],
device_instance=device,
element_uid=multi_level_sensor,
@ -55,44 +55,7 @@ async def async_setup_entry(
class DevoloMultiLevelDeviceEntity(DevoloDeviceEntity):
"""Representation of a multi level sensor within devolo Home Control."""
def __init__(
self,
homecontrol,
device_instance,
element_uid,
multi_level_sensor_property=None,
sync=None,
):
"""Initialize a devolo multi level sensor."""
if multi_level_sensor_property is None:
self._multi_level_sensor_property = (
device_instance.multi_level_sensor_property[element_uid]
)
else:
self._multi_level_sensor_property = multi_level_sensor_property
self._state = self._multi_level_sensor_property.value
self._device_class = DEVICE_CLASS_MAPPING.get(
self._multi_level_sensor_property.sensor_type
)
name = device_instance.item_name
if self._device_class is None:
name += f" {self._multi_level_sensor_property.sensor_type}"
self._unit = self._multi_level_sensor_property.unit
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
name=name,
sync=self._sync if sync is None else sync,
)
"""Abstract representation of a multi level sensor within devolo Home Control."""
@property
def device_class(self) -> str:
@ -102,24 +65,43 @@ class DevoloMultiLevelDeviceEntity(DevoloDeviceEntity):
@property
def state(self):
"""Return the state of the sensor."""
return self._state
return self._value
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity."""
return self._unit
def _sync(self, message=None):
"""Update the multi level sensor state."""
if message[0] == self._multi_level_sensor_property.element_uid:
self._state = self._device_instance.multi_level_sensor_property[
message[0]
].value
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()
else:
_LOGGER.debug("No valid message received: %s", message)
self.schedule_update_ha_state()
class DevoloGenericMultiLevelDeviceEntity(DevoloMultiLevelDeviceEntity):
"""Representation of a generic multi level sensor within devolo Home Control."""
def __init__(
self,
homecontrol,
device_instance,
element_uid,
):
"""Initialize a devolo multi level sensor."""
self._multi_level_sensor_property = device_instance.multi_level_sensor_property[
element_uid
]
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
)
self._device_class = DEVICE_CLASS_MAPPING.get(
self._multi_level_sensor_property.sensor_type
)
self._value = self._multi_level_sensor_property.value
self._unit = self._multi_level_sensor_property.unit
if self._device_class is None:
self._name += f" {self._multi_level_sensor_property.sensor_type}"
class DevoloConsumptionEntity(DevoloMultiLevelDeviceEntity):
@ -127,36 +109,36 @@ class DevoloConsumptionEntity(DevoloMultiLevelDeviceEntity):
def __init__(self, homecontrol, device_instance, element_uid, consumption):
"""Initialize a devolo consumption sensor."""
self._device_instance = device_instance
self.value = getattr(
device_instance.consumption_property[element_uid], consumption
)
self.sensor_type = consumption
self.unit = getattr(
device_instance.consumption_property[element_uid], f"{consumption}_unit"
)
self.element_uid = element_uid
super().__init__(
homecontrol,
device_instance,
element_uid,
multi_level_sensor_property=self,
sync=self._sync,
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
)
self._sensor_type = consumption
self._device_class = DEVICE_CLASS_MAPPING.get(consumption)
self._value = getattr(
device_instance.consumption_property[element_uid], consumption
)
self._unit = getattr(
device_instance.consumption_property[element_uid], f"{consumption}_unit"
)
self._name += f" {consumption}"
@property
def unique_id(self):
"""Return the unique ID of the entity."""
return f"{self._unique_id}_{self.sensor_type}"
return f"{self._unique_id}_{self._sensor_type}"
def _sync(self, message=None):
def _sync(self, message):
"""Update the consumption sensor state."""
if message[0] == self.element_uid:
self._state = getattr(
self._device_instance.consumption_property[self.element_uid],
self.sensor_type,
if message[0] == self._unique_id:
self._value = getattr(
self._device_instance.consumption_property[self._unique_id],
self._sensor_type,
)
elif message[0].startswith("hdm"):
self._available = self._device_instance.is_online()

85
homeassistant/components/devolo_home_control/switch.py
View File

@ -6,6 +6,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
from .devolo_device import DevoloDeviceEntity
_LOGGER = logging.getLogger(__name__)
@ -32,26 +33,16 @@ async def async_setup_entry(
async_add_entities(entities)
class DevoloSwitch(SwitchEntity):
class DevoloSwitch(DevoloDeviceEntity, SwitchEntity):
"""Representation of a switch."""
def __init__(self, homecontrol, device_instance, element_uid):
"""Initialize an devolo Switch."""
self._device_instance = device_instance
# Create the unique ID
self._unique_id = element_uid
self._homecontrol = homecontrol
self._name = self._device_instance.item_name
# This is not doing I/O. It fetches an internal state of the API
self._available = self._device_instance.is_online()
# Get the brand and model information
self._brand = self._device_instance.brand
self._model = self._device_instance.name
super().__init__(
homecontrol=homecontrol,
device_instance=device_instance,
element_uid=element_uid,
)
self._binary_switch_property = self._device_instance.binary_switch_property.get(
self._unique_id
)
@ -64,47 +55,6 @@ class DevoloSwitch(SwitchEntity):
else:
self._consumption = None
self.subscriber = None
async def async_added_to_hass(self):
"""Call when entity is added to hass."""
self.subscriber = Subscriber(
self._device_instance.item_name, callback=self.sync
)
self._homecontrol.publisher.register(
self._device_instance.uid, self.subscriber, self.sync
)
@property
def unique_id(self):
"""Return the unique ID of the switch."""
return self._unique_id
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._device_instance.uid)},
"name": self.name,
"manufacturer": self._brand,
"model": self._model,
}
@property
def device_id(self):
"""Return the ID of this switch."""
return self._unique_id
@property
def name(self):
"""Return the display name of this switch."""
return self._name
@property
def should_poll(self):
"""Return the polling state."""
return False
@property
def is_on(self):
"""Return the state."""
@ -115,11 +65,6 @@ class DevoloSwitch(SwitchEntity):
"""Return the current consumption."""
return self._consumption
@property
def available(self):
"""Return the online state."""
return self._available
def turn_on(self, **kwargs):
"""Switch on the device."""
self._is_on = True
@ -130,7 +75,7 @@ class DevoloSwitch(SwitchEntity):
self._is_on = False
self._binary_switch_property.set(state=False)
def sync(self, message=None):
def _sync(self, message):
"""Update the binary switch state and consumption."""
if message[0].startswith("devolo.BinarySwitch"):
self._is_on = self._device_instance.binary_switch_property[message[0]].state
@ -143,17 +88,3 @@ class DevoloSwitch(SwitchEntity):
else:
_LOGGER.debug("No valid message received: %s", message)
self.schedule_update_ha_state()
class Subscriber:
"""Subscriber class for the publisher in mprm websocket class."""
def __init__(self, name, callback):
"""Initiate the device."""
self.name = name
self.callback = callback
def update(self, message):
"""Trigger hass to update the device."""
_LOGGER.debug('%s got message "%s"', self.name, message)
self.callback(message)