Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 06:05:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Xbox coordinators (#160174)

Browse Source
This commit is contained in:
Manu
2026-01-05 21:31:57 +01:00
committed by GitHub
parent 5b0dab479d
commit 354fafda1a
14 changed files with 290 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
homeassistant/components/xbox/__init__.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import logging
from httpx import HTTPStatusError, RequestError, TimeoutException
@@ -19,13 +20,15 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
)
from homeassistant.helpers.httpx_client import get_async_client
from . import api
from .api import AsyncConfigEntryAuth
from .const import DOMAIN
from .coordinator import (
XboxConfigEntry,
XboxConsolesCoordinator,
XboxConsoleStatusCoordinator,
XboxCoordinators,
XboxUpdateCoordinator,
XboxPresenceCoordinator,
)
_LOGGER = logging.getLogger(__name__)
@@ -44,12 +47,30 @@ PLATFORMS = [
async def async_setup_entry(hass: HomeAssistant, entry: XboxConfigEntry) -> bool:
"""Set up xbox from a config entry."""
coordinator = XboxUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
try:
implementation = await async_get_config_entry_implementation(hass, entry)
except ImplementationUnavailableError as e:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="oauth2_implementation_unavailable",
) from e
consoles = XboxConsolesCoordinator(hass, entry, coordinator)
session = OAuth2Session(hass, entry, implementation)
async_session = get_async_client(hass)
auth = api.AsyncConfigEntryAuth(async_session, session)
client = XboxLiveClient(auth)
entry.runtime_data = XboxCoordinators(coordinator, consoles)
consoles = XboxConsolesCoordinator(hass, entry, client)
await consoles.async_config_entry_first_refresh()
status = XboxConsoleStatusCoordinator(hass, entry, client, consoles.data)
presence = XboxPresenceCoordinator(hass, entry, client)
await asyncio.gather(
status.async_config_entry_first_refresh(),
presence.async_config_entry_first_refresh(),
)
entry.runtime_data = XboxCoordinators(consoles, status, presence)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

2
homeassistant/components/xbox/binary_sensor.py
View File

