diff --git a/homeassistant/components/withings/application_credentials.py b/homeassistant/components/withings/application_credentials.py index e5c401d5e74..1d5b52466c4 100644 --- a/homeassistant/components/withings/application_credentials.py +++ b/homeassistant/components/withings/application_credentials.py @@ -1,15 +1,17 @@ """application_credentials platform for Withings.""" +from typing import Any + from withings_api import AbstractWithingsApi, WithingsAuth from homeassistant.components.application_credentials import ( + AuthImplementation, AuthorizationServer, ClientCredential, ) from homeassistant.core import HomeAssistant from homeassistant.helpers import config_entry_oauth2_flow -from .common import WithingsLocalOAuth2Implementation from .const import DOMAIN @@ -26,3 +28,50 @@ async def async_get_auth_implementation( token_url=f"{AbstractWithingsApi.URL}/v2/oauth2", ), ) + + +class WithingsLocalOAuth2Implementation(AuthImplementation): + """Oauth2 implementation that only uses the external url.""" + + async def _token_request(self, data: dict) -> dict: + """Make a token request and adapt Withings API reply.""" + new_token = await super()._token_request(data) + # Withings API returns habitual token data under json key "body": + # { + # "status": [{integer} Withings API response status], + # "body": { + # "access_token": [{string} Your new access_token], + # "expires_in": [{integer} Access token expiry delay in seconds], + # "token_type": [{string] HTTP Authorization Header format: Bearer], + # "scope": [{string} Scopes the user accepted], + # "refresh_token": [{string} Your new refresh_token], + # "userid": [{string} The Withings ID of the user] + # } + # } + # so we copy that to token root. + if body := new_token.pop("body", None): + new_token.update(body) + return new_token + + async def async_resolve_external_data(self, external_data: Any) -> dict: + """Resolve the authorization code to tokens.""" + return await self._token_request( + { + "action": "requesttoken", + "grant_type": "authorization_code", + "code": external_data["code"], + "redirect_uri": external_data["state"]["redirect_uri"], + } + ) + + async def _async_refresh_token(self, token: dict) -> dict: + """Refresh tokens.""" + new_token = await self._token_request( + { + "action": "requesttoken", + "grant_type": "refresh_token", + "client_id": self.client_id, + "refresh_token": token["refresh_token"], + } + ) + return {**token, **new_token} diff --git a/homeassistant/components/withings/binary_sensor.py b/homeassistant/components/withings/binary_sensor.py index e1351d7c019..976774f23b3 100644 --- a/homeassistant/components/withings/binary_sensor.py +++ b/homeassistant/components/withings/binary_sensor.py @@ -14,13 +14,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .common import ( - BaseWithingsSensor, - UpdateType, - WithingsEntityDescription, - async_get_data_manager, -) +from .common import UpdateType, async_get_data_manager from .const import Measurement +from .entity import BaseWithingsSensor, WithingsEntityDescription @dataclass diff --git a/homeassistant/components/withings/common.py b/homeassistant/components/withings/common.py index 516c306cc0f..3d215567f45 100644 --- a/homeassistant/components/withings/common.py +++ b/homeassistant/components/withings/common.py @@ -28,7 +28,6 @@ from withings_api.common import ( ) from homeassistant.components import webhook -from homeassistant.components.application_credentials import AuthImplementation from homeassistant.components.http import HomeAssistantView from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_WEBHOOK_ID @@ -38,13 +37,11 @@ from homeassistant.helpers.config_entry_oauth2_flow import ( AbstractOAuth2Implementation, OAuth2Session, ) -from homeassistant.helpers.device_registry import DeviceInfo -from homeassistant.helpers.entity import Entity, EntityDescription from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.util import dt as dt_util from . import const -from .const import DOMAIN, Measurement +from .const import Measurement _LOGGER = logging.getLogger(const.LOG_NAMESPACE) _RETRY_COEFFICIENT = 0.5 @@ -64,20 +61,6 @@ class UpdateType(StrEnum): WEBHOOK = "webhook" -@dataclass -class WithingsEntityDescriptionMixin: - """Mixin for describing withings data.""" - - measurement: Measurement - measure_type: NotifyAppli | GetSleepSummaryField | MeasureType - update_type: UpdateType - - -@dataclass -class WithingsEntityDescription(EntityDescription, WithingsEntityDescriptionMixin): - """Immutable class for describing withings data.""" - - @dataclass class WebhookConfig: """Config for a webhook.""" @@ -538,85 +521,6 @@ class DataManager: ) -def get_attribute_unique_id( - description: WithingsEntityDescription, user_id: int -) -> str: - """Get a entity unique id for a user's attribute.""" - return f"withings_{user_id}_{description.measurement.value}" - - -class BaseWithingsSensor(Entity): - """Base class for withings sensors.""" - - _attr_should_poll = False - entity_description: WithingsEntityDescription - _attr_has_entity_name = True - - def __init__( - self, data_manager: DataManager, description: WithingsEntityDescription - ) -> None: - """Initialize the Withings sensor.""" - self._data_manager = data_manager - self.entity_description = description - self._attr_unique_id = get_attribute_unique_id( - description, data_manager.user_id - ) - self._state_data: Any | None = None - self._attr_device_info = DeviceInfo( - identifiers={(DOMAIN, str(data_manager.user_id))}, - name=data_manager.profile, - ) - - @property - def available(self) -> bool: - """Return True if entity is available.""" - if self.entity_description.update_type == UpdateType.POLL: - return self._data_manager.poll_data_update_coordinator.last_update_success - - if self.entity_description.update_type == UpdateType.WEBHOOK: - return self._data_manager.webhook_config.enabled and ( - self.entity_description.measurement - in self._data_manager.webhook_update_coordinator.data - ) - - return True - - @callback - def _on_poll_data_updated(self) -> None: - self._update_state_data( - self._data_manager.poll_data_update_coordinator.data or {} - ) - - @callback - def _on_webhook_data_updated(self) -> None: - self._update_state_data( - self._data_manager.webhook_update_coordinator.data or {} - ) - - def _update_state_data(self, data: dict[Measurement, Any]) -> None: - """Update the state data.""" - self._state_data = data.get(self.entity_description.measurement) - self.async_write_ha_state() - - async def async_added_to_hass(self) -> None: - """Register update dispatcher.""" - if self.entity_description.update_type == UpdateType.POLL: - self.async_on_remove( - self._data_manager.poll_data_update_coordinator.async_add_listener( - self._on_poll_data_updated - ) - ) - self._on_poll_data_updated() - - elif self.entity_description.update_type == UpdateType.WEBHOOK: - self.async_on_remove( - self._data_manager.webhook_update_coordinator.async_add_listener( - self._on_webhook_data_updated - ) - ) - self._on_webhook_data_updated() - - async def async_get_data_manager( hass: HomeAssistant, config_entry: ConfigEntry ) -> DataManager: @@ -680,50 +584,3 @@ def get_all_data_managers(hass: HomeAssistant) -> tuple[DataManager, ...]: def async_remove_data_manager(hass: HomeAssistant, config_entry: ConfigEntry) -> None: """Remove a data manager for a config entry.""" del hass.data[const.DOMAIN][config_entry.entry_id][const.DATA_MANAGER] - - -class WithingsLocalOAuth2Implementation(AuthImplementation): - """Oauth2 implementation that only uses the external url.""" - - async def _token_request(self, data: dict) -> dict: - """Make a token request and adapt Withings API reply.""" - new_token = await super()._token_request(data) - # Withings API returns habitual token data under json key "body": - # { - # "status": [{integer} Withings API response status], - # "body": { - # "access_token": [{string} Your new access_token], - # "expires_in": [{integer} Access token expiry delay in seconds], - # "token_type": [{string] HTTP Authorization Header format: Bearer], - # "scope": [{string} Scopes the user accepted], - # "refresh_token": [{string} Your new refresh_token], - # "userid": [{string} The Withings ID of the user] - # } - # } - # so we copy that to token root. - if body := new_token.pop("body", None): - new_token.update(body) - return new_token - - async def async_resolve_external_data(self, external_data: Any) -> dict: - """Resolve the authorization code to tokens.""" - return await self._token_request( - { - "action": "requesttoken", - "grant_type": "authorization_code", - "code": external_data["code"], - "redirect_uri": external_data["state"]["redirect_uri"], - } - ) - - async def _async_refresh_token(self, token: dict) -> dict: - """Refresh tokens.""" - new_token = await self._token_request( - { - "action": "requesttoken", - "grant_type": "refresh_token", - "client_id": self.client_id, - "refresh_token": token["refresh_token"], - } - ) - return {**token, **new_token} diff --git a/homeassistant/components/withings/entity.py b/homeassistant/components/withings/entity.py new file mode 100644 index 00000000000..a1ad8828b81 --- /dev/null +++ b/homeassistant/components/withings/entity.py @@ -0,0 +1,100 @@ +"""Base entity for Withings.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from withings_api.common import GetSleepSummaryField, MeasureType, NotifyAppli + +from homeassistant.core import callback +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity import Entity, EntityDescription + +from .common import DataManager, UpdateType +from .const import DOMAIN, Measurement + + +@dataclass +class WithingsEntityDescriptionMixin: + """Mixin for describing withings data.""" + + measurement: Measurement + measure_type: NotifyAppli | GetSleepSummaryField | MeasureType + update_type: UpdateType + + +@dataclass +class WithingsEntityDescription(EntityDescription, WithingsEntityDescriptionMixin): + """Immutable class for describing withings data.""" + + +class BaseWithingsSensor(Entity): + """Base class for withings sensors.""" + + _attr_should_poll = False + entity_description: WithingsEntityDescription + _attr_has_entity_name = True + + def __init__( + self, data_manager: DataManager, description: WithingsEntityDescription + ) -> None: + """Initialize the Withings sensor.""" + self._data_manager = data_manager + self.entity_description = description + self._attr_unique_id = ( + f"withings_{data_manager.user_id}_{description.measurement.value}" + ) + self._state_data: Any | None = None + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, str(data_manager.user_id))}, + name=data_manager.profile, + ) + + @property + def available(self) -> bool: + """Return True if entity is available.""" + if self.entity_description.update_type == UpdateType.POLL: + return self._data_manager.poll_data_update_coordinator.last_update_success + + if self.entity_description.update_type == UpdateType.WEBHOOK: + return self._data_manager.webhook_config.enabled and ( + self.entity_description.measurement + in self._data_manager.webhook_update_coordinator.data + ) + + return True + + @callback + def _on_poll_data_updated(self) -> None: + self._update_state_data( + self._data_manager.poll_data_update_coordinator.data or {} + ) + + @callback + def _on_webhook_data_updated(self) -> None: + self._update_state_data( + self._data_manager.webhook_update_coordinator.data or {} + ) + + def _update_state_data(self, data: dict[Measurement, Any]) -> None: + """Update the state data.""" + self._state_data = data.get(self.entity_description.measurement) + self.async_write_ha_state() + + async def async_added_to_hass(self) -> None: + """Register update dispatcher.""" + if self.entity_description.update_type == UpdateType.POLL: + self.async_on_remove( + self._data_manager.poll_data_update_coordinator.async_add_listener( + self._on_poll_data_updated + ) + ) + self._on_poll_data_updated() + + elif self.entity_description.update_type == UpdateType.WEBHOOK: + self.async_on_remove( + self._data_manager.webhook_update_coordinator.async_add_listener( + self._on_webhook_data_updated + ) + ) + self._on_webhook_data_updated() diff --git a/homeassistant/components/withings/sensor.py b/homeassistant/components/withings/sensor.py index 4f98daacc42..e8798adae2f 100644 --- a/homeassistant/components/withings/sensor.py +++ b/homeassistant/components/withings/sensor.py @@ -23,12 +23,7 @@ from homeassistant.const import ( from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .common import ( - BaseWithingsSensor, - UpdateType, - WithingsEntityDescription, - async_get_data_manager, -) +from .common import UpdateType, async_get_data_manager from .const import ( SCORE_POINTS, UOM_BEATS_PER_MINUTE, @@ -37,6 +32,7 @@ from .const import ( UOM_MMHG, Measurement, ) +from .entity import BaseWithingsSensor, WithingsEntityDescription @dataclass diff --git a/tests/components/withings/common.py b/tests/components/withings/common.py index 6bb1b30917c..7680b19e289 100644 --- a/tests/components/withings/common.py +++ b/tests/components/withings/common.py @@ -23,11 +23,10 @@ import homeassistant.components.webhook as webhook from homeassistant.components.withings.common import ( ConfigEntryWithingsApi, DataManager, - WithingsEntityDescription, get_all_data_managers, - get_attribute_unique_id, ) import homeassistant.components.withings.const as const +from homeassistant.components.withings.entity import WithingsEntityDescription from homeassistant.config import async_process_ha_core_config from homeassistant.config_entries import SOURCE_USER, ConfigEntry from homeassistant.const import ( @@ -324,6 +323,6 @@ async def async_get_entity_id( ) -> str | None: """Get an entity id for a user's attribute.""" entity_registry = er.async_get(hass) - unique_id = get_attribute_unique_id(description, user_id) + unique_id = f"withings_{user_id}_{description.measurement.value}" return entity_registry.async_get_entity_id(platform, const.DOMAIN, unique_id) diff --git a/tests/components/withings/test_sensor.py b/tests/components/withings/test_sensor.py index 4cc71df80d7..cf0069c968a 100644 --- a/tests/components/withings/test_sensor.py +++ b/tests/components/withings/test_sensor.py @@ -7,8 +7,8 @@ from syrupy import SnapshotAssertion from withings_api.common import NotifyAppli from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN -from homeassistant.components.withings.common import WithingsEntityDescription from homeassistant.components.withings.const import Measurement +from homeassistant.components.withings.entity import WithingsEntityDescription from homeassistant.components.withings.sensor import SENSORS from homeassistant.core import HomeAssistant, State from homeassistant.helpers import entity_registry as er