jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup eight_sleep (#69171)

Browse Source 
* Cleanup eight_sleep

* Only set data after checking for users
This commit is contained in:
Raman Gupta 2022-04-29 21:17:03 -04:00 committed by GitHub
parent d01666f3a2
commit 0360613ddd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 150 additions and 237 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

204
homeassistant/components/eight_sleep/__init__.py
View File

@ -3,20 +3,12 @@ from __future__ import annotations
from datetime import timedelta
import logging
from typing import Union
from pyeight.eight import EightSleep
from pyeight.user import EightUser
import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_BINARY_SENSORS,
CONF_PASSWORD,
CONF_SENSORS,
CONF_USERNAME,
Platform,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import discovery
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -27,44 +19,24 @@ from homeassistant.helpers.update_coordinator import (
DataUpdateCoordinator,
)
from .const import (
ATTR_HEAT_DURATION,
ATTR_TARGET_HEAT,
DATA_API,
DATA_HEAT,
DATA_USER,
DOMAIN,
NAME_MAP,
SERVICE_HEAT_SET,
)
_LOGGER = logging.getLogger(__name__)
DATA_EIGHT = "eight_sleep"
DATA_HEAT = "heat"
DATA_USER = "user"
DATA_API = "api"
DOMAIN = "eight_sleep"
HEAT_ENTITY = "heat"
USER_ENTITY = "user"
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
HEAT_SCAN_INTERVAL = timedelta(seconds=60)
USER_SCAN_INTERVAL = timedelta(seconds=300)
NAME_MAP = {
"left_current_sleep": "Left Sleep Session",
"left_current_sleep_fitness": "Left Sleep Fitness",
"left_last_sleep": "Left Previous Sleep Session",
"right_current_sleep": "Right Sleep Session",
"right_current_sleep_fitness": "Right Sleep Fitness",
"right_last_sleep": "Right Previous Sleep Session",
}
SENSORS = [
"current_sleep",
"current_sleep_fitness",
"last_sleep",
"bed_state",
"bed_temperature",
"sleep_stage",
]
SERVICE_HEAT_SET = "heat_set"
ATTR_TARGET_HEAT = "target"
ATTR_HEAT_DURATION = "duration"
VALID_TARGET_HEAT = vol.All(vol.Coerce(int), vol.Clamp(min=-100, max=100))
VALID_DURATION = vol.All(vol.Coerce(int), vol.Clamp(min=0, max=28800))
@ -107,15 +79,11 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
user = conf[CONF_USERNAME]
password = conf[CONF_PASSWORD]
if hass.config.time_zone is None:
_LOGGER.error("Timezone is not set in Home Assistant")
return False
eight = EightSleep(
user, password, hass.config.time_zone, async_get_clientsession(hass)
)
timezone = str(hass.config.time_zone)
eight = EightSleep(user, password, timezone, async_get_clientsession(hass))
hass.data.setdefault(DATA_EIGHT, {})[DATA_API] = eight
hass.data.setdefault(DOMAIN, {})
# Authenticate, build sensors
success = await eight.start()
@ -123,43 +91,37 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
# Authentication failed, cannot continue
return False
heat_coordinator = hass.data[DOMAIN][DATA_HEAT] = EightSleepHeatDataCoordinator(
hass, eight
heat_coordinator: DataUpdateCoordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=f"{DOMAIN}_heat",
update_interval=HEAT_SCAN_INTERVAL,
update_method=eight.update_device_data,
)
user_coordinator = hass.data[DOMAIN][DATA_USER] = EightSleepUserDataCoordinator(
hass, eight
user_coordinator: DataUpdateCoordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=f"{DOMAIN}_user",
update_interval=USER_SCAN_INTERVAL,
update_method=eight.update_user_data,
)
await heat_coordinator.async_config_entry_first_refresh()
await user_coordinator.async_config_entry_first_refresh()
# Load sub components
sensors = []
binary_sensors = []
if eight.users:
for user, obj in eight.users.items():
for sensor in SENSORS:
sensors.append((obj.side, sensor))
binary_sensors.append((obj.side, "bed_presence"))
sensors.append((None, "room_temperature"))
else:
if not eight.users:
# No users, cannot continue
return False
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.SENSOR, DOMAIN, {CONF_SENSORS: sensors}, config
)
)
hass.data[DOMAIN] = {
DATA_API: eight,
DATA_HEAT: heat_coordinator,
DATA_USER: user_coordinator,
}
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.BINARY_SENSOR,
DOMAIN,
{CONF_BINARY_SENSORS: binary_sensors},
config,
for platform in PLATFORMS:
hass.async_create_task(
discovery.async_load_platform(hass, platform, DOMAIN, {}, config)
)
)
async def async_service_handler(service: ServiceCall) -> None:
"""Handle eight sleep service calls."""
@ -185,88 +147,32 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
class EightSleepHeatDataCoordinator(DataUpdateCoordinator):
"""Class to retrieve heat data from Eight Sleep."""
def __init__(self, hass: HomeAssistant, api: EightSleep) -> None:
"""Initialize coordinator."""
self.api = api
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}_heat",
update_interval=HEAT_SCAN_INTERVAL,
)
async def _async_update_data(self) -> None:
await self.api.update_device_data()
class EightSleepUserDataCoordinator(DataUpdateCoordinator):
"""Class to retrieve user data from Eight Sleep."""
def __init__(self, hass: HomeAssistant, api: EightSleep) -> None:
"""Initialize coordinator."""
self.api = api
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}_user",
update_interval=USER_SCAN_INTERVAL,
)
async def _async_update_data(self) -> None:
await self.api.update_user_data()
class EightSleepBaseEntity(
CoordinatorEntity[
Union[EightSleepUserDataCoordinator, EightSleepHeatDataCoordinator]
]
):
class EightSleepBaseEntity(CoordinatorEntity[DataUpdateCoordinator]):
"""The base Eight Sleep entity class."""
def __init__(
self,
name: str,
coordinator: EightSleepUserDataCoordinator | EightSleepHeatDataCoordinator,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
side: str | None,
user_id: str | None,
sensor: str,
units: str | None = None,
) -> None:
"""Initialize the data object."""
super().__init__(coordinator)
self._eight = eight
self._side = side
self._user_id = user_id
self._sensor = sensor
self._usrobj: EightUser | None = 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
self._user_obj: EightUser | None = None
if self._user_id:
self._user_obj = self._eight.users[user_id]
mapped_name = NAME_MAP.get(sensor, sensor.replace("_", " ").title())
if self._user_obj is not None:
mapped_name = f"{self._user_obj.side.title()} {mapped_name}"
self._attr_name = f"Eight {mapped_name}"
self._attr_unique_id = (
f"{_get_device_unique_id(eight, self._user_obj)}.{sensor}"
)

