Move xbox coordinator to separate module (#117421)

.coveragerc

@@ -1647,6 +1647,7 @@ omit =
     homeassistant/components/xbox/base_sensor.py
     homeassistant/components/xbox/binary_sensor.py
     homeassistant/components/xbox/browse_media.py
+    homeassistant/components/xbox/coordinator.py
     homeassistant/components/xbox/media_player.py
     homeassistant/components/xbox/remote.py
     homeassistant/components/xbox/sensor.py

homeassistant/components/xbox/__init__.py

@@ -2,23 +2,10 @@
 
 from __future__ import annotations
 
-from contextlib import suppress
-from dataclasses import dataclass
-from datetime import timedelta
 import logging
 
 from xbox.webapi.api.client import XboxLiveClient
-from xbox.webapi.api.provider.catalog.const import SYSTEM_PFN_ID_MAP
+from xbox.webapi.api.provider.smartglass.models import SmartglassConsoleList
-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.smartglass.models import (
-    SmartglassConsoleList,
-    SmartglassConsoleStatus,
-)
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
@@ -28,10 +15,10 @@ from homeassistant.helpers import (
     config_entry_oauth2_flow,
     config_validation as cv,
 )
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from . import api
 from .const import DOMAIN
+from .coordinator import XboxUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -89,142 +76,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         hass.data[DOMAIN].pop(entry.entry_id)
 
     return unload_ok
-
-
-@dataclass
-class ConsoleData:
-    """Xbox console status data."""
-
-    status: SmartglassConsoleStatus
-    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
-
-
-@dataclass
-class XboxData:
-    """Xbox dataclass for update coordinator."""
-
-    consoles: dict[str, ConsoleData]
-    presence: dict[str, PresenceData]
-
-
-class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):  # pylint: disable=hass-enforce-coordinator-module
-    """Store Xbox Console Status."""
-
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        client: XboxLiveClient,
-        consoles: SmartglassConsoleList,
-    ) -> None:
-        """Initialize."""
-        super().__init__(
-            hass,
-            _LOGGER,
-            name=DOMAIN,
-            update_interval=timedelta(seconds=10),
-        )
-        self.data = XboxData({}, {})
-        self.client: XboxLiveClient = client
-        self.consoles: SmartglassConsoleList = consoles
-
-    async def _async_update_data(self) -> XboxData:
-        """Fetch the latest console status."""
-        # 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)
-            )
-
-            _LOGGER.debug(
-                "%s status: %s",
-                console.name,
-                status.dict(),
-            )
-
-            # Setup focus app
-            app_details: Product | None = None
-            if current_state is not None:
-                app_details = current_state.app_details
-
-            if status.focus_app_aumid:
-                if (
-                    not current_state
-                    or status.focus_app_aumid != current_state.status.focus_app_aumid
-                ):
-                    app_id = status.focus_app_aumid.split("!")[0]
-                    id_type = AlternateIdType.PACKAGE_FAMILY_NAME
-                    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
-                        )
-                    )
-                    if catalog_result and catalog_result.products:
-                        app_details = catalog_result.products[0]
-            else:
-                app_details = None
-
-            new_console_data[console.id] = ConsoleData(
-                status=status, app_details=app_details
-            )
-
-        # 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)
-
-        return XboxData(new_console_data, presence_data)
-
-
-def _build_presence_data(person: Person) -> PresenceData:
-    """Build presence data from a person."""
-    active_app: PresenceDetail | None = None
-    with suppress(StopIteration):
-        active_app = next(
-            presence for presence in person.presence_details if presence.is_primary
-        )
-
-    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=active_app is not None and active_app.is_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,
-    )

homeassistant/components/xbox/base_sensor.py

@@ -7,8 +7,8 @@ from yarl import URL
 from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import PresenceData, XboxUpdateCoordinator
 from .const import DOMAIN
+from .coordinator import PresenceData, XboxUpdateCoordinator
 
 
 class XboxBaseSensorEntity(CoordinatorEntity[XboxUpdateCoordinator]):

homeassistant/components/xbox/binary_sensor.py

@@ -10,9 +10,9 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import XboxUpdateCoordinator
 from .base_sensor import XboxBaseSensorEntity
 from .const import DOMAIN
+from .coordinator import XboxUpdateCoordinator
 
 PRESENCE_ATTRIBUTES = ["online", "in_party", "in_game", "in_multiplayer"]
 

homeassistant/components/xbox/coordinator.py

@@ -0,0 +1,167 @@
+"""Coordinator for the xbox integration."""
+
+from __future__ import annotations
+
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import timedelta
+import logging
+
+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.smartglass.models import (
+    SmartglassConsoleList,
+    SmartglassConsoleStatus,
+)
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class ConsoleData:
+    """Xbox console status data."""
+
+    status: SmartglassConsoleStatus
+    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
+
+
+@dataclass
+class XboxData:
+    """Xbox dataclass for update coordinator."""
+
+    consoles: dict[str, ConsoleData]
+    presence: dict[str, PresenceData]
+
+
+class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
+    """Store Xbox Console Status."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        client: XboxLiveClient,
+        consoles: SmartglassConsoleList,
+    ) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=10),
+        )
+        self.data = XboxData({}, {})
+        self.client: XboxLiveClient = client
+        self.consoles: SmartglassConsoleList = consoles
+
+    async def _async_update_data(self) -> XboxData:
+        """Fetch the latest console status."""
+        # 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)
+            )
+
+            _LOGGER.debug(
+                "%s status: %s",
+                console.name,
+                status.dict(),
+            )
+
+            # Setup focus app
+            app_details: Product | None = None
+            if current_state is not None:
+                app_details = current_state.app_details
+
+            if status.focus_app_aumid:
+                if (
+                    not current_state
+                    or status.focus_app_aumid != current_state.status.focus_app_aumid
+                ):
+                    app_id = status.focus_app_aumid.split("!")[0]
+                    id_type = AlternateIdType.PACKAGE_FAMILY_NAME
+                    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
+                        )
+                    )
+                    if catalog_result and catalog_result.products:
+                        app_details = catalog_result.products[0]
+            else:
+                app_details = None
+
+            new_console_data[console.id] = ConsoleData(
+                status=status, app_details=app_details
+            )
+
+        # 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)
+
+        return XboxData(new_console_data, presence_data)
+
+
+def _build_presence_data(person: Person) -> PresenceData:
+    """Build presence data from a person."""
+    active_app: PresenceDetail | None = None
+    with suppress(StopIteration):
+        active_app = next(
+            presence for presence in person.presence_details if presence.is_primary
+        )
+
+    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=active_app is not None and active_app.is_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,
+    )

homeassistant/components/xbox/media_player.py

@@ -27,9 +27,9 @@ from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import ConsoleData, XboxUpdateCoordinator
 from .browse_media import build_item_response
 from .const import DOMAIN
+from .coordinator import ConsoleData, XboxUpdateCoordinator
 
 SUPPORT_XBOX = (
     MediaPlayerEntityFeature.TURN_ON

homeassistant/components/xbox/remote.py

@@ -27,8 +27,8 @@ from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import ConsoleData, XboxUpdateCoordinator
 from .const import DOMAIN
+from .coordinator import ConsoleData, XboxUpdateCoordinator
 
 
 async def async_setup_entry(

homeassistant/components/xbox/sensor.py

@@ -10,9 +10,9 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import XboxUpdateCoordinator
 from .base_sensor import XboxBaseSensorEntity
 from .const import DOMAIN
+from .coordinator import XboxUpdateCoordinator
 
 SENSOR_ATTRIBUTES = ["status", "gamer_score", "account_tier", "gold_tenure"]