Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Functinality to save/restore snapshots for monoprice platform () 

* added functionality to save/restore snapshots to monoprice platform

* renamed monoprice_snapshot, monoprice_restore to snapshot, restore

This is to simplify refactoring of snapshot/restore functionality for monoprice, snapcast and sonos in the future
pull/11274/head
Egor Tsinko 2017-12-22 02:26:34 -07:00 committed by Pascal Vizeli
parent 9e0a765801
commit eeb309aea1
6 changed files with 240 additions and 27 deletions

68
homeassistant/components/media_player/monoprice.py
View File

@ -5,18 +5,21 @@ For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/media_player.monoprice/
"""
import logging
from os import path
import voluptuous as vol
from homeassistant.const import (CONF_NAME, CONF_PORT, STATE_OFF, STATE_ON)
from homeassistant.const import (ATTR_ENTITY_ID, CONF_NAME, CONF_PORT,
STATE_OFF, STATE_ON)
from homeassistant.config import load_yaml_config_file
import homeassistant.helpers.config_validation as cv
from homeassistant.components.media_player import (
MediaPlayerDevice, PLATFORM_SCHEMA, SUPPORT_VOLUME_MUTE,
SUPPORT_SELECT_SOURCE, SUPPORT_TURN_ON, SUPPORT_TURN_OFF,
SUPPORT_VOLUME_SET, SUPPORT_VOLUME_STEP)
DOMAIN, MediaPlayerDevice, MEDIA_PLAYER_SCHEMA, PLATFORM_SCHEMA,
SUPPORT_VOLUME_MUTE, SUPPORT_SELECT_SOURCE, SUPPORT_TURN_ON,
SUPPORT_TURN_OFF, SUPPORT_VOLUME_SET, SUPPORT_VOLUME_STEP)
REQUIREMENTS = ['pymonoprice==0.2']
REQUIREMENTS = ['pymonoprice==0.3']
_LOGGER = logging.getLogger(__name__)
@ -35,6 +38,11 @@ SOURCE_SCHEMA = vol.Schema({
CONF_ZONES = 'zones'
CONF_SOURCES = 'sources'
DATA_MONOPRICE = 'monoprice'
SERVICE_SNAPSHOT = 'snapshot'
SERVICE_RESTORE = 'restore'
# Valid zone ids: 11-16 or 21-26 or 31-36
ZONE_IDS = vol.All(vol.Coerce(int), vol.Any(vol.Range(min=11, max=16),
vol.Range(min=21, max=26),
@ -56,9 +64,9 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
port = config.get(CONF_PORT)
from serial import SerialException
from pymonoprice import Monoprice
from pymonoprice import get_monoprice
try:
monoprice = Monoprice(port)
monoprice = get_monoprice(port)
except SerialException:
_LOGGER.error('Error connecting to Monoprice controller.')
return
@ -66,10 +74,41 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
sources = {source_id: extra[CONF_NAME] for source_id, extra
in config[CONF_SOURCES].items()}
hass.data[DATA_MONOPRICE] = []
for zone_id, extra in config[CONF_ZONES].items():
_LOGGER.info("Adding zone %d - %s", zone_id, extra[CONF_NAME])
add_devices([MonopriceZone(monoprice, sources,
zone_id, extra[CONF_NAME])], True)
hass.data[DATA_MONOPRICE].append(MonopriceZone(monoprice, sources,
zone_id,
extra[CONF_NAME]))
add_devices(hass.data[DATA_MONOPRICE], True)
descriptions = load_yaml_config_file(
path.join(path.dirname(__file__), 'services.yaml'))
def service_handle(service):
"""Handle for services."""
entity_ids = service.data.get(ATTR_ENTITY_ID)
if entity_ids:
devices = [device for device in hass.data[DATA_MONOPRICE]
if device.entity_id in entity_ids]
else:
devices = hass.data[DATA_MONOPRICE]
for device in devices:
if service.service == SERVICE_SNAPSHOT:
device.snapshot()
elif service.service == SERVICE_RESTORE:
device.restore()
hass.services.register(
DOMAIN, SERVICE_SNAPSHOT, service_handle,
descriptions.get(SERVICE_SNAPSHOT), schema=MEDIA_PLAYER_SCHEMA)
hass.services.register(
DOMAIN, SERVICE_RESTORE, service_handle,
descriptions.get(SERVICE_RESTORE), schema=MEDIA_PLAYER_SCHEMA)
class MonopriceZone(MediaPlayerDevice):
@ -90,6 +129,7 @@ class MonopriceZone(MediaPlayerDevice):
self._zone_id = zone_id
self._name = zone_name
self._snapshot = None
self._state = None
self._volume = None
self._source = None
@ -152,6 +192,16 @@ class MonopriceZone(MediaPlayerDevice):
"""List of available input sources."""
return self._source_names
def snapshot(self):
"""Save zone's current state."""
self._snapshot = self._monoprice.zone_status(self._zone_id)
def restore(self):
"""Restore saved state."""
if self._snapshot:
self._monoprice.restore_zone(self._snapshot)
self.schedule_update_ha_state(True)
def select_source(self, source):
"""Set input source."""
if source not in self._source_name_id:

