Add mochad component (#3970)

This commit adds a new component for communicating with mochad[1] a
socket interface for the CM15A and CM19A USB X10 controllers. This
commit leverages the pymochad library to interface with a mochad socket
either on a local or remote machine. Mochad is added as as a generic
platform because it supports multiple different classes of device,
however in this patch only the switch device implemented as a starting
point. Future patches will include other devices types. (although
that's dependent on someone gaining access to those)

[1] https://sourceforge.net/projects/mochad/
This commit is contained in:
Matthew Treinish 2016-10-25 00:49:49 -04:00 committed by Paulus Schoutsen
parent 23f54b07c7
commit 0ff500ca25
7 changed files with 272 additions and 0 deletions

View File

@ -109,6 +109,9 @@ omit =
homeassistant/components/zoneminder.py homeassistant/components/zoneminder.py
homeassistant/components/*/zoneminder.py homeassistant/components/*/zoneminder.py
homeassistant/components/mochad.py
homeassistant/components/*/mochad.py
homeassistant/components/alarm_control_panel/alarmdotcom.py homeassistant/components/alarm_control_panel/alarmdotcom.py
homeassistant/components/alarm_control_panel/concord232.py homeassistant/components/alarm_control_panel/concord232.py
homeassistant/components/alarm_control_panel/nx584.py homeassistant/components/alarm_control_panel/nx584.py

View File

@ -0,0 +1,85 @@
"""
Support for CM15A/CM19A X10 Controller using mochad daemon.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/mochad/
"""
import logging
import voluptuous as vol
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.const import (CONF_HOST, CONF_PORT)
from homeassistant.helpers import config_validation as cv
REQUIREMENTS = ['pymochad==0.1.1']
_LOGGER = logging.getLogger(__name__)
CONTROLLER = None
DOMAIN = 'mochad'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_HOST, default='localhost'): cv.string,
vol.Optional(CONF_PORT, default=1099): cv.port,
})
}, extra=vol.ALLOW_EXTRA)
def setup(hass, config):
"""Setup the mochad platform."""
conf = config[DOMAIN]
host = conf.get(CONF_HOST)
port = conf.get(CONF_PORT)
from pymochad import exceptions
global CONTROLLER
try:
CONTROLLER = MochadCtrl(host, port)
except exceptions.ConfigurationError:
_LOGGER.exception()
return False
def stop_mochad(event):
"""Stop the Mochad service."""
CONTROLLER.disconnect()
def start_mochad(event):
"""Start the Mochad service."""
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_mochad)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_mochad)
return True
class MochadCtrl(object):
"""Mochad controller."""
def __init__(self, host, port):
"""Initialize a PyMochad controller."""
super(MochadCtrl, self).__init__()
self._host = host
self._port = port
from pymochad import controller
self.ctrl = controller.PyMochad(server=self._host, port=self._port)
@property
def host(self):
"""The server where mochad is running."""
return self._host
@property
def port(self):
"""The port mochad is running on."""
return self._port
def disconnect(self):
"""Close the connection to the mochad socket."""
self.ctrl.socket.close()

View File

@ -0,0 +1,81 @@
"""
Contains functionality to use a X10 switch over Mochad.
For more details about this platform, please refer to the documentation at
https://home.assistant.io/components/switch.mochad
"""
import logging
import voluptuous as vol
from homeassistant.components import mochad
from homeassistant.components.switch import SwitchDevice
from homeassistant.const import (CONF_NAME, CONF_PLATFORM)
from homeassistant.helpers import config_validation as cv
DEPENDENCIES = ['mochad']
_LOGGER = logging.getLogger(__name__)
CONF_ADDRESS = 'address'
CONF_DEVICES = 'devices'
PLATFORM_SCHEMA = vol.Schema({
vol.Required(CONF_PLATFORM): mochad.DOMAIN,
CONF_DEVICES: [{
vol.Optional(CONF_NAME): cv.string,
vol.Required(CONF_ADDRESS): cv.x10_address,
vol.Optional('comm_type'): cv.string,
}]
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Setup X10 switches over a mochad controller."""
devs = config.get(CONF_DEVICES)
add_devices([MochadSwitch(
hass, mochad.CONTROLLER.ctrl, dev) for dev in devs])
return True
class MochadSwitch(SwitchDevice):
"""Representation of a X10 switch over Mochad."""
def __init__(self, hass, ctrl, dev):
"""Initialize a Mochad Switch Device."""
from pymochad import device
self._controller = ctrl
self._address = dev[CONF_ADDRESS]
self._name = dev.get(CONF_NAME, 'x10_switch_dev_%s' % self._address)
self._comm_type = dev.get('comm_type', 'pl')
self.device = device.Device(ctrl, self._address,
comm_type=self._comm_type)
self._state = self._get_device_status()
@property
def name(self):
"""Get the name of the switch."""
return self._name
def turn_on(self, **kwargs):
"""Turn the switch on."""
self._state = True
self.device.send_cmd('on')
self._controller.read_data()
def turn_off(self, **kwargs):
"""Turn the switch off."""
self._state = False
self.device.send_cmd('off')
self._controller.read_data()
def _get_device_status(self):
"""Get the status of the switch from mochad."""
status = self.device.get_status().rstrip()
return status == 'on'
@property
def is_on(self):
"""Return true if switch is on."""
return self._state

View File

@ -2,6 +2,7 @@
from collections import OrderedDict from collections import OrderedDict
from datetime import timedelta from datetime import timedelta
import os import os
import re
from urllib.parse import urlparse from urllib.parse import urlparse
from socket import _GLOBAL_DEFAULT_TIMEOUT from socket import _GLOBAL_DEFAULT_TIMEOUT
@ -336,6 +337,14 @@ def url(value: Any) -> str:
raise vol.Invalid('invalid url') raise vol.Invalid('invalid url')
def x10_address(value):
"""Validate an x10 address."""
regex = re.compile(r'([A-Pa-p]{1})(?:[2-9]|1[0-6]?)$')
if not regex.match(value):
raise vol.Invalid('Invalid X10 Address')
return str(value).lower()
def ordered_dict(value_validator, key_validator=match_all): def ordered_dict(value_validator, key_validator=match_all):
"""Validate an ordered dict validator that maintains ordering. """Validate an ordered dict validator that maintains ordering.

View File

@ -370,6 +370,9 @@ pylast==1.6.0
# homeassistant.components.sensor.loopenergy # homeassistant.components.sensor.loopenergy
pyloopenergy==0.0.15 pyloopenergy==0.0.15
# homeassistant.components.mochad
pymochad==0.1.1
# homeassistant.components.device_tracker.netgear # homeassistant.components.device_tracker.netgear
pynetgear==0.3.3 pynetgear==0.3.3

View File

@ -0,0 +1,79 @@
"""The tests for the mochad switch platform."""
import unittest
import unittest.mock as mock
from homeassistant.bootstrap import setup_component
from homeassistant.components import switch
from homeassistant.components.switch import mochad
from tests.common import get_test_home_assistant
class TestMochadSwitchSetup(unittest.TestCase):
"""Test the mochad switch."""
PLATFORM = mochad
COMPONENT = switch
THING = 'switch'
def setUp(self):
"""Setup things to be run when tests are started."""
super(TestMochadSwitchSetup, self).setUp()
self.hass = get_test_home_assistant()
def tearDown(self):
"""Stop everyhing that was started."""
self.hass.stop()
super(TestMochadSwitchSetup, self).tearDown()
@mock.patch('pymochad.controller.PyMochad')
@mock.patch('homeassistant.components.switch.mochad.MochadSwitch')
def test_setup_adds_proper_devices(self, mock_switch, mock_client):
"""Test if setup adds devices."""
good_config = {
'mochad': {},
'switch': {
'platform': 'mochad',
'devices': [
{
'name': 'Switch1',
'address': 'a1',
},
],
}
}
self.assertTrue(setup_component(self.hass, switch.DOMAIN, good_config))
class TestMochadSwitch(unittest.TestCase):
"""Test for mochad switch platform."""
def setUp(self):
"""Setup things to be run when tests are started."""
super(TestMochadSwitch, self).setUp()
self.hass = get_test_home_assistant()
controller_mock = mock.MagicMock()
device_patch = mock.patch('pymochad.device.Device')
device_patch.start()
self.addCleanup(device_patch.stop)
dev_dict = {'address': 'a1', 'name': 'fake_switch'}
self.switch = mochad.MochadSwitch(self.hass, controller_mock,
dev_dict)
def teardown_method(self, method):
"""Stop everything that was started."""
self.hass.stop()
def test_name(self):
"""Test the name."""
self.assertEqual('fake_switch', self.switch.name)
def test_turn_on(self):
"""Test turn_on."""
self.switch.turn_on()
self.switch.device.send_cmd.assert_called_once_with('on')
def test_turn_off(self):
"""Test turn_off."""
self.switch.turn_off()
self.switch.device.send_cmd.assert_called_once_with('off')

View File

@ -300,6 +300,18 @@ def test_temperature_unit():
schema('F') schema('F')
def test_x10_address():
"""Test x10 addr validator."""
schema = vol.Schema(cv.x10_address)
with pytest.raises(vol.Invalid):
schema('Q1')
schema('q55')
schema('garbage_addr')
schema('a1')
schema('C11')
def test_template(): def test_template():
"""Test template validator.""" """Test template validator."""
schema = vol.Schema(cv.template) schema = vol.Schema(cv.template)