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

Netatmo refactor to use pyatmo 7.0.1 (#73482) (#78523)

Browse source 
Co-authored-by: Robert Svensson <Kane610@users.noreply.github.com>
This commit is contained in:
Tobias Sauerwein 2022-09-26 03:55:58 +02:00 • committed by GitHub
parent 1b17c83095
commit 81abeac83e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
37 changed files with 2915 additions and 1279 deletions
Download patch file Download diff file Expand all files Collapse all files

631
homeassistant/components/netatmo/sensor.py
View file

@ -1,9 +1,9 @@
"""Support for the Netatmo Weather Service."""
"""Support for the Netatmo sensors."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import NamedTuple, cast
from typing import cast
import pyatmo
@ -21,15 +21,15 @@ from homeassistant.const import (
DEGREE,
LENGTH_MILLIMETERS,
PERCENTAGE,
POWER_WATT,
PRESSURE_MBAR,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
SOUND_PRESSURE_DB,
SPEED_KILOMETERS_PER_HOUR,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import async_entries_for_config_entry
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
@ -38,20 +38,18 @@ from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
CONF_URL_ENERGY,
CONF_URL_WEATHER,
CONF_WEATHER_AREAS,
DATA_HANDLER,
DOMAIN,
NETATMO_CREATE_BATTERY,
NETATMO_CREATE_ROOM_SENSOR,
NETATMO_CREATE_SENSOR,
NETATMO_CREATE_WEATHER_SENSOR,
SIGNAL_NAME,
TYPE_WEATHER,
)
from .data_handler import (
HOMECOACH_DATA_CLASS_NAME,
PUBLICDATA_DATA_CLASS_NAME,
WEATHERSTATION_DATA_CLASS_NAME,
NetatmoDataHandler,
NetatmoDevice,
)
from .data_handler import HOME, PUBLIC, NetatmoDataHandler, NetatmoDevice, NetatmoRoom
from .helper import NetatmoArea
from .netatmo_entity_base import NetatmoBase
@ -62,10 +60,12 @@ SUPPORTED_PUBLIC_SENSOR_TYPES: tuple[str, ...] = (
"pressure",
"humidity",
"rain",
"windstrength",
"guststrength",
"wind_strength",
"gust_strength",
"sum_rain_1",
"sum_rain_24",
"wind_angle",
"gust_angle",
)
@ -85,7 +85,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="temperature",
name="Temperature",
netatmo_name="Temperature",
netatmo_name="temperature",
entity_registry_enabled_default=True,
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
@ -101,7 +101,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="co2",
name="CO2",
netatmo_name="CO2",
netatmo_name="co2",
native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
entity_registry_enabled_default=True,
state_class=SensorStateClass.MEASUREMENT,
@ -110,7 +110,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="pressure",
name="Pressure",
netatmo_name="Pressure",
netatmo_name="pressure",
entity_registry_enabled_default=True,
native_unit_of_measurement=PRESSURE_MBAR,
state_class=SensorStateClass.MEASUREMENT,
@ -126,7 +126,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="noise",
name="Noise",
netatmo_name="Noise",
netatmo_name="noise",
entity_registry_enabled_default=True,
native_unit_of_measurement=SOUND_PRESSURE_DB,
icon="mdi:volume-high",
@ -135,7 +135,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="humidity",
name="Humidity",
netatmo_name="Humidity",
netatmo_name="humidity",
entity_registry_enabled_default=True,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
@ -144,7 +144,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="rain",
name="Rain",
netatmo_name="Rain",
netatmo_name="rain",
entity_registry_enabled_default=True,
native_unit_of_measurement=LENGTH_MILLIMETERS,
state_class=SensorStateClass.MEASUREMENT,
@ -156,7 +156,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
netatmo_name="sum_rain_1",
entity_registry_enabled_default=False,
native_unit_of_measurement=LENGTH_MILLIMETERS,
state_class=SensorStateClass.TOTAL_INCREASING,
state_class=SensorStateClass.TOTAL,
icon="mdi:weather-rainy",
),
NetatmoSensorEntityDescription(
@ -171,7 +171,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="battery_percent",
name="Battery Percent",
netatmo_name="battery_percent",
netatmo_name="battery",
entity_registry_enabled_default=True,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=PERCENTAGE,
@ -181,14 +181,14 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="windangle",
name="Direction",
netatmo_name="WindAngle",
netatmo_name="wind_direction",
entity_registry_enabled_default=True,
icon="mdi:compass-outline",
),
NetatmoSensorEntityDescription(
key="windangle_value",
name="Angle",
netatmo_name="WindAngle",
netatmo_name="wind_angle",
entity_registry_enabled_default=False,
native_unit_of_measurement=DEGREE,
icon="mdi:compass-outline",
@ -197,7 +197,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="windstrength",
name="Wind Strength",
netatmo_name="WindStrength",
netatmo_name="wind_strength",
entity_registry_enabled_default=True,
native_unit_of_measurement=SPEED_KILOMETERS_PER_HOUR,
icon="mdi:weather-windy",
@ -206,14 +206,14 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="gustangle",
name="Gust Direction",
netatmo_name="GustAngle",
netatmo_name="gust_direction",
entity_registry_enabled_default=False,
icon="mdi:compass-outline",
),
NetatmoSensorEntityDescription(
key="gustangle_value",
name="Gust Angle",
netatmo_name="GustAngle",
netatmo_name="gust_angle",
entity_registry_enabled_default=False,
native_unit_of_measurement=DEGREE,
icon="mdi:compass-outline",
@ -222,7 +222,7 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="guststrength",
name="Gust Strength",
netatmo_name="GustStrength",
netatmo_name="gust_strength",
entity_registry_enabled_default=False,
native_unit_of_measurement=SPEED_KILOMETERS_PER_HOUR,
icon="mdi:weather-windy",
@ -239,39 +239,19 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
NetatmoSensorEntityDescription(
key="rf_status",
name="Radio",
netatmo_name="rf_status",
netatmo_name="rf_strength",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
icon="mdi:signal",
),
NetatmoSensorEntityDescription(
key="rf_status_lvl",
name="Radio Level",
netatmo_name="rf_status",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
),
NetatmoSensorEntityDescription(
key="wifi_status",
name="Wifi",
netatmo_name="wifi_status",
netatmo_name="wifi_strength",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
icon="mdi:wifi",
),
NetatmoSensorEntityDescription(
key="wifi_status_lvl",
name="Wifi Level",
netatmo_name="wifi_status",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
),
NetatmoSensorEntityDescription(
key="health_idx",
name="Health",
@ -279,136 +259,110 @@ SENSOR_TYPES: tuple[NetatmoSensorEntityDescription, ...] = (
entity_registry_enabled_default=True,
icon="mdi:cloud",
),
NetatmoSensorEntityDescription(
key="power",
name="Power",
netatmo_name="power",
entity_registry_enabled_default=True,
native_unit_of_measurement=POWER_WATT,
state_class=SensorStateClass.TOTAL,
device_class=SensorDeviceClass.POWER,
),
)
SENSOR_TYPES_KEYS = [desc.key for desc in SENSOR_TYPES]
MODULE_TYPE_OUTDOOR = "NAModule1"
MODULE_TYPE_WIND = "NAModule2"
MODULE_TYPE_RAIN = "NAModule3"
MODULE_TYPE_INDOOR = "NAModule4"
class BatteryData(NamedTuple):
"""Metadata for a batter."""
full: int
high: int
medium: int
low: int
BATTERY_VALUES = {
MODULE_TYPE_WIND: BatteryData(
full=5590,
high=5180,
medium=4770,
low=4360,
),
MODULE_TYPE_RAIN: BatteryData(
full=5500,
high=5000,
medium=4500,
low=4000,
),
MODULE_TYPE_INDOOR: BatteryData(
full=5500,
high=5280,
medium=4920,
low=4560,
),
MODULE_TYPE_OUTDOOR: BatteryData(
full=5500,
high=5000,
medium=4500,
low=4000,
),
}
PUBLIC = "public"
BATTERY_SENSOR_DESCRIPTION = NetatmoSensorEntityDescription(
key="battery",
name="Battery Percent",
netatmo_name="battery",
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.BATTERY,
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Netatmo weather and homecoach platform."""
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
platform_not_ready = True
"""Set up the Netatmo sensor platform."""
async def find_entities(data_class_name: str) -> list:
"""Find all entities."""
all_module_infos = {}
data = data_handler.data
@callback
def _create_battery_entity(netatmo_device: NetatmoDevice) -> None:
if not hasattr(netatmo_device.device, "battery"):
return
entity = NetatmoClimateBatterySensor(netatmo_device)
async_add_entities([entity])
if data_class_name not in data:
return []
entry.async_on_unload(
async_dispatcher_connect(hass, NETATMO_CREATE_BATTERY, _create_battery_entity)
)
if data[data_class_name] is None:
return []
@callback
def _create_weather_sensor_entity(netatmo_device: NetatmoDevice) -> None:
async_add_entities(
NetatmoWeatherSensor(netatmo_device, description)
for description in SENSOR_TYPES
if description.netatmo_name in netatmo_device.device.features
)
data_class = data[data_class_name]
entry.async_on_unload(
async_dispatcher_connect(
hass, NETATMO_CREATE_WEATHER_SENSOR, _create_weather_sensor_entity
)
)
for station_id in data_class.stations:
for module_id in data_class.get_modules(station_id):
all_module_infos[module_id] = data_class.get_module(module_id)
all_module_infos[station_id] = data_class.get_station(station_id)
entities = []
for module in all_module_infos.values():
if "_id" not in module:
_LOGGER.debug("Skipping module %s", module.get("module_name"))
continue
conditions = [
c.lower()
for c in data_class.get_monitored_conditions(module_id=module["_id"])
if c.lower() in SENSOR_TYPES_KEYS
@callback
def _create_sensor_entity(netatmo_device: NetatmoDevice) -> None:
_LOGGER.debug(
"Adding %s sensor %s",
netatmo_device.device.device_category,
netatmo_device.device.name,
)
async_add_entities(
[
NetatmoSensor(netatmo_device, description)
for description in SENSOR_TYPES
if description.key in netatmo_device.device.features
]
for condition in conditions:
if f"{condition}_value" in SENSOR_TYPES_KEYS:
conditions.append(f"{condition}_value")
elif f"{condition}_lvl" in SENSOR_TYPES_KEYS:
conditions.append(f"{condition}_lvl")
)
entities.extend(
[
NetatmoSensor(data_handler, data_class_name, module, description)
for description in SENSOR_TYPES
if description.key in conditions
]
)
entry.async_on_unload(
async_dispatcher_connect(hass, NETATMO_CREATE_SENSOR, _create_sensor_entity)
)
_LOGGER.debug("Adding weather sensors %s", entities)
return entities
@callback
def _create_room_sensor_entity(netatmo_device: NetatmoRoom) -> None:
async_add_entities(
NetatmoRoomSensor(netatmo_device, description)
for description in SENSOR_TYPES
if description.key in netatmo_device.room.features
)
for data_class_name in (
WEATHERSTATION_DATA_CLASS_NAME,
HOMECOACH_DATA_CLASS_NAME,
):
data_class = data_handler.data.get(data_class_name)
if data_class and data_class.raw_data:
platform_not_ready = False
async_add_entities(await find_entities(data_class_name), True)
entry.async_on_unload(
async_dispatcher_connect(
hass, NETATMO_CREATE_ROOM_SENSOR, _create_room_sensor_entity
)
)
device_registry = dr.async_get(hass)
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
async def add_public_entities(update: bool = True) -> None:
"""Retrieve Netatmo public weather entities."""
entities = {
device.name: device.id
for device in dr.async_entries_for_config_entry(
for device in async_entries_for_config_entry(
device_registry, entry.entry_id
)
if device.model == "Public Weather stations"
if device.model == "Public Weather station"
}
new_entities = []
for area in [
NetatmoArea(**i) for i in entry.options.get(CONF_WEATHER_AREAS, {}).values()
]:
signal_name = f"{PUBLICDATA_DATA_CLASS_NAME}-{area.uuid}"
signal_name = f"{PUBLIC}-{area.uuid}"
if area.area_name in entities:
entities.pop(area.area_name)
@ -422,25 +376,21 @@ async def async_setup_entry(
continue
await data_handler.subscribe(
PUBLICDATA_DATA_CLASS_NAME,
PUBLIC,
signal_name,
None,
lat_ne=area.lat_ne,
lon_ne=area.lon_ne,
lat_sw=area.lat_sw,
lon_sw=area.lon_sw,
area_id=str(area.uuid),
)
data_class = data_handler.data.get(signal_name)
if data_class and data_class.raw_data:
nonlocal platform_not_ready
platform_not_ready = False
new_entities.extend(
[
NetatmoPublicSensor(data_handler, area, description)
for description in SENSOR_TYPES
if description.key in SUPPORTED_PUBLIC_SENSOR_TYPES
if description.netatmo_name in SUPPORTED_PUBLIC_SENSOR_TYPES
]
)
@ -454,68 +404,56 @@ async def async_setup_entry(
hass, f"signal-{DOMAIN}-public-update-{entry.entry_id}", add_public_entities
)
@callback
def _create_entity(netatmo_device: NetatmoDevice) -> None:
entity = NetatmoClimateBatterySensor(netatmo_device)
_LOGGER.debug("Adding climate battery sensor %s", entity)
async_add_entities([entity])
entry.async_on_unload(
async_dispatcher_connect(hass, NETATMO_CREATE_BATTERY, _create_entity)
)
await add_public_entities(False)
if platform_not_ready:
raise PlatformNotReady
class NetatmoWeatherSensor(NetatmoBase, SensorEntity):
"""Implementation of a Netatmo weather/home coach sensor."""
class NetatmoSensor(NetatmoBase, SensorEntity):
"""Implementation of a Netatmo sensor."""
_attr_has_entity_name = True
entity_description: NetatmoSensorEntityDescription
def __init__(
self,
data_handler: NetatmoDataHandler,
data_class_name: str,
module_info: dict,
netatmo_device: NetatmoDevice,
description: NetatmoSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(data_handler)
super().__init__(netatmo_device.data_handler)
self.entity_description = description
self._publishers.append({"name": data_class_name, SIGNAL_NAME: data_class_name})
self._module = netatmo_device.device
self._id = self._module.entity_id
self._station_id = (
self._module.bridge if self._module.bridge is not None else self._id
)
self._device_name = self._module.name
category = getattr(self._module.device_category, "name")
self._publishers.extend(
[
{
"name": category,
SIGNAL_NAME: category,
},
]
)
self._id = module_info["_id"]
self._station_id = module_info.get("main_device", self._id)
station = self._data.get_station(self._station_id)
if not (device := self._data.get_module(self._id)):
# Assume it's a station if module can't be found
device = station
if device["type"] in ("NHC", "NAMain"):
self._device_name = module_info["station_name"]
else:
self._device_name = (
f"{station['station_name']} "
f"{module_info.get('module_name', device['type'])}"
)
self._attr_name = f"{self._device_name} {description.name}"
self._model = device["type"]
self._netatmo_type = TYPE_WEATHER
self._attr_name = f"{description.name}"
self._model = self._module.device_type
self._config_url = CONF_URL_WEATHER
self._attr_unique_id = f"{self._id}-{description.key}"
@property
def _data(self) -> pyatmo.AsyncWeatherStationData:
"""Return data for this entity."""
return cast(
pyatmo.AsyncWeatherStationData,
self.data_handler.data[self._publishers[0]["name"]],
)
if hasattr(self._module, "place"):
place = cast(
pyatmo.modules.base_class.Place, getattr(self._module, "place")
)
if hasattr(place, "location") and place.location is not None:
self._attr_extra_state_attributes.update(
{
ATTR_LATITUDE: place.location.latitude,
ATTR_LONGITUDE: place.location.longitude,
}
)
@property
def available(self) -> bool:
@ -525,46 +463,25 @@ class NetatmoSensor(NetatmoBase, SensorEntity):
@callback
def async_update_callback(self) -> None:
"""Update the entity's state."""
data = self._data.get_last_data(station_id=self._station_id, exclude=3600).get(
self._id
)
if data is None:
if self.state:
_LOGGER.debug(
"No data found for %s - %s (%s)",
self.name,
self._device_name,
self._id,
)
self._attr_native_value = None
if (
state := getattr(self._module, self.entity_description.netatmo_name)
) is None:
return
try:
state = data[self.entity_description.netatmo_name]
if self.entity_description.key in {"temperature", "pressure", "sum_rain_1"}:
self._attr_native_value = round(state, 1)
elif self.entity_description.key in {"windangle_value", "gustangle_value"}:
self._attr_native_value = fix_angle(state)
elif self.entity_description.key in {"windangle", "gustangle"}:
self._attr_native_value = process_angle(fix_angle(state))
elif self.entity_description.key == "rf_status":
self._attr_native_value = process_rf(state)
elif self.entity_description.key == "wifi_status":
self._attr_native_value = process_wifi(state)
elif self.entity_description.key == "health_idx":
self._attr_native_value = process_health(state)
else:
self._attr_native_value = state
except KeyError:
if self.state:
_LOGGER.debug(
"No %s data found for %s",
self.entity_description.key,
self._device_name,
)
self._attr_native_value = None
return
if self.entity_description.netatmo_name in {
"temperature",
"pressure",
"sum_rain_1",
}:
self._attr_native_value = round(state, 1)
elif self.entity_description.netatmo_name == "rf_strength":
self._attr_native_value = process_rf(state)
elif self.entity_description.netatmo_name == "wifi_strength":
self._attr_native_value = process_wifi(state)
elif self.entity_description.netatmo_name == "health_idx":
self._attr_native_value = process_health(state)
else:
self._attr_native_value = state
self.async_write_ha_state()
@ -580,24 +497,25 @@ class NetatmoClimateBatterySensor(NetatmoBase, SensorEntity):
) -> None:
"""Initialize the sensor."""
super().__init__(netatmo_device.data_handler)
self.entity_description = NetatmoSensorEntityDescription(
key="battery_percent",
name="Battery Percent",
netatmo_name="battery_percent",
entity_registry_enabled_default=True,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.BATTERY,
self.entity_description = BATTERY_SENSOR_DESCRIPTION
self._module = cast(pyatmo.modules.NRV, netatmo_device.device)
self._id = netatmo_device.parent_id
self._publishers.extend(
[
{
"name": HOME,
"home_id": netatmo_device.device.home.entity_id,
SIGNAL_NAME: netatmo_device.signal_name,
},
]
)
self._module = netatmo_device.device
self._id = netatmo_device.parent_id
self._attr_name = f"{self._module.name} {self.entity_description.name}"
self._signal_name = netatmo_device.signal_name
self._room_id = self._module.room_id
self._model = getattr(self._module.device_type, "value")
self._config_url = CONF_URL_ENERGY
self._attr_unique_id = (
f"{self._id}-{self._module.entity_id}-{self.entity_description.key}"
@ -613,70 +531,54 @@ class NetatmoClimateBatterySensor(NetatmoBase, SensorEntity):
return
self._attr_available = True
self._attr_native_value = self._process_battery_state()
def _process_battery_state(self) -> int | None:
"""Construct room status."""
if battery_state := self._module.battery_state:
return process_battery_percentage(battery_state)
return None
self._attr_native_value = self._module.battery
def process_battery_percentage(data: str) -> int:
"""Process battery data and return percent (int) for display."""
mapping = {
"max": 100,
"full": 90,
"high": 75,
"medium": 50,
"low": 25,
"very low": 10,
}
return mapping[data]
class NetatmoSensor(NetatmoBase, SensorEntity):
"""Implementation of a Netatmo sensor."""
entity_description: NetatmoSensorEntityDescription
def fix_angle(angle: int) -> int:
"""Fix angle when value is negative."""
if angle < 0:
return 360 + angle
return angle
def __init__(
self,
netatmo_device: NetatmoDevice,
description: NetatmoSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(netatmo_device.data_handler)
self.entity_description = description
self._module = netatmo_device.device
self._id = self._module.entity_id
def process_angle(angle: int) -> str:
"""Process angle and return string for display."""
if angle >= 330:
return "N"
if angle >= 300:
return "NW"
if angle >= 240:
return "W"
if angle >= 210:
return "SW"
if angle >= 150:
return "S"
if angle >= 120:
return "SE"
if angle >= 60:
return "E"
if angle >= 30:
return "NE"
return "N"
self._publishers.extend(
[
{
"name": HOME,
"home_id": netatmo_device.device.home.entity_id,
SIGNAL_NAME: netatmo_device.signal_name,
},
]
)
self._attr_name = f"{self._module.name} {self.entity_description.name}"
self._room_id = self._module.room_id
self._model = getattr(self._module.device_type, "value")
self._config_url = CONF_URL_ENERGY
def process_battery(data: int, model: str) -> str:
"""Process battery data and return string for display."""
battery_data = BATTERY_VALUES[model]
self._attr_unique_id = (
f"{self._id}-{self._module.entity_id}-{self.entity_description.key}"
)
if data >= battery_data.full:
return "Full"
if data >= battery_data.high:
return "High"
if data >= battery_data.medium:
return "Medium"
if data >= battery_data.low:
return "Low"
return "Very Low"
@callback
def async_update_callback(self) -> None:
"""Update the entity's state."""
if (state := getattr(self._module, self.entity_description.key)) is None:
return
self._attr_native_value = state
self.async_write_ha_state()
def process_health(health: int) -> str:
@ -714,9 +616,57 @@ def process_wifi(strength: int) -> str:
return "Full"
class NetatmoRoomSensor(NetatmoBase, SensorEntity):
"""Implementation of a Netatmo room sensor."""
entity_description: NetatmoSensorEntityDescription
def __init__(
self,
netatmo_room: NetatmoRoom,
description: NetatmoSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(netatmo_room.data_handler)
self.entity_description = description
self._room = netatmo_room.room
self._id = self._room.entity_id
self._publishers.extend(
[
{
"name": HOME,
"home_id": netatmo_room.room.home.entity_id,
SIGNAL_NAME: netatmo_room.signal_name,
},
]
)
self._attr_name = f"{self._room.name} {self.entity_description.name}"
self._room_id = self._room.entity_id
self._model = f"{self._room.climate_type}"
self._config_url = CONF_URL_ENERGY
self._attr_unique_id = (
f"{self._id}-{self._room.entity_id}-{self.entity_description.key}"
)
@callback
def async_update_callback(self) -> None:
"""Update the entity's state."""
if (state := getattr(self._room, self.entity_description.key)) is None:
return
self._attr_native_value = state
self.async_write_ha_state()
class NetatmoPublicSensor(NetatmoBase, SensorEntity):
"""Represent a single sensor in a Netatmo."""
_attr_has_entity_name = True
entity_description: NetatmoSensorEntityDescription
def __init__(
@ -729,11 +679,10 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
super().__init__(data_handler)
self.entity_description = description
self._signal_name = f"{PUBLICDATA_DATA_CLASS_NAME}-{area.uuid}"
self._signal_name = f"{PUBLIC}-{area.uuid}"
self._publishers.append(
{
"name": PUBLICDATA_DATA_CLASS_NAME,
"name": PUBLIC,
"lat_ne": area.lat_ne,
"lon_ne": area.lon_ne,
"lat_sw": area.lat_sw,
@ -743,12 +692,14 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
}
)
self._station = data_handler.account.public_weather_areas[str(area.uuid)]
self.area = area
self._mode = area.mode
self._area_name = area.area_name
self._id = self._area_name
self._device_name = f"{self._area_name}"
self._attr_name = f"{self._device_name} {description.name}"
self._attr_name = f"{description.name}"
self._show_on_map = area.show_on_map
self._attr_unique_id = (
f"{self._device_name.replace(' ', '-')}-{description.key}"
@ -762,17 +713,12 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
}
)
@property
def _data(self) -> pyatmo.AsyncPublicData:
"""Return data for this entity."""
return cast(pyatmo.AsyncPublicData, self.data_handler.data[self._signal_name])
async def async_added_to_hass(self) -> None:
"""Entity created."""
await super().async_added_to_hass()
assert self.device_info and "name" in self.device_info
self.data_handler.config_entry.async_on_unload(
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"netatmo-config-{self.device_info['name']}",
@ -790,22 +736,11 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
)
self.area = area
self._signal_name = f"{PUBLICDATA_DATA_CLASS_NAME}-{area.uuid}"
self._publishers = [
{
"name": PUBLICDATA_DATA_CLASS_NAME,
"lat_ne": area.lat_ne,
"lon_ne": area.lon_ne,
"lat_sw": area.lat_sw,
"lon_sw": area.lon_sw,
"area_name": area.area_name,
SIGNAL_NAME: self._signal_name,
}
]
self._signal_name = f"{PUBLIC}-{area.uuid}"
self._mode = area.mode
self._show_on_map = area.show_on_map
await self.data_handler.subscribe(
PUBLICDATA_DATA_CLASS_NAME,
PUBLIC,
self._signal_name,
self.async_update_callback,
lat_ne=area.lat_ne,
@ -819,22 +754,26 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
"""Update the entity's state."""
data = None
if self.entity_description.key == "temperature":
data = self._data.get_latest_temperatures()
elif self.entity_description.key == "pressure":
data = self._data.get_latest_pressures()
elif self.entity_description.key == "humidity":
data = self._data.get_latest_humidities()
elif self.entity_description.key == "rain":
data = self._data.get_latest_rain()
elif self.entity_description.key == "sum_rain_1":
data = self._data.get_60_min_rain()
elif self.entity_description.key == "sum_rain_24":
data = self._data.get_24_h_rain()
elif self.entity_description.key == "windstrength":
data = self._data.get_latest_wind_strengths()
elif self.entity_description.key == "guststrength":
data = self._data.get_latest_gust_strengths()
if self.entity_description.netatmo_name == "temperature":
data = self._station.get_latest_temperatures()
elif self.entity_description.netatmo_name == "pressure":
data = self._station.get_latest_pressures()
elif self.entity_description.netatmo_name == "humidity":
data = self._station.get_latest_humidities()
elif self.entity_description.netatmo_name == "rain":
data = self._station.get_latest_rain()
elif self.entity_description.netatmo_name == "sum_rain_1":
data = self._station.get_60_min_rain()
elif self.entity_description.netatmo_name == "sum_rain_24":
data = self._station.get_24_h_rain()
elif self.entity_description.netatmo_name == "wind_strength":
data = self._station.get_latest_wind_strengths()
elif self.entity_description.netatmo_name == "gust_strength":
data = self._station.get_latest_gust_strengths()
elif self.entity_description.netatmo_name == "wind_angle":
data = self._station.get_latest_wind_angles()
elif self.entity_description.netatmo_name == "gust_angle":
data = self._station.get_latest_gust_angles()
if not data:
if self.available: