mirror of
https://github.com/home-assistant/core.git
synced 2025-07-17 10:17:09 +00:00
Clean up eight_sleep code (#58508)
This commit is contained in:
parent
b60b38c6f6
commit
788a9bd9f7
@ -38,6 +38,7 @@ DOMAIN = "eight_sleep"
|
|||||||
HEAT_ENTITY = "heat"
|
HEAT_ENTITY = "heat"
|
||||||
USER_ENTITY = "user"
|
USER_ENTITY = "user"
|
||||||
|
|
||||||
|
|
||||||
HEAT_SCAN_INTERVAL = timedelta(seconds=60)
|
HEAT_SCAN_INTERVAL = timedelta(seconds=60)
|
||||||
USER_SCAN_INTERVAL = timedelta(seconds=300)
|
USER_SCAN_INTERVAL = timedelta(seconds=300)
|
||||||
|
|
||||||
@ -48,18 +49,9 @@ NAME_MAP = {
|
|||||||
"left_current_sleep": "Left Sleep Session",
|
"left_current_sleep": "Left Sleep Session",
|
||||||
"left_current_sleep_fitness": "Left Sleep Fitness",
|
"left_current_sleep_fitness": "Left Sleep Fitness",
|
||||||
"left_last_sleep": "Left Previous Sleep Session",
|
"left_last_sleep": "Left Previous Sleep Session",
|
||||||
"left_bed_state": "Left Bed State",
|
|
||||||
"left_presence": "Left Bed Presence",
|
|
||||||
"left_bed_temp": "Left Bed Temperature",
|
|
||||||
"left_sleep_stage": "Left Sleep Stage",
|
|
||||||
"right_current_sleep": "Right Sleep Session",
|
"right_current_sleep": "Right Sleep Session",
|
||||||
"right_current_sleep_fitness": "Right Sleep Fitness",
|
"right_current_sleep_fitness": "Right Sleep Fitness",
|
||||||
"right_last_sleep": "Right Previous Sleep Session",
|
"right_last_sleep": "Right Previous Sleep Session",
|
||||||
"right_bed_state": "Right Bed State",
|
|
||||||
"right_presence": "Right Bed Presence",
|
|
||||||
"right_bed_temp": "Right Bed Temperature",
|
|
||||||
"right_sleep_stage": "Right Sleep Stage",
|
|
||||||
"room_temp": "Room Temperature",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
SENSORS = [
|
SENSORS = [
|
||||||
@ -67,7 +59,7 @@ SENSORS = [
|
|||||||
"current_sleep_fitness",
|
"current_sleep_fitness",
|
||||||
"last_sleep",
|
"last_sleep",
|
||||||
"bed_state",
|
"bed_state",
|
||||||
"bed_temp",
|
"bed_temperature",
|
||||||
"sleep_stage",
|
"sleep_stage",
|
||||||
]
|
]
|
||||||
|
|
||||||
@ -104,6 +96,14 @@ CONFIG_SCHEMA = vol.Schema(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _get_device_unique_id(eight: EightSleep, user_obj: EightUser = None) -> str:
|
||||||
|
"""Get the device's unique ID."""
|
||||||
|
unique_id = eight.deviceid
|
||||||
|
if user_obj:
|
||||||
|
unique_id = f"{unique_id}.{user_obj.userid}.{user_obj.side}"
|
||||||
|
return unique_id
|
||||||
|
|
||||||
|
|
||||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||||
"""Set up the Eight Sleep component."""
|
"""Set up the Eight Sleep component."""
|
||||||
|
|
||||||
@ -143,11 +143,11 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|||||||
sensors = []
|
sensors = []
|
||||||
binary_sensors = []
|
binary_sensors = []
|
||||||
if eight.users:
|
if eight.users:
|
||||||
for obj in eight.users.values():
|
for user, obj in eight.users.items():
|
||||||
for sensor in SENSORS:
|
for sensor in SENSORS:
|
||||||
sensors.append(f"{obj.side}_{sensor}")
|
sensors.append((obj.side, sensor))
|
||||||
binary_sensors.append(f"{obj.side}_presence")
|
binary_sensors.append((obj.side, "bed_presence"))
|
||||||
sensors.append("room_temp")
|
sensors.append((None, "room_temperature"))
|
||||||
else:
|
else:
|
||||||
# No users, cannot continue
|
# No users, cannot continue
|
||||||
return False
|
return False
|
||||||
@ -173,9 +173,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|||||||
duration = params.pop(ATTR_HEAT_DURATION, 0)
|
duration = params.pop(ATTR_HEAT_DURATION, 0)
|
||||||
|
|
||||||
for sens in sensor:
|
for sens in sensor:
|
||||||
side = sens.split("_")[1]
|
side = sens[0]
|
||||||
userid = eight.fetch_userid(side)
|
userid = eight.fetch_userid(side)
|
||||||
usrobj: EightUser = eight.users[userid]
|
usrobj = eight.users[userid]
|
||||||
await usrobj.set_heating_level(target, duration)
|
await usrobj.set_heating_level(target, duration)
|
||||||
|
|
||||||
await heat_coordinator.async_request_refresh()
|
await heat_coordinator.async_request_refresh()
|
||||||
@ -199,9 +199,12 @@ class EightSleepHeatDataCoordinator(DataUpdateCoordinator):
|
|||||||
_LOGGER,
|
_LOGGER,
|
||||||
name=f"{DOMAIN}_heat",
|
name=f"{DOMAIN}_heat",
|
||||||
update_interval=HEAT_SCAN_INTERVAL,
|
update_interval=HEAT_SCAN_INTERVAL,
|
||||||
update_method=self.api.update_device_data,
|
update_method=self._async_update_data,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _async_update_data(self) -> None:
|
||||||
|
await self.api.update_device_data()
|
||||||
|
|
||||||
|
|
||||||
class EightSleepUserDataCoordinator(DataUpdateCoordinator):
|
class EightSleepUserDataCoordinator(DataUpdateCoordinator):
|
||||||
"""Class to retrieve user data from Eight Sleep."""
|
"""Class to retrieve user data from Eight Sleep."""
|
||||||
@ -214,14 +217,57 @@ class EightSleepUserDataCoordinator(DataUpdateCoordinator):
|
|||||||
_LOGGER,
|
_LOGGER,
|
||||||
name=f"{DOMAIN}_user",
|
name=f"{DOMAIN}_user",
|
||||||
update_interval=USER_SCAN_INTERVAL,
|
update_interval=USER_SCAN_INTERVAL,
|
||||||
update_method=self.api.update_user_data,
|
update_method=self._async_update_data,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _async_update_data(self) -> None:
|
||||||
|
await self.api.update_user_data()
|
||||||
|
|
||||||
class EightSleepEntity(CoordinatorEntity):
|
|
||||||
"""The Eight Sleep device entity."""
|
|
||||||
|
|
||||||
def __init__(self, coordinator: DataUpdateCoordinator, eight: EightSleep) -> None:
|
class EightSleepBaseEntity(CoordinatorEntity):
|
||||||
|
"""The base Eight Sleep entity class."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
name: str,
|
||||||
|
coordinator: EightSleepUserDataCoordinator | EightSleepHeatDataCoordinator,
|
||||||
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
|
sensor: str,
|
||||||
|
) -> None:
|
||||||
"""Initialize the data object."""
|
"""Initialize the data object."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator)
|
||||||
self._eight = eight
|
self._eight = eight
|
||||||
|
self._side = side
|
||||||
|
self._sensor = sensor
|
||||||
|
self._usrobj: EightUser = None
|
||||||
|
if self._side:
|
||||||
|
self._usrobj = self._eight.users[self._eight.fetch_userid(self._side)]
|
||||||
|
full_sensor_name = self._sensor
|
||||||
|
if self._side is not None:
|
||||||
|
full_sensor_name = f"{self._side}_{full_sensor_name}"
|
||||||
|
mapped_name = NAME_MAP.get(
|
||||||
|
full_sensor_name, full_sensor_name.replace("_", " ").title()
|
||||||
|
)
|
||||||
|
|
||||||
|
self._attr_name = f"{name} {mapped_name}"
|
||||||
|
self._attr_unique_id = (
|
||||||
|
f"{_get_device_unique_id(eight, self._usrobj)}.{self._sensor}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class EightSleepUserEntity(EightSleepBaseEntity):
|
||||||
|
"""The Eight Sleep user entity."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
name: str,
|
||||||
|
coordinator: EightSleepUserDataCoordinator,
|
||||||
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
|
sensor: str,
|
||||||
|
units: str,
|
||||||
|
) -> None:
|
||||||
|
"""Initialize the data object."""
|
||||||
|
super().__init__(name, coordinator, eight, side, sensor)
|
||||||
|
self._units = units
|
||||||
|
@ -1,25 +1,25 @@
|
|||||||
"""Support for Eight Sleep binary sensors."""
|
"""Support for Eight Sleep binary sensors."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from pyeight.eight import EightSleep
|
from pyeight.eight import EightSleep
|
||||||
from pyeight.user import EightUser
|
|
||||||
|
|
||||||
from homeassistant.components.binary_sensor import (
|
from homeassistant.components.binary_sensor import (
|
||||||
DEVICE_CLASS_OCCUPANCY,
|
DEVICE_CLASS_OCCUPANCY,
|
||||||
BinarySensorEntity,
|
BinarySensorEntity,
|
||||||
)
|
)
|
||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
||||||
|
|
||||||
from . import (
|
from . import (
|
||||||
CONF_BINARY_SENSORS,
|
CONF_BINARY_SENSORS,
|
||||||
DATA_API,
|
DATA_API,
|
||||||
DATA_EIGHT,
|
DATA_EIGHT,
|
||||||
DATA_HEAT,
|
DATA_HEAT,
|
||||||
NAME_MAP,
|
EightSleepBaseEntity,
|
||||||
EightSleepEntity,
|
EightSleepHeatDataCoordinator,
|
||||||
)
|
)
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -37,55 +37,40 @@ async def async_setup_platform(
|
|||||||
|
|
||||||
name = "Eight"
|
name = "Eight"
|
||||||
sensors = discovery_info[CONF_BINARY_SENSORS]
|
sensors = discovery_info[CONF_BINARY_SENSORS]
|
||||||
eight = hass.data[DATA_EIGHT][DATA_API]
|
eight: EightSleep = hass.data[DATA_EIGHT][DATA_API]
|
||||||
heat_coordinator = hass.data[DATA_EIGHT][DATA_HEAT]
|
heat_coordinator: EightSleepHeatDataCoordinator = hass.data[DATA_EIGHT][DATA_HEAT]
|
||||||
|
|
||||||
all_sensors = []
|
all_sensors = [
|
||||||
|
EightHeatSensor(name, heat_coordinator, eight, side, sensor)
|
||||||
|
for side, sensor in sensors
|
||||||
|
]
|
||||||
|
|
||||||
for sensor in sensors:
|
async_add_entities(all_sensors)
|
||||||
all_sensors.append(EightHeatSensor(name, heat_coordinator, eight, sensor))
|
|
||||||
|
|
||||||
async_add_entities(all_sensors, True)
|
|
||||||
|
|
||||||
|
|
||||||
class EightHeatSensor(EightSleepEntity, BinarySensorEntity):
|
class EightHeatSensor(EightSleepBaseEntity, BinarySensorEntity):
|
||||||
"""Representation of a Eight Sleep heat-based sensor."""
|
"""Representation of a Eight Sleep heat-based sensor."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
name: str,
|
name: str,
|
||||||
coordinator: DataUpdateCoordinator,
|
coordinator: EightSleepHeatDataCoordinator,
|
||||||
eight: EightSleep,
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
sensor: str,
|
sensor: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator, eight)
|
super().__init__(name, coordinator, eight, side, sensor)
|
||||||
|
|
||||||
self._sensor = sensor
|
|
||||||
self._mapped_name = NAME_MAP.get(self._sensor, self._sensor)
|
|
||||||
self._state = None
|
|
||||||
|
|
||||||
self._side = self._sensor.split("_")[0]
|
|
||||||
self._userid = self._eight.fetch_userid(self._side)
|
|
||||||
self._usrobj: EightUser = self._eight.users[self._userid]
|
|
||||||
|
|
||||||
self._attr_name = f"{name} {self._mapped_name}"
|
|
||||||
self._attr_device_class = DEVICE_CLASS_OCCUPANCY
|
self._attr_device_class = DEVICE_CLASS_OCCUPANCY
|
||||||
|
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Presence Sensor: %s, Side: %s, User: %s",
|
"Presence Sensor: %s, Side: %s, User: %s",
|
||||||
self._sensor,
|
self._sensor,
|
||||||
self._side,
|
self._side,
|
||||||
self._userid,
|
self._usrobj.userid,
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
"""Return true if the binary sensor is on."""
|
"""Return true if the binary sensor is on."""
|
||||||
return bool(self._state)
|
return bool(self._usrobj.bed_presence)
|
||||||
|
|
||||||
@callback
|
|
||||||
def _handle_coordinator_update(self) -> None:
|
|
||||||
"""Handle updated data from the coordinator."""
|
|
||||||
self._state = self._usrobj.bed_presence
|
|
||||||
super()._handle_coordinator_update()
|
|
||||||
|
@ -1,23 +1,16 @@
|
|||||||
"""Support for Eight Sleep sensors."""
|
"""Support for Eight Sleep sensors."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import Mapping
|
|
||||||
import logging
|
import logging
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from pyeight.eight import EightSleep
|
from pyeight.eight import EightSleep
|
||||||
from pyeight.user import EightUser
|
|
||||||
|
|
||||||
from homeassistant.components.sensor import SensorEntity
|
from homeassistant.components.sensor import SensorEntity
|
||||||
from homeassistant.const import (
|
from homeassistant.const import PERCENTAGE, TEMP_CELSIUS, TEMP_FAHRENHEIT
|
||||||
DEVICE_CLASS_TEMPERATURE,
|
from homeassistant.core import HomeAssistant
|
||||||
PERCENTAGE,
|
|
||||||
TEMP_CELSIUS,
|
|
||||||
TEMP_FAHRENHEIT,
|
|
||||||
)
|
|
||||||
from homeassistant.core import HomeAssistant, callback
|
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType
|
||||||
|
|
||||||
from . import (
|
from . import (
|
||||||
CONF_SENSORS,
|
CONF_SENSORS,
|
||||||
@ -25,10 +18,10 @@ from . import (
|
|||||||
DATA_EIGHT,
|
DATA_EIGHT,
|
||||||
DATA_HEAT,
|
DATA_HEAT,
|
||||||
DATA_USER,
|
DATA_USER,
|
||||||
NAME_MAP,
|
EightSleepBaseEntity,
|
||||||
EightSleepEntity,
|
|
||||||
EightSleepHeatDataCoordinator,
|
EightSleepHeatDataCoordinator,
|
||||||
EightSleepUserDataCoordinator,
|
EightSleepUserDataCoordinator,
|
||||||
|
EightSleepUserEntity,
|
||||||
)
|
)
|
||||||
|
|
||||||
ATTR_ROOM_TEMP = "Room Temperature"
|
ATTR_ROOM_TEMP = "Room Temperature"
|
||||||
@ -63,7 +56,7 @@ async def async_setup_platform(
|
|||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
config: ConfigType,
|
config: ConfigType,
|
||||||
async_add_entities: AddEntitiesCallback,
|
async_add_entities: AddEntitiesCallback,
|
||||||
discovery_info: DiscoveryInfoType = None,
|
discovery_info: dict[str, list[tuple[str, str]]] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the eight sleep sensors."""
|
"""Set up the eight sleep sensors."""
|
||||||
if discovery_info is None:
|
if discovery_info is None:
|
||||||
@ -71,7 +64,7 @@ async def async_setup_platform(
|
|||||||
|
|
||||||
name = "Eight"
|
name = "Eight"
|
||||||
sensors = discovery_info[CONF_SENSORS]
|
sensors = discovery_info[CONF_SENSORS]
|
||||||
eight = hass.data[DATA_EIGHT][DATA_API]
|
eight: EightSleep = hass.data[DATA_EIGHT][DATA_API]
|
||||||
heat_coordinator: EightSleepHeatDataCoordinator = hass.data[DATA_EIGHT][DATA_HEAT]
|
heat_coordinator: EightSleepHeatDataCoordinator = hass.data[DATA_EIGHT][DATA_HEAT]
|
||||||
user_coordinator: EightSleepUserDataCoordinator = hass.data[DATA_EIGHT][DATA_USER]
|
user_coordinator: EightSleepUserDataCoordinator = hass.data[DATA_EIGHT][DATA_USER]
|
||||||
|
|
||||||
@ -80,24 +73,26 @@ async def async_setup_platform(
|
|||||||
else:
|
else:
|
||||||
units = "us"
|
units = "us"
|
||||||
|
|
||||||
all_sensors: list[EightSleepEntity] = []
|
all_sensors: list[SensorEntity] = []
|
||||||
|
|
||||||
for sensor in sensors:
|
for side, sensor in sensors:
|
||||||
if "bed_state" in sensor:
|
if sensor == "bed_state":
|
||||||
all_sensors.append(EightHeatSensor(name, heat_coordinator, eight, sensor))
|
|
||||||
elif "room_temp" in sensor:
|
|
||||||
all_sensors.append(
|
all_sensors.append(
|
||||||
EightRoomSensor(name, user_coordinator, eight, sensor, units)
|
EightHeatSensor(name, heat_coordinator, eight, side, sensor)
|
||||||
|
)
|
||||||
|
elif sensor == "room_temperature":
|
||||||
|
all_sensors.append(
|
||||||
|
EightRoomSensor(name, user_coordinator, eight, side, sensor, units)
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
all_sensors.append(
|
all_sensors.append(
|
||||||
EightUserSensor(name, user_coordinator, eight, sensor, units)
|
EightUserSensor(name, user_coordinator, eight, side, sensor, units)
|
||||||
)
|
)
|
||||||
|
|
||||||
async_add_entities(all_sensors, True)
|
async_add_entities(all_sensors)
|
||||||
|
|
||||||
|
|
||||||
class EightHeatSensor(EightSleepEntity, SensorEntity):
|
class EightHeatSensor(EightSleepBaseEntity, SensorEntity):
|
||||||
"""Representation of an eight sleep heat-based sensor."""
|
"""Representation of an eight sleep heat-based sensor."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@ -105,51 +100,27 @@ class EightHeatSensor(EightSleepEntity, SensorEntity):
|
|||||||
name: str,
|
name: str,
|
||||||
coordinator: EightSleepHeatDataCoordinator,
|
coordinator: EightSleepHeatDataCoordinator,
|
||||||
eight: EightSleep,
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
sensor: str,
|
sensor: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator, eight)
|
super().__init__(name, coordinator, eight, side, sensor)
|
||||||
|
self._attr_native_unit_of_measurement = PERCENTAGE
|
||||||
self._sensor = sensor
|
|
||||||
self._mapped_name = NAME_MAP.get(self._sensor, self._sensor)
|
|
||||||
self._name = f"{name} {self._mapped_name}"
|
|
||||||
self._state = None
|
|
||||||
|
|
||||||
self._side = self._sensor.split("_")[0]
|
|
||||||
self._userid = self._eight.fetch_userid(self._side)
|
|
||||||
self._usrobj: EightUser = self._eight.users[self._userid]
|
|
||||||
|
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Heat Sensor: %s, Side: %s, User: %s",
|
"Heat Sensor: %s, Side: %s, User: %s",
|
||||||
self._sensor,
|
self._sensor,
|
||||||
self._side,
|
self._side,
|
||||||
self._userid,
|
self._usrobj.userid,
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self) -> str:
|
def native_value(self) -> int:
|
||||||
"""Return the name of the sensor, if any."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_value(self) -> str | None:
|
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self._state
|
return self._usrobj.heating_level
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_unit_of_measurement(self) -> str:
|
def extra_state_attributes(self) -> dict[str, Any]:
|
||||||
"""Return the unit the value is expressed in."""
|
|
||||||
return PERCENTAGE
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def _handle_coordinator_update(self) -> None:
|
|
||||||
"""Handle updated data from the coordinator."""
|
|
||||||
_LOGGER.debug("Updating Heat sensor: %s", self._sensor)
|
|
||||||
self._state = self._usrobj.heating_level
|
|
||||||
super()._handle_coordinator_update()
|
|
||||||
|
|
||||||
@property
|
|
||||||
def extra_state_attributes(self) -> Mapping[str, Any]:
|
|
||||||
"""Return device state attributes."""
|
"""Return device state attributes."""
|
||||||
return {
|
return {
|
||||||
ATTR_TARGET_HEAT: self._usrobj.target_heating_level,
|
ATTR_TARGET_HEAT: self._usrobj.target_heating_level,
|
||||||
@ -158,7 +129,17 @@ class EightHeatSensor(EightSleepEntity, SensorEntity):
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
class EightUserSensor(EightSleepEntity, SensorEntity):
|
def _get_breakdown_percent(
|
||||||
|
attr: dict[str, Any], key: str, denominator: int | float
|
||||||
|
) -> int | float:
|
||||||
|
"""Get a breakdown percent."""
|
||||||
|
try:
|
||||||
|
return round((attr["breakdown"][key] / denominator) * 100, 2)
|
||||||
|
except ZeroDivisionError:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
class EightUserSensor(EightSleepUserEntity, SensorEntity):
|
||||||
"""Representation of an eight sleep user-based sensor."""
|
"""Representation of an eight sleep user-based sensor."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@ -166,180 +147,138 @@ class EightUserSensor(EightSleepEntity, SensorEntity):
|
|||||||
name: str,
|
name: str,
|
||||||
coordinator: EightSleepUserDataCoordinator,
|
coordinator: EightSleepUserDataCoordinator,
|
||||||
eight: EightSleep,
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
sensor: str,
|
sensor: str,
|
||||||
units: str,
|
units: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator, eight)
|
super().__init__(name, coordinator, eight, side, sensor, units)
|
||||||
|
|
||||||
self._sensor = sensor
|
if self._sensor == "bed_temperature":
|
||||||
self._sensor_root = self._sensor.split("_", 1)[1]
|
self._attr_icon = "mdi:thermometer"
|
||||||
self._mapped_name = NAME_MAP.get(self._sensor, self._sensor)
|
|
||||||
self._name = f"{name} {self._mapped_name}"
|
|
||||||
self._state = None
|
|
||||||
self._attr = None
|
|
||||||
self._units = units
|
|
||||||
|
|
||||||
self._side = self._sensor.split("_", 1)[0]
|
|
||||||
self._userid = self._eight.fetch_userid(self._side)
|
|
||||||
self._usrobj: EightUser = self._eight.users[self._userid]
|
|
||||||
|
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"User Sensor: %s, Side: %s, User: %s",
|
"User Sensor: %s, Side: %s, User: %s",
|
||||||
self._sensor,
|
self._sensor,
|
||||||
self._side,
|
self._side,
|
||||||
self._userid,
|
self._usrobj.userid if self._usrobj else None,
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self) -> str:
|
def native_value(self) -> str | int | float | None:
|
||||||
"""Return the name of the sensor, if any."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_value(self) -> str | None:
|
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self._state
|
if "current" in self._sensor:
|
||||||
|
if "fitness" in self._sensor:
|
||||||
|
return self._usrobj.current_sleep_fitness_score
|
||||||
|
return self._usrobj.current_sleep_score
|
||||||
|
|
||||||
|
if "last" in self._sensor:
|
||||||
|
return self._usrobj.last_sleep_score
|
||||||
|
|
||||||
|
if self._sensor == "bed_temperature":
|
||||||
|
temp = self._usrobj.current_values["bed_temp"]
|
||||||
|
try:
|
||||||
|
if self._units == "si":
|
||||||
|
return round(temp, 2)
|
||||||
|
return round((temp * 1.8) + 32, 2)
|
||||||
|
except TypeError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
if self._sensor == "sleep_stage":
|
||||||
|
return self._usrobj.current_values["stage"]
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_unit_of_measurement(self) -> str | None:
|
def native_unit_of_measurement(self) -> str | None:
|
||||||
"""Return the unit the value is expressed in."""
|
"""Return the unit the value is expressed in."""
|
||||||
if (
|
if self._sensor in ("current_sleep", "last_sleep", "current_sleep_fitness"):
|
||||||
"current_sleep" in self._sensor
|
|
||||||
or "last_sleep" in self._sensor
|
|
||||||
or "current_sleep_fitness" in self._sensor
|
|
||||||
):
|
|
||||||
return "Score"
|
return "Score"
|
||||||
if "bed_temp" in self._sensor:
|
if self._sensor == "bed_temperature":
|
||||||
if self._units == "si":
|
if self._units == "si":
|
||||||
return TEMP_CELSIUS
|
return TEMP_CELSIUS
|
||||||
return TEMP_FAHRENHEIT
|
return TEMP_FAHRENHEIT
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
def _get_rounded_value(
|
||||||
def device_class(self) -> str | None:
|
self, attr: dict[str, Any], key: str, use_units: bool = True
|
||||||
"""Return the class of this device, from component DEVICE_CLASSES."""
|
) -> int | float | None:
|
||||||
if "bed_temp" in self._sensor:
|
"""Get rounded value based on units for given key."""
|
||||||
return DEVICE_CLASS_TEMPERATURE
|
try:
|
||||||
return None
|
if self._units == "si" or not use_units:
|
||||||
|
return round(attr["room_temp"], 2)
|
||||||
|
return round((attr["room_temp"] * 1.8) + 32, 2)
|
||||||
|
except TypeError:
|
||||||
|
return None
|
||||||
|
|
||||||
@callback
|
@property
|
||||||
def _handle_coordinator_update(self) -> None:
|
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||||
"""Handle updated data from the coordinator."""
|
"""Return device state attributes."""
|
||||||
_LOGGER.debug("Updating User sensor: %s", self._sensor)
|
attr = None
|
||||||
if "current" in self._sensor:
|
if "current" in self._sensor:
|
||||||
if "fitness" in self._sensor:
|
if "fitness" in self._sensor:
|
||||||
self._state = self._usrobj.current_sleep_fitness_score
|
attr = self._usrobj.current_fitness_values
|
||||||
self._attr = self._usrobj.current_fitness_values
|
|
||||||
else:
|
else:
|
||||||
self._state = self._usrobj.current_sleep_score
|
attr = self._usrobj.current_values
|
||||||
self._attr = self._usrobj.current_values
|
|
||||||
elif "last" in self._sensor:
|
elif "last" in self._sensor:
|
||||||
self._state = self._usrobj.last_sleep_score
|
attr = self._usrobj.last_values
|
||||||
self._attr = self._usrobj.last_values
|
|
||||||
elif "bed_temp" in self._sensor:
|
|
||||||
temp = self._usrobj.current_values["bed_temp"]
|
|
||||||
try:
|
|
||||||
if self._units == "si":
|
|
||||||
self._state = round(temp, 2)
|
|
||||||
else:
|
|
||||||
self._state = round((temp * 1.8) + 32, 2)
|
|
||||||
except TypeError:
|
|
||||||
self._state = None
|
|
||||||
elif "sleep_stage" in self._sensor:
|
|
||||||
self._state = self._usrobj.current_values["stage"]
|
|
||||||
|
|
||||||
super()._handle_coordinator_update()
|
if attr is None:
|
||||||
|
|
||||||
@property
|
|
||||||
def extra_state_attributes(self) -> Mapping[str, Any] | None:
|
|
||||||
"""Return device state attributes."""
|
|
||||||
if self._attr is None:
|
|
||||||
# Skip attributes if sensor type doesn't support
|
# Skip attributes if sensor type doesn't support
|
||||||
return None
|
return None
|
||||||
|
|
||||||
if "fitness" in self._sensor_root:
|
if "fitness" in self._sensor:
|
||||||
state_attr = {
|
state_attr = {
|
||||||
ATTR_FIT_DATE: self._attr["date"],
|
ATTR_FIT_DATE: attr["date"],
|
||||||
ATTR_FIT_DURATION_SCORE: self._attr["duration"],
|
ATTR_FIT_DURATION_SCORE: attr["duration"],
|
||||||
ATTR_FIT_ASLEEP_SCORE: self._attr["asleep"],
|
ATTR_FIT_ASLEEP_SCORE: attr["asleep"],
|
||||||
ATTR_FIT_OUT_SCORE: self._attr["out"],
|
ATTR_FIT_OUT_SCORE: attr["out"],
|
||||||
ATTR_FIT_WAKEUP_SCORE: self._attr["wakeup"],
|
ATTR_FIT_WAKEUP_SCORE: attr["wakeup"],
|
||||||
}
|
}
|
||||||
return state_attr
|
return state_attr
|
||||||
|
|
||||||
state_attr = {ATTR_SESSION_START: self._attr["date"]}
|
state_attr = {ATTR_SESSION_START: attr["date"]}
|
||||||
state_attr[ATTR_TNT] = self._attr["tnt"]
|
state_attr[ATTR_TNT] = attr["tnt"]
|
||||||
state_attr[ATTR_PROCESSING] = self._attr["processing"]
|
state_attr[ATTR_PROCESSING] = attr["processing"]
|
||||||
|
|
||||||
sleep_time = (
|
if attr.get("breakdown") is not None:
|
||||||
sum(self._attr["breakdown"].values()) - self._attr["breakdown"]["awake"]
|
sleep_time = sum(attr["breakdown"].values()) - attr["breakdown"]["awake"]
|
||||||
)
|
state_attr[ATTR_SLEEP_DUR] = sleep_time
|
||||||
state_attr[ATTR_SLEEP_DUR] = sleep_time
|
state_attr[ATTR_LIGHT_PERC] = _get_breakdown_percent(
|
||||||
try:
|
attr, "light", sleep_time
|
||||||
state_attr[ATTR_LIGHT_PERC] = round(
|
|
||||||
(self._attr["breakdown"]["light"] / sleep_time) * 100, 2
|
|
||||||
)
|
)
|
||||||
except ZeroDivisionError:
|
state_attr[ATTR_DEEP_PERC] = _get_breakdown_percent(
|
||||||
state_attr[ATTR_LIGHT_PERC] = 0
|
attr, "deep", sleep_time
|
||||||
try:
|
|
||||||
state_attr[ATTR_DEEP_PERC] = round(
|
|
||||||
(self._attr["breakdown"]["deep"] / sleep_time) * 100, 2
|
|
||||||
)
|
)
|
||||||
except ZeroDivisionError:
|
state_attr[ATTR_REM_PERC] = _get_breakdown_percent(attr, "rem", sleep_time)
|
||||||
state_attr[ATTR_DEEP_PERC] = 0
|
|
||||||
|
|
||||||
try:
|
room_temp = self._get_rounded_value(attr, "room_temp")
|
||||||
state_attr[ATTR_REM_PERC] = round(
|
bed_temp = self._get_rounded_value(attr, "bed_temp")
|
||||||
(self._attr["breakdown"]["rem"] / sleep_time) * 100, 2
|
|
||||||
|
if "current" in self._sensor:
|
||||||
|
state_attr[ATTR_RESP_RATE] = self._get_rounded_value(
|
||||||
|
attr, "resp_rate", False
|
||||||
)
|
)
|
||||||
except ZeroDivisionError:
|
state_attr[ATTR_HEART_RATE] = self._get_rounded_value(
|
||||||
state_attr[ATTR_REM_PERC] = 0
|
attr, "heart_rate", False
|
||||||
|
)
|
||||||
try:
|
state_attr[ATTR_SLEEP_STAGE] = attr["stage"]
|
||||||
if self._units == "si":
|
|
||||||
room_temp = round(self._attr["room_temp"], 2)
|
|
||||||
else:
|
|
||||||
room_temp = round((self._attr["room_temp"] * 1.8) + 32, 2)
|
|
||||||
except TypeError:
|
|
||||||
room_temp = None
|
|
||||||
|
|
||||||
try:
|
|
||||||
if self._units == "si":
|
|
||||||
bed_temp = round(self._attr["bed_temp"], 2)
|
|
||||||
else:
|
|
||||||
bed_temp = round((self._attr["bed_temp"] * 1.8) + 32, 2)
|
|
||||||
except TypeError:
|
|
||||||
bed_temp = None
|
|
||||||
|
|
||||||
if "current" in self._sensor_root:
|
|
||||||
try:
|
|
||||||
state_attr[ATTR_RESP_RATE] = round(self._attr["resp_rate"], 2)
|
|
||||||
except TypeError:
|
|
||||||
state_attr[ATTR_RESP_RATE] = None
|
|
||||||
try:
|
|
||||||
state_attr[ATTR_HEART_RATE] = round(self._attr["heart_rate"], 2)
|
|
||||||
except TypeError:
|
|
||||||
state_attr[ATTR_HEART_RATE] = None
|
|
||||||
state_attr[ATTR_SLEEP_STAGE] = self._attr["stage"]
|
|
||||||
state_attr[ATTR_ROOM_TEMP] = room_temp
|
state_attr[ATTR_ROOM_TEMP] = room_temp
|
||||||
state_attr[ATTR_BED_TEMP] = bed_temp
|
state_attr[ATTR_BED_TEMP] = bed_temp
|
||||||
elif "last" in self._sensor_root:
|
elif "last" in self._sensor:
|
||||||
try:
|
state_attr[ATTR_AVG_RESP_RATE] = self._get_rounded_value(
|
||||||
state_attr[ATTR_AVG_RESP_RATE] = round(self._attr["resp_rate"], 2)
|
attr, "resp_rate", False
|
||||||
except TypeError:
|
)
|
||||||
state_attr[ATTR_AVG_RESP_RATE] = None
|
state_attr[ATTR_AVG_HEART_RATE] = self._get_rounded_value(
|
||||||
try:
|
attr, "heart_rate", False
|
||||||
state_attr[ATTR_AVG_HEART_RATE] = round(self._attr["heart_rate"], 2)
|
)
|
||||||
except TypeError:
|
|
||||||
state_attr[ATTR_AVG_HEART_RATE] = None
|
|
||||||
state_attr[ATTR_AVG_ROOM_TEMP] = room_temp
|
state_attr[ATTR_AVG_ROOM_TEMP] = room_temp
|
||||||
state_attr[ATTR_AVG_BED_TEMP] = bed_temp
|
state_attr[ATTR_AVG_BED_TEMP] = bed_temp
|
||||||
|
|
||||||
return state_attr
|
return state_attr
|
||||||
|
|
||||||
|
|
||||||
class EightRoomSensor(EightSleepEntity, SensorEntity):
|
class EightRoomSensor(EightSleepUserEntity, SensorEntity):
|
||||||
"""Representation of an eight sleep room sensor."""
|
"""Representation of an eight sleep room sensor."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@ -347,51 +286,25 @@ class EightRoomSensor(EightSleepEntity, SensorEntity):
|
|||||||
name: str,
|
name: str,
|
||||||
coordinator: EightSleepUserDataCoordinator,
|
coordinator: EightSleepUserDataCoordinator,
|
||||||
eight: EightSleep,
|
eight: EightSleep,
|
||||||
|
side: str | None,
|
||||||
sensor: str,
|
sensor: str,
|
||||||
units: str,
|
units: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator, eight)
|
super().__init__(name, coordinator, eight, side, sensor, units)
|
||||||
|
|
||||||
self._sensor = sensor
|
self._attr_icon = "mdi:thermometer"
|
||||||
self._mapped_name = NAME_MAP.get(self._sensor, self._sensor)
|
self._attr_native_unit_of_measurement: str = (
|
||||||
self._name = f"{name} {self._mapped_name}"
|
TEMP_CELSIUS if self._units == "si" else TEMP_FAHRENHEIT
|
||||||
self._state = None
|
)
|
||||||
self._attr = None
|
|
||||||
self._units = units
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self) -> str:
|
def native_value(self) -> int | float | None:
|
||||||
"""Return the name of the sensor, if any."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_value(self) -> str | None:
|
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self._state
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def _handle_coordinator_update(self) -> None:
|
|
||||||
"""Handle updated data from the coordinator."""
|
|
||||||
_LOGGER.debug("Updating Room sensor: %s", self._sensor)
|
|
||||||
temp = self._eight.room_temperature()
|
temp = self._eight.room_temperature()
|
||||||
try:
|
try:
|
||||||
if self._units == "si":
|
if self._units == "si":
|
||||||
self._state = round(temp, 2)
|
return round(temp, 2)
|
||||||
else:
|
return round((temp * 1.8) + 32, 2)
|
||||||
self._state = round((temp * 1.8) + 32, 2)
|
|
||||||
except TypeError:
|
except TypeError:
|
||||||
self._state = None
|
return None
|
||||||
super()._handle_coordinator_update()
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_unit_of_measurement(self) -> str:
|
|
||||||
"""Return the unit the value is expressed in."""
|
|
||||||
if self._units == "si":
|
|
||||||
return TEMP_CELSIUS
|
|
||||||
return TEMP_FAHRENHEIT
|
|
||||||
|
|
||||||
@property
|
|
||||||
def device_class(self) -> str:
|
|
||||||
"""Return the class of this device, from component DEVICE_CLASSES."""
|
|
||||||
return DEVICE_CLASS_TEMPERATURE
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user