sensor.py

Go to the documentation of this file.

"""Support for VersaSense sensor peripheral."""
 
from __future__ import annotations
 
import logging
 
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
from . import DOMAIN
from .const import (
    KEY_CONSUMER,
    KEY_IDENTIFIER,
    KEY_MEASUREMENT,
    KEY_PARENT_MAC,
    KEY_PARENT_NAME,
    KEY_UNIT,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the sensor platform."""
    if discovery_info is None:
        return
 
    consumer = hass.data[DOMAIN][KEY_CONSUMER]
 
    sensor_list = []
 
    for entity_info in discovery_info.values():
        peripheral = hass.data[DOMAIN][entity_info[KEY_PARENT_MAC]][
            entity_info[KEY_IDENTIFIER]
        ]
        parent_name = entity_info[KEY_PARENT_NAME]
        unit = entity_info[KEY_UNIT]
        measurement = entity_info[KEY_MEASUREMENT]
 
        sensor_list.append(
            VSensor(peripheral, parent_name, unit, measurement, consumer)
        )
 
    async_add_entities(sensor_list)
 
 
class VSensor(SensorEntity):
    """Representation of a Sensor."""
 
    def __init__(self, peripheral, parent_name, unit, measurement, consumer):
        """Initialize the sensor."""
        self._state_state = None
        self._available_available = True
        self._name_name = f"{parent_name} {measurement}"
        self._parent_mac_parent_mac = peripheral.parentMac
        self._identifier_identifier = peripheral.identifier
        self._unit_unit = unit
        self._measurement_measurement = measurement
        self.consumerconsumer = consumer
 
    @property
    def unique_id(self):
        """Return the unique id of the sensor."""
        return f"{self._parent_mac}/{self._identifier}/{self._measurement}"
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state
 
    @property
    def native_unit_of_measurement(self):
        """Return the unit of measurement."""
        return self._unit_unit
 
    @property
    def available(self):
        """Return if the sensor is available."""
        return self._available_available
 
    async def async_update(self) -> None:
        """Fetch new state data for the sensor."""
        samples = await self.consumerconsumer.fetchPeripheralSample(
            None, self._identifier_identifier, self._parent_mac_parent_mac
        )
 
        if samples is not None:
            for sample in samples:
                if sample.measurement == self._measurement_measurement:
                    self._available_available = True
                    self._state_state = sample.value
                    break
        else:
            _LOGGER.error("Sample unavailable")
            self._available_available = False
            self._state_state = None