"""Support for Xiaomi Mi Flora BLE plant sensor."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
import btlewrap
from btlewrap import BluetoothBackendException
from miflora import miflora_poller
import voluptuous as vol
from homeassistant.components.sensor import (
PLATFORM_SCHEMA,
STATE_CLASS_MEASUREMENT,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import (
CONDUCTIVITY,
CONF_FORCE_UPDATE,
CONF_MAC,
CONF_MONITORED_CONDITIONS,
CONF_NAME,
CONF_SCAN_INTERVAL,
DEVICE_CLASS_BATTERY,
DEVICE_CLASS_ILLUMINANCE,
DEVICE_CLASS_TEMPERATURE,
EVENT_HOMEASSISTANT_START,
LIGHT_LUX,
PERCENTAGE,
TEMP_CELSIUS,
)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
try:
import bluepy.btle # noqa: F401 pylint: disable=unused-import
BACKEND = btlewrap.BluepyBackend
except ImportError:
BACKEND = btlewrap.GatttoolBackend
_LOGGER = logging.getLogger(__name__)
CONF_ADAPTER = "adapter"
CONF_MEDIAN = "median"
CONF_GO_UNAVAILABLE_TIMEOUT = "go_unavailable_timeout"
DEFAULT_ADAPTER = "hci0"
DEFAULT_FORCE_UPDATE = False
DEFAULT_MEDIAN = 3
DEFAULT_NAME = "Mi Flora"
DEFAULT_GO_UNAVAILABLE_TIMEOUT = timedelta(seconds=7200)
SCAN_INTERVAL = timedelta(seconds=1200)
ATTR_LAST_SUCCESSFUL_UPDATE = "last_successful_update"
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="temperature",
name="Temperature",
native_unit_of_measurement=TEMP_CELSIUS,
device_class=DEVICE_CLASS_TEMPERATURE,
),
SensorEntityDescription(
key="light",
name="Light intensity",
native_unit_of_measurement=LIGHT_LUX,
device_class=DEVICE_CLASS_ILLUMINANCE,
),
SensorEntityDescription(
key="moisture",
name="Moisture",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:water-percent",
),
SensorEntityDescription(
key="conductivity",
name="Conductivity",
native_unit_of_measurement=CONDUCTIVITY,
icon="mdi:flash-circle",
),
SensorEntityDescription(
key="battery",
name="Battery",
native_unit_of_measurement=PERCENTAGE,
device_class=DEVICE_CLASS_BATTERY,
),
)
SENSOR_KEYS: list[str] = [desc.key for desc in SENSOR_TYPES]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_MAC): cv.string,
vol.Optional(CONF_MONITORED_CONDITIONS, default=SENSOR_KEYS): vol.All(
cv.ensure_list, [vol.In(SENSOR_KEYS)]
),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MEDIAN, default=DEFAULT_MEDIAN): cv.positive_int,
vol.Optional(CONF_FORCE_UPDATE, default=DEFAULT_FORCE_UPDATE): cv.boolean,
vol.Optional(CONF_ADAPTER, default=DEFAULT_ADAPTER): cv.string,
vol.Optional(
CONF_GO_UNAVAILABLE_TIMEOUT, default=DEFAULT_GO_UNAVAILABLE_TIMEOUT
): cv.time_period,
}
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the MiFlora sensor."""
backend = BACKEND
_LOGGER.debug("Miflora is using %s backend", backend.__name__)
cache = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL).total_seconds()
poller = miflora_poller.MiFloraPoller(
config[CONF_MAC],
cache_timeout=cache,
adapter=config[CONF_ADAPTER],
backend=backend,
)
force_update = config[CONF_FORCE_UPDATE]
median = config[CONF_MEDIAN]
go_unavailable_timeout = config[CONF_GO_UNAVAILABLE_TIMEOUT]
prefix = config[CONF_NAME]
monitored_conditions = config[CONF_MONITORED_CONDITIONS]
entities = [
MiFloraSensor(
description,
poller,
prefix,
force_update,
median,
go_unavailable_timeout,
)
for description in SENSOR_TYPES
if description.key in monitored_conditions
]
async_add_entities(entities)
class MiFloraSensor(SensorEntity):
"""Implementing the MiFlora sensor."""
_attr_state_class = STATE_CLASS_MEASUREMENT
def __init__(
self,
description: SensorEntityDescription,
poller,
prefix,
force_update,
median,
go_unavailable_timeout,
):
"""Initialize the sensor."""
self.entity_description = description
self.poller = poller
self.data: list[Any] = []
if prefix:
self._attr_name = f"{prefix} {description.name}"
self._attr_force_update = force_update
self.go_unavailable_timeout = go_unavailable_timeout
self.last_successful_update = dt_util.utc_from_timestamp(0)
# Median is used to filter out outliers. median of 3 will filter
# single outliers, while median of 5 will filter double outliers
# Use median_count = 1 if no filtering is required.
self.median_count = median
async def async_added_to_hass(self):
"""Set initial state."""
@callback
def on_startup(_):
self.async_schedule_update_ha_state(True)
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, on_startup)
@property
def available(self):
"""Return True if did update since 2h."""
return self.last_successful_update > (
dt_util.utcnow() - self.go_unavailable_timeout
)
@property
def extra_state_attributes(self):
"""Return the state attributes of the device."""
return {ATTR_LAST_SUCCESSFUL_UPDATE: self.last_successful_update}
def update(self):
"""
Update current conditions.
This uses a rolling median over 3 values to filter out outliers.
"""
try:
_LOGGER.debug("Polling data for %s", self.name)
data = self.poller.parameter_value(self.entity_description.key)
except (OSError, BluetoothBackendException) as err:
_LOGGER.info("Polling error %s: %s", type(err).__name__, err)
return
if data is not None:
_LOGGER.debug("%s = %s", self.name, data)
self.data.append(data)
self.last_successful_update = dt_util.utcnow()
else:
_LOGGER.info("Did not receive any data from Mi Flora sensor %s", self.name)
# Remove old data from median list or set sensor value to None
# if no data is available anymore
if self.data:
self.data = self.data[1:]
else:
self._attr_native_value = None
return
_LOGGER.debug("Data collected: %s", self.data)
if len(self.data) > self.median_count:
self.data = self.data[1:]
if len(self.data) == self.median_count:
median = sorted(self.data)[int((self.median_count - 1) / 2)]
_LOGGER.debug("Median is: %s", median)
self._attr_native_value = median
elif self._attr_native_value is None:
_LOGGER.debug("Set initial state")
self._attr_native_value = self.data[0]
else:
_LOGGER.debug("Not yet enough data for median calculation")