This commit is contained in:
2026-06-30 19:53:54 -04:00
commit b9855e54d3
16 changed files with 1524 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
hello
+165
View File
@@ -0,0 +1,165 @@
"""Support for SleepIQ from SleepNumber."""
import logging
from typing import Any
from asyncsleepiq import (
AsyncSleepIQ,
SleepIQAPIException,
SleepIQLoginException,
SleepIQTimeoutException,
)
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, PRESSURE, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, IS_IN_BED, SLEEP_NUMBER
from .coordinator import (
SleepIQConfigEntry,
SleepIQData,
SleepIQDataUpdateCoordinator,
SleepIQPauseUpdateCoordinator,
SleepIQSleepDataCoordinator,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.LIGHT,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: {
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up sleepiq component."""
if DOMAIN in config:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config[DOMAIN]
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: SleepIQConfigEntry) -> bool:
"""Set up the SleepIQ config entry."""
conf = entry.data
email = conf[CONF_USERNAME]
password = conf[CONF_PASSWORD]
client_session = async_create_clientsession(hass)
gateway = AsyncSleepIQ(client_session=client_session)
try:
await gateway.login(email, password)
except SleepIQLoginException as err:
_LOGGER.error("Could not authenticate with SleepIQ server")
raise ConfigEntryAuthFailed(err) from err
except SleepIQTimeoutException as err:
raise ConfigEntryNotReady(
str(err) or "Timed out during authentication"
) from err
try:
await gateway.init_beds()
except SleepIQTimeoutException as err:
raise ConfigEntryNotReady(
str(err) or "Timed out during initialization"
) from err
except SleepIQAPIException as err:
raise ConfigEntryNotReady(str(err) or "Error reading from SleepIQ API") from err
await _async_migrate_unique_ids(hass, entry, gateway)
coordinator = SleepIQDataUpdateCoordinator(hass, entry, gateway)
pause_coordinator = SleepIQPauseUpdateCoordinator(hass, entry, gateway)
sleep_data_coordinator = SleepIQSleepDataCoordinator(hass, entry, gateway)
# Call the SleepIQ API to refresh data
await coordinator.async_config_entry_first_refresh()
await pause_coordinator.async_config_entry_first_refresh()
await sleep_data_coordinator.async_config_entry_first_refresh()
entry.runtime_data = SleepIQData(
data_coordinator=coordinator,
pause_coordinator=pause_coordinator,
sleep_data_coordinator=sleep_data_coordinator,
client=gateway,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: SleepIQConfigEntry) -> bool:
"""Unload the config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def _async_migrate_unique_ids(
hass: HomeAssistant, entry: ConfigEntry, gateway: AsyncSleepIQ
) -> None:
"""Migrate old unique ids."""
names_to_ids = {
sleeper.name: sleeper.sleeper_id
for bed in gateway.beds.values()
for sleeper in bed.sleepers
}
bed_ids = {bed.id for bed in gateway.beds.values()}
@callback
def _async_migrator(entity_entry: er.RegistryEntry) -> dict[str, Any] | None:
# Old format for sleeper entities was {bed_id}_{sleeper.name}_{sensor_type}.....
# New format is {sleeper.sleeper_id}_{sensor_type}....
sensor_types = [IS_IN_BED, PRESSURE, SLEEP_NUMBER]
old_unique_id = entity_entry.unique_id
parts = old_unique_id.split("_")
# If it doesn't begin with a bed id or end with one of the sensor types,
# it doesn't need to be migrated
if parts[0] not in bed_ids or not old_unique_id.endswith(tuple(sensor_types)):
return None
sensor_type = next(filter(old_unique_id.endswith, sensor_types))
sleeper_name = "_".join(parts[1:]).removesuffix(f"_{sensor_type}")
sleeper_id = names_to_ids.get(sleeper_name)
if not sleeper_id:
return None
new_unique_id = f"{sleeper_id}_{sensor_type}"
_LOGGER.debug(
"Migrating unique_id from [%s] to [%s]",
old_unique_id,
new_unique_id,
)
return {"new_unique_id": new_unique_id}
await er.async_migrate_entries(hass, entry.entry_id, _async_migrator)
@@ -0,0 +1,54 @@
"""Support for SleepIQ sensors."""
from typing import override
from asyncsleepiq import SleepIQBed, SleepIQSleeper
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import ICON_EMPTY, ICON_OCCUPIED, IS_IN_BED
from .coordinator import SleepIQConfigEntry, SleepIQDataUpdateCoordinator
from .entity import SleepIQSleeperEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the SleepIQ bed binary sensors."""
data = entry.runtime_data
async_add_entities(
IsInBedBinarySensor(data.data_coordinator, bed, sleeper)
for bed in data.client.beds.values()
for sleeper in bed.sleepers
)
class IsInBedBinarySensor(
SleepIQSleeperEntity[SleepIQDataUpdateCoordinator], BinarySensorEntity
):
"""Implementation of a SleepIQ presence sensor."""
_attr_device_class = BinarySensorDeviceClass.OCCUPANCY
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
sleeper: SleepIQSleeper,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, bed, sleeper, IS_IN_BED)
@callback
@override
def _async_update_attrs(self) -> None:
"""Update sensor attributes."""
self._attr_is_on = self.sleeper.in_bed
self._attr_icon = ICON_OCCUPIED if self.sleeper.in_bed else ICON_EMPTY
+72
View File
@@ -0,0 +1,72 @@
"""Support for SleepIQ buttons."""
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, override
from asyncsleepiq import SleepIQBed
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SleepIQConfigEntry
from .entity import SleepIQEntity
@dataclass(frozen=True, kw_only=True)
class SleepIQButtonEntityDescription(ButtonEntityDescription):
"""Class to describe a Button entity."""
press_action: Callable[[SleepIQBed], Any]
ENTITY_DESCRIPTIONS = [
SleepIQButtonEntityDescription(
key="calibrate",
name="Calibrate",
press_action=lambda client: client.calibrate(),
icon="mdi:target",
),
SleepIQButtonEntityDescription(
key="stop-pump",
name="Stop Pump",
press_action=lambda client: client.stop_pump(),
icon="mdi:stop",
),
]
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sleep number buttons."""
data = entry.runtime_data
async_add_entities(
SleepNumberButton(bed, ed)
for bed in data.client.beds.values()
for ed in ENTITY_DESCRIPTIONS
)
class SleepNumberButton(SleepIQEntity, ButtonEntity):
"""Representation of an SleepIQ button."""
entity_description: SleepIQButtonEntityDescription
def __init__(
self, bed: SleepIQBed, entity_description: SleepIQButtonEntityDescription
) -> None:
"""Initialize the Button."""
super().__init__(bed)
self._attr_name = f"SleepNumber {bed.name} {entity_description.name}"
self._attr_unique_id = f"{bed.id}-{entity_description.key}"
self.entity_description = entity_description
@override
async def async_press(self) -> None:
"""Press the button."""
await self.entity_description.press_action(self.bed)
+124
View File
@@ -0,0 +1,124 @@
"""Config flow to configure SleepIQ component."""
from collections.abc import Mapping
import logging
from typing import Any, override
from asyncsleepiq import AsyncSleepIQ, SleepIQLoginException, SleepIQTimeoutException
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class SleepIQFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a SleepIQ config flow."""
VERSION = 1
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Import a SleepIQ account as a config entry.
This flow is triggered by 'async_setup' for configured accounts.
"""
await self.async_set_unique_id(import_data[CONF_USERNAME].lower())
self._abort_if_unique_id_configured()
if error := await try_connection(self.hass, import_data):
_LOGGER.error("Could not authenticate with SleepIQ server: %s", error)
return self.async_abort(reason=error)
return self.async_create_entry(
title=import_data[CONF_USERNAME], data=import_data
)
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
# Don't allow multiple instances with the same username
await self.async_set_unique_id(user_input[CONF_USERNAME].lower())
self._abort_if_unique_id_configured()
if error := await try_connection(self.hass, user_input):
errors["base"] = error
else:
return self.async_create_entry(
title=user_input[CONF_USERNAME], data=user_input
)
else:
user_input = {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_USERNAME,
default=user_input.get(CONF_USERNAME),
): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
last_step=True,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm reauth."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
if user_input is not None:
data = {
CONF_USERNAME: reauth_entry.data[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
}
if not (error := await try_connection(self.hass, data)):
return self.async_update_reload_and_abort(reauth_entry, data=data)
errors["base"] = error
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
errors=errors,
description_placeholders={
CONF_USERNAME: reauth_entry.data[CONF_USERNAME],
},
)
async def try_connection(hass: HomeAssistant, user_input: dict[str, Any]) -> str | None:
"""Test if the given credentials can successfully login to SleepIQ."""
client_session = async_get_clientsession(hass)
gateway = AsyncSleepIQ(client_session=client_session)
try:
await gateway.login(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
except SleepIQLoginException:
return "invalid_auth"
except SleepIQTimeoutException:
return "cannot_connect"
return None
+46
View File
@@ -0,0 +1,46 @@
"""Define constants for the SleepIQ component."""
from homeassistant.const import PRESSURE
DATA_SLEEPIQ = "data_sleepiq"
DOMAIN = "sleepiq"
ACTUATOR = "actuator"
CORE_CLIMATE_TIMER = "core_climate_timer"
CORE_CLIMATE = "core_climate"
BED = "bed"
FIRMNESS = "firmness"
ICON_EMPTY = "mdi:bed-empty"
ICON_OCCUPIED = "mdi:bed"
IS_IN_BED = "is_in_bed"
SLEEP_NUMBER = "sleep_number"
FOOT_WARMING_TIMER = "foot_warming_timer"
FOOT_WARMER = "foot_warmer"
SLEEP_SCORE = "sleep_score"
SLEEP_DURATION = "sleep_duration"
HEART_RATE = "heart_rate"
RESPIRATORY_RATE = "respiratory_rate"
HRV = "hrv"
ENTITY_TYPES = {
ACTUATOR: "Position",
CORE_CLIMATE_TIMER: "Core Climate Timer",
CORE_CLIMATE: "Core Climate",
FIRMNESS: "Firmness",
PRESSURE: "Pressure",
IS_IN_BED: "Is In Bed",
SLEEP_NUMBER: "SleepNumber",
FOOT_WARMING_TIMER: "Foot Warming Timer",
FOOT_WARMER: "Foot Warmer",
SLEEP_SCORE: "Sleep Score",
SLEEP_DURATION: "Sleep Duration",
HEART_RATE: "Heart Rate Average",
RESPIRATORY_RATE: "Respiratory Rate Average",
HRV: "Heart Rate Variability",
}
LEFT = "left"
RIGHT = "right"
SIDES = [LEFT, RIGHT]
SLEEPIQ_DATA = "sleepiq_data"
SLEEPIQ_STATUS_COORDINATOR = "sleepiq_status"
+138
View File
@@ -0,0 +1,138 @@
"""Coordinator for SleepIQ."""
import asyncio
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import override
from asyncsleepiq import AsyncSleepIQ, SleepIQAPIException, SleepIQTimeoutException
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
_LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL = timedelta(seconds=60)
LONGER_UPDATE_INTERVAL = timedelta(minutes=5)
SLEEP_DATA_UPDATE_INTERVAL = timedelta(hours=1) # Sleep data doesn't change frequently
type SleepIQConfigEntry = ConfigEntry[SleepIQData]
class SleepIQDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""SleepIQ data update coordinator."""
config_entry: SleepIQConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: SleepIQConfigEntry,
client: AsyncSleepIQ,
) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"{config_entry.data[CONF_USERNAME]}@SleepIQ",
update_interval=UPDATE_INTERVAL,
)
self.client = client
@override
async def _async_update_data(self) -> None:
tasks = [self.client.fetch_bed_statuses()] + [
bed.foundation.update_foundation_status()
for bed in self.client.beds.values()
]
try:
await asyncio.gather(*tasks)
except SleepIQTimeoutException as err:
raise UpdateFailed(f"Timed out fetching SleepIQ data: {err}") from err
except SleepIQAPIException as err:
raise UpdateFailed(f"Failed to fetch SleepIQ data: {err}") from err
class SleepIQPauseUpdateCoordinator(DataUpdateCoordinator[None]):
"""SleepIQ pause update coordinator."""
config_entry: SleepIQConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: SleepIQConfigEntry,
client: AsyncSleepIQ,
) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"{config_entry.data[CONF_USERNAME]}@SleepIQPause",
update_interval=LONGER_UPDATE_INTERVAL,
)
self.client = client
@override
async def _async_update_data(self) -> None:
try:
await asyncio.gather(
*[bed.fetch_pause_mode() for bed in self.client.beds.values()]
)
except SleepIQTimeoutException as err:
raise UpdateFailed(f"Timed out fetching SleepIQ pause data: {err}") from err
except SleepIQAPIException as err:
raise UpdateFailed(f"Failed to fetch SleepIQ pause data: {err}") from err
class SleepIQSleepDataCoordinator(DataUpdateCoordinator[None]):
"""SleepIQ sleep health data coordinator."""
config_entry: SleepIQConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: SleepIQConfigEntry,
client: AsyncSleepIQ,
) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"{config_entry.data[CONF_USERNAME]}@SleepIQSleepData",
update_interval=SLEEP_DATA_UPDATE_INTERVAL,
)
self.client = client
@override
async def _async_update_data(self) -> None:
"""Fetch sleep health data from API via asyncsleepiq library."""
try:
await asyncio.gather(
*[
sleeper.fetch_sleep_data()
for bed in self.client.beds.values()
for sleeper in bed.sleepers
]
)
except SleepIQTimeoutException as err:
raise UpdateFailed(f"Timed out fetching SleepIQ sleep data: {err}") from err
except SleepIQAPIException as err:
raise UpdateFailed(f"Failed to fetch SleepIQ sleep data: {err}") from err
@dataclass
class SleepIQData:
"""Data for the sleepiq integration."""
data_coordinator: SleepIQDataUpdateCoordinator
pause_coordinator: SleepIQPauseUpdateCoordinator
sleep_data_coordinator: SleepIQSleepDataCoordinator
client: AsyncSleepIQ
+105
View File
@@ -0,0 +1,105 @@
"""Entity for the SleepIQ integration."""
from abc import abstractmethod
from typing import override
from asyncsleepiq import SleepIQBed, SleepIQSleeper
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ENTITY_TYPES, ICON_OCCUPIED
from .coordinator import (
SleepIQDataUpdateCoordinator,
SleepIQPauseUpdateCoordinator,
SleepIQSleepDataCoordinator,
)
type _DataCoordinatorType = (
SleepIQDataUpdateCoordinator
| SleepIQPauseUpdateCoordinator
| SleepIQSleepDataCoordinator
)
def device_from_bed(bed: SleepIQBed) -> DeviceInfo:
"""Create a device given a bed."""
return DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, bed.mac_addr)},
manufacturer="SleepNumber",
name=bed.name,
model=bed.model,
)
def sleeper_for_side(bed: SleepIQBed, side: str) -> SleepIQSleeper:
"""Find the sleeper for a side or the first sleeper."""
for sleeper in bed.sleepers:
if sleeper.side == side:
return sleeper
return bed.sleepers[0]
class SleepIQEntity(Entity):
"""Implementation of a SleepIQ entity."""
def __init__(self, bed: SleepIQBed) -> None:
"""Initialize the SleepIQ entity."""
self.bed = bed
self._attr_device_info = device_from_bed(bed)
class SleepIQBedEntity[_SleepIQCoordinatorT: _DataCoordinatorType](
CoordinatorEntity[_SleepIQCoordinatorT]
):
"""Implementation of a SleepIQ sensor."""
_attr_icon = ICON_OCCUPIED
def __init__(
self,
coordinator: _SleepIQCoordinatorT,
bed: SleepIQBed,
) -> None:
"""Initialize the SleepIQ sensor entity."""
super().__init__(coordinator)
self.bed = bed
self._attr_device_info = device_from_bed(bed)
self._async_update_attrs()
@callback
@override
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._async_update_attrs()
super()._handle_coordinator_update()
@callback
@abstractmethod
def _async_update_attrs(self) -> None:
"""Update sensor attributes."""
class SleepIQSleeperEntity[_SleepIQCoordinatorT: _DataCoordinatorType](
SleepIQBedEntity[_SleepIQCoordinatorT]
):
"""Implementation of a SleepIQ sensor."""
_attr_icon = ICON_OCCUPIED
def __init__(
self,
coordinator: _SleepIQCoordinatorT,
bed: SleepIQBed,
sleeper: SleepIQSleeper,
name: str,
) -> None:
"""Initialize the SleepIQ sensor entity."""
self.sleeper = sleeper
super().__init__(coordinator, bed)
self._attr_name = f"SleepNumber {bed.name} {sleeper.name} {ENTITY_TYPES[name]}"
self._attr_unique_id = f"{sleeper.sleeper_id}_{name}"
+21
View File
@@ -0,0 +1,21 @@
{
"entity": {
"sensor": {
"heart_rate_avg": {
"default": "mdi:heart-pulse"
},
"hrv": {
"default": "mdi:heart-flash"
},
"respiratory_rate_avg": {
"default": "mdi:lungs"
},
"sleep_duration": {
"default": "mdi:sleep"
},
"sleep_score": {
"default": "mdi:sleep"
}
}
}
}
+66
View File
@@ -0,0 +1,66 @@
"""Support for SleepIQ outlet lights."""
import logging
from typing import Any, override
from asyncsleepiq import SleepIQBed, SleepIQLight
from homeassistant.components.light import ColorMode, LightEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SleepIQConfigEntry, SleepIQDataUpdateCoordinator
from .entity import SleepIQBedEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the SleepIQ bed lights."""
data = entry.runtime_data
async_add_entities(
SleepIQLightEntity(data.data_coordinator, bed, light)
for bed in data.client.beds.values()
for light in bed.foundation.lights
)
class SleepIQLightEntity(SleepIQBedEntity[SleepIQDataUpdateCoordinator], LightEntity):
"""Representation of a light."""
_attr_color_mode = ColorMode.ONOFF
_attr_supported_color_modes = {ColorMode.ONOFF}
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
light: SleepIQLight,
) -> None:
"""Initialize the light."""
self.light = light
super().__init__(coordinator, bed)
self._attr_name = f"SleepNumber {bed.name} Light {light.outlet_id}"
self._attr_unique_id = f"{bed.id}-light-{light.outlet_id}" # pylint: disable=home-assistant-entity-unique-id-redundant-platform
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on light."""
await self.light.turn_on()
self._handle_coordinator_update()
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off light."""
await self.light.turn_off()
self._handle_coordinator_update()
@callback
@override
def _async_update_attrs(self) -> None:
"""Update light attributes."""
self._attr_is_on = self.light.is_on
@@ -0,0 +1,18 @@
{
"domain": "kitty_sleepiq",
"name": "Kitty SleepIQ",
"issue_tracker": "https://winslowthehedgehog.com/issues",
"version": "2026.06.30",
"codeowners": ["@nuggets_home_maryishan"],
"config_flow": true,
"dhcp": [
{
"macaddress": "64DBA0*"
}
],
"documentation": "https://www.home-assistant.io/integrations/sleepiq",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["asyncsleepiq"],
"requirements": ["asyncsleepiq==1.7.1"]
}
+266
View File
@@ -0,0 +1,266 @@
"""Support for SleepIQ SleepNumber firmness number entities."""
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, cast, override
from asyncsleepiq import (
CoreTemps,
FootWarmingTemps,
SleepIQActuator,
SleepIQBed,
SleepIQCoreClimate,
SleepIQFootWarmer,
SleepIQSleeper,
)
from homeassistant.components.number import (
NumberDeviceClass,
NumberEntity,
NumberEntityDescription,
)
from homeassistant.const import UnitOfTime
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
ACTUATOR,
CORE_CLIMATE_TIMER,
ENTITY_TYPES,
FIRMNESS,
FOOT_WARMING_TIMER,
ICON_OCCUPIED,
)
from .coordinator import SleepIQConfigEntry, SleepIQDataUpdateCoordinator
from .entity import SleepIQBedEntity, sleeper_for_side
@dataclass(frozen=True, kw_only=True)
class SleepIQNumberEntityDescription(NumberEntityDescription):
"""Class to describe a SleepIQ number entity."""
value_fn: Callable[[Any], float]
set_value_fn: Callable[[Any, int], Coroutine[None, None, None]]
get_name_fn: Callable[[SleepIQBed, Any], str]
get_unique_id_fn: Callable[[SleepIQBed, Any], str]
async def _async_set_firmness(sleeper: SleepIQSleeper, firmness: int) -> None:
await sleeper.set_sleepnumber(firmness)
async def _async_set_actuator_position(
actuator: SleepIQActuator, position: int
) -> None:
await actuator.set_position(position)
def _get_actuator_name(bed: SleepIQBed, actuator: SleepIQActuator) -> str:
if actuator.side:
return (
"SleepNumber"
f" {bed.name} {actuator.side_full}"
f" {actuator.actuator_full}"
f" {ENTITY_TYPES[ACTUATOR]}"
)
return f"SleepNumber {bed.name} {actuator.actuator_full} {ENTITY_TYPES[ACTUATOR]}"
def _get_actuator_unique_id(bed: SleepIQBed, actuator: SleepIQActuator) -> str:
if actuator.side:
return f"{bed.id}_{actuator.side.value}_{actuator.actuator}"
return f"{bed.id}_{actuator.actuator}"
def _get_sleeper_name(bed: SleepIQBed, sleeper: SleepIQSleeper) -> str:
return f"SleepNumber {bed.name} {sleeper.name} {ENTITY_TYPES[FIRMNESS]}"
def _get_sleeper_unique_id(bed: SleepIQBed, sleeper: SleepIQSleeper) -> str:
return f"{sleeper.sleeper_id}_{FIRMNESS}"
async def _async_set_foot_warmer_time(
foot_warmer: SleepIQFootWarmer, time: int
) -> None:
temperature = FootWarmingTemps(foot_warmer.temperature)
if temperature != FootWarmingTemps.OFF:
await foot_warmer.turn_on(temperature, time)
foot_warmer.timer = time
def _get_foot_warming_name(bed: SleepIQBed, foot_warmer: SleepIQFootWarmer) -> str:
sleeper = sleeper_for_side(bed, foot_warmer.side)
return f"SleepNumber {bed.name} {sleeper.name} {ENTITY_TYPES[FOOT_WARMING_TIMER]}"
def _get_foot_warming_unique_id(bed: SleepIQBed, foot_warmer: SleepIQFootWarmer) -> str:
return f"{bed.id}_{foot_warmer.side.value}_{FOOT_WARMING_TIMER}"
async def _async_set_core_climate_time(
core_climate: SleepIQCoreClimate, time: int
) -> None:
temperature = CoreTemps(core_climate.temperature)
if temperature != CoreTemps.OFF:
await core_climate.turn_on(temperature, time)
core_climate.timer = time
def _get_core_climate_name(bed: SleepIQBed, core_climate: SleepIQCoreClimate) -> str:
sleeper = sleeper_for_side(bed, core_climate.side)
return f"SleepNumber {bed.name} {sleeper.name} {ENTITY_TYPES[CORE_CLIMATE_TIMER]}"
def _get_core_climate_unique_id(
bed: SleepIQBed, core_climate: SleepIQCoreClimate
) -> str:
return f"{bed.id}_{core_climate.side.value}_{CORE_CLIMATE_TIMER}"
NUMBER_DESCRIPTIONS: dict[str, SleepIQNumberEntityDescription] = {
FIRMNESS: SleepIQNumberEntityDescription(
key=FIRMNESS,
native_min_value=5,
native_max_value=100,
native_step=5,
name=ENTITY_TYPES[FIRMNESS],
icon=ICON_OCCUPIED,
value_fn=lambda sleeper: cast(float, sleeper.sleep_number),
set_value_fn=_async_set_firmness,
get_name_fn=_get_sleeper_name,
get_unique_id_fn=_get_sleeper_unique_id,
),
ACTUATOR: SleepIQNumberEntityDescription(
key=ACTUATOR,
native_min_value=0,
native_max_value=100,
native_step=1,
name=ENTITY_TYPES[ACTUATOR],
icon=ICON_OCCUPIED,
value_fn=lambda actuator: cast(float, actuator.position),
set_value_fn=_async_set_actuator_position,
get_name_fn=_get_actuator_name,
get_unique_id_fn=_get_actuator_unique_id,
),
FOOT_WARMING_TIMER: SleepIQNumberEntityDescription(
key=FOOT_WARMING_TIMER,
native_min_value=30,
native_max_value=360,
native_step=30,
name=ENTITY_TYPES[FOOT_WARMING_TIMER],
icon="mdi:timer",
value_fn=lambda foot_warmer: foot_warmer.timer,
set_value_fn=_async_set_foot_warmer_time,
get_name_fn=_get_foot_warming_name,
get_unique_id_fn=_get_foot_warming_unique_id,
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=NumberDeviceClass.DURATION,
),
CORE_CLIMATE_TIMER: SleepIQNumberEntityDescription(
key=CORE_CLIMATE_TIMER,
native_min_value=0,
native_max_value=SleepIQCoreClimate.max_core_climate_time,
native_step=30,
name=ENTITY_TYPES[CORE_CLIMATE_TIMER],
icon="mdi:timer",
value_fn=lambda core_climate: core_climate.timer,
set_value_fn=_async_set_core_climate_time,
get_name_fn=_get_core_climate_name,
get_unique_id_fn=_get_core_climate_unique_id,
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=NumberDeviceClass.DURATION,
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the SleepIQ bed sensors."""
data = entry.runtime_data
entities: list[SleepIQNumberEntity] = []
for bed in data.client.beds.values():
entities.extend(
SleepIQNumberEntity(
data.data_coordinator,
bed,
sleeper,
NUMBER_DESCRIPTIONS[FIRMNESS],
)
for sleeper in bed.sleepers
)
entities.extend(
SleepIQNumberEntity(
data.data_coordinator,
bed,
actuator,
NUMBER_DESCRIPTIONS[ACTUATOR],
)
for actuator in bed.foundation.actuators
)
entities.extend(
SleepIQNumberEntity(
data.data_coordinator,
bed,
foot_warmer,
NUMBER_DESCRIPTIONS[FOOT_WARMING_TIMER],
)
for foot_warmer in bed.foundation.foot_warmers
)
entities.extend(
SleepIQNumberEntity(
data.data_coordinator,
bed,
core_climate,
NUMBER_DESCRIPTIONS[CORE_CLIMATE_TIMER],
)
for core_climate in bed.foundation.core_climates
)
async_add_entities(entities)
class SleepIQNumberEntity(SleepIQBedEntity[SleepIQDataUpdateCoordinator], NumberEntity):
"""Representation of a SleepIQ number entity."""
entity_description: SleepIQNumberEntityDescription
_attr_icon = "mdi:bed"
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
device: Any,
description: SleepIQNumberEntityDescription,
) -> None:
"""Initialize the number."""
self.entity_description = description
self.device = device
self._attr_name = description.get_name_fn(bed, device)
self._attr_unique_id = description.get_unique_id_fn(bed, device)
if description.icon:
self._attr_icon = description.icon
super().__init__(coordinator, bed)
@callback
@override
def _async_update_attrs(self) -> None:
"""Update number attributes."""
self._attr_native_value = float(self.entity_description.value_fn(self.device))
@override
async def async_set_native_value(self, value: float) -> None:
"""Set the number value."""
await self.entity_description.set_value_fn(self.device, int(value))
self._attr_native_value = value
self.async_write_ha_state()
+182
View File
@@ -0,0 +1,182 @@
"""Support for SleepIQ foundation preset selection."""
from typing import override
from asyncsleepiq import (
CoreTemps,
FootWarmingTemps,
Side,
SleepIQBed,
SleepIQCoreClimate,
SleepIQFootWarmer,
SleepIQPreset,
)
from homeassistant.components.select import SelectEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CORE_CLIMATE, FOOT_WARMER
from .coordinator import SleepIQConfigEntry, SleepIQDataUpdateCoordinator
from .entity import SleepIQBedEntity, SleepIQSleeperEntity, sleeper_for_side
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the SleepIQ foundation preset select entities."""
data = entry.runtime_data
entities: list[SleepIQBedEntity] = []
for bed in data.client.beds.values():
entities.extend(
SleepIQSelectEntity(data.data_coordinator, bed, preset)
for preset in bed.foundation.presets
)
entities.extend(
SleepIQFootWarmingTempSelectEntity(data.data_coordinator, bed, foot_warmer)
for foot_warmer in bed.foundation.foot_warmers
)
entities.extend(
SleepIQCoreTempSelectEntity(data.data_coordinator, bed, core_climate)
for core_climate in bed.foundation.core_climates
)
async_add_entities(entities)
class SleepIQSelectEntity(SleepIQBedEntity[SleepIQDataUpdateCoordinator], SelectEntity):
"""Representation of a SleepIQ select entity."""
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
preset: SleepIQPreset,
) -> None:
"""Initialize the select entity."""
self.preset = preset
self._attr_name = f"SleepNumber {bed.name} Foundation Preset"
self._attr_unique_id = f"{bed.id}_preset"
if preset.side != Side.NONE:
self._attr_name += f" {preset.side_full}"
self._attr_unique_id += f"_{preset.side.value}"
self._attr_options = preset.options
super().__init__(coordinator, bed)
self._async_update_attrs()
@callback
@override
def _async_update_attrs(self) -> None:
"""Update entity attributes."""
self._attr_current_option = self.preset.preset
@override
async def async_select_option(self, option: str) -> None:
"""Change the current preset."""
await self.preset.set_preset(option)
self._attr_current_option = option
self.async_write_ha_state()
class SleepIQFootWarmingTempSelectEntity(
SleepIQSleeperEntity[SleepIQDataUpdateCoordinator], SelectEntity
):
"""Representation of a SleepIQ foot warming temperature select entity."""
_attr_icon = "mdi:heat-wave"
_attr_options = [e.name.lower() for e in FootWarmingTemps]
_attr_translation_key = "foot_warmer_temp"
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
foot_warmer: SleepIQFootWarmer,
) -> None:
"""Initialize the select entity."""
self.foot_warmer = foot_warmer
sleeper = sleeper_for_side(bed, foot_warmer.side)
super().__init__(coordinator, bed, sleeper, FOOT_WARMER)
self._async_update_attrs()
@callback
@override
def _async_update_attrs(self) -> None:
"""Update entity attributes."""
self._attr_current_option = FootWarmingTemps(
self.foot_warmer.temperature
).name.lower()
@override
async def async_select_option(self, option: str) -> None:
"""Change the current preset."""
temperature = FootWarmingTemps[option.upper()]
timer = self.foot_warmer.timer or 120
if temperature == 0:
await self.foot_warmer.turn_off()
else:
await self.foot_warmer.turn_on(temperature, timer)
self._attr_current_option = option
await self.coordinator.async_request_refresh()
self.async_write_ha_state()
class SleepIQCoreTempSelectEntity(
SleepIQSleeperEntity[SleepIQDataUpdateCoordinator], SelectEntity
):
"""Representation of a SleepIQ core climate temperature select entity."""
# Maps to translate between asyncsleepiq and HA's naming preference
SLEEPIQ_TO_HA_CORE_TEMP_MAP = {
CoreTemps.OFF: "off",
CoreTemps.HEATING_PUSH_LOW: "heating_low",
CoreTemps.HEATING_PUSH_MED: "heating_medium",
CoreTemps.HEATING_PUSH_HIGH: "heating_high",
CoreTemps.COOLING_PULL_LOW: "cooling_low",
CoreTemps.COOLING_PULL_MED: "cooling_medium",
CoreTemps.COOLING_PULL_HIGH: "cooling_high",
}
HA_TO_SLEEPIQ_CORE_TEMP_MAP = {v: k for k, v in SLEEPIQ_TO_HA_CORE_TEMP_MAP.items()}
_attr_icon = "mdi:heat-wave"
_attr_options = list(SLEEPIQ_TO_HA_CORE_TEMP_MAP.values())
_attr_translation_key = "core_temps"
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator,
bed: SleepIQBed,
core_climate: SleepIQCoreClimate,
) -> None:
"""Initialize the select entity."""
self.core_climate = core_climate
sleeper = sleeper_for_side(bed, core_climate.side)
super().__init__(coordinator, bed, sleeper, CORE_CLIMATE)
self._async_update_attrs()
@callback
@override
def _async_update_attrs(self) -> None:
"""Update entity attributes."""
sleepiq_option = CoreTemps(self.core_climate.temperature)
self._attr_current_option = self.SLEEPIQ_TO_HA_CORE_TEMP_MAP[sleepiq_option]
@override
async def async_select_option(self, option: str) -> None:
"""Change the current preset."""
temperature = self.HA_TO_SLEEPIQ_CORE_TEMP_MAP[option]
timer = self.core_climate.timer or 240
if temperature == CoreTemps.OFF:
await self.core_climate.turn_off()
else:
await self.core_climate.turn_on(temperature, timer)
self._attr_current_option = option
await self.coordinator.async_request_refresh()
self.async_write_ha_state()
+159
View File
@@ -0,0 +1,159 @@
"""Support for SleepIQ sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from typing import override
from asyncsleepiq import SleepIQBed, SleepIQSleeper
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import PRESSURE, UnitOfTime
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
HEART_RATE,
HRV,
RESPIRATORY_RATE,
SLEEP_DURATION,
SLEEP_NUMBER,
SLEEP_SCORE,
)
from .coordinator import (
SleepIQConfigEntry,
SleepIQDataUpdateCoordinator,
SleepIQSleepDataCoordinator,
)
from .entity import SleepIQSleeperEntity
@dataclass(frozen=True, kw_only=True)
class SleepIQSensorEntityDescription(SensorEntityDescription):
"""Describes SleepIQ sensor entity."""
value_fn: Callable[[SleepIQSleeper], float | int | None]
BED_SENSORS: tuple[SleepIQSensorEntityDescription, ...] = (
SleepIQSensorEntityDescription(
key=PRESSURE,
translation_key="pressure",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda sleeper: sleeper.pressure,
),
SleepIQSensorEntityDescription(
key=SLEEP_NUMBER,
translation_key="sleep_number",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda sleeper: sleeper.sleep_number,
),
)
SLEEP_HEALTH_SENSORS: tuple[SleepIQSensorEntityDescription, ...] = (
SleepIQSensorEntityDescription(
key=SLEEP_SCORE,
translation_key="sleep_score",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="score",
value_fn=lambda sleeper: (
sleeper.sleep_data.sleep_score if sleeper.sleep_data else None
),
),
SleepIQSensorEntityDescription(
key=SLEEP_DURATION,
translation_key="sleep_duration",
device_class=SensorDeviceClass.DURATION,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.HOURS,
suggested_display_precision=1,
value_fn=lambda sleeper: (
round(sleeper.sleep_data.duration / 3600, 1)
if sleeper.sleep_data and sleeper.sleep_data.duration
else None
),
),
SleepIQSensorEntityDescription(
key=HEART_RATE,
translation_key="heart_rate_avg",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="bpm",
value_fn=lambda sleeper: (
sleeper.sleep_data.heart_rate if sleeper.sleep_data else None
),
),
SleepIQSensorEntityDescription(
key=RESPIRATORY_RATE,
translation_key="respiratory_rate_avg",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="brpm",
value_fn=lambda sleeper: (
sleeper.sleep_data.respiratory_rate if sleeper.sleep_data else None
),
),
SleepIQSensorEntityDescription(
key=HRV,
translation_key="hrv",
device_class=SensorDeviceClass.DURATION,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
value_fn=lambda sleeper: sleeper.sleep_data.hrv if sleeper.sleep_data else None,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the SleepIQ bed sensors."""
data = entry.runtime_data
entities: list[SensorEntity] = []
entities.extend(
SleepIQSensorEntity(data.data_coordinator, bed, sleeper, description)
for bed in data.client.beds.values()
for sleeper in bed.sleepers
for description in BED_SENSORS
)
entities.extend(
SleepIQSensorEntity(data.sleep_data_coordinator, bed, sleeper, description)
for bed in data.client.beds.values()
for sleeper in bed.sleepers
for description in SLEEP_HEALTH_SENSORS
)
async_add_entities(entities)
class SleepIQSensorEntity(
SleepIQSleeperEntity[SleepIQDataUpdateCoordinator | SleepIQSleepDataCoordinator],
SensorEntity,
):
"""Representation of a SleepIQ sensor."""
entity_description: SleepIQSensorEntityDescription
def __init__(
self,
coordinator: SleepIQDataUpdateCoordinator | SleepIQSleepDataCoordinator,
bed: SleepIQBed,
sleeper: SleepIQSleeper,
description: SleepIQSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(coordinator, bed, sleeper, description.key)
@callback
@override
def _async_update_attrs(self) -> None:
"""Update sensor attributes."""
self._attr_native_value = self.entity_description.value_fn(self.sleeper)
+50
View File
@@ -0,0 +1,50 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"step": {
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"description": "The SleepIQ integration needs to re-authenticate your account {username}.",
"title": "[%key:common::config_flow::title::reauth%]"
},
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
}
}
}
},
"entity": {
"select": {
"core_temps": {
"state": {
"cooling_high": "Cooling high",
"cooling_low": "Cooling low",
"cooling_medium": "Cooling medium",
"heating_high": "Heating high",
"heating_low": "Heating low",
"heating_medium": "Heating medium",
"off": "[%key:common::state::off%]"
}
},
"foot_warmer_temp": {
"state": {
"high": "[%key:common::state::high%]",
"low": "[%key:common::state::low%]",
"medium": "[%key:common::state::medium%]",
"off": "[%key:common::state::off%]"
}
}
}
}
}
+57
View File
@@ -0,0 +1,57 @@
"""Support for SleepIQ switches."""
from typing import Any, override
from asyncsleepiq import SleepIQBed
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SleepIQConfigEntry, SleepIQPauseUpdateCoordinator
from .entity import SleepIQBedEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: SleepIQConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sleep number switches."""
data = entry.runtime_data
async_add_entities(
SleepNumberPrivateSwitch(data.pause_coordinator, bed)
for bed in data.client.beds.values()
)
class SleepNumberPrivateSwitch(
SleepIQBedEntity[SleepIQPauseUpdateCoordinator], SwitchEntity
):
"""Representation of SleepIQ privacy mode."""
def __init__(
self, coordinator: SleepIQPauseUpdateCoordinator, bed: SleepIQBed
) -> None:
"""Initialize the switch."""
super().__init__(coordinator, bed)
self._attr_name = f"SleepNumber {bed.name} Pause Mode"
self._attr_unique_id = f"{bed.id}-pause-mode"
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch."""
await self.bed.set_pause_mode(True)
self._handle_coordinator_update()
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch."""
await self.bed.set_pause_mode(False)
self._handle_coordinator_update()
@callback
@override
def _async_update_attrs(self) -> None:
"""Update switch attributes."""
self._attr_is_on = self.bed.paused