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

Add service select scene to Yamaha Hifi media player (#36564)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
StevusPrimus 2020-06-08 19:31:58 +02:00 • committed by GitHub
parent 3adfb86a19
commit 5975ec340b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 219 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
homeassistant/components/yamaha/const.py
View file

@ -1,3 +1,4 @@
"""Constants for the Yamaha component."""
DOMAIN = "yamaha"
SERVICE_ENABLE_OUTPUT = "enable_output"
SERVICE_SELECT_SCENE = "select_scene"

143
homeassistant/components/yamaha/media_player.py
View file

@ -22,7 +22,6 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_NAME,
STATE_IDLE,
@ -30,15 +29,17 @@ from homeassistant.const import (
STATE_ON,
STATE_PLAYING,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import config_validation as cv, entity_platform
from .const import DOMAIN, SERVICE_ENABLE_OUTPUT
from .const import SERVICE_ENABLE_OUTPUT, SERVICE_SELECT_SCENE
_LOGGER = logging.getLogger(__name__)
ATTR_ENABLED = "enabled"
ATTR_PORT = "port"
ATTR_SCENE = "scene"
CONF_SOURCE_IGNORE = "source_ignore"
CONF_SOURCE_NAMES = "source_names"
CONF_ZONE_IGNORE = "zone_ignore"
@ -47,12 +48,6 @@ CONF_ZONE_NAMES = "zone_names"
DATA_YAMAHA = "yamaha_known_receivers"
DEFAULT_NAME = "Yamaha Receiver"
MEDIA_PLAYER_SCHEMA = vol.Schema({ATTR_ENTITY_ID: cv.comp_entity_ids})
ENABLE_OUTPUT_SCHEMA = MEDIA_PLAYER_SCHEMA.extend(
{vol.Required(ATTR_ENABLED): cv.boolean, vol.Required(ATTR_PORT): cv.string}
)
SUPPORT_YAMAHA = (
SUPPORT_VOLUME_SET
| SUPPORT_VOLUME_MUTE
@ -79,78 +74,94 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Yamaha platform."""
class YamahaConfigInfo:
"""Configuration Info for Yamaha Receivers."""
# Keep track of configured receivers so that we don't end up
# discovering a receiver dynamically that we have static config
# for. Map each device from its zone_id to an instance since
# YamahaDevice is not hashable (thus not possible to add to a set).
if hass.data.get(DATA_YAMAHA) is None:
hass.data[DATA_YAMAHA] = {}
def __init__(self, config: None, discovery_info: None):
"""Initialize the Configuration Info for Yamaha Receiver."""
self.name = config.get(CONF_NAME)
self.host = config.get(CONF_HOST)
self.ctrl_url = f"http://{self.host}:80/YamahaRemoteControl/ctrl"
self.source_ignore = config.get(CONF_SOURCE_IGNORE)
self.source_names = config.get(CONF_SOURCE_NAMES)
self.zone_ignore = config.get(CONF_ZONE_IGNORE)
self.zone_names = config.get(CONF_ZONE_NAMES)
self.from_discovery = False
if discovery_info is not None:
self.name = discovery_info.get("name")
self.model = discovery_info.get("model_name")
self.ctrl_url = discovery_info.get("control_url")
self.desc_url = discovery_info.get("description_url")
self.zone_ignore = []
self.from_discovery = True
name = config.get(CONF_NAME)
host = config.get(CONF_HOST)
source_ignore = config.get(CONF_SOURCE_IGNORE)
source_names = config.get(CONF_SOURCE_NAMES)
zone_ignore = config.get(CONF_ZONE_IGNORE)
zone_names = config.get(CONF_ZONE_NAMES)
if discovery_info is not None:
name = discovery_info.get("name")
model = discovery_info.get("model_name")
ctrl_url = discovery_info.get("control_url")
desc_url = discovery_info.get("description_url")
def _discovery(config_info):
"""Discover receivers from configuration in the network."""
if config_info.from_discovery:
receivers = rxv.RXV(
ctrl_url, model_name=model, friendly_name=name, unit_desc_url=desc_url
config_info.ctrl_url,
model_name=config_info.model,
friendly_name=config_info.name,
unit_desc_url=config_info.desc_url,
).zone_controllers()
_LOGGER.debug("Receivers: %s", receivers)
# when we are dynamically discovered config is empty
zone_ignore = []
elif host is None:
elif config_info.host is None:
receivers = []
for recv in rxv.find():
receivers.extend(recv.zone_controllers())
else:
ctrl_url = f"http://{host}:80/YamahaRemoteControl/ctrl"
receivers = rxv.RXV(ctrl_url, name).zone_controllers()
receivers = rxv.RXV(config_info.ctrl_url, config_info.name).zone_controllers()
devices = []
return receivers
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Yamaha platform."""
# Keep track of configured receivers so that we don't end up
# discovering a receiver dynamically that we have static config
# for. Map each device from its zone_id .
known_zones = hass.data.setdefault(DATA_YAMAHA, set())
# Get the Infos for configuration from config (YAML) or Discovery
config_info = YamahaConfigInfo(config=config, discovery_info=discovery_info)
# Async check if the Receivers are there in the network
receivers = await hass.async_add_executor_job(_discovery, config_info)
entities = []
for receiver in receivers:
if receiver.zone in zone_ignore:
if receiver.zone in config_info.zone_ignore:
continue
device = YamahaDevice(name, receiver, source_ignore, source_names, zone_names)
entity = YamahaDevice(
config_info.name,
receiver,
config_info.source_ignore,
config_info.source_names,
config_info.zone_names,
)
# Only add device if it's not already added
if device.zone_id not in hass.data[DATA_YAMAHA]:
hass.data[DATA_YAMAHA][device.zone_id] = device
devices.append(device)
if entity.zone_id not in known_zones:
known_zones.add(entity.zone_id)
entities.append(entity)
else:
_LOGGER.debug("Ignoring duplicate receiver: %s", name)
_LOGGER.debug("Ignoring duplicate receiver: %s", config_info.name)
def service_handler(service):
"""Handle for services."""
entity_ids = service.data.get(ATTR_ENTITY_ID)
async_add_entities(entities)
devices = [
device
for device in hass.data[DATA_YAMAHA].values()
if not entity_ids or device.entity_id in entity_ids
]
for device in devices:
port = service.data[ATTR_PORT]
enabled = service.data[ATTR_ENABLED]
device.enable_output(port, enabled)
device.schedule_update_ha_state(True)
hass.services.register(
DOMAIN, SERVICE_ENABLE_OUTPUT, service_handler, schema=ENABLE_OUTPUT_SCHEMA
# Register Service 'select_scene'
platform = entity_platform.current_platform.get()
platform.async_register_entity_service(
SERVICE_SELECT_SCENE, {vol.Required(ATTR_SCENE): cv.string}, "set_scene",
)
# Register Service 'enable_output'
platform.async_register_entity_service(
SERVICE_ENABLE_OUTPUT,
{vol.Required(ATTR_ENABLED): cv.boolean, vol.Required(ATTR_PORT): cv.string},
"enable_output",
)
add_entities(devices)
class YamahaDevice(MediaPlayerEntity):
@ -350,7 +361,6 @@ class YamahaDevice(MediaPlayerEntity):
Yamaha to direct play certain kinds of media. media_type is
treated as the input type that we are setting, and media id is
specific to it.
For the NET RADIO mediatype the format for ``media_id`` is a
"path" in your vtuner hierarchy. For instance:
``Bookmarks>Internet>Radio Paradise``. The separators are
@ -358,12 +368,10 @@ class YamahaDevice(MediaPlayerEntity):
scenes. There is a looping construct built into the yamaha
library to do this with a fallback timeout if the vtuner
service is unresponsive.
NOTE: this might take a while, because the only API interface
for setting the net radio station emulates button pressing and
navigating through the net radio menu hierarchy. And each sub
menu must be fetched by the receiver from the vtuner service.
"""
if media_type == "NET RADIO":
self.receiver.net_radio(media_id)
@ -372,6 +380,13 @@ class YamahaDevice(MediaPlayerEntity):
"""Enable or disable an output port.."""
self.receiver.enable_output(port, enabled)
def set_scene(self, scene):
"""Set the current scene."""
try:
self.receiver.scene = scene
except AssertionError:
_LOGGER.warning("Scene '%s' does not exist!", scene)
def select_sound_mode(self, sound_mode):
"""Set Sound Mode for Receiver.."""
self.receiver.surround_program = sound_mode

9
homeassistant/components/yamaha/services.yaml
View file

@ -10,3 +10,12 @@ enable_output:
enabled:
description: Boolean indicating if port should be enabled or not.
example: true
select_scene:
description: "Select a scene on the receiver"
fields:
entity_id:
description: Name(s) of entities to enable/disable port on.
example: "media_player.yamaha"
scene:
description: Name of the scene. Standard for RX-V437 is 'BD/DVD Movie Viewing', 'TV Viewing', 'NET Audio Listening' or 'Radio Listening'
example: "TV Viewing"

169
tests/components/yamaha/test_media_player.py
View file

@ -1,12 +1,15 @@
"""The tests for the Yamaha Media player platform."""
import unittest
import pytest
import homeassistant.components.media_player as mp
from homeassistant.components.yamaha import media_player as yamaha
from homeassistant.setup import setup_component
from homeassistant.components.yamaha.const import DOMAIN
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.setup import async_setup_component
from tests.async_mock import MagicMock, patch
from tests.common import get_test_home_assistant
from tests.async_mock import MagicMock, PropertyMock, call, patch
CONFIG = {"media_player": {"platform": "yamaha", "host": "127.0.0.1"}}
def _create_zone_mock(name, url):
@ -23,54 +26,142 @@ class FakeYamahaDevice:
"""Initialize the fake Yamaha device."""
self.ctrl_url = ctrl_url
self.name = name
self.zones = zones or []
self._zones = zones or []
def zone_controllers(self):
"""Return controllers for all available zones."""
return self.zones
return self._zones
class TestYamahaMediaPlayer(unittest.TestCase):
"""Test the Yamaha media player."""
@pytest.fixture(name="main_zone")
def main_zone_fixture():
"""Mock the main zone."""
return _create_zone_mock("Main zone", "http://main")
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.main_zone = _create_zone_mock("Main zone", "http://main")
self.device = FakeYamahaDevice(
"http://receiver", "Receiver", zones=[self.main_zone]
@pytest.fixture(name="device")
def device_fixture(main_zone):
"""Mock the yamaha device."""
device = FakeYamahaDevice("http://receiver", "Receiver", zones=[main_zone])
with patch("rxv.RXV", return_value=device):
yield device
async def test_setup_host(hass, device, main_zone):
"""Test set up integration with host."""
assert await async_setup_component(hass, mp.DOMAIN, CONFIG)
await hass.async_block_till_done()
state = hass.states.get("media_player.yamaha_receiver_main_zone")
assert state is not None
assert state.state == "off"
async def test_setup_no_host(hass, device, main_zone):
"""Test set up integration without host."""
with patch("rxv.find", return_value=[device]):
assert await async_setup_component(
hass, mp.DOMAIN, {"media_player": {"platform": "yamaha"}}
)
await hass.async_block_till_done()
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
state = hass.states.get("media_player.yamaha_receiver_main_zone")
def enable_output(self, port, enabled):
"""Enable output on a specific port."""
data = {
"entity_id": "media_player.yamaha_receiver_main_zone",
"port": port,
"enabled": enabled,
}
assert state is not None
assert state.state == "off"
self.hass.services.call(yamaha.DOMAIN, yamaha.SERVICE_ENABLE_OUTPUT, data, True)
def create_receiver(self, mock_rxv):
"""Create a mocked receiver."""
mock_rxv.return_value = self.device
async def test_setup_discovery(hass, device, main_zone):
"""Test set up integration via discovery."""
discovery_info = {
"name": "Yamaha Receiver",
"model_name": "Yamaha",
"control_url": "http://receiver",
"description_url": "http://receiver/description",
}
await async_load_platform(
hass, mp.DOMAIN, "yamaha", discovery_info, {mp.DOMAIN: {}}
)
await hass.async_block_till_done()
config = {"media_player": {"platform": "yamaha", "host": "127.0.0.1"}}
state = hass.states.get("media_player.yamaha_receiver_main_zone")
assert setup_component(self.hass, mp.DOMAIN, config)
self.hass.block_till_done()
assert state is not None
assert state.state == "off"
@patch("rxv.RXV")
def test_enable_output(self, mock_rxv):
"""Test enabling and disabling outputs."""
self.create_receiver(mock_rxv)
self.enable_output("hdmi1", True)
self.main_zone.enable_output.assert_called_with("hdmi1", True)
async def test_setup_zone_ignore(hass, device, main_zone):
"""Test set up integration without host."""
assert await async_setup_component(
hass,
mp.DOMAIN,
{
"media_player": {
"platform": "yamaha",
"host": "127.0.0.1",
"zone_ignore": "Main zone",
}
},
)
await hass.async_block_till_done()
self.enable_output("hdmi2", False)
self.main_zone.enable_output.assert_called_with("hdmi2", False)
state = hass.states.get("media_player.yamaha_receiver_main_zone")
assert state is None
async def test_enable_output(hass, device, main_zone):
"""Test enable output service."""
assert await async_setup_component(hass, mp.DOMAIN, CONFIG)
await hass.async_block_till_done()
port = "hdmi1"
enabled = True
data = {
"entity_id": "media_player.yamaha_receiver_main_zone",
"port": port,
"enabled": enabled,
}
await hass.services.async_call(DOMAIN, yamaha.SERVICE_ENABLE_OUTPUT, data, True)
assert main_zone.enable_output.call_count == 1
assert main_zone.enable_output.call_args == call(port, enabled)
async def test_select_scene(hass, device, main_zone, caplog):
"""Test select scene service."""
scene_prop = PropertyMock(return_value=None)
type(main_zone).scene = scene_prop
assert await async_setup_component(hass, mp.DOMAIN, CONFIG)
await hass.async_block_till_done()
scene = "TV Viewing"
data = {
"entity_id": "media_player.yamaha_receiver_main_zone",
"scene": scene,
}
await hass.services.async_call(DOMAIN, yamaha.SERVICE_SELECT_SCENE, data, True)
assert scene_prop.call_count == 1
assert scene_prop.call_args == call(scene)
scene = "BD/DVD Movie Viewing"
data["scene"] = scene
await hass.services.async_call(DOMAIN, yamaha.SERVICE_SELECT_SCENE, data, True)
assert scene_prop.call_count == 2
assert scene_prop.call_args == call(scene)
scene_prop.side_effect = AssertionError()
missing_scene = "Missing scene"
data["scene"] = missing_scene
await hass.services.async_call(DOMAIN, yamaha.SERVICE_SELECT_SCENE, data, True)
assert f"Scene '{missing_scene}' does not exist!" in caplog.text