47
homeassistant/components/eight_sleep/binary_sensor.py
View File

@ -12,15 +12,10 @@ from homeassistant.components.binary_sensor import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import (
CONF_BINARY_SENSORS,
DATA_API,
DATA_EIGHT,
DATA_HEAT,
EightSleepBaseEntity,
EightSleepHeatDataCoordinator,
)
from . import EightSleepBaseEntity
from .const import DATA_API, DATA_HEAT, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -35,17 +30,16 @@ async def async_setup_platform(
if discovery_info is None:
return
name = "Eight"
sensors = discovery_info[CONF_BINARY_SENSORS]
eight: EightSleep = hass.data[DATA_EIGHT][DATA_API]
heat_coordinator: EightSleepHeatDataCoordinator = hass.data[DATA_EIGHT][DATA_HEAT]
eight: EightSleep = hass.data[DOMAIN][DATA_API]
heat_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_HEAT]
all_sensors = [
EightHeatSensor(name, heat_coordinator, eight, side, sensor)
for side, sensor in sensors
]
entities = []
for user in eight.users.values():
entities.append(
EightHeatSensor(heat_coordinator, eight, user.userid, "bed_presence")
)
async_add_entities(all_sensors)
async_add_entities(entities)
class EightHeatSensor(EightSleepBaseEntity, BinarySensorEntity):
@ -53,25 +47,24 @@ class EightHeatSensor(EightSleepBaseEntity, BinarySensorEntity):
def __init__(
self,
name: str,
coordinator: EightSleepHeatDataCoordinator,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
side: str | None,
user_id: str | None,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(name, coordinator, eight, side, sensor)
super().__init__(coordinator, eight, user_id, sensor)
self._attr_device_class = BinarySensorDeviceClass.OCCUPANCY
assert self._usrobj
assert self._user_obj
_LOGGER.debug(
"Presence Sensor: %s, Side: %s, User: %s",
self._sensor,
self._side,
self._usrobj.userid,
sensor,
self._user_obj.side,
user_id,
)
@property
def is_on(self) -> bool:
"""Return true if the binary sensor is on."""
assert self._usrobj
return bool(self._usrobj.bed_presence)
assert self._user_obj
return bool(self._user_obj.bed_presence)

19
homeassistant/components/eight_sleep/const.py Normal file
View File

@ -0,0 +1,19 @@
"""Eight Sleep constants."""
DATA_HEAT = "heat"
DATA_USER = "user"
DATA_API = "api"
DOMAIN = "eight_sleep"
HEAT_ENTITY = "heat"
USER_ENTITY = "user"
NAME_MAP = {
"current_sleep": "Sleep Session",
"current_sleep_fitness": "Sleep Fitness",
"last_sleep": "Previous Sleep Session",
}
SERVICE_HEAT_SET = "heat_set"
ATTR_TARGET_HEAT = "target"
ATTR_HEAT_DURATION = "duration"

117
homeassistant/components/eight_sleep/sensor.py
View File

@ -11,18 +11,10 @@ from homeassistant.const import PERCENTAGE, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import (
CONF_SENSORS,
DATA_API,
DATA_EIGHT,
DATA_HEAT,
DATA_USER,
EightSleepBaseEntity,
EightSleepHeatDataCoordinator,
EightSleepUserDataCoordinator,
EightSleepUserEntity,
)
from . import EightSleepBaseEntity
from .const import DATA_API, DATA_HEAT, DATA_USER, DOMAIN
ATTR_ROOM_TEMP = "Room Temperature"
ATTR_AVG_ROOM_TEMP = "Average Room Temperature"
@ -51,6 +43,16 @@ ATTR_FIT_WAKEUP_SCORE = "Fitness Wakeup Score"
_LOGGER = logging.getLogger(__name__)
EIGHT_USER_SENSORS = [
"current_sleep",
"current_sleep_fitness",
"last_sleep",
"bed_temperature",
"sleep_stage",
]
EIGHT_HEAT_SENSORS = ["bed_state"]
EIGHT_ROOM_SENSORS = ["room_temperature"]
async def async_setup_platform(
hass: HomeAssistant,
@ -62,11 +64,9 @@ async def async_setup_platform(
if discovery_info is None:
return
name = "Eight"
sensors = discovery_info[CONF_SENSORS]
eight: EightSleep = hass.data[DATA_EIGHT][DATA_API]
heat_coordinator: EightSleepHeatDataCoordinator = hass.data[DATA_EIGHT][DATA_HEAT]
user_coordinator: EightSleepUserDataCoordinator = hass.data[DATA_EIGHT][DATA_USER]
eight: EightSleep = hass.data[DOMAIN][DATA_API]
heat_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_HEAT]
user_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_USER]
if hass.config.units.is_metric:
units = "si"
@ -75,19 +75,17 @@ async def async_setup_platform(
all_sensors: list[SensorEntity] = []
for side, sensor in sensors:
if sensor == "bed_state":
for obj in eight.users.values():
for sensor in EIGHT_USER_SENSORS:
all_sensors.append(
EightHeatSensor(name, heat_coordinator, eight, side, sensor)
EightUserSensor(user_coordinator, eight, obj.userid, sensor, units)
)
elif sensor == "room_temperature":
for sensor in EIGHT_HEAT_SENSORS:
all_sensors.append(
EightRoomSensor(name, user_coordinator, eight, side, sensor, units)
)
else:
all_sensors.append(
EightUserSensor(name, user_coordinator, eight, side, sensor, units)
EightHeatSensor(heat_coordinator, eight, obj.userid, sensor)
)
for sensor in EIGHT_ROOM_SENSORS:
all_sensors.append(EightRoomSensor(user_coordinator, eight, sensor, units))
async_add_entities(all_sensors)
@ -97,38 +95,37 @@ class EightHeatSensor(EightSleepBaseEntity, SensorEntity):
def __init__(
self,
name: str,
coordinator: EightSleepHeatDataCoordinator,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
side: str | None,
user_id: str,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(name, coordinator, eight, side, sensor)
super().__init__(coordinator, eight, user_id, sensor)
self._attr_native_unit_of_measurement = PERCENTAGE
assert self._usrobj
assert self._user_obj
_LOGGER.debug(
"Heat Sensor: %s, Side: %s, User: %s",
self._sensor,
self._side,
self._usrobj.userid,
self._user_obj.side,
self._user_id,
)
@property
def native_value(self) -> int:
"""Return the state of the sensor."""
assert self._usrobj
return self._usrobj.heating_level
assert self._user_obj
return self._user_obj.heating_level
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return device state attributes."""
assert self._usrobj
assert self._user_obj
return {
ATTR_TARGET_HEAT: self._usrobj.target_heating_level,
ATTR_ACTIVE_HEAT: self._usrobj.now_heating,
ATTR_DURATION_HEAT: self._usrobj.heating_remaining,
ATTR_TARGET_HEAT: self._user_obj.target_heating_level,
ATTR_ACTIVE_HEAT: self._user_obj.now_heating,
ATTR_DURATION_HEAT: self._user_obj.heating_remaining,
}
@ -142,20 +139,20 @@ def _get_breakdown_percent(
return 0
class EightUserSensor(EightSleepUserEntity, SensorEntity):
class EightUserSensor(EightSleepBaseEntity, SensorEntity):
"""Representation of an eight sleep user-based sensor."""
def __init__(
self,
name: str,
coordinator: EightSleepUserDataCoordinator,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
side: str | None,
user_id: str,
sensor: str,
units: str,
) -> None:
"""Initialize the sensor."""
super().__init__(name, coordinator, eight, side, sensor, units)
super().__init__(coordinator, eight, user_id, sensor, units)
assert self._user_obj
if self._sensor == "bed_temperature":
self._attr_icon = "mdi:thermometer"
@ -163,26 +160,26 @@ class EightUserSensor(EightSleepUserEntity, SensorEntity):
_LOGGER.debug(
"User Sensor: %s, Side: %s, User: %s",
self._sensor,
self._side,
self._usrobj.userid if self._usrobj else None,
self._user_obj.side,
self._user_id,
)
@property
def native_value(self) -> str | int | float | None:
"""Return the state of the sensor."""
if not self._usrobj:
if not self._user_obj:
return None
if "current" in self._sensor:
if "fitness" in self._sensor:
return self._usrobj.current_sleep_fitness_score
return self._usrobj.current_sleep_score
return self._user_obj.current_sleep_fitness_score
return self._user_obj.current_sleep_score
if "last" in self._sensor:
return self._usrobj.last_sleep_score
return self._user_obj.last_sleep_score
if self._sensor == "bed_temperature":
temp = self._usrobj.current_values["bed_temp"]
temp = self._user_obj.current_values["bed_temp"]
try:
if self._units == "si":
return round(temp, 2)
@ -191,7 +188,7 @@ class EightUserSensor(EightSleepUserEntity, SensorEntity):
return None
if self._sensor == "sleep_stage":
return self._usrobj.current_values["stage"]
return self._user_obj.current_values["stage"]
return None
@ -221,13 +218,13 @@ class EightUserSensor(EightSleepUserEntity, SensorEntity):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return device state attributes."""
attr = None
if "current" in self._sensor and self._usrobj:
if "current" in self._sensor and self._user_obj:
if "fitness" in self._sensor:
attr = self._usrobj.current_fitness_values
attr = self._user_obj.current_fitness_values
else:
attr = self._usrobj.current_values
elif "last" in self._sensor and self._usrobj:
attr = self._usrobj.last_values
attr = self._user_obj.current_values
elif "last" in self._sensor and self._user_obj:
attr = self._user_obj.last_values
if attr is None:
# Skip attributes if sensor type doesn't support
@ -284,20 +281,18 @@ class EightUserSensor(EightSleepUserEntity, SensorEntity):
return state_attr
class EightRoomSensor(EightSleepUserEntity, SensorEntity):
class EightRoomSensor(EightSleepBaseEntity, SensorEntity):
"""Representation of an eight sleep room sensor."""
def __init__(
self,
name: str,
coordinator: EightSleepUserDataCoordinator,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
side: str | None,
sensor: str,
units: str,
) -> None:
"""Initialize the sensor."""
super().__init__(name, coordinator, eight, side, sensor, units)
super().__init__(coordinator, eight, None, sensor, units)
self._attr_icon = "mdi:thermometer"
self._attr_native_unit_of_measurement: str = (