jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 12:17:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement Google Assistant media traits (#35803)

Browse Source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Brynley McDonald 2020-06-03 11:20:59 +12:00 committed by GitHub
parent f94bbdab61
commit 5d6a563ac7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 386 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

197
homeassistant/components/google_assistant/trait.py
View File

@ -42,9 +42,13 @@ from homeassistant.const import (
STATE_ALARM_DISARMED,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
STATE_IDLE,
STATE_LOCKED,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
TEMP_CELSIUS,
@ -52,7 +56,7 @@ from homeassistant.const import (
)
from homeassistant.core import DOMAIN as HA_DOMAIN
from homeassistant.helpers.network import get_url
from homeassistant.util import color as color_util, temperature as temp_util
from homeassistant.util import color as color_util, dt, temperature as temp_util
from .const import (
CHALLENGE_ACK_NEEDED,
@ -85,6 +89,8 @@ TRAIT_OPENCLOSE = f"{PREFIX_TRAITS}OpenClose"
TRAIT_VOLUME = f"{PREFIX_TRAITS}Volume"
TRAIT_ARMDISARM = f"{PREFIX_TRAITS}ArmDisarm"
TRAIT_HUMIDITY_SETTING = f"{PREFIX_TRAITS}HumiditySetting"
TRAIT_TRANSPORT_CONTROL = f"{PREFIX_TRAITS}TransportControl"
TRAIT_MEDIA_STATE = f"{PREFIX_TRAITS}MediaState"
PREFIX_COMMANDS = "action.devices.commands."
COMMAND_ONOFF = f"{PREFIX_COMMANDS}OnOff"
@ -109,6 +115,15 @@ COMMAND_OPENCLOSE = f"{PREFIX_COMMANDS}OpenClose"
COMMAND_SET_VOLUME = f"{PREFIX_COMMANDS}setVolume"
COMMAND_VOLUME_RELATIVE = f"{PREFIX_COMMANDS}volumeRelative"
COMMAND_ARMDISARM = f"{PREFIX_COMMANDS}ArmDisarm"
COMMAND_MEDIA_NEXT = f"{PREFIX_COMMANDS}mediaNext"
COMMAND_MEDIA_PAUSE = f"{PREFIX_COMMANDS}mediaPause"
COMMAND_MEDIA_PREVIOUS = f"{PREFIX_COMMANDS}mediaPrevious"
COMMAND_MEDIA_RESUME = f"{PREFIX_COMMANDS}mediaResume"
COMMAND_MEDIA_SEEK_RELATIVE = f"{PREFIX_COMMANDS}mediaSeekRelative"
COMMAND_MEDIA_SEEK_TO_POSITION = f"{PREFIX_COMMANDS}mediaSeekToPosition"
COMMAND_MEDIA_SHUFFLE = f"{PREFIX_COMMANDS}mediaShuffle"
COMMAND_MEDIA_STOP = f"{PREFIX_COMMANDS}mediaStop"
TRAITS = []
@ -1500,3 +1515,183 @@ def _verify_ack_challenge(data, state, challenge):
return
if not challenge or not challenge.get("ack"):
raise ChallengeNeeded(CHALLENGE_ACK_NEEDED)
MEDIA_COMMAND_SUPPORT_MAPPING = {
COMMAND_MEDIA_NEXT: media_player.SUPPORT_NEXT_TRACK,
COMMAND_MEDIA_PAUSE: media_player.SUPPORT_PAUSE,
COMMAND_MEDIA_PREVIOUS: media_player.SUPPORT_PREVIOUS_TRACK,
COMMAND_MEDIA_RESUME: media_player.SUPPORT_PLAY,
COMMAND_MEDIA_SEEK_RELATIVE: media_player.SUPPORT_SEEK,
COMMAND_MEDIA_SEEK_TO_POSITION: media_player.SUPPORT_SEEK,
COMMAND_MEDIA_SHUFFLE: media_player.SUPPORT_SHUFFLE_SET,
COMMAND_MEDIA_STOP: media_player.SUPPORT_STOP,
}
MEDIA_COMMAND_ATTRIBUTES = {
COMMAND_MEDIA_NEXT: "NEXT",
COMMAND_MEDIA_PAUSE: "PAUSE",
COMMAND_MEDIA_PREVIOUS: "PREVIOUS",
COMMAND_MEDIA_RESUME: "RESUME",
COMMAND_MEDIA_SEEK_RELATIVE: "SEEK_RELATIVE",
COMMAND_MEDIA_SEEK_TO_POSITION: "SEEK_TO_POSITION",
COMMAND_MEDIA_SHUFFLE: "SHUFFLE",
COMMAND_MEDIA_STOP: "STOP",
}
@register_trait
class TransportControlTrait(_Trait):
"""Trait to control media playback.
https://developers.google.com/actions/smarthome/traits/transportcontrol
"""
name = TRAIT_TRANSPORT_CONTROL
commands = [
COMMAND_MEDIA_NEXT,
COMMAND_MEDIA_PAUSE,
COMMAND_MEDIA_PREVIOUS,
COMMAND_MEDIA_RESUME,
COMMAND_MEDIA_SEEK_RELATIVE,
COMMAND_MEDIA_SEEK_TO_POSITION,
COMMAND_MEDIA_SHUFFLE,
COMMAND_MEDIA_STOP,
]
@staticmethod
def supported(domain, features, device_class):
"""Test if state is supported."""
if domain == media_player.DOMAIN:
for feature in MEDIA_COMMAND_SUPPORT_MAPPING.values():
if features & feature:
return True
return False
def sync_attributes(self):
"""Return opening direction."""
response = {}
if self.state.domain == media_player.DOMAIN:
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
support = []
for command, feature in MEDIA_COMMAND_SUPPORT_MAPPING.items():
if features & feature:
support.append(MEDIA_COMMAND_ATTRIBUTES[command])
response["transportControlSupportedCommands"] = support
return response
def query_attributes(self):
"""Return the attributes of this trait for this entity."""
return {}
async def execute(self, command, data, params, challenge):
"""Execute a media command."""
service_attrs = {ATTR_ENTITY_ID: self.state.entity_id}
if command == COMMAND_MEDIA_SEEK_RELATIVE:
service = media_player.SERVICE_MEDIA_SEEK
rel_position = params["relativePositionMs"] / 1000
seconds_since = 0 # Default to 0 seconds
if self.state.state == STATE_PLAYING:
now = dt.utcnow()
upd_at = self.state.attributes.get(
media_player.ATTR_MEDIA_POSITION_UPDATED_AT, now
)
seconds_since = (now - upd_at).total_seconds()
position = self.state.attributes.get(media_player.ATTR_MEDIA_POSITION, 0)
max_position = self.state.attributes.get(
media_player.ATTR_MEDIA_DURATION, 0
)
service_attrs[media_player.ATTR_MEDIA_SEEK_POSITION] = min(
max(position + seconds_since + rel_position, 0), max_position
)
elif command == COMMAND_MEDIA_SEEK_TO_POSITION:
service = media_player.SERVICE_MEDIA_SEEK
max_position = self.state.attributes.get(
media_player.ATTR_MEDIA_DURATION, 0
)
service_attrs[media_player.ATTR_MEDIA_SEEK_POSITION] = min(
max(params["absPositionMs"] / 1000, 0), max_position
)
elif command == COMMAND_MEDIA_NEXT:
service = media_player.SERVICE_MEDIA_NEXT_TRACK
elif command == COMMAND_MEDIA_PAUSE:
service = media_player.SERVICE_MEDIA_PAUSE
elif command == COMMAND_MEDIA_PREVIOUS:
service = media_player.SERVICE_MEDIA_PREVIOUS_TRACK
elif command == COMMAND_MEDIA_RESUME:
service = media_player.SERVICE_MEDIA_PLAY
elif command == COMMAND_MEDIA_SHUFFLE:
service = media_player.SERVICE_SHUFFLE_SET
# Google Assistant only supports enabling shuffle
service_attrs[media_player.ATTR_MEDIA_SHUFFLE] = True
elif command == COMMAND_MEDIA_STOP:
service = media_player.SERVICE_MEDIA_STOP
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
await self.hass.services.async_call(
media_player.DOMAIN,
service,
service_attrs,
blocking=True,
context=data.context,
)
@register_trait
class MediaStateTrait(_Trait):
"""Trait to get media playback state.
https://developers.google.com/actions/smarthome/traits/mediastate
"""
name = TRAIT_MEDIA_STATE
commands = []
activity_lookup = {
STATE_OFF: "INACTIVE",
STATE_IDLE: "STANDBY",
STATE_PLAYING: "ACTIVE",
STATE_ON: "STANDBY",
STATE_PAUSED: "STANDBY",
STATE_STANDBY: "STANDBY",
STATE_UNAVAILABLE: "INACTIVE",
STATE_UNKNOWN: "INACTIVE",
}
playback_lookup = {
STATE_OFF: "STOPPED",
STATE_IDLE: "STOPPED",
STATE_PLAYING: "PLAYING",
STATE_ON: "STOPPED",
STATE_PAUSED: "PAUSED",
STATE_STANDBY: "STOPPED",
STATE_UNAVAILABLE: "STOPPED",
STATE_UNKNOWN: "STOPPED",
}
@staticmethod
def supported(domain, features, device_class):
"""Test if state is supported."""
return domain == media_player.DOMAIN
def sync_attributes(self):
"""Return attributes for a sync request."""
return {"supportActivityState": True, "supportPlaybackState": True}
def query_attributes(self):
"""Return the attributes of this trait for this entity."""
return {
"activityState": self.activity_lookup.get(self.state.state, "INACTIVE"),
"playbackState": self.playback_lookup.get(self.state.state, "STOPPED"),
}

13
tests/components/google_assistant/__init__.py
View File

@ -167,6 +167,8 @@ DEMO_DEVICES = [
"action.devices.traits.OnOff",
"action.devices.traits.Volume",
"action.devices.traits.Modes",
"action.devices.traits.TransportControl",
"action.devices.traits.MediaState",
],
"type": "action.devices.types.SETTOP",
"willReportState": False,
@ -178,6 +180,8 @@ DEMO_DEVICES = [
"action.devices.traits.OnOff",
"action.devices.traits.Volume",
"action.devices.traits.Modes",
"action.devices.traits.TransportControl",
"action.devices.traits.MediaState",
],
"type": "action.devices.types.SETTOP",
"willReportState": False,
@ -185,7 +189,12 @@ DEMO_DEVICES = [
{
"id": "media_player.lounge_room",
"name": {"name": "Lounge room"},
"traits": ["action.devices.traits.OnOff", "action.devices.traits.Modes"],
"traits": [
"action.devices.traits.OnOff",
"action.devices.traits.Modes",
"action.devices.traits.TransportControl",
"action.devices.traits.MediaState",
],
"type": "action.devices.types.SETTOP",
"willReportState": False,
},
@ -196,6 +205,8 @@ DEMO_DEVICES = [
"action.devices.traits.OnOff",
"action.devices.traits.Volume",
"action.devices.traits.Modes",
"action.devices.traits.TransportControl",
"action.devices.traits.MediaState",
],
"type": "action.devices.types.SETTOP",
"willReportState": False,

10
tests/components/google_assistant/test_smart_home.py
View File

@ -769,10 +769,16 @@ async def test_device_media_player(hass, device_class, google_type):
"agentUserId": "test-agent",
"devices": [
{
"attributes": {},
"attributes": {
"supportActivityState": True,
"supportPlaybackState": True,
},
"id": sensor.entity_id,
"name": {"name": sensor.name},
"traits": ["action.devices.traits.OnOff"],
"traits": [
"action.devices.traits.OnOff",
"action.devices.traits.MediaState",
],
"type": google_type,
"willReportState": False,
}

170
tests/components/google_assistant/test_trait.py
View File

@ -1,4 +1,5 @@
"""Tests for the Google Assistant traits."""
from datetime import datetime, timedelta
import logging
import pytest
@ -35,8 +36,13 @@ from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_DISARMED,
STATE_ALARM_PENDING,
STATE_IDLE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
TEMP_CELSIUS,
TEMP_FAHRENHEIT,
@ -1850,3 +1856,167 @@ async def test_humidity_setting_sensor_data(hass, state, ambient):
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_ONOFF, BASIC_DATA, {"on": False}, {})
assert err.value.code == const.ERR_NOT_SUPPORTED
async def test_transport_control(hass):
"""Test the TransportControlTrait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
for feature in trait.MEDIA_COMMAND_SUPPORT_MAPPING.values():
assert trait.TransportControlTrait.supported(media_player.DOMAIN, feature, None)
now = datetime(2020, 1, 1)
trt = trait.TransportControlTrait(
hass,
State(
"media_player.bla",
media_player.STATE_PLAYING,
{
media_player.ATTR_MEDIA_POSITION: 100,
media_player.ATTR_MEDIA_DURATION: 200,
media_player.ATTR_MEDIA_POSITION_UPDATED_AT: now
- timedelta(seconds=10),
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.5,
ATTR_SUPPORTED_FEATURES: media_player.SUPPORT_PLAY
| media_player.SUPPORT_STOP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"transportControlSupportedCommands": ["RESUME", "STOP"]
}
assert trt.query_attributes() == {}
# COMMAND_MEDIA_SEEK_RELATIVE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_SEEK
)
# Patch to avoid time ticking over during the command failing the test
with patch("homeassistant.util.dt.utcnow", return_value=now):
await trt.execute(
trait.COMMAND_MEDIA_SEEK_RELATIVE,
BASIC_DATA,
{"relativePositionMs": 10000},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
# 100s (current position) + 10s (from command) + 10s (from updated_at)
media_player.ATTR_MEDIA_SEEK_POSITION: 120,
}
# COMMAND_MEDIA_SEEK_TO_POSITION
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_SEEK
)
await trt.execute(
trait.COMMAND_MEDIA_SEEK_TO_POSITION, BASIC_DATA, {"absPositionMs": 50000}, {}
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_SEEK_POSITION: 50,
}
# COMMAND_MEDIA_NEXT
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_NEXT_TRACK
)
await trt.execute(trait.COMMAND_MEDIA_NEXT, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_PAUSE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PAUSE
)
await trt.execute(trait.COMMAND_MEDIA_PAUSE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_PREVIOUS
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PREVIOUS_TRACK
)
await trt.execute(trait.COMMAND_MEDIA_PREVIOUS, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_RESUME
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PLAY
)
await trt.execute(trait.COMMAND_MEDIA_RESUME, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_SHUFFLE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_SHUFFLE_SET
)
await trt.execute(trait.COMMAND_MEDIA_SHUFFLE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_SHUFFLE: True,
}
# COMMAND_MEDIA_STOP
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_STOP
)
await trt.execute(trait.COMMAND_MEDIA_STOP, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
@pytest.mark.parametrize(
"state",
(
STATE_OFF,
STATE_IDLE,
STATE_PLAYING,
STATE_ON,
STATE_PAUSED,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
),
)
async def test_media_state(hass, state):
"""Test the MediaStateTrait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.TransportControlTrait.supported(
media_player.DOMAIN, media_player.SUPPORT_PLAY, None
)
trt = trait.MediaStateTrait(
hass,
State(
"media_player.bla",
state,
{
media_player.ATTR_MEDIA_POSITION: 100,
media_player.ATTR_MEDIA_DURATION: 200,
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.5,
ATTR_SUPPORTED_FEATURES: media_player.SUPPORT_PLAY
| media_player.SUPPORT_STOP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"supportActivityState": True,
"supportPlaybackState": True,
}
assert trt.query_attributes() == {
"activityState": trt.activity_lookup.get(state),
"playbackState": trt.playback_lookup.get(state),
}