evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Remote entity to Xbox Integration (#41809)

Browse source
This commit is contained in:
Jason Hunter 2020-10-13 21:00:44 -04:00 • committed by GitHub
parent ba55cb8955
commit 3013f39191
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 246 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

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

113
homeassistant/components/xbox/__init__.py
View file

@ -1,8 +1,18 @@
"""The xbox integration."""
import asyncio
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Dict, Optional
import voluptuous as vol
from xbox.webapi.api.client import XboxLiveClient
from xbox.webapi.api.provider.catalog.const import HOME_APP_IDS, SYSTEM_PFN_ID_MAP
from xbox.webapi.api.provider.catalog.models import AlternateIdType, Product
from xbox.webapi.api.provider.smartglass.models import (
SmartglassConsoleList,
SmartglassConsoleStatus,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
@ -12,10 +22,14 @@ from homeassistant.helpers import (
config_entry_oauth2_flow,
config_validation as cv,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import api, config_flow
from .const import DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
@ -28,7 +42,7 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["media_player"]
PLATFORMS = ["media_player", "remote"]
async def async_setup(hass: HomeAssistant, config: dict):
@ -65,7 +79,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
aiohttp_client.async_get_clientsession(hass), session
)
hass.data[DOMAIN][entry.entry_id] = XboxLiveClient(auth)
client = XboxLiveClient(auth)
consoles: SmartglassConsoleList = await client.smartglass.get_console_list()
_LOGGER.debug(
"Found %d consoles: %s",
len(consoles.result),
consoles.dict(),
)
coordinator = XboxUpdateCoordinator(hass, client, consoles)
await coordinator.async_refresh()
hass.data[DOMAIN][entry.entry_id] = {
"client": XboxLiveClient(auth),
"consoles": consoles,
"coordinator": coordinator,
}
for component in PLATFORMS:
hass.async_create_task(
@ -89,3 +118,83 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
@dataclass
class XboxData:
"""Xbox dataclass for update coordinator."""
status: SmartglassConsoleStatus
app_details: Optional[Product]
class XboxUpdateCoordinator(DataUpdateCoordinator):
"""Store Xbox Console Status."""
def __init__(
self,
hass: HomeAssistantType,
client: XboxLiveClient,
consoles: SmartglassConsoleList,
) -> None:
"""Initialize."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=10),
)
self.data: Dict[str, XboxData] = {}
self.client: XboxLiveClient = client
self.consoles: SmartglassConsoleList = consoles
async def _async_update_data(self) -> Dict[str, XboxData]:
"""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)
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: Optional[Product] = 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:
if not current_state or not current_state.status.focus_app_aumid:
id_type = AlternateIdType.LEGACY_XBOX_PRODUCT_ID
catalog_result = (
await self.client.catalog.get_product_from_alternate_id(
HOME_APP_IDS[id_type], id_type
)
)
app_details = catalog_result.products[0]
new_data[console.id] = XboxData(status=status, app_details=app_details)
return new_data

112
homeassistant/components/xbox/media_player.py
View file

@ -4,14 +4,12 @@ import re
from typing import List, Optional
from xbox.webapi.api.client import XboxLiveClient
from xbox.webapi.api.provider.catalog.const import HOME_APP_IDS, SYSTEM_PFN_ID_MAP
from xbox.webapi.api.provider.catalog.models import AlternateIdType, Image, Product
from xbox.webapi.api.provider.catalog.models import Image
from xbox.webapi.api.provider.smartglass.models import (
PlaybackState,
PowerState,
SmartglassConsole,
SmartglassConsoleList,
SmartglassConsoleStatus,
VolumeDirection,
)
@ -31,7 +29,9 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import STATE_OFF, STATE_ON, STATE_PAUSED, STATE_PLAYING
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import XboxData, XboxUpdateCoordinator
from .browse_media import build_item_response
from .const import DOMAIN
@ -63,29 +63,31 @@ XBOX_STATE_MAP = {
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Xbox media_player from a config entry."""
client: XboxLiveClient = hass.data[DOMAIN][entry.entry_id]
consoles: SmartglassConsoleList = await client.smartglass.get_console_list()
_LOGGER.debug(
"Found %d consoles: %s",
len(consoles.result),
consoles.dict(),
)
client: XboxLiveClient = hass.data[DOMAIN][entry.entry_id]["client"]
consoles: SmartglassConsoleList = hass.data[DOMAIN][entry.entry_id]["consoles"]
coordinator: XboxUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
"coordinator"
]
async_add_entities(
[XboxMediaPlayer(client, console) for console in consoles.result], True
[XboxMediaPlayer(client, console, coordinator) for console in consoles.result]
)
class XboxMediaPlayer(MediaPlayerEntity):
"""Representation of an Xbox device."""
class XboxMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
"""Representation of an Xbox Media Player."""
def __init__(self, client: XboxLiveClient, console: SmartglassConsole) -> None:
"""Initialize the Plex device."""
def __init__(
self,
client: XboxLiveClient,
console: SmartglassConsole,
coordinator: XboxUpdateCoordinator,
) -> None:
"""Initialize the Xbox Media Player."""
super().__init__(coordinator)
self.client: XboxLiveClient = client
self._console: SmartglassConsole = console
self._console_status: SmartglassConsoleStatus = None
self._app_details: Optional[Product] = None
@property
def name(self):
"""Return the device name."""
@ -96,12 +98,18 @@ class XboxMediaPlayer(MediaPlayerEntity):
"""Console device ID."""
return self._console.id
@property
def data(self) -> XboxData:
"""Return coordinator data for this console."""
return self.coordinator.data[self._console.id]
@property
def state(self):
"""State of the player."""
if self._console_status.playback_state in XBOX_STATE_MAP:
return XBOX_STATE_MAP[self._console_status.playback_state]
return XBOX_STATE_MAP[self._console_status.power_state]
status = self.data.status
if status.playback_state in XBOX_STATE_MAP:
return XBOX_STATE_MAP[status.playback_state]
return XBOX_STATE_MAP[status.power_state]
@property
def supported_features(self):
@ -109,33 +117,36 @@ class XboxMediaPlayer(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._console_status.is_tv_configured:
if not self.data.status.is_tv_configured:
active_support &= ~SUPPORT_VOLUME_MUTE & ~SUPPORT_VOLUME_STEP
return active_support
@property
def media_content_type(self):
"""Media content type."""
if self._app_details and self._app_details.product_family == "Games":
app_details = self.data.app_details
if app_details and app_details.product_family == "Games":
return MEDIA_TYPE_GAME
return MEDIA_TYPE_APP
@property
def media_title(self):
"""Title of current playing media."""
if not self._app_details:
app_details = self.data.app_details
if not app_details:
return None
return (
self._app_details.localized_properties[0].product_title
or self._app_details.localized_properties[0].short_title
app_details.localized_properties[0].product_title
or app_details.localized_properties[0].short_title
)
@property
def media_image_url(self):
"""Image url of current playing media."""
if not self._app_details:
app_details = self.data.app_details
if not app_details:
return None
image = _find_media_image(self._app_details.localized_properties[0].images)
image = _find_media_image(app_details.localized_properties[0].images)
if not image:
return None
@ -150,49 +161,6 @@ class XboxMediaPlayer(MediaPlayerEntity):
"""If the image url is remotely accessible."""
return True
async def async_update(self) -> None:
"""Update Xbox state."""
status: SmartglassConsoleStatus = (
await self.client.smartglass.get_console_status(self._console.id)
)
_LOGGER.debug(
"%s status: %s",
self._console.name,
status.dict(),
)
if status.focus_app_aumid:
if (
not self._console_status
or status.focus_app_aumid != self._console_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:
self._app_details = catalog_result.products[0]
else:
self._app_details = None
else:
if self.media_title != "Home":
id_type = AlternateIdType.LEGACY_XBOX_PRODUCT_ID
catalog_result = (
await self.client.catalog.get_product_from_alternate_id(
HOME_APP_IDS[id_type], id_type
)
)
self._app_details = catalog_result.products[0]
self._console_status = status
async def async_turn_on(self):
"""Turn the media player on."""
await self.client.smartglass.wake_up(self._console.id)
@ -237,7 +205,7 @@ class XboxMediaPlayer(MediaPlayerEntity):
return await build_item_response(
self.client,
self._console.id,
self._console_status.is_tv_configured,
self.data.status.is_tv_configured,
media_content_type,
media_content_id,
)

94
homeassistant/components/xbox/remote.py Normal file
View file

@ -0,0 +1,94 @@
"""Xbox Remote support."""
import asyncio
from typing import Any, Iterable
from xbox.webapi.api.client import XboxLiveClient
from xbox.webapi.api.provider.smartglass.models import (
InputKeyType,
PowerState,
SmartglassConsole,
SmartglassConsoleList,
)
from homeassistant.components.remote import (
ATTR_DELAY_SECS,
ATTR_NUM_REPEATS,
DEFAULT_DELAY_SECS,
RemoteEntity,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import XboxData, XboxUpdateCoordinator
from .const import DOMAIN
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Xbox media_player from a config entry."""
client: XboxLiveClient = hass.data[DOMAIN][entry.entry_id]["client"]
consoles: SmartglassConsoleList = hass.data[DOMAIN][entry.entry_id]["consoles"]
coordinator: XboxUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
"coordinator"
]
async_add_entities(
[XboxRemote(client, console, coordinator) for console in consoles.result]
)
class XboxRemote(CoordinatorEntity, RemoteEntity):
"""Representation of an Xbox remote."""
def __init__(
self,
client: XboxLiveClient,
console: SmartglassConsole,
coordinator: XboxUpdateCoordinator,
) -> None:
"""Initialize the Xbox Media Player."""
super().__init__(coordinator)
self.client: XboxLiveClient = client
self._console: SmartglassConsole = console
@property
def name(self):
"""Return the device name."""
return f"{self._console.name} Remote"
@property
def unique_id(self):
"""Console device ID."""
return self._console.id
@property
def data(self) -> XboxData:
"""Return coordinator data for this console."""
return self.coordinator.data[self._console.id]
@property
def is_on(self):
"""Return True if device is on."""
return self.data.status.power_state == PowerState.On
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the Xbox on."""
await self.client.smartglass.wake_up(self._console.id)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the Xbox off."""
await self.client.smartglass.turn_off(self._console.id)
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send controller or text input to the Xbox."""
num_repeats = kwargs[ATTR_NUM_REPEATS]
delay = kwargs.get(ATTR_DELAY_SECS, DEFAULT_DELAY_SECS)
for _ in range(num_repeats):
for single_command in command:
try:
button = InputKeyType(single_command)
await self.client.smartglass.press_button(self._console.id, button)
except ValueError:
await self.client.smartglass.insert_text(
self._console.id, single_command
)
await asyncio.sleep(delay)