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

Add meater cook sensors (#70669)

Browse source
This commit is contained in:
Erik Montnemery 2022-04-26 18:27:06 +02:00 • committed by GitHub
parent cf5f0a415c
commit 90ef6e209c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 143 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
homeassistant/components/meater/__init__.py
View file

@ -9,6 +9,7 @@ from meater import (
ServiceUnavailableError,
TooManyRequestsError,
)
from meater.MeaterApi import MeaterProbe
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
@ -42,13 +43,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
_LOGGER.error("Unable to authenticate with the Meater API: %s", err)
return False
async def async_update_data():
async def async_update_data() -> dict[str, MeaterProbe]:
"""Fetch data from API endpoint."""
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with async_timeout.timeout(10):
devices = await meater_api.get_all_devices()
devices: list[MeaterProbe] = await meater_api.get_all_devices()
except AuthenticationError as err:
raise UpdateFailed("The API call wasn't authenticated") from err
except TooManyRequestsError as err:
@ -56,7 +57,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"Too many requests have been made to the API, rate limiting is in place"
) from err
return devices
return {device.id: device for device in devices}
coordinator = DataUpdateCoordinator(
hass,

179
homeassistant/components/meater/sensor.py
View file

@ -1,7 +1,18 @@
"""The Meater Temperature Probe integration."""
from enum import Enum
from __future__ import annotations
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from meater.MeaterApi import MeaterProbe
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import TEMP_CELSIUS
from homeassistant.core import HomeAssistant, callback
@ -10,10 +21,112 @@ from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.util import dt as dt_util
from .const import DOMAIN
@dataclass
class MeaterSensorEntityDescription(SensorEntityDescription):
"""Describes meater sensor entity."""
available: Callable[
[MeaterProbe | None], bool | type[NotImplementedError]
] = lambda x: NotImplementedError
value: Callable[
[MeaterProbe], datetime | float | str | None | type[NotImplementedError]
] = lambda x: NotImplementedError
def _elapsed_time_to_timestamp(probe: MeaterProbe) -> datetime | None:
"""Convert elapsed time to timestamp."""
if not probe.cook:
return None
return dt_util.utcnow() - timedelta(seconds=probe.cook.time_elapsed)
def _remaining_time_to_timestamp(probe: MeaterProbe) -> datetime | None:
"""Convert remaining time to timestamp."""
if not probe.cook or probe.cook.time_remaining < 0:
return None
return dt_util.utcnow() + timedelta(probe.cook.time_remaining)
SENSOR_TYPES = (
# Ambient temperature
MeaterSensorEntityDescription(
key="ambient",
device_class=SensorDeviceClass.TEMPERATURE,
name="Ambient",
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
available=lambda probe: probe is not None,
value=lambda probe: probe.ambient_temperature,
),
# Internal temperature (probe tip)
MeaterSensorEntityDescription(
key="internal",
device_class=SensorDeviceClass.TEMPERATURE,
name="Internal",
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
available=lambda probe: probe is not None,
value=lambda probe: probe.internal_temperature,
),
# Name of selected meat in user language or user given custom name
MeaterSensorEntityDescription(
key="cook_name",
name="Cooking",
available=lambda probe: probe is not None and probe.cook is not None,
value=lambda probe: probe.cook.name if probe.cook else None,
),
# One of Not Started, Configured, Started, Ready For Resting, Resting,
# Slightly Underdone, Finished, Slightly Overdone, OVERCOOK!. Not translated.
MeaterSensorEntityDescription(
key="cook_state",
name="Cook state",
available=lambda probe: probe is not None and probe.cook is not None,
value=lambda probe: probe.cook.state if probe.cook else None,
),
# Target temperature
MeaterSensorEntityDescription(
key="cook_target_temp",
device_class=SensorDeviceClass.TEMPERATURE,
name="Target",
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
available=lambda probe: probe is not None and probe.cook is not None,
value=lambda probe: probe.cook.target_temperature if probe.cook else None,
),
# Peak temperature
MeaterSensorEntityDescription(
key="cook_peak_temp",
device_class=SensorDeviceClass.TEMPERATURE,
name="Peak",
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
available=lambda probe: probe is not None and probe.cook is not None,
value=lambda probe: probe.cook.peak_temperature if probe.cook else None,
),
# Time since the start of cook in seconds. Default: 0.
MeaterSensorEntityDescription(
key="cook_time_remaining",
device_class=SensorDeviceClass.TIMESTAMP,
name="Remaining time",
available=lambda probe: probe is not None and probe.cook is not None,
value=_remaining_time_to_timestamp,
),
# Remaining time in seconds. When unknown/calculating default is used. Default: -1
MeaterSensorEntityDescription(
key="cook_time_elapsed",
device_class=SensorDeviceClass.TIMESTAMP,
name="Elapsed time",
available=lambda probe: probe is not None and probe.cook is not None,
value=_elapsed_time_to_timestamp,
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
@ -34,20 +147,16 @@ async def async_setup_entry(
# Add entities for temperature probes which we've not yet seen
for dev in devices:
if dev.id in known_probes:
if dev in known_probes:
continue
entities.append(
MeaterProbeTemperature(
coordinator, dev.id, TemperatureMeasurement.Internal
)
entities.extend(
[
MeaterProbeTemperature(coordinator, dev, sensor_description)
for sensor_description in SENSOR_TYPES
]
)
entities.append(
MeaterProbeTemperature(
coordinator, dev.id, TemperatureMeasurement.Ambient
)
)
known_probes.add(dev.id)
known_probes.add(dev)
async_add_entities(entities)
@ -57,16 +166,21 @@ async def async_setup_entry(
coordinator.async_add_listener(async_update_data)
class MeaterProbeTemperature(SensorEntity, CoordinatorEntity):
class MeaterProbeTemperature(
SensorEntity, CoordinatorEntity[DataUpdateCoordinator[dict[str, MeaterProbe]]]
):
"""Meater Temperature Sensor Entity."""
_attr_device_class = SensorDeviceClass.TEMPERATURE
_attr_native_unit_of_measurement = TEMP_CELSIUS
entity_description: MeaterSensorEntityDescription
def __init__(self, coordinator, device_id, temperature_reading_type):
def __init__(
self, coordinator, device_id, description: MeaterSensorEntityDescription
) -> None:
"""Initialise the sensor."""
super().__init__(coordinator)
self._attr_name = f"Meater Probe {temperature_reading_type.name}"
self._attr_name = f"Meater Probe {description.name}"
self._attr_device_info = {
"identifiers": {
# Serial numbers are unique identifiers within a specific domain
@ -76,41 +190,26 @@ class MeaterProbeTemperature(SensorEntity, CoordinatorEntity):
"model": "Meater Probe",
"name": f"Meater Probe {device_id}",
}
self._attr_unique_id = f"{device_id}-{temperature_reading_type}"
self._attr_unique_id = f"{device_id}-{description.key}"
self.device_id = device_id
self.temperature_reading_type = temperature_reading_type
self.entity_description = description
@property
def native_value(self):
"""Return the temperature of the probe."""
# First find the right probe in the collection
device = None
for dev in self.coordinator.data:
if dev.id == self.device_id:
device = dev
if device is None:
if not (device := self.coordinator.data.get(self.device_id)):
return None
if TemperatureMeasurement.Internal == self.temperature_reading_type:
return device.internal_temperature
# Not an internal temperature, must be ambient
return device.ambient_temperature
return self.entity_description.value(device)
@property
def available(self):
"""Return if entity is available."""
# See if the device was returned from the API. If not, it's offline
return self.coordinator.last_update_success and any(
self.device_id == device.id for device in self.coordinator.data
return (
self.coordinator.last_update_success
and self.entity_description.available(
self.coordinator.data.get(self.device_id)
)
)
class TemperatureMeasurement(Enum):
"""Enumeration of possible temperature readings from the probe."""
Internal = 1
Ambient = 2