Vesync air quality (#72658)

This commit is contained in:
Ethan Madden 2022-05-30 13:13:53 -07:00 committed by GitHub
parent 7ecb527648
commit 8c16ac2e47
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 169 additions and 164 deletions

View file

@ -1,25 +1,119 @@
"""Support for power & energy sensors for VeSync outlets."""
from collections.abc import Callable
from dataclasses import dataclass
import logging
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ENERGY_KILO_WATT_HOUR, POWER_WATT
from homeassistant.const import (
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
ENERGY_KILO_WATT_HOUR,
PERCENTAGE,
POWER_WATT,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .common import VeSyncBaseEntity
from .const import DOMAIN, VS_DISCOVERY, VS_SENSORS
from .switch import DEV_TYPE_TO_HA
from .common import VeSyncBaseEntity, VeSyncDevice
from .const import DEV_TYPE_TO_HA, DOMAIN, SKU_TO_BASE_DEVICE, VS_DISCOVERY, VS_SENSORS
_LOGGER = logging.getLogger(__name__)
@dataclass
class VeSyncSensorEntityDescriptionMixin:
"""Mixin for required keys."""
value_fn: Callable[[VeSyncDevice], StateType]
@dataclass
class VeSyncSensorEntityDescription(
SensorEntityDescription, VeSyncSensorEntityDescriptionMixin
):
"""Describe VeSync sensor entity."""
exists_fn: Callable[[VeSyncDevice], bool] = lambda _: True
update_fn: Callable[[VeSyncDevice], None] = lambda _: None
def update_energy(device):
"""Update outlet details and energy usage."""
device.update()
device.update_energy()
def sku_supported(device, supported):
"""Get the base device of which a device is an instance."""
return SKU_TO_BASE_DEVICE.get(device.device_type) in supported
def ha_dev_type(device):
"""Get the homeassistant device_type for a given device."""
return DEV_TYPE_TO_HA.get(device.device_type)
FILTER_LIFE_SUPPORTED = ["LV-PUR131S", "Core200S", "Core300S", "Core400S", "Core600S"]
AIR_QUALITY_SUPPORTED = ["LV-PUR131S", "Core400S", "Core600S"]
PM25_SUPPORTED = ["Core400S", "Core600S"]
SENSORS: tuple[VeSyncSensorEntityDescription, ...] = (
VeSyncSensorEntityDescription(
key="filter-life",
name="Filter Life",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda device: device.details["filter_life"],
exists_fn=lambda device: sku_supported(device, FILTER_LIFE_SUPPORTED),
),
VeSyncSensorEntityDescription(
key="air-quality",
name="Air Quality",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: device.details["air_quality"],
exists_fn=lambda device: sku_supported(device, AIR_QUALITY_SUPPORTED),
),
VeSyncSensorEntityDescription(
key="pm25",
name="PM2.5",
device_class=SensorDeviceClass.PM25,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: device.details["air_quality_value"],
exists_fn=lambda device: sku_supported(device, PM25_SUPPORTED),
),
VeSyncSensorEntityDescription(
key="power",
name="current power",
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=POWER_WATT,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: device.details["power"],
update_fn=update_energy,
exists_fn=lambda device: ha_dev_type(device) == "outlet",
),
VeSyncSensorEntityDescription(
key="energy",
name="energy use today",
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=ENERGY_KILO_WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda device: device.details["energy"],
update_fn=update_energy,
exists_fn=lambda device: ha_dev_type(device) == "outlet",
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -44,107 +138,33 @@ def _setup_entities(devices, async_add_entities):
"""Check if device is online and add entity."""
entities = []
for dev in devices:
if DEV_TYPE_TO_HA.get(dev.device_type) != "outlet":
# Not an outlet that supports energy/power, so do not create sensor entities
continue
entities.append(VeSyncPowerSensor(dev))
entities.append(VeSyncEnergySensor(dev))
for description in SENSORS:
if description.exists_fn(dev):
entities.append(VeSyncSensorEntity(dev, description))
async_add_entities(entities, update_before_add=True)
class VeSyncSensorEntity(VeSyncBaseEntity, SensorEntity):
"""Representation of a sensor describing diagnostics of a VeSync outlet."""
"""Representation of a sensor describing a VeSync device."""
def __init__(self, plug):
entity_description: VeSyncSensorEntityDescription
def __init__(
self,
device: VeSyncDevice,
description: VeSyncSensorEntityDescription,
) -> None:
"""Initialize the VeSync outlet device."""
super().__init__(plug)
self.smartplug = plug
super().__init__(device)
self.entity_description = description
self._attr_name = f"{super().name} {description.name}"
self._attr_unique_id = f"{super().unique_id}-{description.key}"
@property
def entity_category(self):
"""Return the diagnostic entity category."""
return EntityCategory.DIAGNOSTIC
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.device)
class VeSyncPowerSensor(VeSyncSensorEntity):
"""Representation of current power use for a VeSync outlet."""
@property
def unique_id(self):
"""Return unique ID for power sensor on device."""
return f"{super().unique_id}-power"
@property
def name(self):
"""Return sensor name."""
return f"{super().name} current power"
@property
def device_class(self):
"""Return the power device class."""
return SensorDeviceClass.POWER
@property
def native_value(self):
"""Return the current power usage in W."""
return self.smartplug.power
@property
def native_unit_of_measurement(self):
"""Return the Watt unit of measurement."""
return POWER_WATT
@property
def state_class(self):
"""Return the measurement state class."""
return SensorStateClass.MEASUREMENT
def update(self):
"""Update outlet details and energy usage."""
self.smartplug.update()
self.smartplug.update_energy()
class VeSyncEnergySensor(VeSyncSensorEntity):
"""Representation of current day's energy use for a VeSync outlet."""
def __init__(self, plug):
"""Initialize the VeSync outlet device."""
super().__init__(plug)
self.smartplug = plug
@property
def unique_id(self):
"""Return unique ID for power sensor on device."""
return f"{super().unique_id}-energy"
@property
def name(self):
"""Return sensor name."""
return f"{super().name} energy use today"
@property
def device_class(self):
"""Return the energy device class."""
return SensorDeviceClass.ENERGY
@property
def native_value(self):
"""Return the today total energy usage in kWh."""
return self.smartplug.energy_today
@property
def native_unit_of_measurement(self):
"""Return the kWh unit of measurement."""
return ENERGY_KILO_WATT_HOUR
@property
def state_class(self):
"""Return the total_increasing state class."""
return SensorStateClass.TOTAL_INCREASING
def update(self):
"""Update outlet details and energy usage."""
self.smartplug.update()
self.smartplug.update_energy()
def update(self) -> None:
"""Run the update function defined for the sensor."""
return self.entity_description.update_fn(self.device)