Refactor coordinator data update and exception handling in Xbox integration (home-assistant#154848)

Author: tr4nt0r

homeassistant/components/xbox/binary_sensor.py

@@ -7,6 +7,7 @@
 from enum import StrEnum
 from functools import partial
 
+from xbox.webapi.api.provider.people.models import Person
 from yarl import URL
 
 from homeassistant.components.binary_sensor import (
@@ -16,7 +17,7 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from .coordinator import PresenceData, XboxConfigEntry, XboxUpdateCoordinator
+from .coordinator import XboxConfigEntry, XboxUpdateCoordinator
 from .entity import XboxBaseEntity
 
 
@@ -34,11 +35,11 @@ class XboxBinarySensor(StrEnum):
 class XboxBinarySensorEntityDescription(BinarySensorEntityDescription):
     """Xbox binary sensor description."""
 
-    is_on_fn: Callable[[PresenceData], bool | None]
-    entity_picture_fn: Callable[[PresenceData], str | None] | None = None
+    is_on_fn: Callable[[Person], bool | None]
+    entity_picture_fn: Callable[[Person], str | None] | None = None
 
 
-def profile_pic(data: PresenceData) -> str | None:
+def profile_pic(person: Person) -> str | None:
     """Return the gamer pic."""
 
     # Xbox sometimes returns a domain that uses a wrong certificate which
@@ -47,43 +48,67 @@ def profile_pic(data: PresenceData) -> str | None:
     # to point to the correct image, with the correct domain and certificate.
     # We need to also remove the 'mode=Padding' query because with it,
     # it results in an error 400.
-    url = URL(data.display_pic)
+    url = URL(person.display_pic_raw)
     if url.host == "images-eds.xboxlive.com":
         url = url.with_host("images-eds-ssl.xboxlive.com").with_scheme("https")
     query = dict(url.query)
     query.pop("mode", None)
     return str(url.with_query(query))
 
 
+def in_game(person: Person) -> bool:
+    """True if person is in a game."""
+
+    active_app = (
+        next(
+            (presence for presence in person.presence_details if presence.is_primary),
+            None,
+        )
+        if person.presence_details
+        else None
+    )
+    return (
+        active_app is not None and active_app.is_game and active_app.state == "Active"
+    )
+
+
 SENSOR_DESCRIPTIONS: tuple[XboxBinarySensorEntityDescription, ...] = (
     XboxBinarySensorEntityDescription(
         key=XboxBinarySensor.ONLINE,
         translation_key=XboxBinarySensor.ONLINE,
-        is_on_fn=lambda x: x.online,
+        is_on_fn=lambda x: x.presence_state == "Online",
         name=None,
         entity_picture_fn=profile_pic,
     ),
     XboxBinarySensorEntityDescription(
         key=XboxBinarySensor.IN_PARTY,
         translation_key=XboxBinarySensor.IN_PARTY,
-        is_on_fn=lambda x: x.in_party,
+        is_on_fn=(
+            lambda x: bool(x.multiplayer_summary.in_party)
+            if x.multiplayer_summary
+            else None
+        ),
         entity_registry_enabled_default=False,
     ),
     XboxBinarySensorEntityDescription(
         key=XboxBinarySensor.IN_GAME,
         translation_key=XboxBinarySensor.IN_GAME,
-        is_on_fn=lambda x: x.in_game,
+        is_on_fn=in_game,
     ),
     XboxBinarySensorEntityDescription(
         key=XboxBinarySensor.IN_MULTIPLAYER,
         translation_key=XboxBinarySensor.IN_MULTIPLAYER,
-        is_on_fn=lambda x: x.in_multiplayer,
+        is_on_fn=(
+            lambda x: bool(x.multiplayer_summary.in_multiplayer_session)
+            if x.multiplayer_summary
+            else None
+        ),
         entity_registry_enabled_default=False,
     ),
     XboxBinarySensorEntityDescription(
         key=XboxBinarySensor.HAS_GAME_PASS,
         translation_key=XboxBinarySensor.HAS_GAME_PASS,
-        is_on_fn=lambda x: x.has_game_pass,
+        is_on_fn=lambda x: x.detail.has_game_pass if x.detail else None,
     ),
 )
 

homeassistant/components/xbox/coordinator.py

@@ -3,18 +3,14 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from datetime import UTC, datetime, timedelta
+from datetime import timedelta
 import logging
 
 from httpx import HTTPStatusError, RequestError, TimeoutException
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.api.provider.catalog.const import SYSTEM_PFN_ID_MAP
 from xbox.webapi.api.provider.catalog.models import AlternateIdType, Product
-from xbox.webapi.api.provider.people.models import (
-    PeopleResponse,
-    Person,
-    PresenceDetail,
-)
+from xbox.webapi.api.provider.people.models import Person
 from xbox.webapi.api.provider.smartglass.models import (
     SmartglassConsoleList,
     SmartglassConsoleStatus,
@@ -25,7 +21,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import config_entry_oauth2_flow, device_registry as dr
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from . import api
 from .const import DOMAIN
@@ -43,33 +39,12 @@ class ConsoleData:
     app_details: Product | None
 
 
-@dataclass
-class PresenceData:
-    """Xbox user presence data."""
-
-    xuid: str
-    gamertag: str
-    display_pic: str
-    online: bool
-    status: str
-    in_party: bool
-    in_game: bool
-    in_multiplayer: bool
-    gamer_score: str
-    gold_tenure: str | None
-    account_tier: str
-    last_seen: datetime | None
-    following_count: int
-    follower_count: int
-    has_game_pass: bool
-
-
 @dataclass
 class XboxData:
     """Xbox dataclass for update coordinator."""
 
     consoles: dict[str, ConsoleData] = field(default_factory=dict)
-    presence: dict[str, PresenceData] = field(default_factory=dict)
+    presence: dict[str, Person] = field(default_factory=dict)
 
 
 class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
@@ -107,7 +82,6 @@ async def _async_setup(self) -> None:
             raise ConfigEntryNotReady(
                 translation_domain=DOMAIN,
                 translation_key="request_exception",
-                translation_placeholders={"error": str(e)},
             ) from e
 
         session = config_entry_oauth2_flow.OAuth2Session(
@@ -129,7 +103,6 @@ async def _async_setup(self) -> None:
             raise ConfigEntryNotReady(
                 translation_domain=DOMAIN,
                 translation_key="request_exception",
-                translation_placeholders={"error": str(e)},
             ) from e
 
         _LOGGER.debug(
@@ -143,11 +116,20 @@ async def _async_update_data(self) -> XboxData:
         # Update Console Status
         new_console_data: dict[str, ConsoleData] = {}
         for console in self.consoles.result:
-            current_state: ConsoleData | None = self.data.consoles.get(console.id)
-            status: SmartglassConsoleStatus = (
-                await self.client.smartglass.get_console_status(console.id)
-            )
-
+            current_state = self.data.consoles.get(console.id)
+            try:
+                status = await self.client.smartglass.get_console_status(console.id)
+            except TimeoutException as e:
+                raise UpdateFailed(
+                    translation_domain=DOMAIN,
+                    translation_key="timeout_exception",
+                ) from e
+            except (RequestError, HTTPStatusError) as e:
+                _LOGGER.debug("Xbox exception:", exc_info=True)
+                raise UpdateFailed(
+                    translation_domain=DOMAIN,
+                    translation_key="request_exception",
+                ) from e
             _LOGGER.debug(
                 "%s status: %s",
                 console.name,
@@ -169,13 +151,26 @@ async def _async_update_data(self) -> XboxData:
                     if app_id in SYSTEM_PFN_ID_MAP:
                         id_type = AlternateIdType.LEGACY_XBOX_PRODUCT_ID
                         app_id = SYSTEM_PFN_ID_MAP[app_id][id_type]
-                    catalog_result = (
-                        await self.client.catalog.get_product_from_alternate_id(
-                            app_id, id_type
+                    try:
+                        catalog_result = (
+                            await self.client.catalog.get_product_from_alternate_id(
+                                app_id, id_type
+                            )
                         )
-                    )
-                    if catalog_result and catalog_result.products:
-                        app_details = catalog_result.products[0]
+                    except TimeoutException as e:
+                        raise UpdateFailed(
+                            translation_domain=DOMAIN,
+                            translation_key="timeout_exception",
+                        ) from e
+                    except (RequestError, HTTPStatusError) as e:
+                        _LOGGER.debug("Xbox exception:", exc_info=True)
+                        raise UpdateFailed(
+                            translation_domain=DOMAIN,
+                            translation_key="request_exception",
+                        ) from e
+                    else:
+                        if catalog_result.products:
+                            app_details = catalog_result.products[0]
             else:
                 app_details = None
 
@@ -184,19 +179,25 @@ async def _async_update_data(self) -> XboxData:
             )
 
         # Update user presence
-        presence_data: dict[str, PresenceData] = {}
-        batch: PeopleResponse = await self.client.people.get_friends_own_batch(
-            [self.client.xuid]
-        )
-        own_presence: Person = batch.people[0]
-        presence_data[own_presence.xuid] = _build_presence_data(own_presence)
-
-        friends: PeopleResponse = await self.client.people.get_friends_own()
-        for friend in friends.people:
-            if not friend.is_favorite:
-                continue
-
-            presence_data[friend.xuid] = _build_presence_data(friend)
+        try:
+            batch = await self.client.people.get_friends_own_batch([self.client.xuid])
+            friends = await self.client.people.get_friends_own()
+        except TimeoutException as e:
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="timeout_exception",
+            ) from e
+        except (RequestError, HTTPStatusError) as e:
+            _LOGGER.debug("Xbox exception:", exc_info=True)
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="request_exception",
+            ) from e
+        else:
+            presence_data = {self.client.xuid: batch.people[0]}
+            presence_data.update(
+                {friend.xuid: friend for friend in friends.people if friend.is_favorite}
+            )
 
         if (
             self.current_friends
@@ -208,11 +209,11 @@ async def _async_update_data(self) -> XboxData:
 
         return XboxData(new_console_data, presence_data)
 
-    def remove_stale_devices(self, presence_data: dict[str, PresenceData]) -> None:
+    def remove_stale_devices(self, presence_data: dict[str, Person]) -> None:
         """Remove stale devices from registry."""
 
         device_reg = dr.async_get(self.hass)
-        identifiers = {(DOMAIN, person.xuid) for person in presence_data.values()} | {
+        identifiers = {(DOMAIN, xuid) for xuid in set(presence_data)} | {
             (DOMAIN, console.id) for console in self.consoles.result
         }
 
@@ -224,38 +225,3 @@ def remove_stale_devices(self, presence_data: dict[str, PresenceData]) -> None:
                 device_reg.async_update_device(
                     device.id, remove_config_entry_id=self.config_entry.entry_id
                 )
-
-
-def _build_presence_data(person: Person) -> PresenceData:
-    """Build presence data from a person."""
-    active_app: PresenceDetail | None = None
-
-    active_app = next(
-        (presence for presence in person.presence_details if presence.is_primary),
-        None,
-    )
-    in_game = (
-        active_app is not None and active_app.is_game and active_app.state == "Active"
-    )
-
-    return PresenceData(
-        xuid=person.xuid,
-        gamertag=person.gamertag,
-        display_pic=person.display_pic_raw,
-        online=person.presence_state == "Online",
-        status=person.presence_text,
-        in_party=person.multiplayer_summary.in_party > 0,
-        in_game=in_game,
-        in_multiplayer=person.multiplayer_summary.in_multiplayer_session,
-        gamer_score=person.gamer_score,
-        gold_tenure=person.detail.tenure,
-        account_tier=person.detail.account_tier,
-        last_seen=(
-            person.last_seen_date_time_utc.replace(tzinfo=UTC)
-            if person.last_seen_date_time_utc
-            else None
-        ),
-        follower_count=person.detail.follower_count,
-        following_count=person.detail.following_count,
-        has_game_pass=person.detail.has_game_pass,
-    )

homeassistant/components/xbox/entity.py

@@ -7,7 +7,7 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
-from .coordinator import PresenceData, XboxUpdateCoordinator
+from .coordinator import Person, XboxUpdateCoordinator
 
 
 class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
@@ -37,6 +37,6 @@ def __init__(
         )
 
     @property
-    def data(self) -> PresenceData:
+    def data(self) -> Person:
         """Return coordinator data for this console."""
         return self.coordinator.data.presence[self.xuid]