Add Media Source to Xbox Integration (#41833)

* initial implementation of xbox media source

* minor fixes

* fix lint

* add media class map to remove multi-line ternary

.coveragerc

@@ -1004,6 +1004,7 @@ omit =
     homeassistant/components/xbox/api.py
     homeassistant/components/xbox/browse_media.py
     homeassistant/components/xbox/media_player.py
+    homeassistant/components/xbox/media_source.py
     homeassistant/components/xbox/remote.py
     homeassistant/components/xbox_live/sensor.py
     homeassistant/components/xeoma/camera.py

homeassistant/components/media_source/const.py

@@ -16,5 +16,5 @@ MEDIA_CLASS_MAP = {
 }
 URI_SCHEME = "media-source://"
 URI_SCHEME_REGEX = re.compile(
-    r"^media-source:\/\/(?:(?P<domain>(?!.+__)(?!_)[\da-z_]+(?<!_))(?:\/(?P<identifier>(?!\/).+))?)?$"
+    r"^media-source:\/\/(?:(?P<domain>(?!_)[\da-z_]+(?<!_))(?:\/(?P<identifier>(?!\/).+))?)?$"
 )

homeassistant/components/xbox/__init__.py

@@ -152,7 +152,7 @@ class XboxUpdateCoordinator(DataUpdateCoordinator):
         """Fetch the latest console status."""
         new_data: Dict[str, XboxData] = {}
         for console in self.consoles.result:
-            current_state: Optional[XboxData] = self.data.get(console.id, None)
+            current_state: Optional[XboxData] = self.data.get(console.id)
             status: SmartglassConsoleStatus = (
                 await self.client.smartglass.get_console_status(console.id)
             )

homeassistant/components/xbox/manifest.json

@@ -3,7 +3,7 @@
   "name": "Xbox",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/xbox",
-  "requirements": ["xbox-webapi==2.0.7"],
+  "requirements": ["xbox-webapi==2.0.8"],
   "dependencies": ["http"],
   "codeowners": ["@hunterjm"]
 }

homeassistant/components/xbox/media_player.py

@@ -117,8 +117,6 @@ class XboxMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
         active_support = SUPPORT_XBOX
         if self.state not in [STATE_PLAYING, STATE_PAUSED]:
             active_support &= ~SUPPORT_NEXT_TRACK & ~SUPPORT_PREVIOUS_TRACK
-        if not self.data.status.is_tv_configured:
-            active_support &= ~SUPPORT_VOLUME_MUTE & ~SUPPORT_VOLUME_STEP
         return active_support
 
     @property
@@ -231,7 +229,7 @@ class XboxMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
 
         return {
             "identifiers": {(DOMAIN, self._console.id)},
-            "name": self.name,
+            "name": self._console.name,
             "manufacturer": "Microsoft",
             "model": model,
         }

homeassistant/components/xbox/media_source.py

