Remove boilerplate code in favour of attributes in Netatmo integration (#52395)

* Make use of attributes

* Add suggestions

* Update homeassistant/components/netatmo/sensor.py

Co-authored-by: Daniel Hjelseth Høyer <mail@dahoiv.net>

* Update homeassistant/components/netatmo/sensor.py

Co-authored-by: Daniel Hjelseth Høyer <mail@dahoiv.net>

Co-authored-by: Daniel Hjelseth Høyer <mail@dahoiv.net>
pull/52422/head
Tobias Sauerwein 2021-07-02 11:36:37 +02:00 committed by GitHub
parent 2ed1efc3a3
commit 7291228e16
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 92 additions and 145 deletions

View File

@ -125,9 +125,9 @@ class NetatmoCamera(NetatmoBase, Camera):
self._id = camera_id
self._home_id = home_id
self._device_name = self._data.get_camera(camera_id=camera_id).get("name")
self._name = f"{MANUFACTURER} {self._device_name}"
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._model = camera_type
self._unique_id = f"{self._id}-{self._model}"
self._attr_unique_id = f"{self._id}-{self._model}"
self._quality = quality
self._vpnurl = None
self._localurl = None
@ -172,6 +172,9 @@ class NetatmoCamera(NetatmoBase, Camera):
self._status = "on"
elif data[WEBHOOK_PUSH_TYPE] == WEBHOOK_LIGHT_MODE:
self._light_state = data["sub_type"]
self._attr_extra_state_attributes.update(
{"light_state": self._light_state}
)
self.async_write_ha_state()
return
@ -190,20 +193,6 @@ class NetatmoCamera(NetatmoBase, Camera):
_LOGGER.debug("Could not fetch live camera image (%s)", err)
return None
@property
def extra_state_attributes(self):
"""Return the Netatmo-specific camera state attributes."""
return {
"id": self._id,
"status": self._status,
"sd_status": self._sd_status,
"alim_status": self._alim_status,
"is_local": self._is_local,
"vpn_url": self._vpnurl,
"local_url": self._localurl,
"light_state": self._light_state,
}
@property
def available(self):
"""Return True if entity is available."""
@ -273,6 +262,19 @@ class NetatmoCamera(NetatmoBase, Camera):
self._data.outdoor_events.get(self._id, {})
)
self._attr_extra_state_attributes.update(
{
"id": self._id,
"status": self._status,
"sd_status": self._sd_status,
"alim_status": self._alim_status,
"is_local": self._is_local,
"vpn_url": self._vpnurl,
"local_url": self._localurl,
"light_state": self._light_state,
}
)
def process_events(self, events):
"""Add meta data to events."""
for event in events.values():
@ -328,7 +330,7 @@ class NetatmoCamera(NetatmoBase, Camera):
async def _service_set_camera_light(self, **kwargs):
"""Service to set light mode."""
mode = kwargs.get(ATTR_CAMERA_LIGHT_MODE)
_LOGGER.debug("Turn %s camera light for '%s'", mode, self._name)
_LOGGER.debug("Turn %s camera light for '%s'", mode, self._attr_name)
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,

View File

@ -198,7 +198,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
break
self._device_name = self._data.rooms[home_id][room_id]["name"]
self._name = f"{MANUFACTURER} {self._device_name}"
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._current_temperature = None
self._target_temperature = None
self._preset = None
@ -218,7 +218,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
if self._model == NA_THERM:
self._operation_list.append(HVAC_MODE_OFF)
self._unique_id = f"{self._id}-{self._model}"
self._attr_unique_id = f"{self._id}-{self._model}"
async def async_added_to_hass(self) -> None:
"""Entity created."""
@ -253,6 +253,9 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self._selected_schedule = self.hass.data[DOMAIN][DATA_SCHEDULES][
self._home_id
].get(data["schedule_id"])
self._attr_extra_state_attributes.update(
{"selected_schedule": self._selected_schedule}
)
self.async_write_ha_state()
self.data_handler.async_force_update(self._home_status_class)
return
@ -426,24 +429,6 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self.async_write_ha_state()
@property
def extra_state_attributes(self):
"""Return the state attributes of the thermostat."""
attr = {}
if self._battery_level is not None:
attr[ATTR_BATTERY_LEVEL] = self._battery_level
if self._model == NA_VALVE:
attr[ATTR_HEATING_POWER_REQUEST] = self._room_status.get(
"heating_power_request", 0
)
if self._selected_schedule is not None:
attr[ATTR_SELECTED_SCHEDULE] = self._selected_schedule
return attr
async def async_turn_off(self):
"""Turn the entity off."""
if self._model == NA_VALVE:
@ -513,6 +498,19 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self._away = self._hvac_mode == HVAC_MAP_NETATMO[STATE_NETATMO_AWAY]
if self._battery_level is not None:
self._attr_extra_state_attributes[ATTR_BATTERY_LEVEL] = self._battery_level
if self._model == NA_VALVE:
self._attr_extra_state_attributes[
ATTR_HEATING_POWER_REQUEST
] = self._room_status.get("heating_power_request", 0)
if self._selected_schedule is not None:
self._attr_extra_state_attributes[
ATTR_SELECTED_SCHEDULE
] = self._selected_schedule
def _build_room_status(self):
"""Construct room status."""
try:

