mirror of
https://github.com/home-assistant/core.git
synced 2025-04-25 01:38:02 +00:00
Add blackbird media player component (#13549)
This commit is contained in:
koolsb
Sebastian Muszynski
parent
3dc70436f1
commit
37cd63ea5a
213
homeassistant/components/media_player/blackbird.py
Normal file
213
homeassistant/components/media_player/blackbird.py
Normal file
@ -0,0 +1,213 @@
|
||||
"""
|
||||
Support for interfacing with Monoprice Blackbird 4k 8x8 HDBaseT Matrix.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/media_player.blackbird
|
||||
"""
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player import (
|
||||
DOMAIN, MEDIA_PLAYER_SCHEMA, PLATFORM_SCHEMA, SUPPORT_SELECT_SOURCE,
|
||||
SUPPORT_TURN_OFF, SUPPORT_TURN_ON, MediaPlayerDevice)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID, CONF_NAME, CONF_HOST, CONF_PORT, STATE_OFF, STATE_ON)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
REQUIREMENTS = ['pyblackbird==0.5']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SUPPORT_BLACKBIRD = SUPPORT_TURN_ON | SUPPORT_TURN_OFF | \
|
||||
SUPPORT_SELECT_SOURCE
|
||||
|
||||
ZONE_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_NAME): cv.string,
|
||||
})
|
||||
|
||||
SOURCE_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_NAME): cv.string,
|
||||
})
|
||||
|
||||
CONF_ZONES = 'zones'
|
||||
CONF_SOURCES = 'sources'
|
||||
CONF_TYPE = 'type'
|
||||
|
||||
DATA_BLACKBIRD = 'blackbird'
|
||||
|
||||
SERVICE_SETALLZONES = 'blackbird_set_all_zones'
|
||||
ATTR_SOURCE = 'source'
|
||||
|
||||
BLACKBIRD_SETALLZONES_SCHEMA = MEDIA_PLAYER_SCHEMA.extend({
|
||||
vol.Required(ATTR_SOURCE): cv.string
|
||||
})
|
||||
|
||||
|
||||
# Valid zone ids: 1-8
|
||||
ZONE_IDS = vol.All(vol.Coerce(int), vol.Range(min=1, max=8))
|
||||
|
||||
# Valid source ids: 1-8
|
||||
SOURCE_IDS = vol.All(vol.Coerce(int), vol.Range(min=1, max=8))
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Required(CONF_TYPE): vol.In(['serial', 'socket']),
|
||||
vol.Optional(CONF_PORT): cv.string,
|
||||
vol.Optional(CONF_HOST): cv.string,
|
||||
vol.Required(CONF_ZONES): vol.Schema({ZONE_IDS: ZONE_SCHEMA}),
|
||||
vol.Required(CONF_SOURCES): vol.Schema({SOURCE_IDS: SOURCE_SCHEMA}),
|
||||
})
|
||||
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the Monoprice Blackbird 4k 8x8 HDBaseT Matrix platform."""
|
||||
port = config.get(CONF_PORT)
|
||||
host = config.get(CONF_HOST)
|
||||
device_type = config.get(CONF_TYPE)
|
||||
|
||||
import socket
|
||||
from pyblackbird import get_blackbird
|
||||
from serial import SerialException
|
||||
|
||||
if device_type == 'serial':
|
||||
if port is None:
|
||||
_LOGGER.error("No port configured")
|
||||
return
|
||||
try:
|
||||
blackbird = get_blackbird(port)
|
||||
except SerialException:
|
||||
_LOGGER.error("Error connecting to the Blackbird controller")
|
||||
return
|
||||
|
||||
elif device_type == 'socket':
|
||||
try:
|
||||
if host is None:
|
||||
_LOGGER.error("No host configured")
|
||||
return
|
||||
blackbird = get_blackbird(host, False)
|
||||
except socket.timeout:
|
||||
_LOGGER.error("Error connecting to the Blackbird controller")
|
||||
return
|
||||
|
||||
else:
|
||||
_LOGGER.error("Incorrect device type specified")
|
||||
return
|
||||
|
||||
sources = {source_id: extra[CONF_NAME] for source_id, extra
|
||||
in config[CONF_SOURCES].items()}
|
||||
|
||||
hass.data[DATA_BLACKBIRD] = []
|
||||
for zone_id, extra in config[CONF_ZONES].items():
|
||||
_LOGGER.info("Adding zone %d - %s", zone_id, extra[CONF_NAME])
|
||||
hass.data[DATA_BLACKBIRD].append(BlackbirdZone(
|
||||
blackbird, sources, zone_id, extra[CONF_NAME]))
|
||||
|
||||
add_devices(hass.data[DATA_BLACKBIRD], True)
|
||||
|
||||
def service_handle(service):
|
||||
"""Handle for services."""
|
||||
entity_ids = service.data.get(ATTR_ENTITY_ID)
|
||||
source = service.data.get(ATTR_SOURCE)
|
||||
if entity_ids:
|
||||
devices = [device for device in hass.data[DATA_BLACKBIRD]
|
||||
if device.entity_id in entity_ids]
|
||||
|
||||
else:
|
||||
devices = hass.data[DATA_BLACKBIRD]
|
||||
|
||||
for device in devices:
|
||||
if service.service == SERVICE_SETALLZONES:
|
||||
device.set_all_zones(source)
|
||||
|
||||
hass.services.register(DOMAIN, SERVICE_SETALLZONES, service_handle,
|
||||
schema=BLACKBIRD_SETALLZONES_SCHEMA)
|
||||
|
||||
|
||||
class BlackbirdZone(MediaPlayerDevice):
|
||||
"""Representation of a Blackbird matrix zone."""
|
||||
|
||||
def __init__(self, blackbird, sources, zone_id, zone_name):
|
||||
"""Initialize new zone."""
|
||||
self._blackbird = blackbird
|
||||
# dict source_id -> source name
|
||||
self._source_id_name = sources
|
||||
# dict source name -> source_id
|
||||
self._source_name_id = {v: k for k, v in sources.items()}
|
||||
# ordered list of all source names
|
||||
self._source_names = sorted(self._source_name_id.keys(),
|
||||
key=lambda v: self._source_name_id[v])
|
||||
self._zone_id = zone_id
|
||||
self._name = zone_name
|
||||
self._state = None
|
||||
self._source = None
|
||||
|
||||
def update(self):
|
||||
"""Retrieve latest state."""
|
||||
state = self._blackbird.zone_status(self._zone_id)
|
||||
if not state:
|
||||
return False
|
||||
self._state = STATE_ON if state.power else STATE_OFF
|
||||
idx = state.av
|
||||
if idx in self._source_id_name:
|
||||
self._source = self._source_id_name[idx]
|
||||
else:
|
||||
self._source = None
|
||||
return True
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the zone."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the zone."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Return flag of media commands that are supported."""
|
||||
return SUPPORT_BLACKBIRD
|
||||
|
||||
@property
|
||||
def media_title(self):
|
||||
"""Return the current source as media title."""
|
||||
return self._source
|
||||
|
||||
@property
|
||||
def source(self):
|
||||
"""Return the current input source of the device."""
|
||||
return self._source
|
||||
|
||||
@property
|
||||
def source_list(self):
|
||||
"""List of available input sources."""
|
||||
return self._source_names
|
||||
|
||||
def set_all_zones(self, source):
|
||||
"""Set all zones to one source."""
|
||||
_LOGGER.debug("Setting all zones")
|
||||
if source not in self._source_name_id:
|
||||
return
|
||||
idx = self._source_name_id[source]
|
||||
_LOGGER.debug("Setting all zones source to %s", idx)
|
||||
self._blackbird.set_all_zone_source(idx)
|
||||
|
||||
def select_source(self, source):
|
||||
"""Set input source."""
|
||||
if source not in self._source_name_id:
|
||||
return
|
||||
idx = self._source_name_id[source]
|
||||
_LOGGER.debug("Setting zone %d source to %s", self._zone_id, idx)
|
||||
self._blackbird.set_zone_source(self._zone_id, idx)
|
||||
|
||||
def turn_on(self):
|
||||
"""Turn the media player on."""
|
||||
_LOGGER.debug("Turning zone %d on", self._zone_id)
|
||||
self._blackbird.set_zone_power(self._zone_id, True)
|
||||
|
||||
def turn_off(self):
|
||||
"""Turn the media player off."""
|
||||
_LOGGER.debug("Turning zone %d off", self._zone_id)
|
||||
self._blackbird.set_zone_power(self._zone_id, False)
|
||||
@ -402,3 +402,13 @@ songpal_set_sound_setting:
|
||||
value:
|
||||
description: Value to set.
|
||||
example: 'on'
|
||||
|
||||
blackbird_set_all_zones:
|
||||
description: Set all Blackbird zones to a single source.
|
||||
fields:
|
||||
entity_id:
|
||||
description: Name of any blackbird zone.
|
||||
example: 'media_player.zone_1'
|
||||
source:
|
||||
description: Name of source to switch to.
|
||||
example: 'Source 1'
|
||||
|
||||
@ -704,6 +704,9 @@ pyatv==0.3.9
|
||||
# homeassistant.components.sensor.bbox
|
||||
pybbox==0.0.5-alpha
|
||||
|
||||
# homeassistant.components.media_player.blackbird
|
||||
pyblackbird==0.5
|
||||
|
||||
# homeassistant.components.device_tracker.bluetooth_tracker
|
||||
# pybluez==0.22
|
||||
|
||||
|
||||
@ -129,6 +129,9 @@ pushbullet.py==0.11.0
|
||||
# homeassistant.components.canary
|
||||
py-canary==0.5.0
|
||||
|
||||
# homeassistant.components.media_player.blackbird
|
||||
pyblackbird==0.5
|
||||
|
||||
# homeassistant.components.deconz
|
||||
pydeconz==36
|
||||
|
||||
|
||||
@ -68,6 +68,7 @@ TEST_REQUIREMENTS = (
|
||||
'prometheus_client',
|
||||
'pushbullet.py',
|
||||
'py-canary',
|
||||
'pyblackbird',
|
||||
'pydeconz',
|
||||
'pydispatcher',
|
||||
'PyJWT',
|
||||
|
||||
328
tests/components/media_player/test_blackbird.py
Normal file
328
tests/components/media_player/test_blackbird.py
Normal file
@ -0,0 +1,328 @@
|
||||
"""The tests for the Monoprice Blackbird media player platform."""
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import voluptuous as vol
|
||||
|
||||
from collections import defaultdict
|
||||
from homeassistant.components.media_player import (
|
||||
DOMAIN, SUPPORT_TURN_ON, SUPPORT_TURN_OFF, SUPPORT_SELECT_SOURCE)
|
||||
from homeassistant.const import STATE_ON, STATE_OFF
|
||||
|
||||
import tests.common
|
||||
from homeassistant.components.media_player.blackbird import (
|
||||
DATA_BLACKBIRD, PLATFORM_SCHEMA, SERVICE_SETALLZONES, setup_platform)
|
||||
|
||||
|
||||
class AttrDict(dict):
|
||||
"""Helper clas for mocking attributes."""
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
"""Set attribute."""
|
||||
self[name] = value
|
||||
|
||||
def __getattr__(self, item):
|
||||
"""Get attribute."""
|
||||
return self[item]
|
||||
|
||||
|
||||
class MockBlackbird(object):
|
||||
"""Mock for pyblackbird object."""
|
||||
|
||||
def __init__(self):
|
||||
"""Init mock object."""
|
||||
self.zones = defaultdict(lambda: AttrDict(power=True,
|
||||
av=1))
|
||||
|
||||
def zone_status(self, zone_id):
|
||||
"""Get zone status."""
|
||||
status = self.zones[zone_id]
|
||||
status.zone = zone_id
|
||||
return AttrDict(status)
|
||||
|
||||
def set_zone_source(self, zone_id, source_idx):
|
||||
"""Set source for zone."""
|
||||
self.zones[zone_id].av = source_idx
|
||||
|
||||
def set_zone_power(self, zone_id, power):
|
||||
"""Turn zone on/off."""
|
||||
self.zones[zone_id].power = power
|
||||
|
||||
def set_all_zone_source(self, source_idx):
|
||||
"""Set source for all zones."""
|
||||
self.zones[3].av = source_idx
|
||||
|
||||
|
||||
class TestBlackbirdSchema(unittest.TestCase):
|
||||
"""Test Blackbird schema."""
|
||||
|
||||
def test_valid_serial_schema(self):
|
||||
"""Test valid schema."""
|
||||
valid_schema = {
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': '/dev/ttyUSB0',
|
||||
'zones': {1: {'name': 'a'},
|
||||
2: {'name': 'a'},
|
||||
3: {'name': 'a'},
|
||||
4: {'name': 'a'},
|
||||
5: {'name': 'a'},
|
||||
6: {'name': 'a'},
|
||||
7: {'name': 'a'},
|
||||
8: {'name': 'a'},
|
||||
},
|
||||
'sources': {
|
||||
1: {'name': 'a'},
|
||||
2: {'name': 'a'},
|
||||
3: {'name': 'a'},
|
||||
4: {'name': 'a'},
|
||||
5: {'name': 'a'},
|
||||
6: {'name': 'a'},
|
||||
7: {'name': 'a'},
|
||||
8: {'name': 'a'},
|
||||
}
|
||||
}
|
||||
PLATFORM_SCHEMA(valid_schema)
|
||||
|
||||
def test_valid_socket_schema(self):
|
||||
"""Test valid schema."""
|
||||
valid_schema = {
|
||||
'platform': 'blackbird',
|
||||
'type': 'socket',
|
||||
'port': '192.168.1.50',
|
||||
'zones': {1: {'name': 'a'},
|
||||
2: {'name': 'a'},
|
||||
3: {'name': 'a'},
|
||||
4: {'name': 'a'},
|
||||
5: {'name': 'a'},
|
||||
},
|
||||
'sources': {
|
||||
1: {'name': 'a'},
|
||||
2: {'name': 'a'},
|
||||
3: {'name': 'a'},
|
||||
4: {'name': 'a'},
|
||||
}
|
||||
}
|
||||
PLATFORM_SCHEMA(valid_schema)
|
||||
|
||||
def test_invalid_schemas(self):
|
||||
"""Test invalid schemas."""
|
||||
schemas = (
|
||||
{}, # Empty
|
||||
None, # None
|
||||
# Missing type
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {1: {'name': 'a'}},
|
||||
'sources': {1: {'name': 'b'}},
|
||||
},
|
||||
# Invalid zone number
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {11: {'name': 'a'}},
|
||||
'sources': {1: {'name': 'b'}},
|
||||
},
|
||||
# Invalid source number
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {1: {'name': 'a'}},
|
||||
'sources': {9: {'name': 'b'}},
|
||||
},
|
||||
# Zone missing name
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {1: {}},
|
||||
'sources': {1: {'name': 'b'}},
|
||||
},
|
||||
# Source missing name
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {1: {'name': 'a'}},
|
||||
'sources': {1: {}},
|
||||
},
|
||||
# Invalid type
|
||||
{
|
||||
'platform': 'blackbird',
|
||||
'type': 'aaa',
|
||||
'port': 'aaa',
|
||||
'name': 'Name',
|
||||
'zones': {1: {'name': 'a'}},
|
||||
'sources': {1: {'name': 'b'}},
|
||||
},
|
||||
)
|
||||
for value in schemas:
|
||||
with self.assertRaises(vol.MultipleInvalid):
|
||||
PLATFORM_SCHEMA(value)
|
||||
|
||||
|
||||
class TestBlackbirdMediaPlayer(unittest.TestCase):
|
||||
"""Test the media_player module."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up the test case."""
|
||||
self.blackbird = MockBlackbird()
|
||||
self.hass = tests.common.get_test_home_assistant()
|
||||
self.hass.start()
|
||||
# Note, source dictionary is unsorted!
|
||||
with mock.patch('pyblackbird.get_blackbird',
|
||||
new=lambda *a: self.blackbird):
|
||||
setup_platform(self.hass, {
|
||||
'platform': 'blackbird',
|
||||
'type': 'serial',
|
||||
'port': '/dev/ttyUSB0',
|
||||
'zones': {3: {'name': 'Zone name'}},
|
||||
'sources': {1: {'name': 'one'},
|
||||
3: {'name': 'three'},
|
||||
2: {'name': 'two'}},
|
||||
}, lambda *args, **kwargs: None, {})
|
||||
self.hass.block_till_done()
|
||||
self.media_player = self.hass.data[DATA_BLACKBIRD][0]
|
||||
self.media_player.hass = self.hass
|
||||
self.media_player.entity_id = 'media_player.zone_3'
|
||||
|
||||
def tearDown(self):
|
||||
"""Tear down the test case."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_setup_platform(self, *args):
|
||||
"""Test setting up platform."""
|
||||
# One service must be registered
|
||||
self.assertTrue(self.hass.services.has_service(DOMAIN,
|
||||
SERVICE_SETALLZONES))
|
||||
self.assertEqual(len(self.hass.data[DATA_BLACKBIRD]), 1)
|
||||
self.assertEqual(self.hass.data[DATA_BLACKBIRD][0].name, 'Zone name')
|
||||
|
||||
def test_setallzones_service_call_with_entity_id(self):
|
||||
"""Test set all zone source service call with entity id."""
|
||||
self.media_player.update()
|
||||
self.assertEqual('Zone name', self.media_player.name)
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
self.assertEqual('one', self.media_player.source)
|
||||
|
||||
# Call set all zones service
|
||||
self.hass.services.call(DOMAIN, SERVICE_SETALLZONES,
|
||||
{'entity_id': 'media_player.zone_3',
|
||||
'source': 'three'},
|
||||
blocking=True)
|
||||
|
||||
# Check that source was changed
|
||||
self.assertEqual(3, self.blackbird.zones[3].av)
|
||||
self.media_player.update()
|
||||
self.assertEqual('three', self.media_player.source)
|
||||
|
||||
def test_setallzones_service_call_without_entity_id(self):
|
||||
"""Test set all zone source service call without entity id."""
|
||||
self.media_player.update()
|
||||
self.assertEqual('Zone name', self.media_player.name)
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
self.assertEqual('one', self.media_player.source)
|
||||
|
||||
# Call set all zones service
|
||||
self.hass.services.call(DOMAIN, SERVICE_SETALLZONES,
|
||||
{'source': 'three'}, blocking=True)
|
||||
|
||||
# Check that source was changed
|
||||
self.assertEqual(3, self.blackbird.zones[3].av)
|
||||
self.media_player.update()
|
||||
self.assertEqual('three', self.media_player.source)
|
||||
|
||||
def test_update(self):
|
||||
"""Test updating values from blackbird."""
|
||||
self.assertIsNone(self.media_player.state)
|
||||
self.assertIsNone(self.media_player.source)
|
||||
|
||||
self.media_player.update()
|
||||
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
self.assertEqual('one', self.media_player.source)
|
||||
|
||||
def test_name(self):
|
||||
"""Test name property."""
|
||||
self.assertEqual('Zone name', self.media_player.name)
|
||||
|
||||
def test_state(self):
|
||||
"""Test state property."""
|
||||
self.assertIsNone(self.media_player.state)
|
||||
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
|
||||
self.blackbird.zones[3].power = False
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_OFF, self.media_player.state)
|
||||
|
||||
def test_supported_features(self):
|
||||
"""Test supported features property."""
|
||||
self.assertEqual(SUPPORT_TURN_ON | SUPPORT_TURN_OFF |
|
||||
SUPPORT_SELECT_SOURCE,
|
||||
self.media_player.supported_features)
|
||||
|
||||
def test_source(self):
|
||||
"""Test source property."""
|
||||
self.assertIsNone(self.media_player.source)
|
||||
self.media_player.update()
|
||||
self.assertEqual('one', self.media_player.source)
|
||||
|
||||
def test_media_title(self):
|
||||
"""Test media title property."""
|
||||
self.assertIsNone(self.media_player.media_title)
|
||||
self.media_player.update()
|
||||
self.assertEqual('one', self.media_player.media_title)
|
||||
|
||||
def test_source_list(self):
|
||||
"""Test source list property."""
|
||||
# Note, the list is sorted!
|
||||
self.assertEqual(['one', 'two', 'three'],
|
||||
self.media_player.source_list)
|
||||
|
||||
def test_select_source(self):
|
||||
"""Test source selection methods."""
|
||||
self.media_player.update()
|
||||
|
||||
self.assertEqual('one', self.media_player.source)
|
||||
|
||||
self.media_player.select_source('two')
|
||||
self.assertEqual(2, self.blackbird.zones[3].av)
|
||||
self.media_player.update()
|
||||
self.assertEqual('two', self.media_player.source)
|
||||
|
||||
# Trying to set unknown source.
|
||||
self.media_player.select_source('no name')
|
||||
self.assertEqual(2, self.blackbird.zones[3].av)
|
||||
self.media_player.update()
|
||||
self.assertEqual('two', self.media_player.source)
|
||||
|
||||
def test_turn_on(self):
|
||||
"""Testing turning on the zone."""
|
||||
self.blackbird.zones[3].power = False
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_OFF, self.media_player.state)
|
||||
|
||||
self.media_player.turn_on()
|
||||
self.assertTrue(self.blackbird.zones[3].power)
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
|
||||
def test_turn_off(self):
|
||||
"""Testing turning off the zone."""
|
||||
self.blackbird.zones[3].power = True
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_ON, self.media_player.state)
|
||||
|
||||
self.media_player.turn_off()
|
||||
self.assertFalse(self.blackbird.zones[3].power)
|
||||
self.media_player.update()
|
||||
self.assertEqual(STATE_OFF, self.media_player.state)
|
||||
Loading…
x
Reference in New Issue
Block a user