@@ -0,0 +1,283 @@
+"""Netatmo Media Source Implementation."""
+from dataclasses import dataclass
+import logging
+from typing import List, Tuple
+
+# pylint: disable=no-name-in-module
+from pydantic.error_wrappers import ValidationError
+from xbox.webapi.api.client import XboxLiveClient
+from xbox.webapi.api.provider.catalog.models import FieldsTemplate, Image
+from xbox.webapi.api.provider.gameclips.models import GameclipsResponse
+from xbox.webapi.api.provider.screenshots.models import ScreenshotResponse
+from xbox.webapi.api.provider.smartglass.models import InstalledPackage
+
+from homeassistant.components.media_player.const import (
+    MEDIA_CLASS_DIRECTORY,
+    MEDIA_CLASS_GAME,
+    MEDIA_CLASS_IMAGE,
+    MEDIA_CLASS_VIDEO,
+)
+from homeassistant.components.media_source.const import MEDIA_MIME_TYPES
+from homeassistant.components.media_source.models import (
+    BrowseMediaSource,
+    MediaSource,
+    MediaSourceItem,
+    PlayMedia,
+)
+from homeassistant.core import callback
+from homeassistant.helpers.typing import HomeAssistantType
+from homeassistant.util import dt as dt_util
+
+from .browse_media import _find_media_image
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+MIME_TYPE_MAP = {
+    "gameclips": "video/mp4",
+    "screenshots": "image/png",
+}
+
+MEDIA_CLASS_MAP = {
+    "gameclips": MEDIA_CLASS_VIDEO,
+    "screenshots": MEDIA_CLASS_IMAGE,
+}
+
+
+async def async_get_media_source(hass: HomeAssistantType):
+    """Set up Xbox media source."""
+    entry = hass.config_entries.async_entries(DOMAIN)[0]
+    client = hass.data[DOMAIN][entry.entry_id]["client"]
+    return XboxSource(hass, client)
+
+
+@callback
+def async_parse_identifier(
+    item: MediaSourceItem,
+) -> Tuple[str, str, str]:
+    """Parse identifier."""
+    identifier = item.identifier or ""
+    start = ["", "", ""]
+    items = identifier.lstrip("/").split("~~", 2)
+    return tuple(items + start[len(items) :])
+
+
+@dataclass
+class XboxMediaItem:
+    """Represents gameclip/screenshot media."""
+
+    caption: str
+    thumbnail: str
+    uri: str
+    media_class: str
+
+
+class XboxSource(MediaSource):
+    """Provide Xbox screenshots and gameclips as media sources."""
+
+    name: str = "Xbox Game Media"
+
+    def __init__(self, hass: HomeAssistantType, client: XboxLiveClient):
+        """Initialize Netatmo source."""
+        super().__init__(DOMAIN)
+
+        self.hass: HomeAssistantType = hass
+        self.client: XboxLiveClient = client
+
+    async def async_resolve_media(self, item: MediaSourceItem) -> PlayMedia:
+        """Resolve media to a url."""
+        _, category, url = async_parse_identifier(item)
+        _, kind = category.split("#", 1)
+        return PlayMedia(url, MIME_TYPE_MAP[kind])
+
+    async def async_browse_media(
+        self, item: MediaSourceItem, media_types: Tuple[str] = MEDIA_MIME_TYPES
+    ) -> BrowseMediaSource:
+        """Return media."""
+        title, category, _ = async_parse_identifier(item)
+
+        if not title:
+            return await self._build_game_library()
+
+        if not category:
+            return _build_categories(title)
+
+        return await self._build_media_items(title, category)
+
+    async def _build_game_library(self):
+        """Display installed games across all consoles."""
+        apps = await self.client.smartglass.get_installed_apps()
+        games = {
+            game.one_store_product_id: game
+            for game in apps.result
+            if game.is_game and game.title_id
+        }
+
+        app_details = await self.client.catalog.get_products(
+            games.keys(),
+            FieldsTemplate.BROWSE,
+        )
+
+        images = {
+            prod.product_id: prod.localized_properties[0].images
+            for prod in app_details.products
+        }
+
+        return BrowseMediaSource(
+            domain=DOMAIN,
+            identifier="",
+            media_class=MEDIA_CLASS_DIRECTORY,
+            media_content_type=None,
+            title="Xbox Game Media",
+            can_play=False,
+            can_expand=True,
+            children=[_build_game_item(game, images) for game in games.values()],
+            children_media_class=MEDIA_CLASS_GAME,
+        )
+
+    async def _build_media_items(self, title, category):
+        """Fetch requested gameclip/screenshot media."""
+        title_id, _, thumbnail = title.split("#", 2)
+        owner, kind = category.split("#", 1)
+
+        items: List[XboxMediaItem] = []
+        try:
+            if kind == "gameclips":
+                if owner == "my":
+                    response: GameclipsResponse = (
+                        await self.client.gameclips.get_recent_clips_by_xuid(
+                            self.client.xuid, title_id
+                        )
+                    )
+                elif owner == "community":
+                    response: GameclipsResponse = await self.client.gameclips.get_recent_community_clips_by_title_id(
+                        title_id
+                    )
+                else:
+                    return None
+                items = [
+                    XboxMediaItem(
+                        item.user_caption
+                        or dt_util.as_local(
+                            dt_util.parse_datetime(item.date_recorded)
+                        ).strftime("%b. %d, %Y %I:%M %p"),
+                        item.thumbnails[0].uri,
+                        item.game_clip_uris[0].uri,
+                        MEDIA_CLASS_VIDEO,
+                    )
+                    for item in response.game_clips
+                ]
+            elif kind == "screenshots":
+                if owner == "my":
+                    response: ScreenshotResponse = (
+                        await self.client.screenshots.get_recent_screenshots_by_xuid(
+                            self.client.xuid, title_id
+                        )
+                    )
+                elif owner == "community":
+                    response: ScreenshotResponse = await self.client.screenshots.get_recent_community_screenshots_by_title_id(
+                        title_id
+                    )
+                else:
+                    return None
+                items = [
+                    XboxMediaItem(
+                        item.user_caption
+                        or dt_util.as_local(item.date_taken).strftime(
+                            "%b. %d, %Y %I:%M%p"
+                        ),
+                        item.thumbnails[0].uri,
+                        item.screenshot_uris[0].uri,
+                        MEDIA_CLASS_IMAGE,
+                    )
+                    for item in response.screenshots
+                ]
+        except ValidationError:
+            # Unexpected API response
+            pass
+
+        return BrowseMediaSource(
+            domain=DOMAIN,
+            identifier=f"{title}~~{category}",
+            media_class=MEDIA_CLASS_DIRECTORY,
+            media_content_type=None,
+            title=f"{owner.title()} {kind.title()}",
+            can_play=False,
+            can_expand=True,
+            children=[_build_media_item(title, category, item) for item in items],
+            children_media_class=MEDIA_CLASS_MAP[kind],
+            thumbnail=thumbnail,
+        )
+
+
+def _build_game_item(item: InstalledPackage, images: List[Image]):
+    """Build individual game."""
+    thumbnail = ""
+    image = _find_media_image(images.get(item.one_store_product_id, []))
+    if image is not None:
+        thumbnail = image.uri
+        if thumbnail[0] == "/":
+            thumbnail = f"https:{thumbnail}"
+
+    return BrowseMediaSource(
+        domain=DOMAIN,
+        identifier=f"{item.title_id}#{item.name}#{thumbnail}",
+        media_class=MEDIA_CLASS_GAME,
+        media_content_type=None,
+        title=item.name,
+        can_play=False,
+        can_expand=True,
+        children_media_class=MEDIA_CLASS_DIRECTORY,
+        thumbnail=thumbnail,
+    )
+
+
+def _build_categories(title):
+    """Build base categories for Xbox media."""
+    _, name, thumbnail = title.split("#", 2)
+    base = BrowseMediaSource(
+        domain=DOMAIN,
+        identifier=f"{title}",
+        media_class=MEDIA_CLASS_GAME,
+        media_content_type=None,
+        title=name,
+        can_play=False,
+        can_expand=True,
+        children=[],
+        children_media_class=MEDIA_CLASS_DIRECTORY,
+        thumbnail=thumbnail,
+    )
+
+    owners = ["my", "community"]
+    kinds = ["gameclips", "screenshots"]
+    for owner in owners:
+        for kind in kinds:
+            base.children.append(
+                BrowseMediaSource(
+                    domain=DOMAIN,
+                    identifier=f"{title}~~{owner}#{kind}",
+                    media_class=MEDIA_CLASS_DIRECTORY,
+                    media_content_type=None,
+                    title=f"{owner.title()} {kind.title()}",
+                    can_play=False,
+                    can_expand=True,
+                    children_media_class=MEDIA_CLASS_MAP[kind],
+                )
+            )
+
+    return base
+
+
+def _build_media_item(title: str, category: str, item: XboxMediaItem):
+    """Build individual media item."""
+    _, kind = category.split("#", 1)
+    return BrowseMediaSource(
+        domain=DOMAIN,
+        identifier=f"{title}~~{category}~~{item.uri}",
+        media_class=item.media_class,
+        media_content_type=MIME_TYPE_MAP[kind],
+        title=item.caption,
+        can_play=True,
+        can_expand=False,
+        thumbnail=item.thumbnail,
+    )