View File

@ -8,6 +8,7 @@ API = "api"
DOMAIN = "netatmo"
MANUFACTURER = "Netatmo"
DEFAULT_ATTRIBUTION = f"Data provided by {MANUFACTURER}"
PLATFORMS = [CAMERA_DOMAIN, CLIMATE_DOMAIN, LIGHT_DOMAIN, SENSOR_DOMAIN]

View File

@ -80,9 +80,9 @@ class NetatmoLight(NetatmoBase, LightEntity):
self._home_id = home_id
self._model = camera_type
self._device_name = self._data.get_camera(camera_id).get("name")
self._name = f"{MANUFACTURER} {self._device_name}"
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._is_on = False
self._unique_id = f"{self._id}-light"
self._attr_unique_id = f"{self._id}-light"
async def async_added_to_hass(self) -> None:
"""Entity created."""
@ -126,7 +126,7 @@ class NetatmoLight(NetatmoBase, LightEntity):
async def async_turn_on(self, **kwargs):
"""Turn camera floodlight on."""
_LOGGER.debug("Turn camera '%s' on", self._name)
_LOGGER.debug("Turn camera '%s' on", self._attr_name)
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,
@ -135,7 +135,7 @@ class NetatmoLight(NetatmoBase, LightEntity):
async def async_turn_off(self, **kwargs):
"""Turn camera floodlight into auto mode."""
_LOGGER.debug("Turn camera '%s' to auto mode", self._name)
_LOGGER.debug("Turn camera '%s' to auto mode", self._attr_name)
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,

View File

@ -1,10 +1,18 @@
"""Base class for Netatmo entities."""
from __future__ import annotations
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers.entity import Entity
from .const import DATA_DEVICE_IDS, DOMAIN, MANUFACTURER, MODELS, SIGNAL_NAME
from .const import (
DATA_DEVICE_IDS,
DEFAULT_ATTRIBUTION,
DOMAIN,
MANUFACTURER,
MODELS,
SIGNAL_NAME,
)
from .data_handler import PUBLICDATA_DATA_CLASS_NAME, NetatmoDataHandler
@ -20,8 +28,9 @@ class NetatmoBase(Entity):
self._device_name = None
self._id = None
self._model = None
self._name = None
self._unique_id = None
self._attr_name = None
self._attr_unique_id = None
self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION}
async def async_added_to_hass(self) -> None:
"""Entity created."""
@ -84,16 +93,6 @@ class NetatmoBase(Entity):
"""Return data for this entity."""
return self.data_handler.data[self._data_classes[0]["name"]]
@property
def unique_id(self):
"""Return the unique ID of this entity."""
return self._unique_id
@property
def name(self):
"""Return the name of this entity."""
return self._name
@property
def device_info(self):
"""Return the device info for the sensor."""

View File

