jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-26 02:07:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove Legacy Works With Nest (#96111)

Browse Source 
* Remove Legacy Works With Nest

* Simplify nest configuration

* Cleanup legacy nest config entries
This commit is contained in:
Allen Porter 2023-07-09 19:38:05 -07:00 committed by GitHub
parent 1c54b2e025
commit c4a39bbfb1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
28 changed files with 704 additions and 2701 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -755,7 +755,6 @@ omit =
homeassistant/components/neato/switch.py
homeassistant/components/neato/vacuum.py
homeassistant/components/nederlandse_spoorwegen/sensor.py
homeassistant/components/nest/legacy/*
homeassistant/components/netdata/sensor.py
homeassistant/components/netgear/__init__.py
homeassistant/components/netgear/button.py

34
homeassistant/components/nest/__init__.py
View File

@ -46,23 +46,22 @@ from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.helpers.entity_registry import async_entries_for_device
from homeassistant.helpers.typing import ConfigType
from . import api, config_flow
from . import api
from .const import (
CONF_PROJECT_ID,
CONF_SUBSCRIBER_ID,
CONF_SUBSCRIBER_ID_IMPORTED,
DATA_DEVICE_MANAGER,
DATA_NEST_CONFIG,
DATA_SDM,
DATA_SUBSCRIBER,
DOMAIN,
)
from .events import EVENT_NAME_MAP, NEST_EVENT
from .legacy import async_setup_legacy, async_setup_legacy_entry
from .media_source import (
async_get_media_event_store,
async_get_media_source_devices,
@ -114,15 +113,20 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass.http.register_view(NestEventMediaView(hass))
hass.http.register_view(NestEventMediaThumbnailView(hass))
if DOMAIN not in config:
return True # ConfigMode.SDM_APPLICATION_CREDENTIALS
hass.data[DOMAIN][DATA_NEST_CONFIG] = config[DOMAIN]
config_mode = config_flow.get_config_mode(hass)
if config_mode == config_flow.ConfigMode.LEGACY:
return await async_setup_legacy(hass, config)
if DOMAIN in config and CONF_PROJECT_ID not in config[DOMAIN]:
ir.async_create_issue(
hass,
DOMAIN,
"legacy_nest_deprecated",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="legacy_nest_deprecated",
translation_placeholders={
"documentation_url": "https://www.home-assistant.io/integrations/nest/",
},
)
return False
return True
@ -167,9 +171,9 @@ class SignalUpdateCallback:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Nest from a config entry with dispatch between old/new flows."""
config_mode = config_flow.get_config_mode(hass)
if DATA_SDM not in entry.data or config_mode == config_flow.ConfigMode.LEGACY:
return await async_setup_legacy_entry(hass, entry)
if DATA_SDM not in entry.data:
hass.async_create_task(hass.config_entries.async_remove(entry.entry_id))
return False
if entry.unique_id != entry.data[CONF_PROJECT_ID]:
hass.config_entries.async_update_entry(

223
homeassistant/components/nest/camera.py
View File

@ -1,19 +1,228 @@
"""Support for Nest cameras that dispatches between API versions."""
"""Support for Google Nest SDM Cameras."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
import datetime
import functools
import logging
from pathlib import Path
from google_nest_sdm.camera_traits import (
CameraImageTrait,
CameraLiveStreamTrait,
RtspStream,
StreamingProtocol,
)
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.exceptions import ApiException
from homeassistant.components.camera import Camera, CameraEntityFeature, StreamType
from homeassistant.components.stream import CONF_EXTRA_PART_WAIT_TIME
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util.dt import utcnow
from .camera_sdm import async_setup_sdm_entry
from .const import DATA_SDM
from .legacy.camera import async_setup_legacy_entry
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
_LOGGER = logging.getLogger(__name__)
PLACEHOLDER = Path(__file__).parent / "placeholder.png"
# Used to schedule an alarm to refresh the stream before expiration
STREAM_EXPIRATION_BUFFER = datetime.timedelta(seconds=30)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the cameras."""
if DATA_SDM not in entry.data:
await async_setup_legacy_entry(hass, entry, async_add_entities)
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities = []
for device in device_manager.devices.values():
if (
CameraImageTrait.NAME in device.traits
or CameraLiveStreamTrait.NAME in device.traits
):
entities.append(NestCamera(device))
async_add_entities(entities)
class NestCamera(Camera):
"""Devices that support cameras."""
_attr_has_entity_name = True
_attr_name = None
def __init__(self, device: Device) -> None:
"""Initialize the camera."""
super().__init__()
self._device = device
self._device_info = NestDeviceInfo(device)
self._stream: RtspStream | None = None
self._create_stream_url_lock = asyncio.Lock()
self._stream_refresh_unsub: Callable[[], None] | None = None
self._attr_is_streaming = CameraLiveStreamTrait.NAME in self._device.traits
self.stream_options[CONF_EXTRA_PART_WAIT_TIME] = 3
@property
def unique_id(self) -> str:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return f"{self._device.name}-camera"
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
return self._device_info.device_info
@property
def brand(self) -> str | None:
"""Return the camera brand."""
return self._device_info.device_brand
@property
def model(self) -> str | None:
"""Return the camera model."""
return self._device_info.device_model
@property
def supported_features(self) -> CameraEntityFeature:
"""Flag supported features."""
supported_features = CameraEntityFeature(0)
if CameraLiveStreamTrait.NAME in self._device.traits:
supported_features |= CameraEntityFeature.STREAM
return supported_features
@property
def frontend_stream_type(self) -> StreamType | None:
"""Return the type of stream supported by this camera."""
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.WEB_RTC in trait.supported_protocols:
return StreamType.WEB_RTC
return super().frontend_stream_type
@property
def available(self) -> bool:
"""Return True if entity is available."""
# Cameras are marked unavailable on stream errors in #54659 however nest
# streams have a high error rate (#60353). Given nest streams are so flaky,
# marking the stream unavailable has other side effects like not showing
# the camera image which sometimes are still able to work. Until the
# streams are fixed, just leave the streams as available.
return True
async def stream_source(self) -> str | None:
"""Return the source of the stream."""
if not self.supported_features & CameraEntityFeature.STREAM:
return None
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.RTSP not in trait.supported_protocols:
return None
async with self._create_stream_url_lock:
if not self._stream:
_LOGGER.debug("Fetching stream url")
try:
self._stream = await trait.generate_rtsp_stream()
except ApiException as err:
raise HomeAssistantError(f"Nest API error: {err}") from err
self._schedule_stream_refresh()
assert self._stream
if self._stream.expires_at < utcnow():
_LOGGER.warning("Stream already expired")
return self._stream.rtsp_stream_url
def _schedule_stream_refresh(self) -> None:
"""Schedules an alarm to refresh the stream url before expiration."""
assert self._stream
_LOGGER.debug("New stream url expires at %s", self._stream.expires_at)
refresh_time = self._stream.expires_at - STREAM_EXPIRATION_BUFFER
# Schedule an alarm to extend the stream
if self._stream_refresh_unsub is not None:
self._stream_refresh_unsub()
self._stream_refresh_unsub = async_track_point_in_utc_time(
self.hass,
self._handle_stream_refresh,
refresh_time,
)
async def _handle_stream_refresh(self, now: datetime.datetime) -> None:
"""Alarm that fires to check if the stream should be refreshed."""
if not self._stream:
return
await async_setup_sdm_entry(hass, entry, async_add_entities)
_LOGGER.debug("Extending stream url")
try:
self._stream = await self._stream.extend_rtsp_stream()
except ApiException as err:
_LOGGER.debug("Failed to extend stream: %s", err)
# Next attempt to catch a url will get a new one
self._stream = None
if self.stream:
await self.stream.stop()
self.stream = None
return
# Update the stream worker with the latest valid url
if self.stream:
self.stream.update_source(self._stream.rtsp_stream_url)
self._schedule_stream_refresh()
async def async_will_remove_from_hass(self) -> None:
"""Invalidates the RTSP token when unloaded."""
if self._stream:
_LOGGER.debug("Invalidating stream")
try:
await self._stream.stop_rtsp_stream()
except ApiException as err:
_LOGGER.debug(
"Failed to revoke stream token, will rely on ttl: %s", err
)
if self._stream_refresh_unsub:
self._stream_refresh_unsub()
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
# Use the thumbnail from RTSP stream, or a placeholder if stream is
# not supported (e.g. WebRTC)
stream = await self.async_create_stream()
if stream:
return await stream.async_get_image(width, height)
return await self.hass.async_add_executor_job(self.placeholder_image)
@classmethod
@functools.cache
def placeholder_image(cls) -> bytes:
"""Return placeholder image to use when no stream is available."""
return PLACEHOLDER.read_bytes()
async def async_handle_web_rtc_offer(self, offer_sdp: str) -> str | None:
"""Return the source of the stream."""
trait: CameraLiveStreamTrait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.WEB_RTC not in trait.supported_protocols:
return await super().async_handle_web_rtc_offer(offer_sdp)
try:
stream = await trait.generate_web_rtc_stream(offer_sdp)
except ApiException as err:
raise HomeAssistantError(f"Nest API error: {err}") from err
return stream.answer_sdp

228
homeassistant/components/nest/camera_sdm.py
View File

@ -1,228 +0,0 @@
"""Support for Google Nest SDM Cameras."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
import datetime
import functools
import logging
from pathlib import Path
from google_nest_sdm.camera_traits import (
CameraImageTrait,
CameraLiveStreamTrait,
RtspStream,
StreamingProtocol,
)
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.exceptions import ApiException
from homeassistant.components.camera import Camera, CameraEntityFeature, StreamType
from homeassistant.components.stream import CONF_EXTRA_PART_WAIT_TIME
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util.dt import utcnow
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
_LOGGER = logging.getLogger(__name__)
PLACEHOLDER = Path(__file__).parent / "placeholder.png"
# Used to schedule an alarm to refresh the stream before expiration
STREAM_EXPIRATION_BUFFER = datetime.timedelta(seconds=30)
async def async_setup_sdm_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the cameras."""
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities = []
for device in device_manager.devices.values():
if (
CameraImageTrait.NAME in device.traits
or CameraLiveStreamTrait.NAME in device.traits
):
entities.append(NestCamera(device))
async_add_entities(entities)
class NestCamera(Camera):
"""Devices that support cameras."""
_attr_has_entity_name = True
_attr_name = None
def __init__(self, device: Device) -> None:
"""Initialize the camera."""
super().__init__()
self._device = device
self._device_info = NestDeviceInfo(device)
self._stream: RtspStream | None = None
self._create_stream_url_lock = asyncio.Lock()
self._stream_refresh_unsub: Callable[[], None] | None = None
self._attr_is_streaming = CameraLiveStreamTrait.NAME in self._device.traits
self.stream_options[CONF_EXTRA_PART_WAIT_TIME] = 3
@property
def unique_id(self) -> str:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return f"{self._device.name}-camera"
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
return self._device_info.device_info
@property
def brand(self) -> str | None:
"""Return the camera brand."""
return self._device_info.device_brand
@property
def model(self) -> str | None:
"""Return the camera model."""
return self._device_info.device_model
@property
def supported_features(self) -> CameraEntityFeature:
"""Flag supported features."""
supported_features = CameraEntityFeature(0)
if CameraLiveStreamTrait.NAME in self._device.traits:
supported_features |= CameraEntityFeature.STREAM
return supported_features
@property
def frontend_stream_type(self) -> StreamType | None:
"""Return the type of stream supported by this camera."""
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.WEB_RTC in trait.supported_protocols:
return StreamType.WEB_RTC
return super().frontend_stream_type
@property
def available(self) -> bool:
"""Return True if entity is available."""
# Cameras are marked unavailable on stream errors in #54659 however nest
# streams have a high error rate (#60353). Given nest streams are so flaky,
# marking the stream unavailable has other side effects like not showing
# the camera image which sometimes are still able to work. Until the
# streams are fixed, just leave the streams as available.
return True
async def stream_source(self) -> str | None:
"""Return the source of the stream."""
if not self.supported_features & CameraEntityFeature.STREAM:
return None
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.RTSP not in trait.supported_protocols:
return None
async with self._create_stream_url_lock:
if not self._stream:
_LOGGER.debug("Fetching stream url")
try:
self._stream = await trait.generate_rtsp_stream()
except ApiException as err:
raise HomeAssistantError(f"Nest API error: {err}") from err
self._schedule_stream_refresh()
assert self._stream
if self._stream.expires_at < utcnow():
_LOGGER.warning("Stream already expired")
return self._stream.rtsp_stream_url
def _schedule_stream_refresh(self) -> None:
"""Schedules an alarm to refresh the stream url before expiration."""
assert self._stream
_LOGGER.debug("New stream url expires at %s", self._stream.expires_at)
refresh_time = self._stream.expires_at - STREAM_EXPIRATION_BUFFER
# Schedule an alarm to extend the stream
if self._stream_refresh_unsub is not None:
self._stream_refresh_unsub()
self._stream_refresh_unsub = async_track_point_in_utc_time(
self.hass,
self._handle_stream_refresh,
refresh_time,
)
async def _handle_stream_refresh(self, now: datetime.datetime) -> None:
"""Alarm that fires to check if the stream should be refreshed."""
if not self._stream:
return
_LOGGER.debug("Extending stream url")
try:
self._stream = await self._stream.extend_rtsp_stream()
except ApiException as err:
_LOGGER.debug("Failed to extend stream: %s", err)
# Next attempt to catch a url will get a new one
self._stream = None
if self.stream:
await self.stream.stop()
self.stream = None
return
# Update the stream worker with the latest valid url
if self.stream:
self.stream.update_source(self._stream.rtsp_stream_url)
self._schedule_stream_refresh()
async def async_will_remove_from_hass(self) -> None:
"""Invalidates the RTSP token when unloaded."""
if self._stream:
_LOGGER.debug("Invalidating stream")
try:
await self._stream.stop_rtsp_stream()
except ApiException as err:
_LOGGER.debug(
"Failed to revoke stream token, will rely on ttl: %s", err
)
if self._stream_refresh_unsub:
self._stream_refresh_unsub()
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
# Use the thumbnail from RTSP stream, or a placeholder if stream is
# not supported (e.g. WebRTC)
stream = await self.async_create_stream()
if stream:
return await stream.async_get_image(width, height)
return await self.hass.async_add_executor_job(self.placeholder_image)
@classmethod
@functools.cache
def placeholder_image(cls) -> bytes:
"""Return placeholder image to use when no stream is available."""
return PLACEHOLDER.read_bytes()
async def async_handle_web_rtc_offer(self, offer_sdp: str) -> str | None:
"""Return the source of the stream."""
trait: CameraLiveStreamTrait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.WEB_RTC not in trait.supported_protocols:
return await super().async_handle_web_rtc_offer(offer_sdp)
try:
stream = await trait.generate_web_rtc_stream(offer_sdp)
except ApiException as err:
raise HomeAssistantError(f"Nest API error: {err}") from err
return stream.answer_sdp

354
homeassistant/components/nest/climate.py
View File

@ -1,19 +1,357 @@
"""Support for Nest climate that dispatches between API versions."""
"""Support for Google Nest SDM climate devices."""
from __future__ import annotations
from typing import Any, cast
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.device_traits import FanTrait, TemperatureTrait
from google_nest_sdm.exceptions import ApiException
from google_nest_sdm.thermostat_traits import (
ThermostatEcoTrait,
ThermostatHeatCoolTrait,
ThermostatHvacTrait,
ThermostatModeTrait,
ThermostatTemperatureSetpointTrait,
)
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
ATTR_TARGET_TEMP_HIGH,
ATTR_TARGET_TEMP_LOW,
FAN_OFF,
FAN_ON,
PRESET_ECO,
PRESET_NONE,
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .climate_sdm import async_setup_sdm_entry
from .const import DATA_SDM
from .legacy.climate import async_setup_legacy_entry
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
# Mapping for sdm.devices.traits.ThermostatMode mode field
THERMOSTAT_MODE_MAP: dict[str, HVACMode] = {
"OFF": HVACMode.OFF,
"HEAT": HVACMode.HEAT,
"COOL": HVACMode.COOL,
"HEATCOOL": HVACMode.HEAT_COOL,
}
THERMOSTAT_INV_MODE_MAP = {v: k for k, v in THERMOSTAT_MODE_MAP.items()}
# Mode for sdm.devices.traits.ThermostatEco
THERMOSTAT_ECO_MODE = "MANUAL_ECO"
# Mapping for sdm.devices.traits.ThermostatHvac status field
THERMOSTAT_HVAC_STATUS_MAP = {
"OFF": HVACAction.OFF,
"HEATING": HVACAction.HEATING,
"COOLING": HVACAction.COOLING,
}
THERMOSTAT_RANGE_MODES = [HVACMode.HEAT_COOL, HVACMode.AUTO]
PRESET_MODE_MAP = {
"MANUAL_ECO": PRESET_ECO,
"OFF": PRESET_NONE,
}
PRESET_INV_MODE_MAP = {v: k for k, v in PRESET_MODE_MAP.items()}
FAN_MODE_MAP = {
"ON": FAN_ON,
"OFF": FAN_OFF,
}
FAN_INV_MODE_MAP = {v: k for k, v in FAN_MODE_MAP.items()}
FAN_INV_MODES = list(FAN_INV_MODE_MAP)
MAX_FAN_DURATION = 43200 # 15 hours is the max in the SDM API
MIN_TEMP = 10
MAX_TEMP = 32
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the climate platform."""
if DATA_SDM not in entry.data:
await async_setup_legacy_entry(hass, entry, async_add_entities)
"""Set up the client entities."""
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities = []
for device in device_manager.devices.values():
if ThermostatHvacTrait.NAME in device.traits:
entities.append(ThermostatEntity(device))
async_add_entities(entities)
class ThermostatEntity(ClimateEntity):
"""A nest thermostat climate entity."""
_attr_min_temp = MIN_TEMP
_attr_max_temp = MAX_TEMP
_attr_has_entity_name = True
_attr_should_poll = False
_attr_name = None
def __init__(self, device: Device) -> None:
"""Initialize ThermostatEntity."""
self._device = device
self._device_info = NestDeviceInfo(device)
@property
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return self._device.name
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
return self._device_info.device_info
@property
def available(self) -> bool:
"""Return device availability."""
return self._device_info.available
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self._attr_supported_features = self._get_supported_features()
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
@property
def temperature_unit(self) -> str:
"""Return the unit of temperature measurement for the system."""
return UnitOfTemperature.CELSIUS
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
if TemperatureTrait.NAME not in self._device.traits:
return None
trait: TemperatureTrait = self._device.traits[TemperatureTrait.NAME]
return trait.ambient_temperature_celsius
@property
def target_temperature(self) -> float | None:
"""Return the temperature currently set to be reached."""
if not (trait := self._target_temperature_trait):
return None
if self.hvac_mode == HVACMode.HEAT:
return trait.heat_celsius
if self.hvac_mode == HVACMode.COOL:
return trait.cool_celsius
return None
@property
def target_temperature_high(self) -> float | None:
"""Return the upper bound target temperature."""
if self.hvac_mode != HVACMode.HEAT_COOL:
return None
if not (trait := self._target_temperature_trait):
return None
return trait.cool_celsius
@property
def target_temperature_low(self) -> float | None:
"""Return the lower bound target temperature."""
if self.hvac_mode != HVACMode.HEAT_COOL:
return None
if not (trait := self._target_temperature_trait):
return None
return trait.heat_celsius
@property
def _target_temperature_trait(
self,
) -> ThermostatHeatCoolTrait | None:
"""Return the correct trait with a target temp depending on mode."""
if (
self.preset_mode == PRESET_ECO
and ThermostatEcoTrait.NAME in self._device.traits
):
return cast(
ThermostatEcoTrait, self._device.traits[ThermostatEcoTrait.NAME]
)
if ThermostatTemperatureSetpointTrait.NAME in self._device.traits:
return cast(
ThermostatTemperatureSetpointTrait,
self._device.traits[ThermostatTemperatureSetpointTrait.NAME],
)
return None
@property
def hvac_mode(self) -> HVACMode:
"""Return the current operation (e.g. heat, cool, idle)."""
hvac_mode = HVACMode.OFF
if ThermostatModeTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatModeTrait.NAME]
if trait.mode in THERMOSTAT_MODE_MAP:
hvac_mode = THERMOSTAT_MODE_MAP[trait.mode]
return hvac_mode
@property
def hvac_modes(self) -> list[HVACMode]:
"""List of available operation modes."""
supported_modes = []
for mode in self._get_device_hvac_modes:
if mode in THERMOSTAT_MODE_MAP:
supported_modes.append(THERMOSTAT_MODE_MAP[mode])
return supported_modes
@property
def _get_device_hvac_modes(self) -> set[str]:
"""Return the set of SDM API hvac modes supported by the device."""
modes = []
if ThermostatModeTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatModeTrait.NAME]
modes.extend(trait.available_modes)
return set(modes)
@property
def hvac_action(self) -> HVACAction | None:
"""Return the current HVAC action (heating, cooling)."""
trait = self._device.traits[ThermostatHvacTrait.NAME]
if trait.status == "OFF" and self.hvac_mode != HVACMode.OFF:
return HVACAction.IDLE
return THERMOSTAT_HVAC_STATUS_MAP.get(trait.status)
@property
def preset_mode(self) -> str:
"""Return the current active preset."""
if ThermostatEcoTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatEcoTrait.NAME]
return PRESET_MODE_MAP.get(trait.mode, PRESET_NONE)
return PRESET_NONE
@property
def preset_modes(self) -> list[str]:
"""Return the available presets."""
modes = []
if ThermostatEcoTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatEcoTrait.NAME]
for mode in trait.available_modes:
if mode in PRESET_MODE_MAP:
modes.append(PRESET_MODE_MAP[mode])
return modes
@property
def fan_mode(self) -> str:
"""Return the current fan mode."""
if (
self.supported_features & ClimateEntityFeature.FAN_MODE
and FanTrait.NAME in self._device.traits
):
trait = self._device.traits[FanTrait.NAME]
return FAN_MODE_MAP.get(trait.timer_mode, FAN_OFF)
return FAN_OFF
@property
def fan_modes(self) -> list[str]:
"""Return the list of available fan modes."""
if (
self.supported_features & ClimateEntityFeature.FAN_MODE
and FanTrait.NAME in self._device.traits
):
return FAN_INV_MODES
return []
def _get_supported_features(self) -> ClimateEntityFeature:
"""Compute the bitmap of supported features from the current state."""
features = ClimateEntityFeature(0)
if HVACMode.HEAT_COOL in self.hvac_modes:
features |= ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
if HVACMode.HEAT in self.hvac_modes or HVACMode.COOL in self.hvac_modes:
features |= ClimateEntityFeature.TARGET_TEMPERATURE
if ThermostatEcoTrait.NAME in self._device.traits:
features |= ClimateEntityFeature.PRESET_MODE
if FanTrait.NAME in self._device.traits:
# Fan trait may be present without actually support fan mode
fan_trait = self._device.traits[FanTrait.NAME]
if fan_trait.timer_mode is not None:
features |= ClimateEntityFeature.FAN_MODE
return features
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode."""
if hvac_mode not in self.hvac_modes:
raise ValueError(f"Unsupported hvac_mode '{hvac_mode}'")
api_mode = THERMOSTAT_INV_MODE_MAP[hvac_mode]
trait = self._device.traits[ThermostatModeTrait.NAME]
try:
await trait.set_mode(api_mode)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} HVAC mode to {hvac_mode}: {err}"
) from err
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
hvac_mode = self.hvac_mode
if kwargs.get(ATTR_HVAC_MODE) is not None:
hvac_mode = kwargs[ATTR_HVAC_MODE]
await self.async_set_hvac_mode(hvac_mode)
low_temp = kwargs.get(ATTR_TARGET_TEMP_LOW)
high_temp = kwargs.get(ATTR_TARGET_TEMP_HIGH)
temp = kwargs.get(ATTR_TEMPERATURE)
if ThermostatTemperatureSetpointTrait.NAME not in self._device.traits:
raise HomeAssistantError(
f"Error setting {self.entity_id} temperature to {kwargs}: "
"Unable to find setpoint trait."
)
trait = self._device.traits[ThermostatTemperatureSetpointTrait.NAME]
try:
if self.preset_mode == PRESET_ECO or hvac_mode == HVACMode.HEAT_COOL:
if low_temp and high_temp:
await trait.set_range(low_temp, high_temp)
elif hvac_mode == HVACMode.COOL and temp:
await trait.set_cool(temp)
elif hvac_mode == HVACMode.HEAT and temp:
await trait.set_heat(temp)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} temperature to {kwargs}: {err}"
) from err
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new target preset mode."""
if preset_mode not in self.preset_modes:
raise ValueError(f"Unsupported preset_mode '{preset_mode}'")
if self.preset_mode == preset_mode: # API doesn't like duplicate preset modes
return
await async_setup_sdm_entry(hass, entry, async_add_entities)
trait = self._device.traits[ThermostatEcoTrait.NAME]
try:
await trait.set_mode(PRESET_INV_MODE_MAP[preset_mode])
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} preset mode to {preset_mode}: {err}"
) from err
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan_mode '{fan_mode}'")
if fan_mode == FAN_ON and self.hvac_mode == HVACMode.OFF:
raise ValueError(
"Cannot turn on fan, please set an HVAC mode (e.g. heat/cool) first"
)
trait = self._device.traits[FanTrait.NAME]
duration = None
if fan_mode != FAN_OFF:
duration = MAX_FAN_DURATION
try:
await trait.set_timer(FAN_INV_MODE_MAP[fan_mode], duration=duration)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} fan mode to {fan_mode}: {err}"
) from err

357
homeassistant/components/nest/climate_sdm.py
View File

@ -1,357 +0,0 @@
"""Support for Google Nest SDM climate devices."""
from __future__ import annotations
from typing import Any, cast
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.device_traits import FanTrait, TemperatureTrait
from google_nest_sdm.exceptions import ApiException
from google_nest_sdm.thermostat_traits import (
ThermostatEcoTrait,
ThermostatHeatCoolTrait,
ThermostatHvacTrait,
ThermostatModeTrait,
ThermostatTemperatureSetpointTrait,
)
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
ATTR_TARGET_TEMP_HIGH,
ATTR_TARGET_TEMP_LOW,
FAN_OFF,
FAN_ON,
PRESET_ECO,
PRESET_NONE,
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
# Mapping for sdm.devices.traits.ThermostatMode mode field
THERMOSTAT_MODE_MAP: dict[str, HVACMode] = {
"OFF": HVACMode.OFF,
"HEAT": HVACMode.HEAT,
"COOL": HVACMode.COOL,
"HEATCOOL": HVACMode.HEAT_COOL,
}
THERMOSTAT_INV_MODE_MAP = {v: k for k, v in THERMOSTAT_MODE_MAP.items()}
# Mode for sdm.devices.traits.ThermostatEco
THERMOSTAT_ECO_MODE = "MANUAL_ECO"
# Mapping for sdm.devices.traits.ThermostatHvac status field
THERMOSTAT_HVAC_STATUS_MAP = {
"OFF": HVACAction.OFF,
"HEATING": HVACAction.HEATING,
"COOLING": HVACAction.COOLING,
}
THERMOSTAT_RANGE_MODES = [HVACMode.HEAT_COOL, HVACMode.AUTO]
PRESET_MODE_MAP = {
"MANUAL_ECO": PRESET_ECO,
"OFF": PRESET_NONE,
}
PRESET_INV_MODE_MAP = {v: k for k, v in PRESET_MODE_MAP.items()}
FAN_MODE_MAP = {
"ON": FAN_ON,
"OFF": FAN_OFF,
}
FAN_INV_MODE_MAP = {v: k for k, v in FAN_MODE_MAP.items()}
FAN_INV_MODES = list(FAN_INV_MODE_MAP)
MAX_FAN_DURATION = 43200 # 15 hours is the max in the SDM API
MIN_TEMP = 10
MAX_TEMP = 32
async def async_setup_sdm_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the client entities."""
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities = []
for device in device_manager.devices.values():
if ThermostatHvacTrait.NAME in device.traits:
entities.append(ThermostatEntity(device))
async_add_entities(entities)
class ThermostatEntity(ClimateEntity):
"""A nest thermostat climate entity."""
_attr_min_temp = MIN_TEMP
_attr_max_temp = MAX_TEMP
_attr_has_entity_name = True
_attr_should_poll = False
_attr_name = None
def __init__(self, device: Device) -> None:
"""Initialize ThermostatEntity."""
self._device = device
self._device_info = NestDeviceInfo(device)
@property
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return self._device.name
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
return self._device_info.device_info
@property
def available(self) -> bool:
"""Return device availability."""
return self._device_info.available
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self._attr_supported_features = self._get_supported_features()
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
@property
def temperature_unit(self) -> str:
"""Return the unit of temperature measurement for the system."""
return UnitOfTemperature.CELSIUS
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
if TemperatureTrait.NAME not in self._device.traits:
return None
trait: TemperatureTrait = self._device.traits[TemperatureTrait.NAME]
return trait.ambient_temperature_celsius
@property
def target_temperature(self) -> float | None:
"""Return the temperature currently set to be reached."""
if not (trait := self._target_temperature_trait):
return None
if self.hvac_mode == HVACMode.HEAT:
return trait.heat_celsius
if self.hvac_mode == HVACMode.COOL:
return trait.cool_celsius
return None
@property
def target_temperature_high(self) -> float | None:
"""Return the upper bound target temperature."""
if self.hvac_mode != HVACMode.HEAT_COOL:
return None
if not (trait := self._target_temperature_trait):
return None
return trait.cool_celsius
@property
def target_temperature_low(self) -> float | None:
"""Return the lower bound target temperature."""
if self.hvac_mode != HVACMode.HEAT_COOL:
return None
if not (trait := self._target_temperature_trait):
return None
return trait.heat_celsius
@property
def _target_temperature_trait(
self,
) -> ThermostatHeatCoolTrait | None:
"""Return the correct trait with a target temp depending on mode."""
if (
self.preset_mode == PRESET_ECO
and ThermostatEcoTrait.NAME in self._device.traits
):
return cast(
ThermostatEcoTrait, self._device.traits[ThermostatEcoTrait.NAME]
)
if ThermostatTemperatureSetpointTrait.NAME in self._device.traits:
return cast(
ThermostatTemperatureSetpointTrait,
self._device.traits[ThermostatTemperatureSetpointTrait.NAME],
)
return None
@property
def hvac_mode(self) -> HVACMode:
"""Return the current operation (e.g. heat, cool, idle)."""
hvac_mode = HVACMode.OFF
if ThermostatModeTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatModeTrait.NAME]
if trait.mode in THERMOSTAT_MODE_MAP:
hvac_mode = THERMOSTAT_MODE_MAP[trait.mode]
return hvac_mode
@property
def hvac_modes(self) -> list[HVACMode]:
"""List of available operation modes."""
supported_modes = []
for mode in self._get_device_hvac_modes:
if mode in THERMOSTAT_MODE_MAP:
supported_modes.append(THERMOSTAT_MODE_MAP[mode])
return supported_modes
@property
def _get_device_hvac_modes(self) -> set[str]:
"""Return the set of SDM API hvac modes supported by the device."""
modes = []
if ThermostatModeTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatModeTrait.NAME]
modes.extend(trait.available_modes)
return set(modes)
@property
def hvac_action(self) -> HVACAction | None:
"""Return the current HVAC action (heating, cooling)."""
trait = self._device.traits[ThermostatHvacTrait.NAME]
if trait.status == "OFF" and self.hvac_mode != HVACMode.OFF:
return HVACAction.IDLE
return THERMOSTAT_HVAC_STATUS_MAP.get(trait.status)
@property
def preset_mode(self) -> str:
"""Return the current active preset."""
if ThermostatEcoTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatEcoTrait.NAME]
return PRESET_MODE_MAP.get(trait.mode, PRESET_NONE)
return PRESET_NONE
@property
def preset_modes(self) -> list[str]:
"""Return the available presets."""
modes = []
if ThermostatEcoTrait.NAME in self._device.traits:
trait = self._device.traits[ThermostatEcoTrait.NAME]
for mode in trait.available_modes:
if mode in PRESET_MODE_MAP:
modes.append(PRESET_MODE_MAP[mode])
return modes
@property
def fan_mode(self) -> str:
"""Return the current fan mode."""
if (
self.supported_features & ClimateEntityFeature.FAN_MODE
and FanTrait.NAME in self._device.traits
):
trait = self._device.traits[FanTrait.NAME]
return FAN_MODE_MAP.get(trait.timer_mode, FAN_OFF)
return FAN_OFF
@property
def fan_modes(self) -> list[str]:
"""Return the list of available fan modes."""
if (
self.supported_features & ClimateEntityFeature.FAN_MODE
and FanTrait.NAME in self._device.traits
):
return FAN_INV_MODES
return []
def _get_supported_features(self) -> ClimateEntityFeature:
"""Compute the bitmap of supported features from the current state."""
features = ClimateEntityFeature(0)
if HVACMode.HEAT_COOL in self.hvac_modes:
features |= ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
if HVACMode.HEAT in self.hvac_modes or HVACMode.COOL in self.hvac_modes:
features |= ClimateEntityFeature.TARGET_TEMPERATURE
if ThermostatEcoTrait.NAME in self._device.traits:
features |= ClimateEntityFeature.PRESET_MODE
if FanTrait.NAME in self._device.traits:
# Fan trait may be present without actually support fan mode
fan_trait = self._device.traits[FanTrait.NAME]
if fan_trait.timer_mode is not None:
features |= ClimateEntityFeature.FAN_MODE
return features
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode."""
if hvac_mode not in self.hvac_modes:
raise ValueError(f"Unsupported hvac_mode '{hvac_mode}'")
api_mode = THERMOSTAT_INV_MODE_MAP[hvac_mode]
trait = self._device.traits[ThermostatModeTrait.NAME]
try:
await trait.set_mode(api_mode)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} HVAC mode to {hvac_mode}: {err}"
) from err
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
hvac_mode = self.hvac_mode
if kwargs.get(ATTR_HVAC_MODE) is not None:
hvac_mode = kwargs[ATTR_HVAC_MODE]
await self.async_set_hvac_mode(hvac_mode)
low_temp = kwargs.get(ATTR_TARGET_TEMP_LOW)
high_temp = kwargs.get(ATTR_TARGET_TEMP_HIGH)
temp = kwargs.get(ATTR_TEMPERATURE)
if ThermostatTemperatureSetpointTrait.NAME not in self._device.traits:
raise HomeAssistantError(
f"Error setting {self.entity_id} temperature to {kwargs}: "
"Unable to find setpoint trait."
)
trait = self._device.traits[ThermostatTemperatureSetpointTrait.NAME]
try:
if self.preset_mode == PRESET_ECO or hvac_mode == HVACMode.HEAT_COOL:
if low_temp and high_temp:
await trait.set_range(low_temp, high_temp)
elif hvac_mode == HVACMode.COOL and temp:
await trait.set_cool(temp)
elif hvac_mode == HVACMode.HEAT and temp:
await trait.set_heat(temp)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} temperature to {kwargs}: {err}"
) from err
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new target preset mode."""
if preset_mode not in self.preset_modes:
raise ValueError(f"Unsupported preset_mode '{preset_mode}'")
if self.preset_mode == preset_mode: # API doesn't like duplicate preset modes
return
trait = self._device.traits[ThermostatEcoTrait.NAME]
try:
await trait.set_mode(PRESET_INV_MODE_MAP[preset_mode])
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} preset mode to {preset_mode}: {err}"
) from err
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan_mode '{fan_mode}'")
if fan_mode == FAN_ON and self.hvac_mode == HVACMode.OFF:
raise ValueError(
"Cannot turn on fan, please set an HVAC mode (e.g. heat/cool) first"
)
trait = self._device.traits[FanTrait.NAME]
duration = None
if fan_mode != FAN_OFF:
duration = MAX_FAN_DURATION
try:
await trait.set_timer(FAN_INV_MODE_MAP[fan_mode], duration=duration)
except ApiException as err:
raise HomeAssistantError(
f"Error setting {self.entity_id} fan mode to {fan_mode}: {err}"
) from err

187
homeassistant/components/nest/config_flow.py
View File

@ -9,15 +9,10 @@ some overrides to custom steps inserted in the middle of the flow.
"""
from __future__ import annotations
import asyncio
from collections import OrderedDict
from collections.abc import Iterable, Mapping
from enum import Enum
import logging
import os
from typing import Any
import async_timeout
from google_nest_sdm.exceptions import (
ApiException,
AuthException,
@ -28,12 +23,9 @@ from google_nest_sdm.structure import InfoTrait, Structure
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.util import get_random_string
from homeassistant.util.json import JsonObjectType, load_json_object
from . import api
from .const import (
@ -71,69 +63,12 @@ DEVICE_ACCESS_CONSOLE_EDIT_URL = (
_LOGGER = logging.getLogger(__name__)
class ConfigMode(Enum):
"""Integration configuration mode."""
SDM = 1 # SDM api with configuration.yaml
LEGACY = 2 # "Works with Nest" API
SDM_APPLICATION_CREDENTIALS = 3 # Config entry only
def get_config_mode(hass: HomeAssistant) -> ConfigMode:
"""Return the integration configuration mode."""
if DOMAIN not in hass.data or not (
config := hass.data[DOMAIN].get(DATA_NEST_CONFIG)
):
return ConfigMode.SDM_APPLICATION_CREDENTIALS
if CONF_PROJECT_ID in config:
return ConfigMode.SDM
return ConfigMode.LEGACY
def _generate_subscription_id(cloud_project_id: str) -> str:
"""Create a new subscription id."""
rnd = get_random_string(SUBSCRIPTION_RAND_LENGTH)
return SUBSCRIPTION_FORMAT.format(cloud_project_id=cloud_project_id, rnd=rnd)
@callback
def register_flow_implementation(
hass: HomeAssistant,
domain: str,
name: str,
gen_authorize_url: str,
convert_code: str,
) -> None:
"""Register a flow implementation for legacy api.
domain: Domain of the component responsible for the implementation.
name: Name of the component.
gen_authorize_url: Coroutine function to generate the authorize url.
convert_code: Coroutine function to convert a code to an access token.
"""
if DATA_FLOW_IMPL not in hass.data:
hass.data[DATA_FLOW_IMPL] = OrderedDict()
hass.data[DATA_FLOW_IMPL][domain] = {
"domain": domain,
"name": name,
"gen_authorize_url": gen_authorize_url,
"convert_code": convert_code,
}
class NestAuthError(HomeAssistantError):
"""Base class for Nest auth errors."""
class CodeInvalid(NestAuthError):
"""Raised when invalid authorization code."""
class UnexpectedStateError(HomeAssistantError):
"""Raised when the config flow is invoked in a 'should not happen' case."""
def generate_config_title(structures: Iterable[Structure]) -> str | None:
"""Pick a user friendly config title based on the Google Home name(s)."""
names: list[str] = []
@ -160,11 +95,6 @@ class NestFlowHandler(
# Possible name to use for config entry based on the Google Home name
self._structure_config_title: str | None = None
@property
def config_mode(self) -> ConfigMode:
"""Return the configuration type for this flow."""
return get_config_mode(self.hass)
def _async_reauth_entry(self) -> ConfigEntry | None:
"""Return existing entry for reauth."""
if self.source != SOURCE_REAUTH or not (
@ -206,7 +136,6 @@ class NestFlowHandler(
async def async_oauth_create_entry(self, data: dict[str, Any]) -> FlowResult:
"""Complete OAuth setup and finish pubsub or finish."""
_LOGGER.debug("Finishing post-oauth configuration")
assert self.config_mode != ConfigMode.LEGACY, "Step only supported for SDM API"
self._data.update(data)
if self.source == SOURCE_REAUTH:
_LOGGER.debug("Skipping Pub/Sub configuration")
@ -215,7 +144,6 @@ class NestFlowHandler(
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Perform reauth upon an API authentication error."""
assert self.config_mode != ConfigMode.LEGACY, "Step only supported for SDM API"
self._data.update(entry_data)
return await self.async_step_reauth_confirm()
@ -224,7 +152,6 @@ class NestFlowHandler(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm reauth dialog."""
assert self.config_mode != ConfigMode.LEGACY, "Step only supported for SDM API"
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
@ -233,8 +160,6 @@ class NestFlowHandler(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
if self.config_mode == ConfigMode.LEGACY:
return await self.async_step_init(user_input)
self._data[DATA_SDM] = {}
if self.source == SOURCE_REAUTH:
return await super().async_step_user(user_input)
@ -391,7 +316,6 @@ class NestFlowHandler(
async def async_step_finish(self, data: dict[str, Any] | None = None) -> FlowResult:
"""Create an entry for the SDM flow."""
_LOGGER.debug("Creating/updating configuration entry")
assert self.config_mode != ConfigMode.LEGACY, "Step only supported for SDM API"
# Update existing config entry when in the reauth flow.
if entry := self._async_reauth_entry():
self.hass.config_entries.async_update_entry(
@ -404,114 +328,3 @@ class NestFlowHandler(
if self._structure_config_title:
title = self._structure_config_title
return self.async_create_entry(title=title, data=self._data)
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow start."""
assert (
self.config_mode == ConfigMode.LEGACY
), "Step only supported for legacy API"
flows = self.hass.data.get(DATA_FLOW_IMPL, {})
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if not flows:
return self.async_abort(reason="missing_configuration")
if len(flows) == 1:
self.flow_impl = list(flows)[0]
return await self.async_step_link()
if user_input is not None:
self.flow_impl = user_input["flow_impl"]
return await self.async_step_link()
return self.async_show_form(
step_id="init",
data_schema=vol.Schema({vol.Required("flow_impl"): vol.In(list(flows))}),
)
async def async_step_link(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Attempt to link with the Nest account.
Route the user to a website to authenticate with Nest. Depending on
implementation type we expect a pin or an external component to
deliver the authentication code.
"""
assert (
self.config_mode == ConfigMode.LEGACY
), "Step only supported for legacy API"
flow = self.hass.data[DATA_FLOW_IMPL][self.flow_impl]
errors = {}
if user_input is not None:
try:
async with async_timeout.timeout(10):
tokens = await flow["convert_code"](user_input["code"])
return self._entry_from_tokens(
f"Nest (via {flow['name']})", flow, tokens
)
except asyncio.TimeoutError:
errors["code"] = "timeout"
except CodeInvalid:
errors["code"] = "invalid_pin"
except NestAuthError:
errors["code"] = "unknown"
except Exception: # pylint: disable=broad-except
errors["code"] = "internal_error"
_LOGGER.exception("Unexpected error resolving code")
try:
async with async_timeout.timeout(10):
url = await flow["gen_authorize_url"](self.flow_id)
except asyncio.TimeoutError:
return self.async_abort(reason="authorize_url_timeout")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected error generating auth url")
return self.async_abort(reason="unknown_authorize_url_generation")
return self.async_show_form(
step_id="link",
description_placeholders={"url": url},
data_schema=vol.Schema({vol.Required("code"): str}),
errors=errors,
)
async def async_step_import(self, info: dict[str, Any]) -> FlowResult:
"""Import existing auth from Nest."""
assert (
self.config_mode == ConfigMode.LEGACY
), "Step only supported for legacy API"
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
config_path = info["nest_conf_path"]
if not await self.hass.async_add_executor_job(os.path.isfile, config_path):
self.flow_impl = DOMAIN # type: ignore[assignment]
return await self.async_step_link()
flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN]
tokens = await self.hass.async_add_executor_job(load_json_object, config_path)
return self._entry_from_tokens(
"Nest (import from configuration.yaml)", flow, tokens
)
@callback
def _entry_from_tokens(
self, title: str, flow: dict[str, Any], tokens: JsonObjectType
) -> FlowResult:
"""Create an entry from tokens."""
return self.async_create_entry(
title=title, data={"tokens": tokens, "impl_domain": flow["domain"]}
)

432
homeassistant/components/nest/legacy/__init__.py
View File

@ -1,432 +0,0 @@
"""Support for Nest devices."""
# mypy: ignore-errors
from datetime import datetime, timedelta
import logging
import threading
from nest import Nest
from nest.nest import APIError, AuthorizationError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_FILENAME,
CONF_STRUCTURE,
EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.dispatcher import async_dispatcher_connect, dispatcher_send
from homeassistant.helpers.entity import DeviceInfo, Entity
from . import local_auth
from .const import DATA_NEST, DATA_NEST_CONFIG, DOMAIN, SIGNAL_NEST_UPDATE
_CONFIGURING = {}
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.CAMERA,
Platform.CLIMATE,
Platform.SENSOR,
]
# Configuration for the legacy nest API
SERVICE_CANCEL_ETA = "cancel_eta"
SERVICE_SET_ETA = "set_eta"
NEST_CONFIG_FILE = "nest.conf"
ATTR_ETA = "eta"
ATTR_ETA_WINDOW = "eta_window"
ATTR_STRUCTURE = "structure"
ATTR_TRIP_ID = "trip_id"
AWAY_MODE_AWAY = "away"
AWAY_MODE_HOME = "home"
ATTR_AWAY_MODE = "away_mode"
SERVICE_SET_AWAY_MODE = "set_away_mode"
# Services for the legacy API
SET_AWAY_MODE_SCHEMA = vol.Schema(
{
vol.Required(ATTR_AWAY_MODE): vol.In([AWAY_MODE_AWAY, AWAY_MODE_HOME]),
vol.Optional(ATTR_STRUCTURE): vol.All(cv.ensure_list, [cv.string]),
}
)
SET_ETA_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ETA): cv.time_period,
vol.Optional(ATTR_TRIP_ID): cv.string,
vol.Optional(ATTR_ETA_WINDOW): cv.time_period,
vol.Optional(ATTR_STRUCTURE): vol.All(cv.ensure_list, [cv.string]),
}
)
CANCEL_ETA_SCHEMA = vol.Schema(
{
vol.Required(ATTR_TRIP_ID): cv.string,
vol.Optional(ATTR_STRUCTURE): vol.All(cv.ensure_list, [cv.string]),
}
)
def nest_update_event_broker(hass, nest):
"""Dispatch SIGNAL_NEST_UPDATE to devices when nest stream API received data.
Used for the legacy nest API.
Runs in its own thread.
"""
_LOGGER.debug("Listening for nest.update_event")
while hass.is_running:
nest.update_event.wait()
if not hass.is_running:
break
nest.update_event.clear()
_LOGGER.debug("Dispatching nest data update")
dispatcher_send(hass, SIGNAL_NEST_UPDATE)
_LOGGER.debug("Stop listening for nest.update_event")
async def async_setup_legacy(hass: HomeAssistant, config: dict) -> bool:
"""Set up Nest components using the legacy nest API."""
if DOMAIN not in config:
return True
ir.async_create_issue(
hass,
DOMAIN,
"legacy_nest_deprecated",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="legacy_nest_deprecated",
translation_placeholders={
"documentation_url": "https://www.home-assistant.io/integrations/nest/",
},
)
conf = config[DOMAIN]
local_auth.initialize(hass, conf[CONF_CLIENT_ID], conf[CONF_CLIENT_SECRET])
filename = config.get(CONF_FILENAME, NEST_CONFIG_FILE)
access_token_cache_file = hass.config.path(filename)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={"nest_conf_path": access_token_cache_file},
)
)
# Store config to be used during entry setup
hass.data[DATA_NEST_CONFIG] = conf
return True
async def async_setup_legacy_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Nest from legacy config entry."""
nest = Nest(access_token=entry.data["tokens"]["access_token"])
_LOGGER.debug("proceeding with setup")
conf = hass.data.get(DATA_NEST_CONFIG, {})
hass.data[DATA_NEST] = NestLegacyDevice(hass, conf, nest)
if not await hass.async_add_executor_job(hass.data[DATA_NEST].initialize):
return False
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
def validate_structures(target_structures):
all_structures = [structure.name for structure in nest.structures]
for target in target_structures:
if target not in all_structures:
_LOGGER.info("Invalid structure: %s", target)
def set_away_mode(service):
"""Set the away mode for a Nest structure."""
if ATTR_STRUCTURE in service.data:
target_structures = service.data[ATTR_STRUCTURE]
validate_structures(target_structures)
else:
target_structures = hass.data[DATA_NEST].local_structure
for structure in nest.structures:
if structure.name in target_structures:
_LOGGER.info(
"Setting away mode for: %s to: %s",
structure.name,
service.data[ATTR_AWAY_MODE],
)
structure.away = service.data[ATTR_AWAY_MODE]
def set_eta(service):
"""Set away mode to away and include ETA for a Nest structure."""
if ATTR_STRUCTURE in service.data:
target_structures = service.data[ATTR_STRUCTURE]
validate_structures(target_structures)
else:
target_structures = hass.data[DATA_NEST].local_structure
for structure in nest.structures:
if structure.name in target_structures:
if structure.thermostats:
_LOGGER.info(
"Setting away mode for: %s to: %s",
structure.name,
AWAY_MODE_AWAY,
)
structure.away = AWAY_MODE_AWAY
now = datetime.utcnow()
trip_id = service.data.get(
ATTR_TRIP_ID, f"trip_{int(now.timestamp())}"
)
eta_begin = now + service.data[ATTR_ETA]
eta_window = service.data.get(ATTR_ETA_WINDOW, timedelta(minutes=1))
eta_end = eta_begin + eta_window
_LOGGER.info(
(
"Setting ETA for trip: %s, "
"ETA window starts at: %s and ends at: %s"
),
trip_id,
eta_begin,
eta_end,
)
structure.set_eta(trip_id, eta_begin, eta_end)
else:
_LOGGER.info(
"No thermostats found in structure: %s, unable to set ETA",
structure.name,
)
def cancel_eta(service):
"""Cancel ETA for a Nest structure."""
if ATTR_STRUCTURE in service.data:
target_structures = service.data[ATTR_STRUCTURE]
validate_structures(target_structures)
else:
target_structures = hass.data[DATA_NEST].local_structure
for structure in nest.structures:
if structure.name in target_structures:
if structure.thermostats:
trip_id = service.data[ATTR_TRIP_ID]
_LOGGER.info("Cancelling ETA for trip: %s", trip_id)
structure.cancel_eta(trip_id)
else:
_LOGGER.info(
"No thermostats found in structure: %s, unable to cancel ETA",
structure.name,
)
hass.services.async_register(
DOMAIN, SERVICE_SET_AWAY_MODE, set_away_mode, schema=SET_AWAY_MODE_SCHEMA
)
hass.services.async_register(
DOMAIN, SERVICE_SET_ETA, set_eta, schema=SET_ETA_SCHEMA
)
hass.services.async_register(
DOMAIN, SERVICE_CANCEL_ETA, cancel_eta, schema=CANCEL_ETA_SCHEMA
)
@callback
def start_up(event):
"""Start Nest update event listener."""
threading.Thread(
name="Nest update listener",
target=nest_update_event_broker,
args=(hass, nest),
).start()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, start_up)
@callback
def shut_down(event):
"""Stop Nest update event listener."""
nest.update_event.set()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, shut_down)
)
_LOGGER.debug("async_setup_nest is done")
return True
class NestLegacyDevice:
"""Structure Nest functions for hass for legacy API."""
def __init__(self, hass, conf, nest):
"""Init Nest Devices."""
self.hass = hass
self.nest = nest
self.local_structure = conf.get(CONF_STRUCTURE)
def initialize(self):
"""Initialize Nest."""
try:
# Do not optimize next statement, it is here for initialize
# persistence Nest API connection.
structure_names = [s.name for s in self.nest.structures]
if self.local_structure is None:
self.local_structure = structure_names
except (AuthorizationError, APIError, OSError) as err:
_LOGGER.error("Connection error while access Nest web service: %s", err)
return False
return True
def structures(self):
"""Generate a list of structures."""
try:
for structure in self.nest.structures:
if structure.name not in self.local_structure:
_LOGGER.debug(
"Ignoring structure %s, not in %s",
structure.name,
self.local_structure,
)
continue
yield structure
except (AuthorizationError, APIError, OSError) as err:
_LOGGER.error("Connection error while access Nest web service: %s", err)
def thermostats(self):
"""Generate a list of thermostats."""
return self._devices("thermostats")
def smoke_co_alarms(self):
"""Generate a list of smoke co alarms."""
return self._devices("smoke_co_alarms")
def cameras(self):
"""Generate a list of cameras."""
return self._devices("cameras")
def _devices(self, device_type):
"""Generate a list of Nest devices."""
try:
for structure in self.nest.structures:
if structure.name not in self.local_structure:
_LOGGER.debug(
"Ignoring structure %s, not in %s",
structure.name,
self.local_structure,
)
continue
for device in getattr(structure, device_type, []):
try:
# Do not optimize next statement,
# it is here for verify Nest API permission.
device.name_long
except KeyError:
_LOGGER.warning(
(
"Cannot retrieve device name for [%s]"
", please check your Nest developer "
"account permission settings"
),
device.serial,
)
continue
yield (structure, device)
except (AuthorizationError, APIError, OSError) as err:
_LOGGER.error("Connection error while access Nest web service: %s", err)
class NestSensorDevice(Entity):
"""Representation of a Nest sensor."""
_attr_should_poll = False
def __init__(self, structure, device, variable):
"""Initialize the sensor."""
self.structure = structure
self.variable = variable
if device is not None:
# device specific
self.device = device
self._name = f"{self.device.name_long} {self.variable.replace('_', ' ')}"
else:
# structure only
self.device = structure
self._name = f"{self.structure.name} {self.variable.replace('_', ' ')}"
self._state = None
self._unit = None
@property
def name(self):
"""Return the name of the nest, if any."""
return self._name
@property
def unique_id(self):
"""Return unique id based on device serial and variable."""
return f"{self.device.serial}-{self.variable}"
@property
def device_info(self) -> DeviceInfo:
"""Return information about the device."""
if not hasattr(self.device, "name_long"):
name = self.structure.name
model = "Structure"
else:
name = self.device.name_long
if self.device.is_thermostat:
model = "Thermostat"
elif self.device.is_camera:
model = "Camera"
elif self.device.is_smoke_co_alarm:
model = "Nest Protect"
else:
model = None
return DeviceInfo(
identifiers={(DOMAIN, self.device.serial)},
manufacturer="Nest Labs",
model=model,
name=name,
)
def update(self):
"""Do not use NestSensorDevice directly."""
raise NotImplementedError
async def async_added_to_hass(self):
"""Register update signal handler."""
async def async_update_state():
"""Update sensor state."""
await self.async_update_ha_state(True)
self.async_on_remove(
async_dispatcher_connect(self.hass, SIGNAL_NEST_UPDATE, async_update_state)
)

164
homeassistant/components/nest/legacy/binary_sensor.py
View File

@ -1,164 +0,0 @@
"""Support for Nest Thermostat binary sensors."""
# mypy: ignore-errors
from itertools import chain
import logging
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_BINARY_SENSORS, CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import NestSensorDevice
from .const import DATA_NEST, DATA_NEST_CONFIG
_LOGGER = logging.getLogger(__name__)
BINARY_TYPES = {"online": BinarySensorDeviceClass.CONNECTIVITY}
CLIMATE_BINARY_TYPES = {
"fan": None,
"is_using_emergency_heat": "heat",
"is_locked": None,
"has_leaf": None,
}
CAMERA_BINARY_TYPES = {
"motion_detected": BinarySensorDeviceClass.MOTION,
"sound_detected": BinarySensorDeviceClass.SOUND,
"person_detected": BinarySensorDeviceClass.OCCUPANCY,
}
STRUCTURE_BINARY_TYPES = {"away": None}
STRUCTURE_BINARY_STATE_MAP = {"away": {"away": True, "home": False}}
_BINARY_TYPES_DEPRECATED = [
"hvac_ac_state",
"hvac_aux_heater_state",
"hvac_heater_state",
"hvac_heat_x2_state",
"hvac_heat_x3_state",
"hvac_alt_heat_state",
"hvac_alt_heat_x2_state",
"hvac_emer_heat_state",
]
_VALID_BINARY_SENSOR_TYPES = {
**BINARY_TYPES,
**CLIMATE_BINARY_TYPES,
**CAMERA_BINARY_TYPES,
**STRUCTURE_BINARY_TYPES,
}
async def async_setup_legacy_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Nest binary sensor based on a config entry."""
nest = hass.data[DATA_NEST]
discovery_info = hass.data.get(DATA_NEST_CONFIG, {}).get(CONF_BINARY_SENSORS, {})
# Add all available binary sensors if no Nest binary sensor config is set
if discovery_info == {}:
conditions = _VALID_BINARY_SENSOR_TYPES
else:
conditions = discovery_info.get(CONF_MONITORED_CONDITIONS, {})
for variable in conditions:
if variable in _BINARY_TYPES_DEPRECATED:
wstr = (
f"{variable} is no a longer supported "
"monitored_conditions. See "
"https://www.home-assistant.io/integrations/binary_sensor.nest/ "
"for valid options."
)
_LOGGER.error(wstr)
def get_binary_sensors():
"""Get the Nest binary sensors."""
sensors = []
for structure in nest.structures():
sensors += [
NestBinarySensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_BINARY_TYPES
]
device_chain = chain(nest.thermostats(), nest.smoke_co_alarms(), nest.cameras())
for structure, device in device_chain:
sensors += [
NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in BINARY_TYPES
]
sensors += [
NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CLIMATE_BINARY_TYPES and device.is_thermostat
]
if device.is_camera:
sensors += [
NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CAMERA_BINARY_TYPES
]
for activity_zone in device.activity_zones:
sensors += [
NestActivityZoneSensor(structure, device, activity_zone)
]
return sensors
async_add_entities(await hass.async_add_executor_job(get_binary_sensors), True)
class NestBinarySensor(NestSensorDevice, BinarySensorEntity):
"""Represents a Nest binary sensor."""
@property
def is_on(self):
"""Return true if the binary sensor is on."""
return self._state
@property
def device_class(self):
"""Return the device class of the binary sensor."""
return _VALID_BINARY_SENSOR_TYPES.get(self.variable)
def update(self):
"""Retrieve latest state."""
value = getattr(self.device, self.variable)
if self.variable in STRUCTURE_BINARY_TYPES:
self._state = bool(STRUCTURE_BINARY_STATE_MAP[self.variable].get(value))
else:
self._state = bool(value)
class NestActivityZoneSensor(NestBinarySensor):
"""Represents a Nest binary sensor for activity in a zone."""
def __init__(self, structure, device, zone):
"""Initialize the sensor."""
super().__init__(structure, device, "")
self.zone = zone
self._name = f"{self._name} {self.zone.name} activity"
@property
def unique_id(self):
"""Return unique id based on camera serial and zone id."""
return f"{self.device.serial}-{self.zone.zone_id}"
@property
def device_class(self):
"""Return the device class of the binary sensor."""
return BinarySensorDeviceClass.MOTION
def update(self):
"""Retrieve latest state."""
self._state = self.device.has_ongoing_motion_in_zone(self.zone.zone_id)

147
homeassistant/components/nest/legacy/camera.py
View File

@ -1,147 +0,0 @@
"""Support for Nest Cameras."""
# mypy: ignore-errors
from __future__ import annotations
from datetime import timedelta
import logging
import requests
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera, CameraEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.dt import utcnow
from .const import DATA_NEST, DOMAIN
_LOGGER = logging.getLogger(__name__)
NEST_BRAND = "Nest"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
async def async_setup_legacy_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Nest sensor based on a config entry."""
camera_devices = await hass.async_add_executor_job(hass.data[DATA_NEST].cameras)
cameras = [NestCamera(structure, device) for structure, device in camera_devices]
async_add_entities(cameras, True)
class NestCamera(Camera):
"""Representation of a Nest Camera."""
_attr_should_poll = True # Cameras default to False
_attr_supported_features = CameraEntityFeature.ON_OFF
def __init__(self, structure, device):
"""Initialize a Nest Camera."""
super().__init__()
self.structure = structure
self.device = device
self._location = None
self._name = None
self._online = None
self._is_streaming = None
self._is_video_history_enabled = False
# Default to non-NestAware subscribed, but will be fixed during update
self._time_between_snapshots = timedelta(seconds=30)
self._last_image = None
self._next_snapshot_at = None
@property
def name(self):
"""Return the name of the nest, if any."""
return self._name
@property
def unique_id(self):
"""Return the serial number."""
return self.device.device_id
@property
def device_info(self) -> DeviceInfo:
"""Return information about the device."""
return DeviceInfo(
identifiers={(DOMAIN, self.device.device_id)},
manufacturer="Nest Labs",
model="Camera",
name=self.device.name_long,
)
@property
def is_recording(self):
"""Return true if the device is recording."""
return self._is_streaming
@property
def brand(self):
"""Return the brand of the camera."""
return NEST_BRAND
@property
def is_on(self):
"""Return true if on."""
return self._online and self._is_streaming
def turn_off(self):
"""Turn off camera."""
_LOGGER.debug("Turn off camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = False
def turn_on(self):
"""Turn on camera."""
if not self._online:
_LOGGER.error("Camera %s is offline", self._name)
return
_LOGGER.debug("Turn on camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = True
def update(self):
"""Cache value from Python-nest."""
self._location = self.device.where
self._name = self.device.name
self._online = self.device.online
self._is_streaming = self.device.is_streaming
self._is_video_history_enabled = self.device.is_video_history_enabled
if self._is_video_history_enabled:
# NestAware allowed 10/min
self._time_between_snapshots = timedelta(seconds=6)
else:
# Otherwise, 2/min
self._time_between_snapshots = timedelta(seconds=30)
def _ready_for_snapshot(self, now):
return self._next_snapshot_at is None or now > self._next_snapshot_at
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
now = utcnow()
if self._ready_for_snapshot(now):
url = self.device.snapshot_url
try:
response = requests.get(url, timeout=10)
except requests.exceptions.RequestException as error:
_LOGGER.error("Error getting camera image: %s", error)
return None
self._next_snapshot_at = now + self._time_between_snapshots
self._last_image = response.content
return self._last_image

339
homeassistant/components/nest/legacy/climate.py
View File

@ -1,339 +0,0 @@
"""Legacy Works with Nest climate implementation."""
# mypy: ignore-errors
import logging
from nest.nest import APIError
import voluptuous as vol
from homeassistant.components.climate import (
ATTR_TARGET_TEMP_HIGH,
ATTR_TARGET_TEMP_LOW,
FAN_AUTO,
FAN_ON,
PLATFORM_SCHEMA,
PRESET_AWAY,
PRESET_ECO,
PRESET_NONE,
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, CONF_SCAN_INTERVAL, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_NEST, DOMAIN, SIGNAL_NEST_UPDATE
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{vol.Optional(CONF_SCAN_INTERVAL): vol.All(vol.Coerce(int), vol.Range(min=1))}
)
NEST_MODE_HEAT_COOL = "heat-cool"
NEST_MODE_ECO = "eco"
NEST_MODE_HEAT = "heat"
NEST_MODE_COOL = "cool"
NEST_MODE_OFF = "off"
MODE_HASS_TO_NEST = {
HVACMode.AUTO: NEST_MODE_HEAT_COOL,
HVACMode.HEAT: NEST_MODE_HEAT,
HVACMode.COOL: NEST_MODE_COOL,
HVACMode.OFF: NEST_MODE_OFF,
}
MODE_NEST_TO_HASS = {v: k for k, v in MODE_HASS_TO_NEST.items()}
ACTION_NEST_TO_HASS = {
"off": HVACAction.IDLE,
"heating": HVACAction.HEATING,
"cooling": HVACAction.COOLING,
}
PRESET_AWAY_AND_ECO = "Away and Eco"
PRESET_MODES = [PRESET_NONE, PRESET_AWAY, PRESET_ECO, PRESET_AWAY_AND_ECO]
async def async_setup_legacy_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Nest climate device based on a config entry."""
temp_unit = hass.config.units.temperature_unit
thermostats = await hass.async_add_executor_job(hass.data[DATA_NEST].thermostats)
all_devices = [
NestThermostat(structure, device, temp_unit)
for structure, device in thermostats
]
async_add_entities(all_devices, True)
class NestThermostat(ClimateEntity):
"""Representation of a Nest thermostat."""
_attr_should_poll = False
def __init__(self, structure, device, temp_unit):
"""Initialize the thermostat."""
self._unit = temp_unit
self.structure = structure
self.device = device
self._fan_modes = [FAN_ON, FAN_AUTO]
# Set the default supported features
self._attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.PRESET_MODE
)
# Not all nest devices support cooling and heating remove unused
self._operation_list = []
if self.device.can_heat and self.device.can_cool:
self._operation_list.append(HVACMode.AUTO)
self._attr_supported_features |= (
ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
)
# Add supported nest thermostat features
if self.device.can_heat:
self._operation_list.append(HVACMode.HEAT)
if self.device.can_cool:
self._operation_list.append(HVACMode.COOL)
self._operation_list.append(HVACMode.OFF)
# feature of device
self._has_fan = self.device.has_fan
if self._has_fan:
self._attr_supported_features |= ClimateEntityFeature.FAN_MODE
# data attributes
self._away = None
self._location = None
self._name = None
self._humidity = None
self._target_temperature = None
self._temperature = None
self._temperature_scale = None
self._mode = None
self._action = None
self._fan = None
self._eco_temperature = None
self._is_locked = None
self._locked_temperature = None
self._min_temperature = None
self._max_temperature = None
async def async_added_to_hass(self):
"""Register update signal handler."""
async def async_update_state():
"""Update device state."""
await self.async_update_ha_state(True)
self.async_on_remove(
async_dispatcher_connect(self.hass, SIGNAL_NEST_UPDATE, async_update_state)
)
@property
def unique_id(self):
"""Return unique ID for this device."""
return self.device.serial
@property
def device_info(self) -> DeviceInfo:
"""Return information about the device."""
return DeviceInfo(
identifiers={(DOMAIN, self.device.device_id)},
manufacturer="Nest Labs",
model="Thermostat",
name=self.device.name_long,
sw_version=self.device.software_version,
)
@property
def name(self):
"""Return the name of the nest, if any."""
return self._name
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return self._temperature_scale
@property
def current_temperature(self):
"""Return the current temperature."""
return self._temperature
@property
def hvac_mode(self) -> HVACMode:
"""Return current operation ie. heat, cool, idle."""
if self._mode == NEST_MODE_ECO:
if self.device.previous_mode in MODE_NEST_TO_HASS:
return MODE_NEST_TO_HASS[self.device.previous_mode]
# previous_mode not supported so return the first compatible mode
return self._operation_list[0]
return MODE_NEST_TO_HASS[self._mode]
@property
def hvac_action(self) -> HVACAction:
"""Return the current hvac action."""
return ACTION_NEST_TO_HASS[self._action]
@property
def target_temperature(self):
"""Return the temperature we try to reach."""
if self._mode not in (NEST_MODE_HEAT_COOL, NEST_MODE_ECO):
return self._target_temperature
return None
@property
def target_temperature_low(self):
"""Return the lower bound temperature we try to reach."""
if self._mode == NEST_MODE_ECO:
return self._eco_temperature[0]
if self._mode == NEST_MODE_HEAT_COOL:
return self._target_temperature[0]
return None
@property
def target_temperature_high(self):
"""Return the upper bound temperature we try to reach."""
if self._mode == NEST_MODE_ECO:
return self._eco_temperature[1]
if self._mode == NEST_MODE_HEAT_COOL:
return self._target_temperature[1]
return None
def set_temperature(self, **kwargs):
"""Set new target temperature."""
temp = None
target_temp_low = kwargs.get(ATTR_TARGET_TEMP_LOW)
target_temp_high = kwargs.get(ATTR_TARGET_TEMP_HIGH)
if self._mode == NEST_MODE_HEAT_COOL:
if target_temp_low is not None and target_temp_high is not None:
temp = (target_temp_low, target_temp_high)
_LOGGER.debug("Nest set_temperature-output-value=%s", temp)
else:
temp = kwargs.get(ATTR_TEMPERATURE)
_LOGGER.debug("Nest set_temperature-output-value=%s", temp)
try:
if temp is not None:
self.device.target = temp
except APIError as api_error:
_LOGGER.error("An error occurred while setting temperature: %s", api_error)
# restore target temperature
self.schedule_update_ha_state(True)
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set operation mode."""
self.device.mode = MODE_HASS_TO_NEST[hvac_mode]
@property
def hvac_modes(self) -> list[HVACMode]:
"""List of available operation modes."""
return self._operation_list
@property
def preset_mode(self):
"""Return current preset mode."""
if self._away and self._mode == NEST_MODE_ECO:
return PRESET_AWAY_AND_ECO
if self._away:
return PRESET_AWAY
if self._mode == NEST_MODE_ECO:
return PRESET_ECO
return PRESET_NONE
@property
def preset_modes(self):
"""Return preset modes."""
return PRESET_MODES
def set_preset_mode(self, preset_mode):
"""Set preset mode."""
if preset_mode == self.preset_mode:
return
need_away = preset_mode in (PRESET_AWAY, PRESET_AWAY_AND_ECO)
need_eco = preset_mode in (PRESET_ECO, PRESET_AWAY_AND_ECO)
is_away = self._away
is_eco = self._mode == NEST_MODE_ECO
if is_away != need_away:
self.structure.away = need_away
if is_eco != need_eco:
if need_eco:
self.device.mode = NEST_MODE_ECO
else:
self.device.mode = self.device.previous_mode
@property
def fan_mode(self):
"""Return whether the fan is on."""
if self._has_fan:
# Return whether the fan is on
return FAN_ON if self._fan else FAN_AUTO
# No Fan available so disable slider
return None
@property
def fan_modes(self):
"""List of available fan modes."""
if self._has_fan:
return self._fan_modes
return None
def set_fan_mode(self, fan_mode):
"""Turn fan on/off."""
if self._has_fan:
self.device.fan = fan_mode.lower()
@property
def min_temp(self):
"""Identify min_temp in Nest API or defaults if not available."""
return self._min_temperature
@property
def max_temp(self):
"""Identify max_temp in Nest API or defaults if not available."""
return self._max_temperature
def update(self):
"""Cache value from Python-nest."""
self._location = self.device.where
self._name = self.device.name
self._humidity = self.device.humidity
self._temperature = self.device.temperature
self._mode = self.device.mode
self._action = self.device.hvac_state
self._target_temperature = self.device.target
self._fan = self.device.fan
self._away = self.structure.away == "away"
self._eco_temperature = self.device.eco_temperature
self._locked_temperature = self.device.locked_temperature
self._min_temperature = self.device.min_temperature
self._max_temperature = self.device.max_temperature
self._is_locked = self.device.is_locked
if self.device.temperature_scale == "C":
self._temperature_scale = UnitOfTemperature.CELSIUS
else:
self._temperature_scale = UnitOfTemperature.FAHRENHEIT

6
homeassistant/components/nest/legacy/const.py
View File

@ -1,6 +0,0 @@
"""Constants used by the legacy Nest component."""
DOMAIN = "nest"
DATA_NEST = "nest"
DATA_NEST_CONFIG = "nest_config"
SIGNAL_NEST_UPDATE = "nest_update"

52
homeassistant/components/nest/legacy/local_auth.py
View File

@ -1,52 +0,0 @@
"""Local Nest authentication for the legacy api."""
# mypy: ignore-errors
import asyncio
from functools import partial
from http import HTTPStatus
from nest.nest import AUTHORIZE_URL, AuthorizationError, NestAuth
from homeassistant.core import callback
from ..config_flow import CodeInvalid, NestAuthError, register_flow_implementation
from .const import DOMAIN
@callback
def initialize(hass, client_id, client_secret):
"""Initialize a local auth provider."""
register_flow_implementation(
hass,
DOMAIN,
"configuration.yaml",
partial(generate_auth_url, client_id),
partial(resolve_auth_code, hass, client_id, client_secret),
)
async def generate_auth_url(client_id, flow_id):
"""Generate an authorize url."""
return AUTHORIZE_URL.format(client_id, flow_id)
async def resolve_auth_code(hass, client_id, client_secret, code):
"""Resolve an authorization code."""
result = asyncio.Future()
auth = NestAuth(
client_id=client_id,
client_secret=client_secret,
auth_callback=result.set_result,
)
auth.pin = code
try:
await hass.async_add_executor_job(auth.login)
return await result
except AuthorizationError as err:
if err.response.status_code == HTTPStatus.UNAUTHORIZED:
raise CodeInvalid() from err
raise NestAuthError(
f"Unknown error: {err} ({err.response.status_code})"
) from err

233
homeassistant/components/nest/legacy/sensor.py
View File

@ -1,233 +0,0 @@
"""Support for Nest Thermostat sensors for the legacy API."""
# mypy: ignore-errors
import logging
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_MONITORED_CONDITIONS,
CONF_SENSORS,
PERCENTAGE,
STATE_OFF,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import NestSensorDevice
from .const import DATA_NEST, DATA_NEST_CONFIG
SENSOR_TYPES = ["humidity", "operation_mode", "hvac_state"]
TEMP_SENSOR_TYPES = ["temperature", "target"]
PROTECT_SENSOR_TYPES = [
"co_status",
"smoke_status",
"battery_health",
# color_status: "gray", "green", "yellow", "red"
"color_status",
]
STRUCTURE_SENSOR_TYPES = ["eta"]
STATE_HEAT = "heat"
STATE_COOL = "cool"
# security_state is structure level sensor, but only meaningful when
# Nest Cam exist
STRUCTURE_CAMERA_SENSOR_TYPES = ["security_state"]
_VALID_SENSOR_TYPES = (
SENSOR_TYPES
+ TEMP_SENSOR_TYPES
+ PROTECT_SENSOR_TYPES
+ STRUCTURE_SENSOR_TYPES
+ STRUCTURE_CAMERA_SENSOR_TYPES
)
SENSOR_UNITS = {"humidity": PERCENTAGE}
SENSOR_DEVICE_CLASSES = {"humidity": SensorDeviceClass.HUMIDITY}
SENSOR_STATE_CLASSES = {"humidity": SensorStateClass.MEASUREMENT}
VARIABLE_NAME_MAPPING = {"eta": "eta_begin", "operation_mode": "mode"}
VALUE_MAPPING = {
"hvac_state": {"heating": STATE_HEAT, "cooling": STATE_COOL, "off": STATE_OFF}
}
SENSOR_TYPES_DEPRECATED = ["last_ip", "local_ip", "last_connection", "battery_level"]
DEPRECATED_WEATHER_VARS = [
"weather_humidity",
"weather_temperature",
"weather_condition",
"wind_speed",
"wind_direction",
]
_SENSOR_TYPES_DEPRECATED = SENSOR_TYPES_DEPRECATED + DEPRECATED_WEATHER_VARS
_LOGGER = logging.getLogger(__name__)
async def async_setup_legacy_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Nest sensor based on a config entry."""
nest = hass.data[DATA_NEST]
discovery_info = hass.data.get(DATA_NEST_CONFIG, {}).get(CONF_SENSORS, {})
# Add all available sensors if no Nest sensor config is set
if discovery_info == {}:
conditions = _VALID_SENSOR_TYPES
else:
conditions = discovery_info.get(CONF_MONITORED_CONDITIONS, {})
for variable in conditions:
if variable in _SENSOR_TYPES_DEPRECATED:
if variable in DEPRECATED_WEATHER_VARS:
wstr = (
f"Nest no longer provides weather data like {variable}. See "
"https://www.home-assistant.io/integrations/#weather "
"for a list of other weather integrations to use."
)
else:
wstr = (
f"{variable} is no a longer supported "
"monitored_conditions. See "
"https://www.home-assistant.io/integrations/"
"binary_sensor.nest/ for valid options."
)
_LOGGER.error(wstr)
def get_sensors():
"""Get the Nest sensors."""
all_sensors = []
for structure in nest.structures():
all_sensors += [
NestBasicSensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_SENSOR_TYPES
]
for structure, device in nest.thermostats():
all_sensors += [
NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in SENSOR_TYPES
]
all_sensors += [
NestTempSensor(structure, device, variable)
for variable in conditions
if variable in TEMP_SENSOR_TYPES
]
for structure, device in nest.smoke_co_alarms():
all_sensors += [
NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in PROTECT_SENSOR_TYPES
]
structures_has_camera = {}
for structure, _ in nest.cameras():
structures_has_camera[structure] = True
for structure in structures_has_camera:
all_sensors += [
NestBasicSensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_CAMERA_SENSOR_TYPES
]
return all_sensors
async_add_entities(await hass.async_add_executor_job(get_sensors), True)
class NestBasicSensor(NestSensorDevice, SensorEntity):
"""Representation a basic Nest sensor."""
@property
def native_unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return self._unit
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def device_class(self):
"""Return the device class of the sensor."""
return SENSOR_DEVICE_CLASSES.get(self.variable)
@property
def state_class(self):
"""Return the state class of the sensor."""
return SENSOR_STATE_CLASSES.get(self.variable)
def update(self):
"""Retrieve latest state."""
self._unit = SENSOR_UNITS.get(self.variable)
if self.variable in VARIABLE_NAME_MAPPING:
self._state = getattr(self.device, VARIABLE_NAME_MAPPING[self.variable])
elif self.variable in VALUE_MAPPING:
state = getattr(self.device, self.variable)
self._state = VALUE_MAPPING[self.variable].get(state, state)
elif self.variable in PROTECT_SENSOR_TYPES and self.variable != "color_status":
# keep backward compatibility
state = getattr(self.device, self.variable)
self._state = state.capitalize() if state is not None else None
else:
self._state = getattr(self.device, self.variable)
class NestTempSensor(NestSensorDevice, SensorEntity):
"""Representation of a Nest Temperature sensor."""
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def native_unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return self._unit
@property
def device_class(self):
"""Return the device class of the sensor."""
return SensorDeviceClass.TEMPERATURE
@property
def state_class(self):
"""Return the state class of the sensor."""
return SensorStateClass.MEASUREMENT
def update(self):
"""Retrieve latest state."""
if self.device.temperature_scale == "C":
self._unit = UnitOfTemperature.CELSIUS
else:
self._unit = UnitOfTemperature.FAHRENHEIT
if (temp := getattr(self.device, self.variable)) is None:
self._state = None
if isinstance(temp, tuple):
low, high = temp
self._state = f"{int(low)}-{int(high)}"
else:
self._state = round(temp, 1)

2
homeassistant/components/nest/manifest.json
View File

@ -20,5 +20,5 @@
"iot_class": "cloud_push",
"loggers": ["google_nest_sdm", "nest"],
"quality_scale": "platinum",
"requirements": ["python-nest==4.2.0", "google-nest-sdm==2.2.5"]
"requirements": ["google-nest-sdm==2.2.5"]
}

100
homeassistant/components/nest/sensor.py
View File

@ -1,20 +1,104 @@
"""Support for Nest sensors that dispatches between API versions."""
"""Support for Google Nest SDM sensors."""
from __future__ import annotations
import logging
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.device_traits import HumidityTrait, TemperatureTrait
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_SDM
from .legacy.sensor import async_setup_legacy_entry
from .sensor_sdm import async_setup_sdm_entry
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
_LOGGER = logging.getLogger(__name__)
DEVICE_TYPE_MAP = {
"sdm.devices.types.CAMERA": "Camera",
"sdm.devices.types.DISPLAY": "Display",
"sdm.devices.types.DOORBELL": "Doorbell",
"sdm.devices.types.THERMOSTAT": "Thermostat",
}
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the sensors."""
if DATA_SDM not in entry.data:
await async_setup_legacy_entry(hass, entry, async_add_entities)
return
await async_setup_sdm_entry(hass, entry, async_add_entities)
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities: list[SensorEntity] = []
for device in device_manager.devices.values():
if TemperatureTrait.NAME in device.traits:
entities.append(TemperatureSensor(device))
if HumidityTrait.NAME in device.traits:
entities.append(HumiditySensor(device))
async_add_entities(entities)
class SensorBase(SensorEntity):
"""Representation of a dynamically updated Sensor."""
_attr_should_poll = False
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_has_entity_name = True
def __init__(self, device: Device) -> None:
"""Initialize the sensor."""
self._device = device
self._device_info = NestDeviceInfo(device)
self._attr_unique_id = f"{device.name}-{self.device_class}"
self._attr_device_info = self._device_info.device_info
@property
def available(self) -> bool:
"""Return the device availability."""
return self._device_info.available
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
class TemperatureSensor(SensorBase):
"""Representation of a Temperature Sensor."""
_attr_device_class = SensorDeviceClass.TEMPERATURE
_attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
@property
def native_value(self) -> float:
"""Return the state of the sensor."""
trait: TemperatureTrait = self._device.traits[TemperatureTrait.NAME]
# Round for display purposes because the API returns 5 decimal places.
# This can be removed if the SDM API issue is fixed, or a frontend
# display fix is added for all integrations.
return float(round(trait.ambient_temperature_celsius, 1))
class HumiditySensor(SensorBase):
"""Representation of a Humidity Sensor."""
_attr_device_class = SensorDeviceClass.HUMIDITY
_attr_native_unit_of_measurement = PERCENTAGE
@property
def native_value(self) -> int:
"""Return the state of the sensor."""
trait: HumidityTrait = self._device.traits[HumidityTrait.NAME]
# Cast without loss of precision because the API always returns an integer.
return int(trait.ambient_humidity_percent)

104
homeassistant/components/nest/sensor_sdm.py
View File

@ -1,104 +0,0 @@
"""Support for Google Nest SDM sensors."""
from __future__ import annotations
import logging
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.device_traits import HumidityTrait, TemperatureTrait
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_DEVICE_MANAGER, DOMAIN
from .device_info import NestDeviceInfo
_LOGGER = logging.getLogger(__name__)
DEVICE_TYPE_MAP = {
"sdm.devices.types.CAMERA": "Camera",
"sdm.devices.types.DISPLAY": "Display",
"sdm.devices.types.DOORBELL": "Doorbell",
"sdm.devices.types.THERMOSTAT": "Thermostat",
}
async def async_setup_sdm_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the sensors."""
device_manager: DeviceManager = hass.data[DOMAIN][entry.entry_id][
DATA_DEVICE_MANAGER
]
entities: list[SensorEntity] = []
for device in device_manager.devices.values():
if TemperatureTrait.NAME in device.traits:
entities.append(TemperatureSensor(device))
if HumidityTrait.NAME in device.traits:
entities.append(HumiditySensor(device))
async_add_entities(entities)
class SensorBase(SensorEntity):
"""Representation of a dynamically updated Sensor."""
_attr_should_poll = False
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_has_entity_name = True
def __init__(self, device: Device) -> None:
"""Initialize the sensor."""
self._device = device
self._device_info = NestDeviceInfo(device)
self._attr_unique_id = f"{device.name}-{self.device_class}"
self._attr_device_info = self._device_info.device_info
@property
def available(self) -> bool:
"""Return the device availability."""
return self._device_info.available
async def async_added_to_hass(self) -> None:
"""Run when entity is added to register update signal handler."""
self.async_on_remove(
self._device.add_update_listener(self.async_write_ha_state)
)
class TemperatureSensor(SensorBase):
"""Representation of a Temperature Sensor."""
_attr_device_class = SensorDeviceClass.TEMPERATURE
_attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
@property
def native_value(self) -> float:
"""Return the state of the sensor."""
trait: TemperatureTrait = self._device.traits[TemperatureTrait.NAME]
# Round for display purposes because the API returns 5 decimal places.
# This can be removed if the SDM API issue is fixed, or a frontend
# display fix is added for all integrations.
return float(round(trait.ambient_temperature_celsius, 1))
class HumiditySensor(SensorBase):
"""Representation of a Humidity Sensor."""
_attr_device_class = SensorDeviceClass.HUMIDITY
_attr_native_unit_of_measurement = PERCENTAGE
@property
def native_value(self) -> int:
"""Return the state of the sensor."""
trait: HumidityTrait = self._device.traits[HumidityTrait.NAME]
# Cast without loss of precision because the API always returns an integer.
return int(trait.ambient_humidity_percent)

18
homeassistant/components/nest/strings.json
View File

@ -35,27 +35,9 @@
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The Nest integration needs to re-authenticate your account"
},
"init": {
"title": "Authentication Provider",
"description": "[%key:common::config_flow::title::oauth2_pick_implementation%]",
"data": {
"flow_impl": "Provider"
}
},
"link": {
"title": "Link Nest Account",
"description": "To link your Nest account, [authorize your account]({url}).\n\nAfter authorization, copy-paste the provided PIN code below.",
"data": {
"code": "[%key:common::config_flow::data::pin%]"
}
}
},
"error": {
"timeout": "Timeout validating code",
"invalid_pin": "Invalid PIN",
"unknown": "[%key:common::config_flow::error::unknown%]",
"internal_error": "Internal error validating code",
"bad_project_id": "Please enter a valid Cloud Project ID (check Cloud Console)",
"wrong_project_id": "Please enter a valid Cloud Project ID (was same as Device Access Project ID)",
"subscriber_error": "Unknown subscriber error, see logs"

3
requirements_all.txt
View File

@ -2119,9 +2119,6 @@ python-mpd2==3.0.5
# homeassistant.components.mystrom
python-mystrom==2.2.0
# homeassistant.components.nest
python-nest==4.2.0
# homeassistant.components.swiss_public_transport
python-opendata-transport==0.3.0

3
requirements_test_all.txt
View File

@ -1554,9 +1554,6 @@ python-miio==0.5.12
# homeassistant.components.mystrom
python-mystrom==2.2.0
# homeassistant.components.nest
python-nest==4.2.0
# homeassistant.components.otbr
# homeassistant.components.thread
python-otbr-api==2.2.0

0
tests/components/nest/test_camera_sdm.py → tests/components/nest/test_camera.py
View File

0
tests/components/nest/test_climate_sdm.py → tests/components/nest/test_climate.py
View File

0
tests/components/nest/test_config_flow_sdm.py → tests/components/nest/test_config_flow.py
View File

242
tests/components/nest/test_config_flow_legacy.py
View File

@ -1,242 +0,0 @@
"""Tests for the Nest config flow."""
import asyncio
from unittest.mock import patch
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.nest import DOMAIN, config_flow
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .common import TEST_CONFIG_LEGACY
from tests.common import MockConfigEntry
CONFIG = TEST_CONFIG_LEGACY.config
async def test_abort_if_single_instance_allowed(hass: HomeAssistant) -> None:
"""Test we abort if Nest is already setup."""
existing_entry = MockConfigEntry(domain=DOMAIN, data={})
existing_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
async def test_full_flow_implementation(hass: HomeAssistant) -> None:
"""Test registering an implementation and finishing flow works."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
# Register an additional implementation to select from during the flow
config_flow.register_flow_implementation(
hass, "test-other", "Test Other", None, None
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"flow_impl": "nest"},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
assert (
result["description_placeholders"]
.get("url")
.startswith("https://home.nest.com/login/oauth2?client_id=some-client-id")
)
def mock_login(auth):
assert auth.pin == "123ABC"
auth.auth_callback({"access_token": "yoo"})
with patch(
"homeassistant.components.nest.legacy.local_auth.NestAuth.login", new=mock_login
), patch(
"homeassistant.components.nest.async_setup_legacy_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123ABC"}
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"]["tokens"] == {"access_token": "yoo"}
assert result["data"]["impl_domain"] == "nest"
assert result["title"] == "Nest (via configuration.yaml)"
async def test_not_pick_implementation_if_only_one(hass: HomeAssistant) -> None:
"""Test we pick the default implementation when registered."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
async def test_abort_if_timeout_generating_auth_url(hass: HomeAssistant) -> None:
"""Test we abort if generating authorize url fails."""
with patch(
"homeassistant.components.nest.legacy.local_auth.generate_auth_url",
side_effect=asyncio.TimeoutError,
):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "authorize_url_timeout"
async def test_abort_if_exception_generating_auth_url(hass: HomeAssistant) -> None:
"""Test we abort if generating authorize url blows up."""
with patch(
"homeassistant.components.nest.legacy.local_auth.generate_auth_url",
side_effect=ValueError,
):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "unknown_authorize_url_generation"
async def test_verify_code_timeout(hass: HomeAssistant) -> None:
"""Test verify code timing out."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
with patch(
"homeassistant.components.nest.legacy.local_auth.NestAuth.login",
side_effect=asyncio.TimeoutError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123ABC"}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
assert result["errors"] == {"code": "timeout"}
async def test_verify_code_invalid(hass: HomeAssistant) -> None:
"""Test verify code invalid."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
with patch(
"homeassistant.components.nest.legacy.local_auth.NestAuth.login",
side_effect=config_flow.CodeInvalid,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123ABC"}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
assert result["errors"] == {"code": "invalid_pin"}
async def test_verify_code_unknown_error(hass: HomeAssistant) -> None:
"""Test verify code unknown error."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
with patch(
"homeassistant.components.nest.legacy.local_auth.NestAuth.login",
side_effect=config_flow.NestAuthError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123ABC"}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
assert result["errors"] == {"code": "unknown"}
async def test_verify_code_exception(hass: HomeAssistant) -> None:
"""Test verify code blows up."""
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
with patch(
"homeassistant.components.nest.legacy.local_auth.NestAuth.login",
side_effect=ValueError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123ABC"}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
assert result["errors"] == {"code": "internal_error"}
async def test_step_import(hass: HomeAssistant) -> None:
"""Test that we trigger import when configuring with client."""
with patch("os.path.isfile", return_value=False):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
flow = hass.config_entries.flow.async_progress()[0]
result = await hass.config_entries.flow.async_configure(flow["flow_id"])
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "link"
async def test_step_import_with_token_cache(hass: HomeAssistant) -> None:
"""Test that we import existing token cache."""
with patch("os.path.isfile", return_value=True), patch(
"homeassistant.components.nest.config_flow.load_json_object",
return_value={"access_token": "yo"},
), patch(
"homeassistant.components.nest.async_setup_legacy_entry", return_value=True
) as mock_setup:
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {"impl_domain": "nest", "tokens": {"access_token": "yo"}}

17
tests/components/nest/test_diagnostics.py
View File

@ -9,8 +9,6 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .common import TEST_CONFIG_LEGACY
from tests.components.diagnostics import (
get_diagnostics_for_config_entry,
get_diagnostics_for_device,
@ -146,21 +144,6 @@ async def test_setup_susbcriber_failure(
assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == {}
@pytest.mark.parametrize("nest_test_config", [TEST_CONFIG_LEGACY])
async def test_legacy_config_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry,
setup_base_platform,
) -> None:
"""Test config entry diagnostics for legacy integration doesn't fail."""
with patch("homeassistant.components.nest.legacy.Nest"):
await setup_base_platform()
assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == {}
async def test_camera_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,

28
tests/components/nest/test_init_sdm.py → tests/components/nest/test_init.py
View File

@ -22,15 +22,20 @@ import pytest
from homeassistant.components.nest import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .common import (
PROJECT_ID,
SUBSCRIBER_ID,
TEST_CONFIG_ENTRY_LEGACY,
TEST_CONFIG_LEGACY,
TEST_CONFIGFLOW_APP_CREDS,
FakeSubscriber,
YieldFixture,
)
from tests.common import MockConfigEntry
PLATFORM = "sensor"
@ -276,3 +281,26 @@ async def test_migrate_unique_id(
assert config_entry.state is ConfigEntryState.LOADED
assert config_entry.unique_id == PROJECT_ID
@pytest.mark.parametrize("nest_test_config", [TEST_CONFIG_LEGACY])
async def test_legacy_works_with_nest_yaml(
hass: HomeAssistant,
config: dict[str, Any],
config_entry: MockConfigEntry,
) -> None:
"""Test integration won't start with legacy works with nest yaml config."""
config_entry.add_to_hass(hass)
assert not await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
@pytest.mark.parametrize("nest_test_config", [TEST_CONFIG_ENTRY_LEGACY])
async def test_legacy_works_with_nest_cleanup(
hass: HomeAssistant, setup_platform
) -> None:
"""Test legacy works with nest config entries are silently removed once yaml is removed."""
await setup_platform()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 0

76
tests/components/nest/test_init_legacy.py
View File

@ -1,76 +0,0 @@
"""Test basic initialization for the Legacy Nest API using mocks for the Nest python library."""
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
from homeassistant.core import HomeAssistant
from .common import TEST_CONFIG_ENTRY_LEGACY, TEST_CONFIG_LEGACY
DOMAIN = "nest"
@pytest.fixture
def nest_test_config():
"""Fixture to specify the overall test fixture configuration."""
return TEST_CONFIG_LEGACY
def make_thermostat():
"""Make a mock thermostat with dummy values."""
device = MagicMock()
type(device).device_id = PropertyMock(return_value="a.b.c.d.e.f.g")
type(device).name = PropertyMock(return_value="My Thermostat")
type(device).name_long = PropertyMock(return_value="My Thermostat")
type(device).serial = PropertyMock(return_value="serial-number")
type(device).mode = "off"
type(device).hvac_state = "off"
type(device).target = PropertyMock(return_value=31.0)
type(device).temperature = PropertyMock(return_value=30.1)
type(device).min_temperature = PropertyMock(return_value=10.0)
type(device).max_temperature = PropertyMock(return_value=50.0)
type(device).humidity = PropertyMock(return_value=40.4)
type(device).software_version = PropertyMock(return_value="a.b.c")
return device
@pytest.mark.parametrize(
"nest_test_config", [TEST_CONFIG_LEGACY, TEST_CONFIG_ENTRY_LEGACY]
)
async def test_thermostat(hass: HomeAssistant, setup_base_platform) -> None:
"""Test simple initialization for thermostat entities."""
thermostat = make_thermostat()
structure = MagicMock()
type(structure).name = PropertyMock(return_value="My Room")
type(structure).thermostats = PropertyMock(return_value=[thermostat])
type(structure).eta = PropertyMock(return_value="away")
nest = MagicMock()
type(nest).structures = PropertyMock(return_value=[structure])
with patch("homeassistant.components.nest.legacy.Nest", return_value=nest), patch(
"homeassistant.components.nest.legacy.sensor._VALID_SENSOR_TYPES",
["humidity", "temperature"],
), patch(
"homeassistant.components.nest.legacy.binary_sensor._VALID_BINARY_SENSOR_TYPES",
{"fan": None},
):
await setup_base_platform()
climate = hass.states.get("climate.my_thermostat")
assert climate is not None
assert climate.state == "off"
temperature = hass.states.get("sensor.my_thermostat_temperature")
assert temperature is not None
assert temperature.state == "-1.1"
humidity = hass.states.get("sensor.my_thermostat_humidity")
assert humidity is not None
assert humidity.state == "40.4"
fan = hass.states.get("binary_sensor.my_thermostat_fan")
assert fan is not None
assert fan.state == "on"

51
tests/components/nest/test_local_auth.py
View File

@ -1,51 +0,0 @@
"""Test Nest local auth."""
from urllib.parse import parse_qsl
import pytest
import requests_mock
from requests_mock import create_response
from homeassistant.components.nest import config_flow, const
from homeassistant.components.nest.legacy import local_auth
@pytest.fixture
def registered_flow(hass):
"""Mock a registered flow."""
local_auth.initialize(hass, "TEST-CLIENT-ID", "TEST-CLIENT-SECRET")
return hass.data[config_flow.DATA_FLOW_IMPL][const.DOMAIN]
async def test_generate_auth_url(registered_flow) -> None:
"""Test generating an auth url.
Mainly testing that it doesn't blow up.
"""
url = await registered_flow["gen_authorize_url"]("TEST-FLOW-ID")
assert url is not None
async def test_convert_code(
requests_mock: requests_mock.Mocker, registered_flow
) -> None:
"""Test converting a code."""
from nest.nest import ACCESS_TOKEN_URL
def token_matcher(request):
"""Match a fetch token request."""
if request.url != ACCESS_TOKEN_URL:
return None
assert dict(parse_qsl(request.text)) == {
"client_id": "TEST-CLIENT-ID",
"client_secret": "TEST-CLIENT-SECRET",
"code": "TEST-CODE",
"grant_type": "authorization_code",
}
return create_response(request, json={"access_token": "TEST-ACCESS-TOKEN"})
requests_mock.add_matcher(token_matcher)
tokens = await registered_flow["convert_code"]("TEST-CODE")
assert tokens == {"access_token": "TEST-ACCESS-TOKEN"}