@@ -112,7 +112,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox Live friends."""
coordinator = entry.runtime_data.status
coordinator = entry.runtime_data.presence
if TYPE_CHECKING:
assert entry.unique_id

2
homeassistant/components/xbox/config_flow.py
View File

@@ -122,7 +122,7 @@ class FriendSubentryFlowHandler(ConfigSubentryFlow):
if config_entry.state is not ConfigEntryState.LOADED:
return self.async_abort(reason="config_entry_not_loaded")
client = config_entry.runtime_data.status.client
client = config_entry.runtime_data.presence.client
friends_list = await client.people.get_friends_own()
if user_input is not None:

286
homeassistant/components/xbox/coordinator.py
View File

@@ -2,10 +2,12 @@
from __future__ import annotations
from abc import abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from http import HTTPStatus
import logging
from typing import ClassVar
from httpx import HTTPStatusError, RequestError, TimeoutException
from pythonxbox.api.client import XboxLiveClient
@@ -13,23 +15,15 @@ from pythonxbox.api.provider.catalog.const import SYSTEM_PFN_ID_MAP
from pythonxbox.api.provider.catalog.models import AlternateIdType, Product
from pythonxbox.api.provider.people.models import Person
from pythonxbox.api.provider.smartglass.models import (
SmartglassConsoleList,
SmartglassConsole,
SmartglassConsoleStatus,
)
from pythonxbox.api.provider.titlehub.models import Title
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
OAuth2Session,
async_get_config_entry_implementation,
)
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from . import api
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@@ -47,9 +41,8 @@ class ConsoleData:
@dataclass
class XboxData:
"""Xbox dataclass for update coordinator."""
"""Xbox dataclass for presence update coordinator."""
consoles: dict[str, ConsoleData] = field(default_factory=dict)
presence: dict[str, Person] = field(default_factory=dict)
title_info: dict[str, Title] = field(default_factory=dict)
@@ -58,21 +51,22 @@ class XboxData:
class XboxCoordinators:
"""Xbox coordinators."""
status: XboxUpdateCoordinator
consoles: XboxConsolesCoordinator
status: XboxConsoleStatusCoordinator
presence: XboxPresenceCoordinator
class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
"""Store Xbox Console Status."""
class XboxBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Base coordinator for Xbox."""
config_entry: XboxConfigEntry
consoles: SmartglassConsoleList
client: XboxLiveClient
_update_inverval: timedelta
def __init__(
self,
hass: HomeAssistant,
config_entry: XboxConfigEntry,
client: XboxLiveClient,
) -> None:
"""Initialize."""
super().__init__(
@@ -80,77 +74,86 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(seconds=15),
update_interval=self._update_interval,
)
self.data = XboxData()
self.current_friends: set[str] = set()
self.title_data: dict[str, Title] = {}
self.client = client
async def _async_setup(self) -> None:
"""Set up coordinator."""
try:
implementation = await async_get_config_entry_implementation(
self.hass, self.config_entry
)
except ImplementationUnavailableError as e:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="oauth2_implementation_unavailable",
) from e
@abstractmethod
async def update_data(self) -> _DataT:
"""Update coordinator data."""
session = OAuth2Session(self.hass, self.config_entry, implementation)
async_session = get_async_client(self.hass)
auth = api.AsyncConfigEntryAuth(async_session, session)
self.client = XboxLiveClient(auth)
async def _async_update_data(self) -> _DataT:
"""Fetch console data."""
try:
self.consoles = await self.client.smartglass.get_console_list()
return await self.update_data()
except TimeoutException as e:
raise ConfigEntryNotReady(
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="timeout_exception",
) from e
except (RequestError, HTTPStatusError) as e:
_LOGGER.debug("Xbox exception:", exc_info=True)
raise ConfigEntryNotReady(
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="request_exception",
) from e
class XboxConsolesCoordinator(XboxBaseCoordinator[dict[str, SmartglassConsole]]):
"""Update list of Xbox consoles."""
config_entry: XboxConfigEntry
_update_interval = timedelta(minutes=10)
async def update_data(self) -> dict[str, SmartglassConsole]:
"""Fetch console data."""
consoles = await self.client.smartglass.get_console_list()
_LOGGER.debug(
"Found %d consoles: %s",
len(self.consoles.result),
self.consoles.model_dump(),
"Found %d consoles: %s", len(consoles.result), consoles.model_dump()
)
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 = 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,
status.model_dump(),
)
return {console.id: console for console in consoles.result}
class XboxConsoleStatusCoordinator(XboxBaseCoordinator[dict[str, ConsoleData]]):
"""Update Xbox console Status."""
config_entry: XboxConfigEntry
_update_interval = timedelta(seconds=10)
def __init__(
self,
hass: HomeAssistant,
config_entry: XboxConfigEntry,
client: XboxLiveClient,
consoles: dict[str, SmartglassConsole],
) -> None:
"""Initialize."""
super().__init__(hass, config_entry, client)
self.data: dict[str, ConsoleData] = {}
self.consoles: dict[str, SmartglassConsole] | None = consoles
async def update_data(self) -> dict[str, ConsoleData]:
"""Fetch console data."""
consoles: list[SmartglassConsole] = list(self.async_contexts())
if not consoles and self.consoles is not None:
consoles = list(self.consoles.values())
self.consoles = None
data: dict[str, ConsoleData] = {}
for console in consoles:
status = await self.client.smartglass.get_console_status(console.id)
_LOGGER.debug("%s status: %s", console.name, status.model_dump())
# Setup focus app
app_details: Product | None = None
if current_state is not None:
if (current_state := self.data.get(console.id)) is not None:
app_details = current_state.app_details
if status.focus_app_aumid:
@@ -163,51 +166,44 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[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]
try:
catalog_result = (
await self.client.catalog.get_product_from_alternate_id(
app_id, id_type
)
catalog_result = (
await self.client.catalog.get_product_from_alternate_id(
app_id, id_type
)
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]
)
if 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
)
data[console.id] = ConsoleData(status=status, app_details=app_details)
# Update user presence
try:
batch = await self.client.people.get_friends_by_xuid(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})
return data
class XboxPresenceCoordinator(XboxBaseCoordinator[XboxData]):
"""Update list of Xbox consoles."""
config_entry: XboxConfigEntry
_update_interval = timedelta(seconds=30)
title_data: ClassVar[dict[str, Title]] = {}
async def update_data(self) -> XboxData:
"""Fetch presence data."""
batch = await self.client.people.get_friends_by_xuid(self.client.xuid)
friends = await self.client.people.get_friends_own()
presence_data = {self.client.xuid: batch.people[0]}
presence_data.update(
{
friend.xuid: friend
for friend in friends.people
if friend.xuid in self.friend_subentries()
}
)
# retrieve title details
for person in presence_data.values():
@@ -229,40 +225,27 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
title = await self.client.titlehub.get_title_info(
presence_detail.title_id
)
except TimeoutException as e:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="timeout_exception",
) from e
except HTTPStatusError as e:
_LOGGER.debug("Xbox exception:", exc_info=True)
if e.response.status_code == HTTPStatus.NOT_FOUND:
continue
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="request_exception",
) from e
except RequestError as e:
_LOGGER.debug("Xbox exception:", exc_info=True)
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="request_exception",
) from e
raise
self.title_data[person.xuid] = title.titles[0]
else:
self.title_data.pop(person.xuid, None)
person.last_seen_date_time_utc = self.last_seen_timestamp(person)
return XboxData(new_console_data, presence_data, self.title_data)
return XboxData(presence_data, self.title_data)
def last_seen_timestamp(self, person: Person) -> datetime | None:
"""Returns the most recent of two timestamps."""
# The Xbox API constantly fluctuates the "last seen" timestamp between two close values,
# causing unnecessary updates. We only accept the most recent one as valild to prevent this.
if not (prev_data := self.data.presence.get(person.xuid)):
return person.last_seen_date_time_utc
prev_dt = prev_data.last_seen_date_time_utc
prev_dt = (
prev_data.last_seen_date_time_utc
if self.data and (prev_data := self.data.presence.get(person.xuid))
else None
)
cur_dt = person.last_seen_date_time_utc
if prev_dt and cur_dt:
@@ -270,51 +253,10 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
return cur_dt
def configured_as_entry(self) -> set[str]:
"""Get xuids of configured entries."""
def friend_subentries(self) -> set[str]:
"""Get configured friend subentries."""
return {
entry.unique_id
for entry in self.hass.config_entries.async_entries(DOMAIN)
if entry.unique_id is not None
friend.unique_id
for friend in self.config_entry.subentries.values()
if friend.unique_id
}
class XboxConsolesCoordinator(DataUpdateCoordinator[SmartglassConsoleList]):
"""Update list of Xbox consoles."""
config_entry: XboxConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: XboxConfigEntry,
coordinator: XboxUpdateCoordinator,
) -> None:
"""Initialize."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(minutes=10),
)
self.client = coordinator.client
self.async_set_updated_data(coordinator.consoles)
async def _async_update_data(self) -> SmartglassConsoleList:
"""Fetch console data."""
try:
return await self.client.smartglass.get_console_list()
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

17
homeassistant/components/xbox/diagnostics.py
View File

@@ -29,12 +29,9 @@ async def async_get_config_entry_diagnostics(
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = config_entry.runtime_data.status
consoles_coordinator = config_entry.runtime_data.consoles
presence = [
async_redact_data(person.model_dump(), TO_REDACT)
for person in coordinator.data.presence.values()
for person in config_entry.runtime_data.presence.data.presence.values()
]
consoles_status = [
{
@@ -43,10 +40,16 @@ async def async_get_config_entry_diagnostics(
console.app_details.model_dump() if console.app_details else None
),
}
for console in coordinator.data.consoles.values()
for console in config_entry.runtime_data.status.data.values()
]
consoles_list = [
console.model_dump()
for console in config_entry.runtime_data.consoles.data.values()
]
title_info = [
title.model_dump()
for title in config_entry.runtime_data.presence.data.title_info.values()
]
consoles_list = consoles_coordinator.data.model_dump()
title_info = [title.model_dump() for title in coordinator.data.title_info.values()]
return {
"consoles_status": consoles_status,

18
homeassistant/components/xbox/entity.py
View File

@@ -18,7 +18,11 @@ from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import ConsoleData, XboxUpdateCoordinator
from .coordinator import (
ConsoleData,
XboxConsoleStatusCoordinator,
XboxPresenceCoordinator,
)
MAP_MODEL = {
ConsoleType.XboxOne: "Xbox One",
@@ -41,7 +45,7 @@ class XboxBaseEntityDescription(EntityDescription):
deprecated: bool | None = None
class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
class XboxBaseEntity(CoordinatorEntity[XboxPresenceCoordinator]):
"""Base Sensor for the Xbox Integration."""
_attr_has_entity_name = True
@@ -49,7 +53,7 @@ class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
def __init__(
self,
coordinator: XboxUpdateCoordinator,
coordinator: XboxPresenceCoordinator,
xuid: str,
entity_description: XboxBaseEntityDescription,
) -> None:
@@ -106,7 +110,7 @@ class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
return super().available and self.xuid in self.coordinator.data.presence
class XboxConsoleBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
class XboxConsoleBaseEntity(CoordinatorEntity[XboxConsoleStatusCoordinator]):
"""Console base entity for the Xbox integration."""
_attr_has_entity_name = True
@@ -114,11 +118,11 @@ class XboxConsoleBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
def __init__(
self,
console: SmartglassConsole,
coordinator: XboxUpdateCoordinator,
coordinator: XboxConsoleStatusCoordinator,
) -> None:
"""Initialize the Xbox Console entity."""
super().__init__(coordinator)
super().__init__(coordinator, console)
self.client = coordinator.client
self._console = console
@@ -135,7 +139,7 @@ class XboxConsoleBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
@property
def data(self) -> ConsoleData:
"""Return coordinator data for this console."""
return self.coordinator.data.consoles[self._console.id]
return self.coordinator.data[self._console.id]
def check_deprecated_entity(

6
homeassistant/components/xbox/image.py
View File

@@ -15,7 +15,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import XboxConfigEntry, XboxUpdateCoordinator
from .coordinator import XboxConfigEntry, XboxPresenceCoordinator
from .entity import XboxBaseEntity, XboxBaseEntityDescription, profile_pic
PARALLEL_UPDATES = 0
@@ -64,7 +64,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox images."""
coordinator = config_entry.runtime_data.status
coordinator = config_entry.runtime_data.presence
if TYPE_CHECKING:
assert config_entry.unique_id
async_add_entities(
@@ -95,7 +95,7 @@ class XboxImageEntity(XboxBaseEntity, ImageEntity):
def __init__(
self,
hass: HomeAssistant,
coordinator: XboxUpdateCoordinator,
coordinator: XboxPresenceCoordinator,
xuid: str,
entity_description: XboxImageEntityDescription,
) -> None:

29
homeassistant/components/xbox/media_player.py
View File

@@ -18,7 +18,7 @@ from homeassistant.components.media_player import (
MediaPlayerState,
MediaType,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .browse_media import build_item_response
@@ -57,15 +57,28 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox media_player from a config entry."""
devices_added: set[str] = set()
coordinator = entry.runtime_data.status
status = entry.runtime_data.status
consoles = entry.runtime_data.consoles
async_add_entities(
[
XboxMediaPlayer(console, coordinator)
for console in coordinator.consoles.result
]
)
@callback
def add_entities() -> None:
nonlocal devices_added
new_devices = set(consoles.data) - devices_added
if new_devices:
async_add_entities(
[
XboxMediaPlayer(consoles.data[console_id], status)
for console_id in new_devices
]
)
devices_added |= new_devices
entry.async_on_unload(consoles.async_add_listener(add_entities))
add_entities()
class XboxMediaPlayer(XboxConsoleBaseEntity, MediaPlayerEntity):

2
homeassistant/components/xbox/media_source.py
View File

@@ -641,7 +641,7 @@ class XboxSource(MediaSource):
def gamerpic(config_entry: XboxConfigEntry) -> str | None:
"""Return gamerpic."""
coordinator = config_entry.runtime_data.status
coordinator = config_entry.runtime_data.presence
if TYPE_CHECKING:
assert config_entry.unique_id
person = coordinator.data.presence[coordinator.client.xuid]

28
homeassistant/components/xbox/remote.py
View File

@@ -15,7 +15,7 @@ from homeassistant.components.remote import (
DEFAULT_DELAY_SECS,
RemoteEntity,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import XboxConfigEntry
@@ -46,11 +46,29 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox media_player from a config entry."""
coordinator = entry.runtime_data.status
devices_added: set[str] = set()
async_add_entities(
[XboxRemote(console, coordinator) for console in coordinator.consoles.result]
)
coordinator = entry.runtime_data.status
consoles = entry.runtime_data.consoles
@callback
def add_entities() -> None:
nonlocal devices_added
new_devices = set(consoles.data) - devices_added
if new_devices:
async_add_entities(
[
XboxRemote(consoles.data[console_id], coordinator)
for console_id in new_devices
]
)
devices_added |= new_devices
entry.async_on_unload(consoles.async_add_listener(add_entities))
add_entities()
class XboxRemote(XboxConsoleBaseEntity, RemoteEntity):

51
homeassistant/components/xbox/sensor.py
View File

@@ -21,7 +21,7 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.const import CONF_NAME, UnitOfInformation
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
@@ -253,12 +253,12 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox Live friends."""
coordinator = config_entry.runtime_data.status
presence = config_entry.runtime_data.presence
if TYPE_CHECKING:
assert config_entry.unique_id
async_add_entities(
[
XboxSensorEntity(coordinator, config_entry.unique_id, description)
XboxSensorEntity(presence, config_entry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if check_deprecated_entity(
hass, config_entry.unique_id, description, SENSOR_DOMAIN
@@ -268,31 +268,44 @@ async def async_setup_entry(
for subentry_id, subentry in config_entry.subentries.items():
async_add_entities(
[
XboxSensorEntity(coordinator, subentry.unique_id, description)
XboxSensorEntity(presence, subentry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if subentry.unique_id
and check_deprecated_entity(
hass, subentry.unique_id, description, SENSOR_DOMAIN
)
and subentry.unique_id in coordinator.data.presence
and subentry.unique_id in presence.data.presence
and subentry.subentry_type == "friend"
],
config_subentry_id=subentry_id,
)
consoles_coordinator = config_entry.runtime_data.consoles
consoles = config_entry.runtime_data.consoles
async_add_entities(
[
XboxStorageDeviceSensorEntity(
console, storage_device, consoles_coordinator, description
devices_added: set[str] = set()
@callback
def add_entities() -> None:
nonlocal devices_added
new_devices = set(consoles.data) - devices_added
if new_devices:
async_add_entities(
[
XboxStorageDeviceSensorEntity(
consoles.data[console_id], storage_device, consoles, description
)
for description in STORAGE_SENSOR_DESCRIPTIONS
for console_id in new_devices
if (storage_devices := consoles.data[console_id].storage_devices)
for storage_device in storage_devices
]
)
for description in STORAGE_SENSOR_DESCRIPTIONS
for console in coordinator.consoles.result
if console.storage_devices
for storage_device in console.storage_devices
]
)
devices_added |= new_devices
config_entry.async_on_unload(consoles.async_add_listener(add_entities))
add_entities()
class XboxSensorEntity(XboxBaseEntity, SensorEntity):
@@ -344,9 +357,9 @@ class XboxStorageDeviceSensorEntity(
@property
def data(self) -> StorageDevice | None:
"""Storage device data."""
consoles = self.coordinator.data.result
console = next((c for c in consoles if c.id == self._console.id), None)
if not console or not console.storage_devices:
console = self.coordinator.data[self._console.id]
if not console.storage_devices:
return None
return next(

95
tests/components/xbox/snapshots/test_diagnostics.ambr
View File

@@ -1,58 +1,51 @@
# serializer version: 1
# name: test_diagnostics
dict({
'consoles_list': dict({
'agent_user_id': None,
'result': list([
dict({
'console_streaming_enabled': False,
'console_type': 'XboxOneX',
'digital_assistant_remote_control_enabled': True,
'id': 'ABCDEFG',
'name': 'XONEX',
'power_state': 'ConnectedStandby',
'remote_management_enabled': True,
'storage_devices': list([
dict({
'free_space_bytes': 236267835392.0,
'is_default': True,
'storage_device_id': '1',
'storage_device_name': 'Internal',
'total_space_bytes': 838592360448.0,
}),
]),
}),
dict({
'console_streaming_enabled': False,
'console_type': 'XboxOne',
'digital_assistant_remote_control_enabled': True,
'id': 'HIJKLMN',
'name': 'XONE',
'power_state': 'ConnectedStandby',
'remote_management_enabled': True,
'storage_devices': list([
dict({
'free_space_bytes': 147163541504.0,
'is_default': False,
'storage_device_id': '2',
'storage_device_name': 'Internal',
'total_space_bytes': 391915761664.0,
}),
dict({
'free_space_bytes': 3200714067968.0,
'is_default': True,
'storage_device_id': '3',
'storage_device_name': 'External',
'total_space_bytes': 4000787029504.0,
}),
]),
}),
]),
'status': dict({
'error_code': 'OK',
'error_message': None,
'consoles_list': list([
dict({
'console_streaming_enabled': False,
'console_type': 'XboxOneX',
'digital_assistant_remote_control_enabled': True,
'id': 'ABCDEFG',
'name': 'XONEX',
'power_state': 'ConnectedStandby',
'remote_management_enabled': True,
'storage_devices': list([
dict({
'free_space_bytes': 236267835392.0,
'is_default': True,
'storage_device_id': '1',
'storage_device_name': 'Internal',
'total_space_bytes': 838592360448.0,
}),
]),
}),
}),
dict({
'console_streaming_enabled': False,
'console_type': 'XboxOne',
'digital_assistant_remote_control_enabled': True,
'id': 'HIJKLMN',
'name': 'XONE',
'power_state': 'ConnectedStandby',
'remote_management_enabled': True,
'storage_devices': list([
dict({
'free_space_bytes': 147163541504.0,
'is_default': False,
'storage_device_id': '2',
'storage_device_name': 'Internal',
'total_space_bytes': 391915761664.0,
}),
dict({
'free_space_bytes': 3200714067968.0,
'is_default': True,
'storage_device_id': '3',
'storage_device_name': 'External',
'total_space_bytes': 4000787029504.0,
}),
]),
}),
]),
'consoles_status': list([
dict({
'app_details': dict({

4
tests/components/xbox/test_image.py
View File

@@ -115,12 +115,12 @@ async def test_load_image_from_url(
"rgWHJigthrlsHCxEOMG9UGNdojCYasYt6MJHBjmxmtuAHJeo.sOkUiPmg4JHXvOS82c3UOrvdJTDaCKwCwHPJ0t0Plha8oHFC1i_o-&format=png"
).respond(status_code=HTTPStatus.OK, content_type="image/png", content=b"Test2")
freezer.tick(timedelta(seconds=15))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (state := hass.states.get("image.gsr_ae_gamerpic"))
assert state.state == "2025-06-16T00:00:15+00:00"
assert state.state == "2025-06-16T00:00:30+00:00"
access_token = state.attributes["access_token"]
assert (

2
tests/components/xbox/test_init.py
View File

@@ -59,7 +59,7 @@ async def test_config_implementation_not_available(
"""Test implementation not available."""
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.xbox.coordinator.async_get_config_entry_implementation",
"homeassistant.components.xbox.async_get_config_entry_implementation",
side_effect=ImplementationUnavailableError,
):
await hass.config_entries.async_setup(config_entry.entry_id)