Split Withings common file out to their own file (#100150)

* Split common out in logical pieces

* Split common out in logical pieces

* Split common out in logical pieces

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}

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 (
+from .common import UpdateType, async_get_data_manager
-    BaseWithingsSensor,
-    UpdateType,
-    WithingsEntityDescription,
-    async_get_data_manager,
-)
 from .const import Measurement
+from .entity import BaseWithingsSensor, WithingsEntityDescription
 
 
 @dataclass

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}

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()

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 (
+from .common import UpdateType, async_get_data_manager
-    BaseWithingsSensor,
-    UpdateType,
-    WithingsEntityDescription,
-    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

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)

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