homeassistant/components/xbox/remote.py

@@ -1,5 +1,6 @@
 """Xbox Remote support."""
 import asyncio
+import re
 from typing import Any, Iterable
 
 from xbox.webapi.api.client import XboxLiveClient
@@ -92,3 +93,20 @@ class XboxRemote(CoordinatorEntity, RemoteEntity):
                         self._console.id, single_command
                     )
                 await asyncio.sleep(delay)
+
+    @property
+    def device_info(self):
+        """Return a device description for device registry."""
+        # Turns "XboxOneX" into "Xbox One X" for display
+        matches = re.finditer(
+            ".+?(?:(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])|$)",
+            self._console.console_type,
+        )
+        model = " ".join([m.group(0) for m in matches])
+
+        return {
+            "identifiers": {(DOMAIN, self._console.id)},
+            "name": self._console.name,
+            "manufacturer": "Microsoft",
+            "model": model,
+        }

requirements_all.txt

@@ -2291,7 +2291,7 @@ wolf_smartset==0.1.6
 xbee-helper==0.0.7
 
 # homeassistant.components.xbox
-xbox-webapi==2.0.7
+xbox-webapi==2.0.8
 
 # homeassistant.components.xbox_live
 xboxapi==2.0.1

requirements_test_all.txt

@@ -1081,7 +1081,7 @@ wled==0.4.4
 wolf_smartset==0.1.6
 
 # homeassistant.components.xbox
-xbox-webapi==2.0.7
+xbox-webapi==2.0.8
 
 # homeassistant.components.bluesound
 # homeassistant.components.rest