@ -352,56 +352,30 @@ class NetatmoSensor(NetatmoBase, SensorEntity):
f"{module_info.get('module_name', device['type'])}"
)
self._name = (
self._attr_name = (
f"{MANUFACTURER} {self._device_name} {SENSOR_TYPES[sensor_type][0]}"
)
self.type = sensor_type
self._state = None
self._device_class = SENSOR_TYPES[self.type][4]
self._icon = SENSOR_TYPES[self.type][3]
self._unit_of_measurement = SENSOR_TYPES[self.type][2]
self._attr_device_class = SENSOR_TYPES[self.type][4]
self._attr_icon = SENSOR_TYPES[self.type][3]
self._attr_unit_of_measurement = SENSOR_TYPES[self.type][2]
self._model = device["type"]
self._unique_id = f"{self._id}-{self.type}"
self._enabled_default = SENSOR_TYPES[self.type][5]
@property
def icon(self):
"""Icon to use in the frontend, if any."""
return self._icon
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
self._attr_unique_id = f"{self._id}-{self.type}"
self._attr_entity_registry_enabled_default = SENSOR_TYPES[self.type][5]
@property
def available(self):
"""Return entity availability."""
return self._state is not None
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self._enabled_default
return self._attr_state is not None
@callback
def async_update_callback(self):
"""Update the entity's state."""
if self._data is None:
if self._state is None:
if self._attr_state is None:
return
_LOGGER.warning("No data from update")
self._state = None
self._attr_state = None
return
data = self._data.get_last_data(station_id=self._station_id, exclude=3600).get(
@ -409,36 +383,36 @@ class NetatmoSensor(NetatmoBase, SensorEntity):
)
if data is None:
if self._state:
if self._attr_state:
_LOGGER.debug(
"No data found for %s - %s (%s)",
self.name,
self._device_name,
self._id,
)
self._state = None
self._attr_state = None
return
try:
state = data[SENSOR_TYPES[self.type][1]]
if self.type in {"temperature", "pressure", "sum_rain_1"}:
self._state = round(state, 1)
self._attr_state = round(state, 1)
elif self.type in {"windangle_value", "gustangle_value"}:
self._state = fix_angle(state)
self._attr_state = fix_angle(state)
elif self.type in {"windangle", "gustangle"}:
self._state = process_angle(fix_angle(state))
self._attr_state = process_angle(fix_angle(state))
elif self.type == "rf_status":
self._state = process_rf(state)
self._attr_state = process_rf(state)
elif self.type == "wifi_status":
self._state = process_wifi(state)
self._attr_state = process_wifi(state)
elif self.type == "health_idx":
self._state = process_health(state)
self._attr_state = process_health(state)
else:
self._state = state
self._attr_state = state
except KeyError:
if self._state:
if self._attr_state:
_LOGGER.debug("No %s data found for %s", self.type, self._device_name)
self._state = None
self._attr_state = None
return
self.async_write_ha_state()
@ -550,50 +524,22 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
self._area_name = area.area_name
self._id = self._area_name
self._device_name = f"{self._area_name}"
self._name = f"{MANUFACTURER} {self._device_name} {SENSOR_TYPES[self.type][0]}"
self._state = None
self._device_class = SENSOR_TYPES[self.type][4]
self._icon = SENSOR_TYPES[self.type][3]
self._unit_of_measurement = SENSOR_TYPES[self.type][2]
self._attr_name = (
f"{MANUFACTURER} {self._device_name} {SENSOR_TYPES[self.type][0]}"
)
self._attr_device_class = SENSOR_TYPES[self.type][4]
self._attr_icon = SENSOR_TYPES[self.type][3]
self._attr_unit_of_measurement = SENSOR_TYPES[self.type][2]
self._show_on_map = area.show_on_map
self._unique_id = f"{self._device_name.replace(' ', '-')}-{self.type}"
self._attr_unique_id = f"{self._device_name.replace(' ', '-')}-{self.type}"
self._model = PUBLIC
@property
def icon(self):
"""Icon to use in the frontend."""
return self._icon
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
@property
def extra_state_attributes(self):
"""Return the attributes of the device."""
attrs = {}
if self._show_on_map:
attrs[ATTR_LATITUDE] = (self.area.lat_ne + self.area.lat_sw) / 2
attrs[ATTR_LONGITUDE] = (self.area.lon_ne + self.area.lon_sw) / 2
return attrs
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity."""
return self._unit_of_measurement
@property
def available(self):
"""Return True if entity is available."""
return self._state is not None
self._attr_extra_state_attributes.update(
{
ATTR_LATITUDE: (self.area.lat_ne + self.area.lat_sw) / 2,
ATTR_LONGITUDE: (self.area.lon_ne + self.area.lon_sw) / 2,
}
)
@property
def _data(self):
@ -668,18 +614,19 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
data = self._data.get_latest_gust_strengths()
if data is None:
if self._state is None:
if self._attr_state is None:
return
_LOGGER.debug(
"No station provides %s data in the area %s", self.type, self._area_name
)
self._state = None
self._attr_state = None
return
if values := [x for x in data.values() if x is not None]:
if self._mode == "avg":
self._state = round(sum(values) / len(values), 1)
self._attr_state = round(sum(values) / len(values), 1)
elif self._mode == "max":
self._state = max(values)
self._attr_state = max(values)
self._attr_available = self._attr_state is not None
self.async_write_ha_state()