github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add Xbox Integration (#41697) 

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Jason Hunter 2020-10-13 09:37:01 -04:00 committed by GitHub
parent 12a6d10168
commit 9877e8e25b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 547 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -1002,6 +1002,9 @@ omit =
homeassistant/components/worldtidesinfo/sensor.py
homeassistant/components/worxlandroid/sensor.py
homeassistant/components/x10/light.py
homeassistant/components/xbox/__init__.py
homeassistant/components/xbox/api.py
homeassistant/components/xbox/media_player.py
homeassistant/components/xbox_live/sensor.py
homeassistant/components/xeoma/camera.py
homeassistant/components/xfinity/device_tracker.py

1
CODEOWNERS
View File

@ -499,6 +499,7 @@ homeassistant/components/wled/* @frenck
homeassistant/components/wolflink/* @adamkrol93
homeassistant/components/workday/* @fabaff
homeassistant/components/worldclock/* @fabaff
homeassistant/components/xbox/* @hunterjm
homeassistant/components/xbox_live/* @MartinHjelmare
homeassistant/components/xfinity/* @cisasteelersfan
homeassistant/components/xiaomi_aqara/* @danielhiversen @syssi

91
homeassistant/components/xbox/__init__.py Normal file
View File

@ -0,0 +1,91 @@
"""The xbox integration."""
import asyncio
import voluptuous as vol
from xbox.webapi.api.client import XboxLiveClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
aiohttp_client,
config_entry_oauth2_flow,
config_validation as cv,
)
from . import api, config_flow
from .const import DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
}
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["media_player"]
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the xbox component."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
config_flow.OAuth2FlowHandler.async_register_implementation(
hass,
config_entry_oauth2_flow.LocalOAuth2Implementation(
hass,
DOMAIN,
config[DOMAIN][CONF_CLIENT_ID],
config[DOMAIN][CONF_CLIENT_SECRET],
OAUTH2_AUTHORIZE,
OAUTH2_TOKEN,
),
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up xbox from a config entry."""
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
auth = api.AsyncConfigEntryAuth(
aiohttp_client.async_get_clientsession(hass), session
)
hass.data[DOMAIN][entry.entry_id] = XboxLiveClient(auth)
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

39
homeassistant/components/xbox/api.py Normal file
View File

@ -0,0 +1,39 @@
"""API for xbox bound to Home Assistant OAuth."""
from aiohttp import ClientSession
from xbox.webapi.authentication.manager import AuthenticationManager
from xbox.webapi.authentication.models import OAuth2TokenResponse
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.util.dt import utc_from_timestamp
class AsyncConfigEntryAuth(AuthenticationManager):
"""Provide xbox authentication tied to an OAuth2 based config entry."""
def __init__(
self,
websession: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
):
"""Initialize xbox auth."""
# Leaving out client credentials as they are handled by Home Assistant
super().__init__(websession, "", "", "")
self._oauth_session = oauth_session
self.oauth = self._get_oauth_token()
async def refresh_tokens(self) -> None:
"""Return a valid access token."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
self.oauth = self._get_oauth_token()
# This will skip the OAuth refresh and only refresh User and XSTS tokens
await super().refresh_tokens()
def _get_oauth_token(self) -> OAuth2TokenResponse:
tokens = {**self._oauth_session.token}
issued = tokens["expires_at"] - tokens["expires_in"]
del tokens["expires_at"]
token_response = OAuth2TokenResponse.parse_obj(tokens)
token_response.issued = utc_from_timestamp(issued)
return token_response

38
homeassistant/components/xbox/config_flow.py Normal file
View File

@ -0,0 +1,38 @@
"""Config flow for xbox."""
import logging
from homeassistant import config_entries
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle xbox OAuth2 authentication."""
DOMAIN = DOMAIN
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
@property
def extra_authorize_data(self) -> dict:
"""Extra data that needs to be appended to the authorize url."""
scopes = ["Xboxlive.signin", "Xboxlive.offline_access"]
return {"scope": " ".join(scopes)}
async def async_step_user(self, user_input=None):
"""Handle a flow start."""
await self.async_set_unique_id(DOMAIN)
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
return await super().async_step_user(user_input)

6
homeassistant/components/xbox/const.py Normal file
View File

@ -0,0 +1,6 @@
"""Constants for the xbox integration."""
DOMAIN = "xbox"
OAUTH2_AUTHORIZE = "https://login.live.com/oauth20_authorize.srf"
OAUTH2_TOKEN = "https://login.live.com/oauth20_token.srf"

9
homeassistant/components/xbox/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "xbox",
"name": "Xbox",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/xbox",
"requirements": ["xbox-webapi==2.0.7"],
"dependencies": ["http"],
"codeowners": ["@hunterjm"]
}

247
homeassistant/components/xbox/media_player.py Normal file
View File

@ -0,0 +1,247 @@
"""Xbox Media Player Support."""
import logging
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.smartglass.models import (
PlaybackState,
PowerState,
SmartglassConsole,
SmartglassConsoleList,
SmartglassConsoleStatus,
VolumeDirection,
)
from homeassistant.components.media_player import MediaPlayerEntity
from homeassistant.components.media_player.const import (
MEDIA_TYPE_APP,
MEDIA_TYPE_GAME,
SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE,
SUPPORT_PLAY,
SUPPORT_PREVIOUS_TRACK,
SUPPORT_TURN_OFF,
SUPPORT_TURN_ON,
SUPPORT_VOLUME_MUTE,
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import STATE_OFF, STATE_ON, STATE_PAUSED, STATE_PLAYING
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SUPPORT_XBOX = (
SUPPORT_TURN_ON
| SUPPORT_TURN_OFF
| SUPPORT_PREVIOUS_TRACK
| SUPPORT_NEXT_TRACK
| SUPPORT_PLAY
| SUPPORT_PAUSE
| SUPPORT_VOLUME_STEP
| SUPPORT_VOLUME_MUTE
)
XBOX_STATE_MAP = {
PlaybackState.Playing: STATE_PLAYING,
PlaybackState.Paused: STATE_PAUSED,
PowerState.On: STATE_ON,
PowerState.SystemUpdate: STATE_OFF,
PowerState.ConnectedStandby: STATE_OFF,
PowerState.Off: STATE_OFF,
PowerState.Unknown: None,
}
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()
async_add_entities(
[XboxMediaPlayer(client, console) for console in consoles.result], True
)
class XboxMediaPlayer(MediaPlayerEntity):
"""Representation of an Xbox device."""
def __init__(self, client: XboxLiveClient, console: SmartglassConsole) -> None:
"""Initialize the Plex device."""
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."""
return self._console.name
@property
def unique_id(self):
"""Console device ID."""
return 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]
@property
def supported_features(self):
"""Flag media player features that are supported."""
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:
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":
return MEDIA_TYPE_GAME
return MEDIA_TYPE_APP
@property
def media_title(self):
"""Title of current playing media."""
if not self._app_details:
return None
return (
self._app_details.localized_properties[0].product_title
or self._app_details.localized_properties[0].short_title
)
@property
def media_image_url(self):
"""Image url of current playing media."""
if not self._app_details:
return None
image = _find_media_image(self._app_details.localized_properties[0].images)
if not image:
return None
url = image.uri
if url[0] == "/":
url = f"http:{url}"
return url
@property
def media_image_remotely_accessible(self) -> bool:
"""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)
)
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)
async def async_turn_off(self):
"""Turn the media player off."""
await self.client.smartglass.turn_off(self._console.id)
async def async_mute_volume(self, mute):
"""Mute the volume."""
if mute:
await self.client.smartglass.mute(self._console.id)
else:
await self.client.smartglass.unmute(self._console.id)
async def async_volume_up(self):
"""Turn volume up for media player."""
await self.client.smartglass.volume(self._console.id, VolumeDirection.Up)
async def async_volume_down(self):
"""Turn volume down for media player."""
await self.client.smartglass.volume(self._console.id, VolumeDirection.Down)
async def async_media_play(self):
"""Send play command."""
await self.client.smartglass.play(self._console.id)
async def async_media_pause(self):
"""Send pause command."""
await self.client.smartglass.pause(self._console.id)
async def async_media_previous_track(self):
"""Send previous track command."""
await self.client.smartglass.previous(self._console.id)
async def async_media_next_track(self):
"""Send next track command."""
await self.client.smartglass.next(self._console.id)
@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.name,
"manufacturer": "Microsoft",
"model": model,
}
def _find_media_image(images=List[Image]) -> Optional[Image]:
purpose_order = ["FeaturePromotionalSquareArt", "Tile", "Logo", "BoxArt"]
for purpose in purpose_order:
for image in images:
if (
image.image_purpose == purpose
and image.width == image.height
and image.width >= 300
):
return image
return None

18
homeassistant/components/xbox/strings.json Normal file
View File

@ -0,0 +1,18 @@
{
"title": "xbox",
"config": {
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
}
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
}
}

18
homeassistant/components/xbox/translations/en.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"abort": {
"authorize_url_timeout": "Timeout generating authorize URL.",
"missing_configuration": "The component is not configured. Please follow the documentation.",
"single_instance_allowed": "Already configured. Only a single configuration possible."
},
"create_entry": {
"default": "Successfully authenticated"
},
"step": {
"pick_implementation": {
"title": "Pick Authentication Method"
}
}
},
"title": "xbox"
}

1
homeassistant/generated/config_flows.py
View File

@ -217,6 +217,7 @@ FLOWS = [
"withings",
"wled",
"wolflink",
"xbox",
"xiaomi_aqara",
"xiaomi_miio",
"yeelight",

3
requirements_all.txt
View File

@ -2287,6 +2287,9 @@ wolf_smartset==0.1.6
# homeassistant.components.xbee
xbee-helper==0.0.7
# homeassistant.components.xbox
xbox-webapi==2.0.7
# homeassistant.components.xbox_live
xboxapi==2.0.1

3
requirements_test_all.txt
View File

@ -1077,6 +1077,9 @@ wled==0.4.4
# homeassistant.components.wolflink
wolf_smartset==0.1.6
# homeassistant.components.xbox
xbox-webapi==2.0.7
# homeassistant.components.bluesound
# homeassistant.components.rest
# homeassistant.components.startca

1
tests/components/xbox/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the xbox integration."""

69
tests/components/xbox/test_config_flow.py Normal file
View File

@ -0,0 +1,69 @@
"""Test the xbox config flow."""
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.xbox.const import DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
from homeassistant.helpers import config_entry_oauth2_flow
from tests.async_mock import patch
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
async def test_abort_if_existing_entry(hass):
"""Check flow abort when an entry already exist."""
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
"xbox", context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
async def test_full_flow(hass, aiohttp_client, aioclient_mock, current_request):
"""Check full flow."""
assert await setup.async_setup_component(
hass,
"xbox",
{
"xbox": {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
"http": {"base_url": "https://example.com"},
},
)
result = await hass.config_entries.flow.async_init(
"xbox", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(hass, {"flow_id": result["flow_id"]})
scope = "+".join(["Xboxlive.signin", "Xboxlive.offline_access"])
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={scope}"
)
client = await aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.xbox.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1