jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 09:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Qwikswitch sensors (#13622)

Browse Source
This commit is contained in:
Johann Kellerman 2018-04-08 21:59:19 +02:00 committed by GitHub
parent ef16c53e46
commit b01dceaff2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 182 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.coveragerc
View File

@ -190,8 +190,8 @@ omit =
homeassistant/components/pilight.py
homeassistant/components/*/pilight.py
homeassistant/components/qwikswitch.py
homeassistant/components/*/qwikswitch.py
homeassistant/components/switch/qwikswitch.py
homeassistant/components/light/qwikswitch.py
homeassistant/components/rachio.py
homeassistant/components/*/rachio.py

4
homeassistant/components/light/qwikswitch.py
View File

@ -27,9 +27,9 @@ class QSLight(QSToggleEntity, Light):
@property
def brightness(self):
"""Return the brightness of this light (0-255)."""
return self._qsusb[self.qsid, 1] if self._dim else None
return self.device.value if self.device.is_dimmer else None
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_BRIGHTNESS if self._dim else 0
return SUPPORT_BRIGHTNESS if self.device.is_dimmer else 0

116
homeassistant/components/qwikswitch.py
View File

@ -18,7 +18,7 @@ from homeassistant.helpers.entity import Entity
from homeassistant.components.light import ATTR_BRIGHTNESS
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['pyqwikswitch==0.6']
REQUIREMENTS = ['pyqwikswitch==0.7']
_LOGGER = logging.getLogger(__name__)
@ -34,17 +34,48 @@ CONFIG_SCHEMA = vol.Schema({
vol.Coerce(str),
vol.Optional(CONF_DIMMER_ADJUST, default=1): CV_DIM_VALUE,
vol.Optional(CONF_BUTTON_EVENTS, default=[]): cv.ensure_list_csv,
vol.Optional(CONF_SENSORS, default={}): vol.Schema({cv.slug: str}),
vol.Optional(CONF_SENSORS, default=[]): vol.All(
cv.ensure_list, [vol.Schema({
vol.Required('id'): str,
vol.Optional('channel', default=1): int,
vol.Required('name'): str,
vol.Required('type'): str,
})]),
vol.Optional(CONF_SWITCHES, default=[]): vol.All(
cv.ensure_list, [str])
})}, extra=vol.ALLOW_EXTRA)
class QSToggleEntity(Entity):
"""Representation of a Qwikswitch Entity.
class QSEntity(Entity):
"""Qwikswitch Entity base."""
Implement base QS methods. Modeled around HA ToggleEntity[1] & should only
be used in a class that extends both QSToggleEntity *and* ToggleEntity.
def __init__(self, qsid, name):
"""Initialize the QSEntity."""
self._name = name
self.qsid = qsid
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def poll(self):
"""QS sensors gets packets in update_packet."""
return False
def update_packet(self, packet):
"""Receive update packet from QSUSB. Match dispather_send signature."""
self.async_schedule_update_ha_state()
async def async_added_to_hass(self):
"""Listen for updates from QSUSb via dispatcher."""
self.hass.helpers.dispatcher.async_dispatcher_connect(
self.qsid, self.update_packet)
class QSToggleEntity(QSEntity):
"""Representation of a Qwikswitch Toggle Entity.
Implemented:
- QSLight extends QSToggleEntity and Light[2] (ToggleEntity[1])
@ -57,52 +88,28 @@ class QSToggleEntity(Entity):
def __init__(self, qsid, qsusb):
"""Initialize the ToggleEntity."""
from pyqwikswitch import (QS_NAME, QSDATA, QS_TYPE, QSType)
self.qsid = qsid
self._qsusb = qsusb.devices
dev = qsusb.devices[qsid]
self._dim = dev[QS_TYPE] == QSType.dimmer
self._name = dev[QSDATA][QS_NAME]
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def name(self):
"""Return the name of the light."""
return self._name
self.device = qsusb.devices[qsid]
super().__init__(qsid, self.device.name)
@property
def is_on(self):
"""Check if device is on (non-zero)."""
return self._qsusb[self.qsid, 1] > 0
return self.device.value > 0
async def async_turn_on(self, **kwargs):
"""Turn the device on."""
new = kwargs.get(ATTR_BRIGHTNESS, 255)
self._qsusb.set_value(self.qsid, new)
self.hass.data[DOMAIN].devices.set_value(self.qsid, new)
async def async_turn_off(self, **_):
"""Turn the device off."""
self._qsusb.set_value(self.qsid, 0)
def _update(self, _packet=None):
"""Schedule an update - match dispather_send signature."""
self.async_schedule_update_ha_state()
async def async_added_to_hass(self):
"""Listen for updates from QSUSb via dispatcher."""
self.hass.helpers.dispatcher.async_dispatcher_connect(
self.qsid, self._update)
self.hass.data[DOMAIN].devices.set_value(self.qsid, 0)
async def async_setup(hass, config):
"""Qwiskswitch component setup."""
from pyqwikswitch.async_ import QSUsb
from pyqwikswitch import (
CMD_BUTTONS, QS_CMD, QS_ID, QS_TYPE, QSType)
from pyqwikswitch import CMD_BUTTONS, QS_CMD, QS_ID, QSType
# Add cmd's to in /&listen packets will fire events
# By default only buttons of type [TOGGLE,SCENE EXE,LEVEL]
@ -112,8 +119,8 @@ async def async_setup(hass, config):
url = config[DOMAIN][CONF_URL]
dimmer_adjust = config[DOMAIN][CONF_DIMMER_ADJUST]
sensors = config[DOMAIN]['sensors']
switches = config[DOMAIN]['switches']
sensors = config[DOMAIN][CONF_SENSORS]
switches = config[DOMAIN][CONF_SWITCHES]
def callback_value_changed(_qsd, qsid, _val):
"""Update entity values based on device change."""
@ -131,17 +138,17 @@ async def async_setup(hass, config):
hass.data[DOMAIN] = qsusb
_new = {'switch': [], 'light': [], 'sensor': sensors}
for _id, item in qsusb.devices:
if _id in switches:
if item[QS_TYPE] != QSType.relay:
for qsid, dev in qsusb.devices.items():
if qsid in switches:
if dev.qstype != QSType.relay:
_LOGGER.warning(
"You specified a switch that is not a relay %s", _id)
"You specified a switch that is not a relay %s", qsid)
continue
_new['switch'].append(_id)
elif item[QS_TYPE] in [QSType.relay, QSType.dimmer]:
_new['light'].append(_id)
_new['switch'].append(qsid)
elif dev.qstype in (QSType.relay, QSType.dimmer):
_new['light'].append(qsid)
else:
_LOGGER.warning("Ignored unknown QSUSB device: %s", item)
_LOGGER.warning("Ignored unknown QSUSB device: %s", dev)
continue
# Load platforms
@ -149,24 +156,21 @@ async def async_setup(hass, config):
if comp_conf:
load_platform(hass, comp_name, DOMAIN, {DOMAIN: comp_conf}, config)
def callback_qs_listen(item):
def callback_qs_listen(qspacket):
"""Typically a button press or update signal."""
# If button pressed, fire a hass event
if QS_ID in item:
if item.get(QS_CMD, '') in cmd_buttons:
if QS_ID in qspacket:
if qspacket.get(QS_CMD, '') in cmd_buttons:
hass.bus.async_fire(
'qwikswitch.button.{}'.format(item[QS_ID]), item)
'qwikswitch.button.{}'.format(qspacket[QS_ID]), qspacket)
return
# Private method due to bad __iter__ design in qsusb
# qsusb.devices returns a list of tuples
if item[QS_ID] not in \
qsusb.devices._data: # pylint: disable=protected-access
if qspacket[QS_ID] not in qsusb.devices:
# Not a standard device in, component can handle packet
# i.e. sensors
_LOGGER.debug("Dispatch %s ((%s))", item[QS_ID], item)
_LOGGER.debug("Dispatch %s ((%s))", qspacket[QS_ID], qspacket)
hass.helpers.dispatcher.async_dispatcher_send(
item[QS_ID], item)
qspacket[QS_ID], qspacket)
# Update all ha_objects
hass.async_add_job(qsusb.update_from_devices)

54
homeassistant/components/sensor/qwikswitch.py
View File

@ -6,8 +6,7 @@ https://home-assistant.io/components/sensor.qwikswitch/
"""
import logging
from homeassistant.components.qwikswitch import DOMAIN as QWIKSWITCH
from homeassistant.helpers.entity import Entity
from homeassistant.components.qwikswitch import DOMAIN as QWIKSWITCH, QSEntity
DEPENDENCIES = [QWIKSWITCH]
@ -15,55 +14,48 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, _, add_devices, discovery_info=None):
"""Add lights from the main Qwikswitch component."""
"""Add sensor from the main Qwikswitch component."""
if discovery_info is None:
return
qsusb = hass.data[QWIKSWITCH]
_LOGGER.debug("Setup qwikswitch.sensor %s, %s", qsusb, discovery_info)
devs = [QSSensor(name, qsid)
for name, qsid in discovery_info[QWIKSWITCH].items()]
devs = [QSSensor(sensor) for sensor in discovery_info[QWIKSWITCH]]
add_devices(devs)
class QSSensor(Entity):
class QSSensor(QSEntity):
"""Sensor based on a Qwikswitch relay/dimmer module."""
_val = {}
_val = None
def __init__(self, sensor_name, sensor_id):
def __init__(self, sensor):
"""Initialize the sensor."""
self._name = sensor_name
self.qsid = sensor_id
from pyqwikswitch import SENSORS
super().__init__(sensor['id'], sensor['name'])
self.channel = sensor['channel']
self.sensor_type = sensor['type']
self._decode, self.unit = SENSORS[self.sensor_type]
if isinstance(self.unit, type):
self.unit = "{}:{}".format(self.sensor_type, self.channel)
def update_packet(self, packet):
"""Receive update packet from QSUSB."""
_LOGGER.debug("Update %s (%s): %s", self.entity_id, self.qsid, packet)
self._val = packet
self.async_schedule_update_ha_state()
val = self._decode(packet.get('data'), channel=self.channel)
_LOGGER.debug("Update %s (%s) decoded as %s: %s: %s",
self.entity_id, self.qsid, val, self.channel, packet)
if val is not None:
self._val = val
self.async_schedule_update_ha_state()
@property
def state(self):
"""Return the value of the sensor."""
return self._val.get('data', 0)
@property
def device_state_attributes(self):
"""Return the state attributes of the sensor."""
return self._val
return str(self._val)
@property
def unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return None
@property
def poll(self):
"""QS sensors gets packets in update_packet."""
return False
async def async_added_to_hass(self):
"""Listen for updates from QSUSb via dispatcher."""
# Part of Entity/ToggleEntity
self.hass.helpers.dispatcher.async_dispatcher_connect(
self.qsid, self.update_packet)
return self.unit

2
requirements_all.txt
View File

@ -885,7 +885,7 @@ pyowm==2.8.0
pypollencom==1.1.1
# homeassistant.components.qwikswitch
pyqwikswitch==0.6
pyqwikswitch==0.7
# homeassistant.components.rainbird
pyrainbird==0.1.3

3
requirements_test_all.txt
View File

@ -145,6 +145,9 @@ pymonoprice==0.3
# homeassistant.components.binary_sensor.nx584
pynx584==0.4
# homeassistant.components.qwikswitch
pyqwikswitch==0.7
# homeassistant.components.sensor.darksky
# homeassistant.components.weather.darksky
python-forecastio==1.4.0

1
script/gen_requirements_all.py
View File

@ -73,6 +73,7 @@ TEST_REQUIREMENTS = (
'pylitejet',
'pymonoprice',
'pynx584',
'pyqwikswitch',
'python-forecastio',
'pyunifi',
'pywebpush',

90
tests/components/sensor/test_qwikswitch.py Normal file
View File

@ -0,0 +1,90 @@
"""Test qwikswitch sensors."""
import asyncio
import logging
import pytest
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.components.qwikswitch import DOMAIN as QWIKSWITCH
from homeassistant.bootstrap import async_setup_component
from tests.test_util.aiohttp import mock_aiohttp_client
_LOGGER = logging.getLogger(__name__)
class AiohttpClientMockResponseList(list):
"""List that fires an event on empty pop, for aiohttp Mocker."""
def decode(self, _):
"""Return next item from list."""
try:
res = list.pop(self)
_LOGGER.debug("MockResponseList popped %s: %s", res, self)
return res
except IndexError:
_LOGGER.debug("MockResponseList empty")
return ""
async def wait_till_empty(self, hass):
"""Wait until empty."""
while self:
await asyncio.sleep(1)
await hass.async_block_till_done()
await hass.async_block_till_done()
LISTEN = AiohttpClientMockResponseList()
@pytest.fixture
def aioclient_mock():
"""HTTP client listen and devices."""
devices = """[
{"id":"@000001","name":"Switch 1","type":"rel","val":"OFF",
"time":"1522777506","rssi":"51%"},
{"id":"@000002","name":"Light 2","type":"rel","val":"ON",
"time":"1522777507","rssi":"45%"},
{"id":"@000003","name":"Dim 3","type":"dim","val":"280c00",
"time":"1522777544","rssi":"62%"}]"""
with mock_aiohttp_client() as mock_session:
mock_session.get("http://127.0.0.1:2020/&listen", content=LISTEN)
mock_session.get("http://127.0.0.1:2020/&device", text=devices)
yield mock_session
# @asyncio.coroutine
async def test_sensor_device(hass, aioclient_mock):
"""Test a sensor device."""
config = {
'qwikswitch': {
'sensors': {
'name': 's1',
'id': '@a00001',
'channel': 1,
'type': 'imod',
}
}
}
await async_setup_component(hass, QWIKSWITCH, config)
await hass.async_block_till_done()
state_obj = hass.states.get('sensor.s1')
assert state_obj
assert state_obj.state == 'None'
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
LISTEN.append( # Close
"""{"id":"@a00001","cmd":"","data":"4e0e1601","rssi":"61%"}""")
await hass.async_block_till_done()
state_obj = hass.states.get('sensor.s1')
assert state_obj.state == 'True'
# Causes a 30second delay: can be uncommented when upstream library
# allows cancellation of asyncio.sleep(30) on failed packet ("")
# LISTEN.append( # Open
# """{"id":"@a00001","cmd":"","data":"4e0e1701","rssi":"61%"}""")
# await LISTEN.wait_till_empty(hass)
# state_obj = hass.states.get('sensor.s1')
# assert state_obj.state == 'False'