Merge pull request #788 from happyleavesaoc/command_switch_state

add statecmd to command_switch
This commit is contained in:
happyleavesaoc 2015-12-31 18:43:02 -05:00
commit 9e83a80215
2 changed files with 215 additions and 8 deletions

View file

@ -10,6 +10,8 @@ import logging
import subprocess
from homeassistant.components.switch import SwitchDevice
from homeassistant.const import CONF_VALUE_TEMPLATE
from homeassistant.util import template
_LOGGER = logging.getLogger(__name__)
@ -24,20 +26,30 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
for dev_name, properties in switches.items():
devices.append(
CommandSwitch(
hass,
properties.get('name', dev_name),
properties.get('oncmd', 'true'),
properties.get('offcmd', 'true')))
properties.get('offcmd', 'true'),
properties.get('statecmd', False),
properties.get(CONF_VALUE_TEMPLATE, False)))
add_devices_callback(devices)
class CommandSwitch(SwitchDevice):
""" Represents a switch that can be togggled using shell commands. """
def __init__(self, name, command_on, command_off):
# pylint: disable=too-many-arguments
def __init__(self, hass, name, command_on, command_off,
command_state, value_template):
self._hass = hass
self._name = name
self._state = False
self._command_on = command_on
self._command_off = command_off
self._command_state = command_state
self._value_template = value_template
@staticmethod
def _switch(command):
@ -51,10 +63,27 @@ class CommandSwitch(SwitchDevice):
return success
@staticmethod
def _query_state_value(command):
""" Execute state command for return value. """
_LOGGER.info('Running state command: %s', command)
try:
return_value = subprocess.check_output(command, shell=True)
return return_value.strip().decode('utf-8')
except subprocess.CalledProcessError:
_LOGGER.error('Command failed: %s', command)
@staticmethod
def _query_state_code(command):
""" Execute state command for return code. """
_LOGGER.info('Running state command: %s', command)
return subprocess.call(command, shell=True) == 0
@property
def should_poll(self):
""" No polling needed. """
return False
""" Only poll if we have statecmd. """
return self._command_state is not None
@property
def name(self):
@ -66,14 +95,34 @@ class CommandSwitch(SwitchDevice):
""" True if device is on. """
return self._state
def _query_state(self):
""" Query for state. """
if not self._command_state:
_LOGGER.error('No state command specified')
return
if self._value_template:
return CommandSwitch._query_state_value(self._command_state)
return CommandSwitch._query_state_code(self._command_state)
def update(self):
""" Update device state. """
if self._command_state:
payload = str(self._query_state())
if self._value_template:
payload = template.render_with_possible_json_value(
self._hass, self._value_template, payload)
self._state = (payload.lower() == "true")
def turn_on(self, **kwargs):
""" Turn the device on. """
if CommandSwitch._switch(self._command_on):
if (CommandSwitch._switch(self._command_on) and
not self._command_state):
self._state = True
self.update_ha_state()
self.update_ha_state()
def turn_off(self, **kwargs):
""" Turn the device off. """
if CommandSwitch._switch(self._command_off):
if (CommandSwitch._switch(self._command_off) and
not self._command_state):
self._state = False
self.update_ha_state()
self.update_ha_state()

View file

@ -0,0 +1,158 @@
"""
tests.components.switch.test_command_switch
~~~~~~~~~~~~~~~~~~~~~~~~
Tests command switch.
"""
import json
import os
import tempfile
import unittest
from homeassistant import core
from homeassistant.const import STATE_ON, STATE_OFF
import homeassistant.components.switch as switch
class TestCommandSwitch(unittest.TestCase):
""" Test the command switch. """
def setUp(self): # pylint: disable=invalid-name
self.hass = core.HomeAssistant()
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_state_none(self):
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, 'switch_status')
test_switch = {
'oncmd': 'echo 1 > {}'.format(path),
'offcmd': 'echo 0 > {}'.format(path),
}
self.assertTrue(switch.setup(self.hass, {
'switch': {
'platform': 'command_switch',
'switches': {
'test': test_switch
}
}
}))
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
switch.turn_on(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_ON, state.state)
switch.turn_off(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
def test_state_value(self):
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, 'switch_status')
test_switch = {
'statecmd': 'cat {}'.format(path),
'oncmd': 'echo 1 > {}'.format(path),
'offcmd': 'echo 0 > {}'.format(path),
'value_template': '{{ value=="1" }}'
}
self.assertTrue(switch.setup(self.hass, {
'switch': {
'platform': 'command_switch',
'switches': {
'test': test_switch
}
}
}))
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
switch.turn_on(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_ON, state.state)
switch.turn_off(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
def test_state_json_value(self):
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, 'switch_status')
oncmd = json.dumps({'status': 'ok'})
offcmd = json.dumps({'status': 'nope'})
test_switch = {
'statecmd': 'cat {}'.format(path),
'oncmd': 'echo \'{}\' > {}'.format(oncmd, path),
'offcmd': 'echo \'{}\' > {}'.format(offcmd, path),
'value_template': '{{ value_json.status=="ok" }}'
}
self.assertTrue(switch.setup(self.hass, {
'switch': {
'platform': 'command_switch',
'switches': {
'test': test_switch
}
}
}))
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
switch.turn_on(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_ON, state.state)
switch.turn_off(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
def test_state_code(self):
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, 'switch_status')
test_switch = {
'statecmd': 'cat {}'.format(path),
'oncmd': 'echo 1 > {}'.format(path),
'offcmd': 'echo 0 > {}'.format(path),
}
self.assertTrue(switch.setup(self.hass, {
'switch': {
'platform': 'command_switch',
'switches': {
'test': test_switch
}
}
}))
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_OFF, state.state)
switch.turn_on(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_ON, state.state)
switch.turn_off(self.hass, 'switch.test')
self.hass.pool.block_till_done()
state = self.hass.states.get('switch.test')
self.assertEqual(STATE_ON, state.state)