14
homeassistant/components/media_player/services.yaml
View File

@ -107,6 +107,20 @@ media_seek:
description: Position to seek to. The format is platform dependent.
example: 100
monoprice_snapshot:
description: Take a snapshot of the media player zone.
fields:
entity_id:
description: Name(s) of entities that will be snapshot. Platform dependent.
example: 'media_player.living_room'
monoprice_restore:
description: Restore a snapshot of the media player zone.
fields:
entity_id:
description: Name(s) of entities that will be restored. Platform dependent.
example: 'media_player.living_room'
play_media:
description: Send the media player the command for playing media.
fields:

2
requirements_all.txt
View File

@ -756,7 +756,7 @@ pymochad==0.1.1
pymodbus==1.3.1
# homeassistant.components.media_player.monoprice
pymonoprice==0.2
pymonoprice==0.3
# homeassistant.components.media_player.yamaha_musiccast
pymusiccast==0.1.6

3
requirements_test_all.txt
View File

@ -127,6 +127,9 @@ pydispatcher==2.0.5
# homeassistant.components.litejet
pylitejet==0.1
# homeassistant.components.media_player.monoprice
pymonoprice==0.3
# homeassistant.components.alarm_control_panel.nx584
# homeassistant.components.binary_sensor.nx584
pynx584==0.4

1
script/gen_requirements_all.py
View File

@ -66,6 +66,7 @@ TEST_REQUIREMENTS = (
'pydispatcher',
'PyJWT',
'pylitejet',
'pymonoprice',
'pynx584',
'python-forecastio',
'pyunifi',

179
tests/components/media_player/test_monoprice.py
View File

@ -1,27 +1,30 @@
"""The tests for Monoprice Media player platform."""
import unittest
from unittest import mock
import voluptuous as vol
from collections import defaultdict
from homeassistant.components.media_player import (
SUPPORT_TURN_ON, SUPPORT_TURN_OFF, SUPPORT_VOLUME_MUTE,
DOMAIN, SUPPORT_TURN_ON, SUPPORT_TURN_OFF, SUPPORT_VOLUME_MUTE,
SUPPORT_VOLUME_SET, SUPPORT_VOLUME_STEP, SUPPORT_SELECT_SOURCE)
from homeassistant.const import STATE_ON, STATE_OFF
import tests.common
from homeassistant.components.media_player.monoprice import (
MonopriceZone, PLATFORM_SCHEMA)
DATA_MONOPRICE, PLATFORM_SCHEMA, SERVICE_SNAPSHOT,
SERVICE_RESTORE, setup_platform)
class MockState(object):
"""Mock for zone state object."""
class AttrDict(dict):
"""Helper class for mocking attributes."""
def __init__(self):
"""Init zone state."""
self.power = True
self.volume = 0
self.mute = True
self.source = 1
def __setattr__(self, name, value):
"""Set attribute."""
self[name] = value
def __getattr__(self, item):
"""Get attribute."""
return self[item]
class MockMonoprice(object):
@ -29,11 +32,16 @@ class MockMonoprice(object):
def __init__(self):
"""Init mock object."""
self.zones = defaultdict(lambda *a: MockState())
self.zones = defaultdict(lambda: AttrDict(power=True,
volume=0,
mute=True,
source=1))
def zone_status(self, zone_id):
"""Get zone status."""
return self.zones[zone_id]
status = self.zones[zone_id]
status.zone = zone_id
return AttrDict(status)
def set_source(self, zone_id, source_idx):
"""Set source for zone."""
@ -51,6 +59,10 @@ class MockMonoprice(object):
"""Set volume for zone."""
self.zones[zone_id].volume = volume
def restore_zone(self, zone):
"""Restore zone status."""
self.zones[zone.zone] = AttrDict(zone)
class TestMonopriceSchema(unittest.TestCase):
"""Test Monoprice schema."""
@ -147,11 +159,144 @@ class TestMonopriceMediaPlayer(unittest.TestCase):
def setUp(self):
"""Set up the test case."""
self.monoprice = MockMonoprice()
self.hass = tests.common.get_test_home_assistant()
self.hass.start()
# Note, source dictionary is unsorted!
self.media_player = MonopriceZone(self.monoprice, {1: 'one',
3: 'three',
2: 'two'},
12, 'Zone name')
with mock.patch('pymonoprice.get_monoprice',
new=lambda *a: self.monoprice):
setup_platform(self.hass, {
'platform': 'monoprice',
'port': '/dev/ttyS0',
'name': 'Name',
'zones': {12: {'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_MONOPRICE][0]
self.media_player.hass = self.hass
self.media_player.entity_id = 'media_player.zone_1'
def tearDown(self):
"""Tear down the test case."""
self.hass.stop()
def test_setup_platform(self, *args):
"""Test setting up platform."""
# Two services must be registered
self.assertTrue(self.hass.services.has_service(DOMAIN,
SERVICE_RESTORE))
self.assertTrue(self.hass.services.has_service(DOMAIN,
SERVICE_SNAPSHOT))
self.assertEqual(len(self.hass.data[DATA_MONOPRICE]), 1)
self.assertEqual(self.hass.data[DATA_MONOPRICE][0].name, 'Zone name')
def test_service_calls_with_entity_id(self):
"""Test snapshot save/restore service calls."""
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_ON, self.media_player.state)
self.assertEqual(0.0, self.media_player.volume_level, 0.0001)
self.assertTrue(self.media_player.is_volume_muted)
self.assertEqual('one', self.media_player.source)
# Saving default values
self.hass.services.call(DOMAIN, SERVICE_SNAPSHOT,
{'entity_id': 'media_player.zone_1'},
blocking=True)
# self.hass.block_till_done()
# Changing media player to new state
self.media_player.set_volume_level(1)
self.media_player.select_source('two')
self.media_player.mute_volume(False)
self.media_player.turn_off()
# Checking that values were indeed changed
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_OFF, self.media_player.state)
self.assertEqual(1.0, self.media_player.volume_level, 0.0001)
self.assertFalse(self.media_player.is_volume_muted)
self.assertEqual('two', self.media_player.source)
# Restoring wrong media player to its previous state
# Nothing should be done
self.hass.services.call(DOMAIN, SERVICE_RESTORE,
{'entity_id': 'not_existing'},
blocking=True)
# self.hass.block_till_done()
# Checking that values were not (!) restored
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_OFF, self.media_player.state)
self.assertEqual(1.0, self.media_player.volume_level, 0.0001)
self.assertFalse(self.media_player.is_volume_muted)
self.assertEqual('two', self.media_player.source)
# Restoring media player to its previous state
self.hass.services.call(DOMAIN, SERVICE_RESTORE,
{'entity_id': 'media_player.zone_1'},
blocking=True)
self.hass.block_till_done()
# Checking that values were restored
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_ON, self.media_player.state)
self.assertEqual(0.0, self.media_player.volume_level, 0.0001)
self.assertTrue(self.media_player.is_volume_muted)
self.assertEqual('one', self.media_player.source)
def test_service_calls_without_entity_id(self):
"""Test snapshot save/restore service calls."""
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_ON, self.media_player.state)
self.assertEqual(0.0, self.media_player.volume_level, 0.0001)
self.assertTrue(self.media_player.is_volume_muted)
self.assertEqual('one', self.media_player.source)
# Restoring media player
# since there is no snapshot, nothing should be done
self.hass.services.call(DOMAIN, SERVICE_RESTORE, blocking=True)
self.hass.block_till_done()
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_ON, self.media_player.state)
self.assertEqual(0.0, self.media_player.volume_level, 0.0001)
self.assertTrue(self.media_player.is_volume_muted)
self.assertEqual('one', self.media_player.source)
# Saving default values
self.hass.services.call(DOMAIN, SERVICE_SNAPSHOT, blocking=True)
self.hass.block_till_done()
# Changing media player to new state
self.media_player.set_volume_level(1)
self.media_player.select_source('two')
self.media_player.mute_volume(False)
self.media_player.turn_off()
# Checking that values were indeed changed
self.media_player.update()
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_OFF, self.media_player.state)
self.assertEqual(1.0, self.media_player.volume_level, 0.0001)
self.assertFalse(self.media_player.is_volume_muted)
self.assertEqual('two', self.media_player.source)
# Restoring media player to its previous state
self.hass.services.call(DOMAIN, SERVICE_RESTORE, blocking=True)
self.hass.block_till_done()
# Checking that values were restored
self.assertEqual('Zone name', self.media_player.name)
self.assertEqual(STATE_ON, self.media_player.state)
self.assertEqual(0.0, self.media_player.volume_level, 0.0001)
self.assertTrue(self.media_player.is_volume_muted)
self.assertEqual('one', self.media_player.source)
def test_update(self):
"""Test updating values from monoprice."""