evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Stable and asynchronous KNX library. (#8725)

Browse source 
* First draft of XKNX module for Home-Assistant

* XKNX does now take path of xknx.yaml as parameter

* small fix, telegram_received_callback has different signature

* changed method of registering callbacks of devices

* removed non async command lines from xknx

* telegram_received_cb not needed within HASS module

* updated requirements

* Configuration if XKNX should connect via Routing or Tunneling

* bumping version to 0.6.1

* small fix within xknx plugin

* bumped version

* XKNX-Switches are now BinarySensors and Logic from Sensor was moved to BinarySensor

* renamed Outlet to Switch

* pylint

* configuration of KNX lights via HASS config, yay!

* changed name of attribute

* Added configuration for xknx to switch component

* added support for sensors within hass configuration

* added support for climate within hass configuration

* Thermostat -> Climate

* added configuration support for binary_sensors

* renamed Shutter to Cover

* added configuration support for cover

* restructured file structure according to HASS requirements

* pylint

* pylint

* pylint

* pylint

* pylint

* pylint

* updated version

* pylint

* pylint

* pylint

* added setpoint support for climate devices

* devices are now in a different module

* more asyncio :-)

* pydocstyle

* pydocstyle

* added actions to binary_sensor

* allow more than one automation

* readded requirement

* Modifications suggested by hound

* Modifications suggested by hound

* Modifications suggested by hound

* Modifications suggested by hound

* xknx now imported as local import

* hound *sigh*

* lint

* 'fixed' coverage.

* next try for getting gen_requirements_all.py working

* removed blank line

* XKNX 0.7.1 with logging functionality, replaced some print() calls with _LOGGER

* updated requirements_all.txt

* Fixes issue https://github.com/XKNX/xknx/issues/51

* https://github.com/XKNX/xknx/issues/52 added raw access to KNX bus from HASS component.

* bumped version - 0.7.3 contains some bugfixes

* bumped version - 0.7.3 contains some bugfixes

* setting setpoint within climate device has to be async

* bumped version to 0.7.4

* bumped version

* https://github.com/XKNX/xknx/issues/48 Adding HVAC support.

* pylint suggestions

* Made target temperature and set point required attributes

* renamed value_type to type within sensor configuration

* Issue https://github.com/XKNX/xknx/issues/52 : added filter functionality for not flooding the event bus.

* suggestions by pylint

* Added notify support for knx platform.

* logging error if discovery_info is None.

* review suggestions by @armills

* line too long

* Using discovery_info to notifiy component which devices should be added.

* moved XKNX automation to main level.

* renamed xknx component to knx.

* reverted change within .coveragerc

* changed dependency

* updated docstrings.

* updated version of xknx within requirements_all.txt

* moved requirement to correct position

* renamed configuration attribute

* added @callback-decorator and async_prefix.

* added @callback decorator and async_ prefix to register_callbacks functions

* fixed typo

* pylint suggestions

* added angle position and invert_position and invert_angle to cover.knx

* typo

* bumped version within requirements_all.txt

* bumped version

* Added support for HVAC controller status
This commit is contained in:
Julius Mittenzwei 2017-09-07 09:11:55 +02:00 • committed by Paulus Schoutsen
parent 9a7089bad3
commit 77d0ad1797
10 changed files with 1041 additions and 887 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

142
homeassistant/components/binary_sensor/knx.py
View file

@ -1,21 +1,145 @@
"""
Contains functionality to use a KNX group address as a binary.
Support for KNX/IP binary sensors.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.knx/
"""
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.knx import (KNXConfig, KNXGroupAddress)
import asyncio
import voluptuous as vol
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES, \
KNXAutomation
from homeassistant.components.binary_sensor import PLATFORM_SCHEMA, \
BinarySensorDevice
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
CONF_ADDRESS = 'address'
CONF_DEVICE_CLASS = 'device_class'
CONF_SIGNIFICANT_BIT = 'significant_bit'
CONF_DEFAULT_SIGNIFICANT_BIT = 1
CONF_AUTOMATION = 'automation'
CONF_HOOK = 'hook'
CONF_DEFAULT_HOOK = 'on'
CONF_COUNTER = 'counter'
CONF_DEFAULT_COUNTER = 1
CONF_ACTION = 'action'
CONF__ACTION = 'turn_off_action'
DEFAULT_NAME = 'KNX Binary Sensor'
DEPENDENCIES = ['knx']
AUTOMATION_SCHEMA = vol.Schema({
vol.Optional(CONF_HOOK, default=CONF_DEFAULT_HOOK): cv.string,
vol.Optional(CONF_COUNTER, default=CONF_DEFAULT_COUNTER): cv.port,
vol.Required(CONF_ACTION, default=None): cv.SCRIPT_SCHEMA
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the KNX binary sensor platform."""
add_devices([KNXSwitch(hass, KNXConfig(config))])
AUTOMATIONS_SCHEMA = vol.All(
cv.ensure_list,
[AUTOMATION_SCHEMA]
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_DEVICE_CLASS): cv.string,
vol.Optional(CONF_SIGNIFICANT_BIT, default=CONF_DEFAULT_SIGNIFICANT_BIT):
cv.positive_int,
vol.Optional(CONF_AUTOMATION, default=None): AUTOMATIONS_SCHEMA,
})
class KNXSwitch(KNXGroupAddress, BinarySensorDevice):
"""Representation of a KNX binary sensor device."""
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up binary sensor(s) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
pass
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up binary sensors for KNX platform configured via xknx.yaml."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXBinarySensor(hass, device))
add_devices(entities)
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up binary senor for KNX platform configured within plattform."""
name = config.get(CONF_NAME)
import xknx
binary_sensor = xknx.devices.BinarySensor(
hass.data[DATA_KNX].xknx,
name=name,
group_address=config.get(CONF_ADDRESS),
device_class=config.get(CONF_DEVICE_CLASS),
significant_bit=config.get(CONF_SIGNIFICANT_BIT))
hass.data[DATA_KNX].xknx.devices.add(binary_sensor)
entity = KNXBinarySensor(hass, binary_sensor)
automations = config.get(CONF_AUTOMATION)
if automations is not None:
for automation in automations:
counter = automation.get(CONF_COUNTER)
hook = automation.get(CONF_HOOK)
action = automation.get(CONF_ACTION)
entity.automations.append(KNXAutomation(
hass=hass, device=binary_sensor, hook=hook,
action=action, counter=counter))
add_devices([entity])
class KNXBinarySensor(BinarySensorDevice):
"""Representation of a KNX binary sensor."""
def __init__(self, hass, device):
"""Initialization of KNXBinarySensor."""
self.device = device
self.hass = hass
self.async_register_callbacks()
self.automations = []
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
@property
def should_poll(self):
"""No polling needed within KNX."""
return False
@property
def device_class(self):
"""Return the class of this sensor."""
return self.device.device_class
@property
def is_on(self):
"""Return true if the binary sensor is on."""
return self.device.is_on()

182
homeassistant/components/climate/knx.py
View file

@ -1,68 +1,136 @@
"""
Support for KNX thermostats.
Support for KNX/IP climate devices.
For more details about this platform, please refer to the documentation
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/climate.knx/
"""
import logging
import asyncio
import voluptuous as vol
from homeassistant.components.climate import (ClimateDevice, PLATFORM_SCHEMA)
from homeassistant.components.knx import (KNXConfig, KNXMultiAddressDevice)
from homeassistant.const import (CONF_NAME, TEMP_CELSIUS, ATTR_TEMPERATURE)
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.components.climate import PLATFORM_SCHEMA, ClimateDevice
from homeassistant.const import CONF_NAME, TEMP_CELSIUS, ATTR_TEMPERATURE
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
CONF_ADDRESS = 'address'
CONF_SETPOINT_ADDRESS = 'setpoint_address'
CONF_TEMPERATURE_ADDRESS = 'temperature_address'
CONF_TARGET_TEMPERATURE_ADDRESS = 'target_temperature_address'
CONF_OPERATION_MODE_ADDRESS = 'operation_mode_address'
CONF_OPERATION_MODE_STATE_ADDRESS = 'operation_mode_state_address'
CONF_CONTROLLER_STATUS_ADDRESS = 'controller_status_address'
CONF_CONTROLLER_STATUS_STATE_ADDRESS = 'controller_status_state_address'
CONF_OPERATION_MODE_FROST_PROTECTION_ADDRESS = \
'operation_mode_frost_protection_address'
CONF_OPERATION_MODE_NIGHT_ADDRESS = 'operation_mode_night_address'
CONF_OPERATION_MODE_COMFORT_ADDRESS = 'operation_mode_comfort_address'
DEFAULT_NAME = 'KNX Thermostat'
DEFAULT_NAME = 'KNX Climate'
DEPENDENCIES = ['knx']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_SETPOINT_ADDRESS): cv.string,
vol.Required(CONF_TEMPERATURE_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_TARGET_TEMPERATURE_ADDRESS): cv.string,
vol.Optional(CONF_OPERATION_MODE_ADDRESS): cv.string,
vol.Optional(CONF_OPERATION_MODE_STATE_ADDRESS): cv.string,
vol.Optional(CONF_CONTROLLER_STATUS_ADDRESS): cv.string,
vol.Optional(CONF_CONTROLLER_STATUS_STATE_ADDRESS): cv.string,
vol.Optional(CONF_OPERATION_MODE_FROST_PROTECTION_ADDRESS): cv.string,
vol.Optional(CONF_OPERATION_MODE_NIGHT_ADDRESS): cv.string,
vol.Optional(CONF_OPERATION_MODE_COMFORT_ADDRESS): cv.string,
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Create and add an entity based on the configuration."""
add_devices([KNXThermostat(hass, KNXConfig(config))])
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up climate(s) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
class KNXThermostat(KNXMultiAddressDevice, ClimateDevice):
"""Representation of a KNX thermostat.
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up climates for KNX platform configured within plattform."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXClimate(hass, device))
add_devices(entities)
A KNX thermostat will has the following parameters:
- temperature (current temperature)
- setpoint (target temperature in HASS terms)
- operation mode selection (comfort/night/frost protection)
This version supports only polling. Messages from the KNX bus do not
automatically update the state of the thermostat (to be implemented
in future releases)
"""
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up climate for KNX platform configured within plattform."""
import xknx
climate = xknx.devices.Climate(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address_temperature=config.get(
CONF_TEMPERATURE_ADDRESS),
group_address_target_temperature=config.get(
CONF_TARGET_TEMPERATURE_ADDRESS),
group_address_setpoint=config.get(
CONF_SETPOINT_ADDRESS),
group_address_operation_mode=config.get(
CONF_OPERATION_MODE_ADDRESS),
group_address_operation_mode_state=config.get(
CONF_OPERATION_MODE_STATE_ADDRESS),
group_address_controller_status=config.get(
CONF_CONTROLLER_STATUS_ADDRESS),
group_address_controller_status_state=config.get(
CONF_CONTROLLER_STATUS_STATE_ADDRESS),
group_address_operation_mode_protection=config.get(
CONF_OPERATION_MODE_FROST_PROTECTION_ADDRESS),
group_address_operation_mode_night=config.get(
CONF_OPERATION_MODE_NIGHT_ADDRESS),
group_address_operation_mode_comfort=config.get(
CONF_OPERATION_MODE_COMFORT_ADDRESS))
hass.data[DATA_KNX].xknx.devices.add(climate)
add_devices([KNXClimate(hass, climate)])
def __init__(self, hass, config):
"""Initialize the thermostat based on the given configuration."""
KNXMultiAddressDevice.__init__(
self, hass, config, ['temperature', 'setpoint'], ['mode'])
self._unit_of_measurement = TEMP_CELSIUS # KNX always used celsius
class KNXClimate(ClimateDevice):
"""Representation of a KNX climate."""
def __init__(self, hass, device):
"""Initialization of KNXClimate."""
self.device = device
self.hass = hass
self.async_register_callbacks()
self._unit_of_measurement = TEMP_CELSIUS
self._away = False # not yet supported
self._is_fan_on = False # not yet supported
self._current_temp = None
self._target_temp = None
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
@property
def should_poll(self):
"""Return the polling state, is needed for the KNX thermostat."""
return True
"""No polling needed within KNX."""
return False
@property
def temperature_unit(self):
@ -72,32 +140,42 @@ class KNXThermostat(KNXMultiAddressDevice, ClimateDevice):
@property
def current_temperature(self):
"""Return the current temperature."""
return self._current_temp
return self.device.temperature
@property
def target_temperature(self):
"""Return the temperature we try to reach."""
return self._target_temp
if self.device.supports_target_temperature:
return self.device.target_temperature
return None
def set_temperature(self, **kwargs):
@asyncio.coroutine
def async_set_temperature(self, **kwargs):
"""Set new target temperature."""
temperature = kwargs.get(ATTR_TEMPERATURE)
if temperature is None:
return
from knxip.conversion import float_to_knx2
if self.device.supports_target_temperature:
yield from self.device.set_target_temperature(temperature)
self.set_value('setpoint', float_to_knx2(temperature))
_LOGGER.debug("Set target temperature to %s", temperature)
@property
def current_operation(self):
"""Return current operation ie. heat, cool, idle."""
if self.device.supports_operation_mode:
return self.device.operation_mode.value
return None
def set_operation_mode(self, operation_mode):
@property
def operation_list(self):
"""Return the list of available operation modes."""
return [operation_mode.value for
operation_mode in
self.device.get_supported_operation_modes()]
@asyncio.coroutine
def async_set_operation_mode(self, operation_mode):
"""Set operation mode."""
raise NotImplementedError()
def update(self):
"""Update KNX climate."""
from knxip.conversion import knx2_to_float
super().update()
self._current_temp = knx2_to_float(self.value('temperature'))
self._target_temp = knx2_to_float(self.value('setpoint'))
if self.device.supports_operation_mode:
from xknx.knx import HVACOperationMode
knx_operation_mode = HVACOperationMode(operation_mode)
yield from self.device.set_operation_mode(knx_operation_mode)

320
homeassistant/components/cover/knx.py
View file

@ -1,185 +1,239 @@
"""
Support for KNX covers.
Support for KNX/IP covers.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/cover.knx/
"""
import logging
import asyncio
import voluptuous as vol
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.helpers.event import async_track_utc_time_change
from homeassistant.components.cover import (
CoverDevice, PLATFORM_SCHEMA, ATTR_POSITION, DEVICE_CLASSES_SCHEMA,
SUPPORT_OPEN, SUPPORT_CLOSE, SUPPORT_SET_POSITION, SUPPORT_STOP,
SUPPORT_SET_TILT_POSITION
)
from homeassistant.components.knx import (KNXConfig, KNXMultiAddressDevice)
from homeassistant.const import (CONF_NAME, CONF_DEVICE_CLASS)
CoverDevice, PLATFORM_SCHEMA, SUPPORT_OPEN, SUPPORT_CLOSE,
SUPPORT_SET_POSITION, SUPPORT_STOP, SUPPORT_SET_TILT_POSITION,
ATTR_POSITION, ATTR_TILT_POSITION)
from homeassistant.core import callback
from homeassistant.const import CONF_NAME
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
CONF_GETPOSITION_ADDRESS = 'getposition_address'
CONF_SETPOSITION_ADDRESS = 'setposition_address'
CONF_GETANGLE_ADDRESS = 'getangle_address'
CONF_SETANGLE_ADDRESS = 'setangle_address'
CONF_STOP = 'stop_address'
CONF_UPDOWN = 'updown_address'
CONF_MOVE_LONG_ADDRESS = 'move_long_address'
CONF_MOVE_SHORT_ADDRESS = 'move_short_address'
CONF_POSITION_ADDRESS = 'position_address'
CONF_POSITION_STATE_ADDRESS = 'position_state_address'
CONF_ANGLE_ADDRESS = 'angle_address'
CONF_ANGLE_STATE_ADDRESS = 'angle_state_address'
CONF_TRAVELLING_TIME_DOWN = 'travelling_time_down'
CONF_TRAVELLING_TIME_UP = 'travelling_time_up'
CONF_INVERT_POSITION = 'invert_position'
CONF_INVERT_ANGLE = 'invert_angle'
DEFAULT_TRAVEL_TIME = 25
DEFAULT_NAME = 'KNX Cover'
DEPENDENCIES = ['knx']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_UPDOWN): cv.string,
vol.Required(CONF_STOP): cv.string,
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_GETPOSITION_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_SETPOSITION_ADDRESS): cv.string,
vol.Optional(CONF_MOVE_LONG_ADDRESS): cv.string,
vol.Optional(CONF_MOVE_SHORT_ADDRESS): cv.string,
vol.Optional(CONF_POSITION_ADDRESS): cv.string,
vol.Optional(CONF_POSITION_STATE_ADDRESS): cv.string,
vol.Optional(CONF_ANGLE_ADDRESS): cv.string,
vol.Optional(CONF_ANGLE_STATE_ADDRESS): cv.string,
vol.Optional(CONF_TRAVELLING_TIME_DOWN, default=DEFAULT_TRAVEL_TIME):
cv.positive_int,
vol.Optional(CONF_TRAVELLING_TIME_UP, default=DEFAULT_TRAVEL_TIME):
cv.positive_int,
vol.Optional(CONF_INVERT_POSITION, default=False): cv.boolean,
vol.Inclusive(CONF_GETANGLE_ADDRESS, 'angle'): cv.string,
vol.Inclusive(CONF_SETANGLE_ADDRESS, 'angle'): cv.string,
vol.Optional(CONF_INVERT_ANGLE, default=False): cv.boolean,
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Create and add an entity based on the configuration."""
add_devices([KNXCover(hass, KNXConfig(config))])
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up cover(s) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
class KNXCover(KNXMultiAddressDevice, CoverDevice):
"""Representation of a KNX cover. e.g. a rollershutter."""
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up covers for KNX platform configured via xknx.yaml."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXCover(hass, device))
add_devices(entities)
def __init__(self, hass, config):
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up cover for KNX platform configured within plattform."""
import xknx
cover = xknx.devices.Cover(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address_long=config.get(CONF_MOVE_LONG_ADDRESS),
group_address_short=config.get(CONF_MOVE_SHORT_ADDRESS),
group_address_position_state=config.get(
CONF_POSITION_STATE_ADDRESS),
group_address_angle=config.get(CONF_ANGLE_ADDRESS),
group_address_angle_state=config.get(CONF_ANGLE_STATE_ADDRESS),
group_address_position=config.get(CONF_POSITION_ADDRESS),
travel_time_down=config.get(CONF_TRAVELLING_TIME_DOWN),
travel_time_up=config.get(CONF_TRAVELLING_TIME_UP))
invert_position = config.get(CONF_INVERT_POSITION)
invert_angle = config.get(CONF_INVERT_ANGLE)
hass.data[DATA_KNX].xknx.devices.add(cover)
add_devices([KNXCover(hass, cover, invert_position, invert_angle)])
class KNXCover(CoverDevice):
"""Representation of a KNX cover."""
def __init__(self, hass, device, invert_position=False,
invert_angle=False):
"""Initialize the cover."""
KNXMultiAddressDevice.__init__(
self, hass, config,
['updown', 'stop'], # required
optional=['setposition', 'getposition',
'getangle', 'setangle']
)
self._device_class = config.config.get(CONF_DEVICE_CLASS)
self._invert_position = config.config.get(CONF_INVERT_POSITION)
self._invert_angle = config.config.get(CONF_INVERT_ANGLE)
self._hass = hass
self._current_pos = None
self._target_pos = None
self._current_tilt = None
self._target_tilt = None
self._supported_features = SUPPORT_OPEN | SUPPORT_CLOSE | \
SUPPORT_SET_POSITION | SUPPORT_STOP
self.device = device
self.invert_position = invert_position
self.invert_angle = invert_angle
self.hass = hass
self.async_register_callbacks()
# Tilt is only supported, if there is a angle get and set address
if CONF_SETANGLE_ADDRESS in config.config:
_LOGGER.debug("%s: Tilt supported at addresses %s, %s",
self.name, config.config.get(CONF_SETANGLE_ADDRESS),
config.config.get(CONF_GETANGLE_ADDRESS))
self._supported_features = self._supported_features | \
SUPPORT_SET_TILT_POSITION
self._unsubscribe_auto_updater = None
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
@property
def should_poll(self):
"""Polling is needed for the KNX cover."""
return True
"""No polling needed within KNX."""
return False
@property
def supported_features(self):
"""Flag supported features."""
return self._supported_features
supported_features = SUPPORT_OPEN | SUPPORT_CLOSE | \
SUPPORT_SET_POSITION | SUPPORT_STOP
if self.device.supports_angle:
supported_features |= SUPPORT_SET_TILT_POSITION
return supported_features
@property
def current_cover_position(self):
"""Return the current position of the cover."""
return int(self.from_knx_position(
self.device.current_position(),
self.invert_position))
@property
def is_closed(self):
"""Return if the cover is closed."""
if self.current_cover_position is not None:
if self.current_cover_position > 0:
return False
else:
return True
return self.device.is_closed()
@property
def current_cover_position(self):
"""Return current position of cover.
@asyncio.coroutine
def async_close_cover(self, **kwargs):
"""Close the cover."""
if not self.device.is_closed():
yield from self.device.set_down()
self.start_auto_updater()
None is unknown, 0 is closed, 100 is fully open.
"""
return self._current_pos
@asyncio.coroutine
def async_open_cover(self, **kwargs):
"""Open the cover."""
if not self.device.is_open():
yield from self.device.set_up()
self.start_auto_updater()
@property
def target_position(self):
"""Return the position we are trying to reach: 0 - 100."""
return self._target_pos
@asyncio.coroutine
def async_set_cover_position(self, **kwargs):
"""Move the cover to a specific position."""
if ATTR_POSITION in kwargs:
position = kwargs[ATTR_POSITION]
knx_position = self.to_knx_position(position, self.invert_position)
yield from self.device.set_position(knx_position)
self.start_auto_updater()
@asyncio.coroutine
def async_stop_cover(self, **kwargs):
"""Stop the cover."""
yield from self.device.stop()
self.stop_auto_updater()
@property
def current_cover_tilt_position(self):
"""Return current position of cover.
"""Return current tilt position of cover."""
if not self.device.supports_angle:
return None
return int(self.from_knx_position(
self.device.angle,
self.invert_angle))
None is unknown, 0 is closed, 100 is fully open.
"""
return self._current_tilt
@asyncio.coroutine
def async_set_cover_tilt_position(self, **kwargs):
"""Move the cover tilt to a specific position."""
if ATTR_TILT_POSITION in kwargs:
position = kwargs[ATTR_TILT_POSITION]
knx_position = self.to_knx_position(position, self.invert_angle)
yield from self.device.set_angle(knx_position)
@property
def target_tilt(self):
"""Return the tilt angle (in %) we are trying to reach: 0 - 100."""
return self._target_tilt
def start_auto_updater(self):
"""Start the autoupdater to update HASS while cover is moving."""
if self._unsubscribe_auto_updater is None:
self._unsubscribe_auto_updater = async_track_utc_time_change(
self.hass, self.auto_updater_hook)
def set_cover_position(self, **kwargs):
"""Set new target position."""
position = kwargs.get(ATTR_POSITION)
if position is None:
return
def stop_auto_updater(self):
"""Stop the autoupdater."""
if self._unsubscribe_auto_updater is not None:
self._unsubscribe_auto_updater()
self._unsubscribe_auto_updater = None
if self._invert_position:
position = 100-position
@callback
def auto_updater_hook(self, now):
"""Callback for autoupdater."""
# pylint: disable=unused-argument
self.hass.async_add_job(self.async_update_ha_state())
if self.device.position_reached():
self.stop_auto_updater()
self._target_pos = position
self.set_percentage('setposition', position)
_LOGGER.debug("%s: Set target position to %d", self.name, position)
self.hass.add_job(self.device.auto_stop_if_necessary())
def update(self):
"""Update device state."""
super().update()
value = self.get_percentage('getposition')
if value is not None:
self._current_pos = value
if self._invert_position:
self._current_pos = 100-value
_LOGGER.debug("%s: position = %d", self.name, value)
@staticmethod
def from_knx_position(raw, invert):
"""Convert KNX position [0...255] to hass position [100...0]."""
position = round((raw/256)*100)
if not invert:
position = 100 - position
return position
if self._supported_features & SUPPORT_SET_TILT_POSITION:
value = self.get_percentage('getangle')
if value is not None:
self._current_tilt = value
if self._invert_angle:
self._current_tilt = 100-value
_LOGGER.debug("%s: tilt = %d", self.name, value)
def open_cover(self, **kwargs):
"""Open the cover."""
_LOGGER.debug("%s: open: updown = 0", self.name)
self.set_int_value('updown', 0)
def close_cover(self, **kwargs):
"""Close the cover."""
_LOGGER.debug("%s: open: updown = 1", self.name)
self.set_int_value('updown', 1)
def stop_cover(self, **kwargs):
"""Stop the cover movement."""
_LOGGER.debug("%s: stop: stop = 1", self.name)
self.set_int_value('stop', 1)
def set_cover_tilt_position(self, tilt_position, **kwargs):
"""Move the cover til to a specific position."""
if self._invert_angle:
tilt_position = 100-tilt_position
self._target_tilt = round(tilt_position, -1)
self.set_percentage('setangle', tilt_position)
@property
def device_class(self):
"""Return the class of this device, from component DEVICE_CLASSES."""
return self._device_class
@staticmethod
def to_knx_position(value, invert):
"""Convert hass position [100...0] to KNX position [0...255]."""
knx_position = round(value/100*255.4)
if not invert:
knx_position = 255-knx_position
print(value, " -> ", knx_position)
return knx_position

642
homeassistant/components/knx.py
View file

@ -1,495 +1,255 @@
"""
Support for KNX components.
For more details about this component, please refer to the documentation at
Connects to KNX platform.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/knx/
"""
import logging
import os
import asyncio
import voluptuous as vol
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP, CONF_HOST, CONF_PORT)
from homeassistant.helpers.entity import Entity
from homeassistant.config import load_yaml_config_file
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, \
CONF_HOST, CONF_PORT
from homeassistant.helpers.script import Script
REQUIREMENTS = ['knxip==0.5']
DOMAIN = "knx"
DATA_KNX = "data_knx"
CONF_KNX_CONFIG = "config_file"
CONF_KNX_ROUTING = "routing"
CONF_KNX_TUNNELING = "tunneling"
CONF_KNX_LOCAL_IP = "local_ip"
CONF_KNX_FIRE_EVENT = "fire_event"
CONF_KNX_FIRE_EVENT_FILTER = "fire_event_filter"
SERVICE_KNX_SEND = "send"
SERVICE_KNX_ATTR_ADDRESS = "address"
SERVICE_KNX_ATTR_PAYLOAD = "payload"
ATTR_DISCOVER_DEVICES = 'devices'
_LOGGER = logging.getLogger(__name__)
DEFAULT_HOST = '0.0.0.0'
DEFAULT_PORT = 3671
DOMAIN = 'knx'
REQUIREMENTS = ['xknx==0.7.13']
EVENT_KNX_FRAME_RECEIVED = 'knx_frame_received'
EVENT_KNX_FRAME_SEND = 'knx_frame_send'
TUNNELING_SCHEMA = vol.Schema({
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Required(CONF_KNX_LOCAL_IP): cv.string,
})
KNXTUNNEL = None
KNX_ADDRESS = "address"
KNX_DATA = "data"
KNX_GROUP_WRITE = "group_write"
CONF_LISTEN = "listen"
ROUTING_SCHEMA = vol.Schema({
vol.Required(CONF_KNX_LOCAL_IP): cv.string,
})
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_LISTEN, default=[]):
vol.All(cv.ensure_list, [cv.string]),
}),
vol.Optional(CONF_KNX_CONFIG): cv.string,
vol.Exclusive(CONF_KNX_ROUTING, 'connection_type'): ROUTING_SCHEMA,
vol.Exclusive(CONF_KNX_TUNNELING, 'connection_type'):
TUNNELING_SCHEMA,
vol.Inclusive(CONF_KNX_FIRE_EVENT, 'fire_ev'):
cv.boolean,
vol.Inclusive(CONF_KNX_FIRE_EVENT_FILTER, 'fire_ev'):
vol.All(
cv.ensure_list,
[cv.string])
})
}, extra=vol.ALLOW_EXTRA)
KNX_WRITE_SCHEMA = vol.Schema({
vol.Required(KNX_ADDRESS): vol.All(cv.ensure_list, [cv.string]),
vol.Required(KNX_DATA): vol.All(cv.ensure_list, [cv.byte])
SERVICE_KNX_SEND_SCHEMA = vol.Schema({
vol.Required(SERVICE_KNX_ATTR_ADDRESS): cv.string,
vol.Required(SERVICE_KNX_ATTR_PAYLOAD): vol.Any(
cv.positive_int, [cv.positive_int]),
})
def setup(hass, config):
"""Set up the connection to the KNX IP interface."""
global KNXTUNNEL
from knxip.ip import KNXIPTunnel
from knxip.core import KNXException, parse_group_address
host = config[DOMAIN].get(CONF_HOST)
port = config[DOMAIN].get(CONF_PORT)
if host == '0.0.0.0':
_LOGGER.debug("Will try to auto-detect KNX/IP gateway")
KNXTUNNEL = KNXIPTunnel(host, port)
@asyncio.coroutine
def async_setup(hass, config):
"""Set up knx component."""
from xknx.exceptions import XKNXException
try:
res = KNXTUNNEL.connect()
_LOGGER.debug("Res = %s", res)
if not res:
_LOGGER.error("Could not connect to KNX/IP interface %s", host)
return False
hass.data[DATA_KNX] = KNXModule(hass, config)
yield from hass.data[DATA_KNX].start()
except KNXException as ex:
_LOGGER.exception("Can't connect to KNX/IP interface: %s", ex)
KNXTUNNEL = None
except XKNXException as ex:
_LOGGER.exception("Can't connect to KNX interface: %s", ex)
return False
_LOGGER.info("KNX IP tunnel to %s:%i established", host, port)
for component, discovery_type in (
('switch', 'Switch'),
('climate', 'Climate'),
('cover', 'Cover'),
('light', 'Light'),
('sensor', 'Sensor'),
('binary_sensor', 'BinarySensor'),
('notify', 'Notification')):
found_devices = _get_devices(hass, discovery_type)
hass.async_add_job(
discovery.async_load_platform(hass, component, DOMAIN, {
ATTR_DISCOVER_DEVICES: found_devices
}, config))
descriptions = load_yaml_config_file(
os.path.join(os.path.dirname(__file__), 'services.yaml'))
def received_knx_event(address, data):
"""Process received KNX message."""
if len(data) == 1:
data = data[0]
hass.bus.fire('knx_event', {
'address': address,
'data': data
})
for listen in config[DOMAIN].get(CONF_LISTEN):
_LOGGER.debug("Registering listener for %s", listen)
try:
KNXTUNNEL.register_listener(parse_group_address(listen),
received_knx_event)
except KNXException as knxexception:
_LOGGER.error("Can't register KNX listener for address %s (%s)",
listen, knxexception)
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, close_tunnel)
# Listen to KNX events and send them to the bus
def handle_group_write(call):
"""Bridge knx_frame_send events to the KNX bus."""
# parameters are pre-validated using KNX_WRITE_SCHEMA
addrlist = call.data.get("address")
knxdata = call.data.get("data")
knxaddrlist = []
for addr in addrlist:
try:
_LOGGER.debug("Found %s", addr)
knxaddr = int(addr)
except ValueError:
knxaddr = None
if knxaddr is None:
try:
knxaddr = parse_group_address(addr)
except KNXException:
_LOGGER.error("KNX address format incorrect: %s", addr)
knxaddrlist.append(knxaddr)
for addr in knxaddrlist:
KNXTUNNEL.group_write(addr, knxdata)
# Listen for when knx_frame_send event is fired
hass.services.register(DOMAIN,
KNX_GROUP_WRITE,
handle_group_write,
descriptions[DOMAIN][KNX_GROUP_WRITE],
schema=KNX_WRITE_SCHEMA)
hass.services.async_register(
DOMAIN, SERVICE_KNX_SEND,
hass.data[DATA_KNX].service_send_to_knx_bus,
schema=SERVICE_KNX_SEND_SCHEMA)
return True
def close_tunnel(_data):
"""Close the NKX tunnel connection on shutdown."""
global KNXTUNNEL
KNXTUNNEL.disconnect()
KNXTUNNEL = None
def _get_devices(hass, discovery_type):
return list(
map(lambda device: device.name,
filter(
lambda device: type(device).__name__ == discovery_type,
hass.data[DATA_KNX].xknx.devices)))
class KNXConfig(object):
"""Handle the fetching of configuration from the config file."""
def __init__(self, config):
"""Initialize the configuration."""
from knxip.core import parse_group_address
self.config = config
self.should_poll = config.get('poll', True)
if config.get('address'):
self._address = parse_group_address(config.get('address'))
else:
self._address = None
if self.config.get('state_address'):
self._state_address = parse_group_address(
self.config.get('state_address'))
else:
self._state_address = None
@property
def name(self):
"""Return the name given to the entity."""
return self.config['name']
@property
def address(self):
"""Return the address of the device as an integer value.
3 types of addresses are supported:
integer - 0-65535
2 level - a/b
3 level - a/b/c
"""
return self._address
@property
def state_address(self):
"""Return the group address the device sends its current state to.
Some KNX devices can send the current state to a seperate
group address. This makes send e.g. when an actuator can
be switched but also have a timer functionality.
"""
return self._state_address
class KNXGroupAddress(Entity):
"""Representation of devices connected to a KNX group address."""
class KNXModule(object):
"""Representation of KNX Object."""
def __init__(self, hass, config):
"""Initialize the device."""
self._config = config
self._state = False
self._data = None
_LOGGER.debug(
"Initalizing KNX group address for %s (%s)",
self.name, self.address
)
"""Initialization of KNXModule."""
self.hass = hass
self.config = config
self.initialized = False
self.init_xknx()
self.register_callbacks()
def handle_knx_message(addr, data):
"""Handle an incoming KNX frame.
def init_xknx(self):
"""Initialization of KNX object."""
from xknx import XKNX
self.xknx = XKNX(
config=self.config_file(),
loop=self.hass.loop)
Handle an incoming frame and update our status if it contains
information relating to this device.
"""
if (addr == self.state_address) or (addr == self.address):
self._state = data[0]
self.schedule_update_ha_state()
@asyncio.coroutine
def start(self):
"""Start KNX object. Connect to tunneling or Routing device."""
connection_config = self.connection_config()
yield from self.xknx.start(
state_updater=True,
connection_config=connection_config)
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.stop)
self.initialized = True
KNXTUNNEL.register_listener(self.address, handle_knx_message)
if self.state_address:
KNXTUNNEL.register_listener(self.state_address, handle_knx_message)
@asyncio.coroutine
def stop(self, event):
"""Stop KNX object. Disconnect from tunneling or Routing device."""
yield from self.xknx.stop()
@property
def name(self):
"""Return the entity's display name."""
return self._config.name
def config_file(self):
"""Resolve and return the full path of xknx.yaml if configured."""
config_file = self.config[DOMAIN].get(CONF_KNX_CONFIG)
if not config_file:
return None
if not config_file.startswith("/"):
return self.hass.config.path(config_file)
return config_file
@property
def config(self):
"""Return the entity's configuration."""
return self._config
def connection_config(self):
"""Return the connection_config."""
if CONF_KNX_TUNNELING in self.config[DOMAIN]:
return self.connection_config_tunneling()
elif CONF_KNX_ROUTING in self.config[DOMAIN]:
return self.connection_config_routing()
return self.connection_config_auto()
@property
def should_poll(self):
"""Return the state of the polling, if needed."""
return self._config.should_poll
def connection_config_routing(self):
"""Return the connection_config if routing is configured."""
from xknx.io import ConnectionConfig, ConnectionType
local_ip = \
self.config[DOMAIN][CONF_KNX_ROUTING].get(CONF_KNX_LOCAL_IP)
return ConnectionConfig(
connection_type=ConnectionType.ROUTING,
local_ip=local_ip)
@property
def is_on(self):
"""Return True if the value is not 0 is on, else False."""
return self._state != 0
def connection_config_tunneling(self):
"""Return the connection_config if tunneling is configured."""
from xknx.io import ConnectionConfig, ConnectionType, \
DEFAULT_MCAST_PORT
gateway_ip = \
self.config[DOMAIN][CONF_KNX_TUNNELING].get(CONF_HOST)
gateway_port = \
self.config[DOMAIN][CONF_KNX_TUNNELING].get(CONF_PORT)
local_ip = \
self.config[DOMAIN][CONF_KNX_TUNNELING].get(CONF_KNX_LOCAL_IP)
if gateway_port is None:
gateway_port = DEFAULT_MCAST_PORT
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING,
gateway_ip=gateway_ip,
gateway_port=gateway_port,
local_ip=local_ip)
@property
def address(self):
"""Return the KNX group address."""
return self._config.address
def connection_config_auto(self):
"""Return the connection_config if auto is configured."""
# pylint: disable=no-self-use
from xknx.io import ConnectionConfig
return ConnectionConfig()
@property
def state_address(self):
"""Return the KNX group address."""
return self._config.state_address
def register_callbacks(self):
"""Register callbacks within XKNX object."""
if CONF_KNX_FIRE_EVENT in self.config[DOMAIN] and \
self.config[DOMAIN][CONF_KNX_FIRE_EVENT]:
from xknx.knx import AddressFilter
address_filters = list(map(
AddressFilter,
self.config[DOMAIN][CONF_KNX_FIRE_EVENT_FILTER]))
self.xknx.telegram_queue.register_telegram_received_cb(
self.telegram_received_cb, address_filters)
@property
def cache(self):
"""Return the name given to the entity."""
return self._config.config.get('cache', True)
def group_write(self, value):
"""Write to the group address."""
KNXTUNNEL.group_write(self.address, [value])
def update(self):
"""Get the state from KNX bus or cache."""
from knxip.core import KNXException
try:
if self.state_address:
res = KNXTUNNEL.group_read(
self.state_address, use_cache=self.cache)
else:
res = KNXTUNNEL.group_read(self.address, use_cache=self.cache)
if res:
self._state = res[0]
self._data = res
else:
_LOGGER.debug(
"%s: unable to read from KNX address: %s (None)",
self.name, self.address
)
except KNXException:
_LOGGER.exception(
"%s: unable to read from KNX address: %s",
self.name, self.address
)
return False
class KNXMultiAddressDevice(Entity):
"""Representation of devices connected to a multiple KNX group address.
This is needed for devices like dimmers or shutter actuators as they have
to be controlled by multiple group addresses.
"""
def __init__(self, hass, config, required, optional=None):
"""Initialize the device.
The namelist argument lists the required addresses. E.g. for a dimming
actuators, the namelist might look like:
onoff_address: 0/0/1
brightness_address: 0/0/2
"""
from knxip.core import parse_group_address, KNXException
self.names = {}
self.values = {}
self._config = config
self._state = False
self._data = None
_LOGGER.debug(
"%s: initalizing KNX multi address device",
self.name
)
settings = self._config.config
if config.address:
_LOGGER.debug(
"%s: base address: address=%s",
self.name, settings.get('address')
)
self.names[config.address] = 'base'
if config.state_address:
_LOGGER.debug(
"%s, state address: state_address=%s",
self.name, settings.get('state_address')
)
self.names[config.state_address] = 'state'
# parse required addresses
for name in required:
paramname = '{}{}'.format(name, '_address')
addr = settings.get(paramname)
if addr is None:
_LOGGER.error(
"%s: Required KNX group address %s missing",
self.name, paramname
)
raise KNXException(
"%s: Group address for {} missing in "
"configuration for {}".format(
self.name, paramname
)
)
_LOGGER.debug(
"%s: (required parameter) %s=%s",
self.name, paramname, addr
)
addr = parse_group_address(addr)
self.names[addr] = name
# parse optional addresses
for name in optional:
paramname = '{}{}'.format(name, '_address')
addr = settings.get(paramname)
_LOGGER.debug(
"%s: (optional parameter) %s=%s",
self.name, paramname, addr
)
if addr:
try:
addr = parse_group_address(addr)
except KNXException:
_LOGGER.exception(
"%s: cannot parse group address %s",
self.name, addr
)
self.names[addr] = name
@property
def name(self):
"""Return the entity's display name."""
return self._config.name
@property
def config(self):
"""Return the entity's configuration."""
return self._config
@property
def should_poll(self):
"""Return the state of the polling, if needed."""
return self._config.should_poll
@property
def cache(self):
"""Return the name given to the entity."""
return self._config.config.get('cache', True)
def has_attribute(self, name):
"""Check if the attribute with the given name is defined.
This is mostly important for optional addresses.
"""
for attributename in self.names.values():
if attributename == name:
return True
@asyncio.coroutine
def telegram_received_cb(self, telegram):
"""Callback invoked after a KNX telegram was received."""
self.hass.bus.fire('knx_event', {
'address': telegram.group_address.str(),
'data': telegram.payload.value
})
# False signals XKNX to proceed with processing telegrams.
return False
def set_percentage(self, name, percentage):
"""Set a percentage in knx for a given attribute.
@asyncio.coroutine
def service_send_to_knx_bus(self, call):
"""Service for sending an arbitray KNX message to the KNX bus."""
from xknx.knx import Telegram, Address, DPTBinary, DPTArray
attr_payload = call.data.get(SERVICE_KNX_ATTR_PAYLOAD)
attr_address = call.data.get(SERVICE_KNX_ATTR_ADDRESS)
DPT_Scaling / DPT 5.001 is a single byte scaled percentage
"""
percentage = abs(percentage) # only accept positive values
scaled_value = percentage * 255 / 100
value = min(255, scaled_value)
return self.set_int_value(name, value)
def calculate_payload(attr_payload):
"""Calculate payload depending on type of attribute."""
if isinstance(attr_payload, int):
return DPTBinary(attr_payload)
return DPTArray(attr_payload)
payload = calculate_payload(attr_payload)
address = Address(attr_address)
def get_percentage(self, name):
"""Get a percentage from knx for a given attribute.
telegram = Telegram()
telegram.payload = payload
telegram.group_address = address
yield from self.xknx.telegrams.put(telegram)
DPT_Scaling / DPT 5.001 is a single byte scaled percentage
"""
value = self.get_int_value(name)
percentage = round(value * 100 / 255)
return percentage
def set_int_value(self, name, value, num_bytes=1):
"""Set an integer value for a given attribute."""
# KNX packets are big endian
value = round(value) # only accept integers
b_value = value.to_bytes(num_bytes, byteorder='big')
return self.set_value(name, list(b_value))
class KNXAutomation():
"""Wrapper around xknx.devices.ActionCallback object.."""
def get_int_value(self, name):
"""Get an integer value for a given attribute."""
# KNX packets are big endian
summed_value = 0
raw_value = self.value(name)
try:
# convert raw value in bytes
for val in raw_value:
summed_value *= 256
summed_value += val
except TypeError:
# pknx returns a non-iterable type for unsuccessful reads
pass
def __init__(self, hass, device, hook, action, counter=1):
"""Initialize Automation class."""
self.hass = hass
self.device = device
script_name = "{} turn ON script".format(device.get_name())
self.script = Script(hass, action, script_name)
return summed_value
def value(self, name):
"""Return the value to a given named attribute."""
from knxip.core import KNXException
addr = None
for attributeaddress, attributename in self.names.items():
if attributename == name:
addr = attributeaddress
if addr is None:
_LOGGER.error("%s: attribute '%s' undefined",
self.name, name)
_LOGGER.debug(
"%s: defined attributes: %s",
self.name, str(self.names)
)
return False
try:
res = KNXTUNNEL.group_read(addr, use_cache=self.cache)
except KNXException:
_LOGGER.exception(
"%s: unable to read from KNX address: %s",
self.name, addr
)
return False
return res
def set_value(self, name, value):
"""Set the value of a given named attribute."""
from knxip.core import KNXException
addr = None
for attributeaddress, attributename in self.names.items():
if attributename == name:
addr = attributeaddress
if addr is None:
_LOGGER.error("%s: attribute '%s' undefined",
self.name, name)
_LOGGER.debug(
"%s: defined attributes: %s",
self.name, str(self.names)
)
return False
try:
KNXTUNNEL.group_write(addr, value)
except KNXException:
_LOGGER.exception(
"%s: unable to write to KNX address: %s",
self.name, addr
)
return False
return True
import xknx
self.action = xknx.devices.ActionCallback(
hass.data[DATA_KNX].xknx,
self.script.async_run,
hook=hook,
counter=counter)
device.actions.append(self.action)

192
homeassistant/components/light/knx.py
View file

@ -1,17 +1,17 @@
"""
Support KNX Lighting actuators.
Support for KNX/IP lights.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/Light.knx/
https://home-assistant.io/components/light.knx/
"""
import logging
import asyncio
import voluptuous as vol
from homeassistant.components.knx import (KNXConfig, KNXMultiAddressDevice)
from homeassistant.components.light import (Light, PLATFORM_SCHEMA,
SUPPORT_BRIGHTNESS,
ATTR_BRIGHTNESS)
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.components.light import PLATFORM_SCHEMA, Light, \
SUPPORT_BRIGHTNESS, ATTR_BRIGHTNESS
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
CONF_ADDRESS = 'address'
@ -19,8 +19,6 @@ CONF_STATE_ADDRESS = 'state_address'
CONF_BRIGHTNESS_ADDRESS = 'brightness_address'
CONF_BRIGHTNESS_STATE_ADDRESS = 'brightness_state_address'
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'KNX Light'
DEPENDENCIES = ['knx']
@ -33,84 +31,136 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the KNX light platform."""
add_devices([KNXLight(hass, KNXConfig(config))])
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up light(s) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
class KNXLight(KNXMultiAddressDevice, Light):
"""Representation of a KNX Light device."""
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up lights for KNX platform configured via xknx.yaml."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXLight(hass, device))
add_devices(entities)
def __init__(self, hass, config):
"""Initialize the cover."""
KNXMultiAddressDevice.__init__(
self, hass, config,
[], # required
optional=['state', 'brightness', 'brightness_state']
)
self._hass = hass
self._supported_features = 0
if CONF_BRIGHTNESS_ADDRESS in config.config:
_LOGGER.debug("%s is dimmable", self.name)
self._supported_features = self._supported_features | \
SUPPORT_BRIGHTNESS
self._brightness = None
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up light for KNX platform configured within plattform."""
import xknx
light = xknx.devices.Light(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address_switch=config.get(CONF_ADDRESS),
group_address_switch_state=config.get(CONF_STATE_ADDRESS),
group_address_brightness=config.get(CONF_BRIGHTNESS_ADDRESS),
group_address_brightness_state=config.get(
CONF_BRIGHTNESS_STATE_ADDRESS))
hass.data[DATA_KNX].xknx.devices.add(light)
add_devices([KNXLight(hass, light)])
def turn_on(self, **kwargs):
"""Turn the switch on.
This sends a value 1 to the group address of the device
"""
_LOGGER.debug("%s: turn on", self.name)
self.set_value('base', [1])
self._state = 1
class KNXLight(Light):
"""Representation of a KNX light."""
if ATTR_BRIGHTNESS in kwargs:
self._brightness = kwargs[ATTR_BRIGHTNESS]
_LOGGER.debug("turn_on requested brightness for light: %s is: %s ",
self.name, self._brightness)
assert self._brightness <= 255
self.set_value("brightness", [self._brightness])
def __init__(self, hass, device):
"""Initialization of KNXLight."""
self.device = device
self.hass = hass
self.async_register_callbacks()
if not self.should_poll:
self.schedule_update_ha_state()
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
def turn_off(self, **kwargs):
"""Turn the switch off.
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
This sends a value 1 to the group address of the device
"""
_LOGGER.debug("%s: turn off", self.name)
self.set_value('base', [0])
self._state = 0
if not self.should_poll:
self.schedule_update_ha_state()
@property
def should_poll(self):
"""No polling needed within KNX."""
return False
@property
def brightness(self):
"""Return the brightness of this light between 0..255."""
return self.device.brightness \
if self.device.supports_dimming else \
None
@property
def xy_color(self):
"""Return the XY color value [float, float]."""
return None
@property
def rgb_color(self):
"""Return the RBG color value."""
return None
@property
def color_temp(self):
"""Return the CT color temperature."""
return None
@property
def white_value(self):
"""Return the white value of this light between 0..255."""
return None
@property
def effect_list(self):
"""Return the list of supported effects."""
return None
@property
def effect(self):
"""Return the current effect."""
return None
@property
def is_on(self):
"""Return True if the value is not 0 is on, else False."""
return self._state != 0
"""Return true if light is on."""
return self.device.state
@property
def supported_features(self):
"""Flag supported features."""
return self._supported_features
flags = 0
if self.device.supports_dimming:
flags |= SUPPORT_BRIGHTNESS
return flags
def update(self):
"""Update device state."""
super().update()
if self.has_attribute('brightness_state'):
value = self.value('brightness_state')
if value is not None:
self._brightness = int.from_bytes(value, byteorder='little')
_LOGGER.debug("%s: brightness = %d",
self.name, self._brightness)
@asyncio.coroutine
def async_turn_on(self, **kwargs):
"""Turn the light on."""
if ATTR_BRIGHTNESS in kwargs and self.device.supports_dimming:
yield from self.device.set_brightness(int(kwargs[ATTR_BRIGHTNESS]))
else:
yield from self.device.set_on()
if self.has_attribute('state'):
self._state = self.value("state")[0]
_LOGGER.debug("%s: state = %d", self.name, self._state)
def should_poll(self):
"""No polling needed for a KNX light."""
return False
@asyncio.coroutine
def async_turn_off(self, **kwargs):
"""Turn the light off."""
yield from self.device.set_off()

13
homeassistant/components/notify/__init__.py
View file

@ -82,8 +82,6 @@ def async_setup(hass, config):
"""Set up a notify platform."""
if p_config is None:
p_config = {}
if discovery_info is None:
discovery_info = {}
platform = yield from async_prepare_setup_platform(
hass, config, DOMAIN, p_type)
@ -105,8 +103,12 @@ def async_setup(hass, config):
raise HomeAssistantError("Invalid notify platform.")
if notify_service is None:
_LOGGER.error(
"Failed to initialize notification service %s", p_type)
# Platforms can decide not to create a service based
# on discovery data.
if discovery_info is None:
_LOGGER.error(
"Failed to initialize notification service %s",
p_type)
return
except Exception: # pylint: disable=broad-except
@ -115,6 +117,9 @@ def async_setup(hass, config):
notify_service.hass = hass
if discovery_info is None:
discovery_info = {}
@asyncio.coroutine
def async_notify_message(service):
"""Handle sending notification message service calls."""

99
homeassistant/components/notify/knx.py Normal file
View file

@ -0,0 +1,99 @@
"""
KNX/IP notification service.
For more details about this platform, please refer to the documentation
https://home-assistant.io/components/notify.knx/
"""
import asyncio
import voluptuous as vol
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.components.notify import PLATFORM_SCHEMA, \
BaseNotificationService
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
CONF_ADDRESS = 'address'
DEFAULT_NAME = 'KNX Notify'
DEPENDENCIES = ['knx']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string
})
@asyncio.coroutine
def async_get_service(hass, config, discovery_info=None):
"""Get the KNX notification service."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
return async_get_service_discovery(hass, discovery_info) \
if discovery_info is not None else \
async_get_service_config(hass, config)
@callback
def async_get_service_discovery(hass, discovery_info):
"""Set up notifications for KNX platform configured via xknx.yaml."""
notification_devices = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
notification_devices.append(device)
return \
KNXNotificationService(hass, notification_devices) \
if notification_devices else \
None
@callback
def async_get_service_config(hass, config):
"""Set up notification for KNX platform configured within plattform."""
import xknx
notification = xknx.devices.Notification(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address=config.get(CONF_ADDRESS))
hass.data[DATA_KNX].xknx.devices.add(notification)
return KNXNotificationService(hass, [notification, ])
class KNXNotificationService(BaseNotificationService):
"""Implement demo notification service."""
def __init__(self, hass, devices):
"""Initialize the service."""
self.hass = hass
self.devices = devices
@property
def targets(self):
"""Return a dictionary of registered targets."""
ret = {}
for device in self.devices:
ret[device.name] = device.name
return ret
@asyncio.coroutine
def async_send_message(self, message="", **kwargs):
"""Send a notification to knx bus."""
if "target" in kwargs:
yield from self._async_send_to_device(message, kwargs["target"])
else:
yield from self._async_send_to_all_devices(message)
@asyncio.coroutine
def _async_send_to_all_devices(self, message):
"""Send a notification to knx bus to all connected devices."""
for device in self.devices:
yield from device.set(message)
@asyncio.coroutine
def _async_send_to_device(self, message, names):
"""Send a notification to knx bus to device with given names."""
for device in self.devices:
if device.name in names:
yield from device.set(message)

227
homeassistant/components/sensor/knx.py
View file

@ -1,184 +1,111 @@
"""
Sensors of a KNX Device.
Support for KNX/IP sensors.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/knx/
https://home-assistant.io/components/sensor.knx/
"""
from enum import Enum
import logging
import asyncio
import voluptuous as vol
from homeassistant.const import (
CONF_NAME, CONF_MAXIMUM, CONF_MINIMUM,
CONF_TYPE, TEMP_CELSIUS
)
from homeassistant.components.knx import (KNXConfig, KNXGroupAddress)
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.helpers.entity import Entity
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
CONF_ADDRESS = 'address'
CONF_TYPE = 'type'
DEFAULT_NAME = 'KNX Sensor'
DEPENDENCIES = ['knx']
DEFAULT_NAME = "KNX sensor"
CONF_TEMPERATURE = 'temperature'
CONF_ADDRESS = 'address'
CONF_ILLUMINANCE = 'illuminance'
CONF_PERCENTAGE = 'percentage'
CONF_SPEED_MS = 'speed_ms'
class KNXAddressType(Enum):
"""Enum to indicate conversion type for the KNX address."""
FLOAT = 1
PERCENT = 2
# define the fixed settings required for each sensor type
FIXED_SETTINGS_MAP = {
# Temperature as defined in KNX Standard 3.10 - 9.001 DPT_Value_Temp
CONF_TEMPERATURE: {
'unit': TEMP_CELSIUS,
'default_minimum': -273,
'default_maximum': 670760,
'address_type': KNXAddressType.FLOAT
},
# Speed m/s as defined in KNX Standard 3.10 - 9.005 DPT_Value_Wsp
CONF_SPEED_MS: {
'unit': 'm/s',
'default_minimum': 0,
'default_maximum': 670760,
'address_type': KNXAddressType.FLOAT
},
# Luminance(LUX) as defined in KNX Standard 3.10 - 9.004 DPT_Value_Lux
CONF_ILLUMINANCE: {
'unit': 'lx',
'default_minimum': 0,
'default_maximum': 670760,
'address_type': KNXAddressType.FLOAT
},
# Percentage(%) as defined in KNX Standard 3.10 - 5.001 DPT_Scaling
CONF_PERCENTAGE: {
'unit': '%',
'default_minimum': 0,
'default_maximum': 100,
'address_type': KNXAddressType.PERCENT
}
}
SENSOR_TYPES = set(FIXED_SETTINGS_MAP.keys())
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_TYPE): vol.In(SENSOR_TYPES),
vol.Required(CONF_ADDRESS): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MINIMUM): vol.Coerce(float),
vol.Optional(CONF_MAXIMUM): vol.Coerce(float)
vol.Optional(CONF_TYPE): cv.string,
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the KNX Sensor platform."""
add_devices([KNXSensor(hass, KNXConfig(config))])
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up sensor(s) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
class KNXSensor(KNXGroupAddress):
"""Representation of a KNX Sensor device."""
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up sensors for KNX platform configured via xknx.yaml."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXSensor(hass, device))
add_devices(entities)
def __init__(self, hass, config):
"""Initialize a KNX Float Sensor."""
# set up the KNX Group address
KNXGroupAddress.__init__(self, hass, config)
device_type = config.config.get(CONF_TYPE)
sensor_config = FIXED_SETTINGS_MAP.get(device_type)
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up sensor for KNX platform configured within plattform."""
import xknx
sensor = xknx.devices.Sensor(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address=config.get(CONF_ADDRESS),
value_type=config.get(CONF_TYPE))
hass.data[DATA_KNX].xknx.devices.add(sensor)
add_devices([KNXSensor(hass, sensor)])
if not sensor_config:
raise NotImplementedError()
# set up the conversion function based on the address type
address_type = sensor_config.get('address_type')
if address_type == KNXAddressType.FLOAT:
self.convert = convert_float
elif address_type == KNXAddressType.PERCENT:
self.convert = convert_percent
else:
raise NotImplementedError()
class KNXSensor(Entity):
"""Representation of a KNX sensor."""
# other settings
self._unit_of_measurement = sensor_config.get('unit')
default_min = float(sensor_config.get('default_minimum'))
default_max = float(sensor_config.get('default_maximum'))
self._minimum_value = config.config.get(CONF_MINIMUM, default_min)
self._maximum_value = config.config.get(CONF_MAXIMUM, default_max)
_LOGGER.debug(
"%s: configured additional settings: unit=%s, "
"min=%f, max=%f, type=%s",
self.name, self._unit_of_measurement,
self._minimum_value, self._maximum_value, str(address_type)
)
def __init__(self, hass, device):
"""Initialization of KNXSensor."""
self.device = device
self.hass = hass
self.async_register_callbacks()
self._value = None
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
@property
def should_poll(self):
"""No polling needed within KNX."""
return False
@property
def state(self):
"""Return the Value of the KNX Sensor."""
return self._value
"""Return the state of the sensor."""
return self.device.resolve_state()
@property
def unit_of_measurement(self):
"""Return the defined Unit of Measurement for the KNX Sensor."""
return self._unit_of_measurement
def update(self):
"""Update KNX sensor."""
super().update()
self._value = None
if self._data:
if self._data == 0:
value = 0
else:
value = self.convert(self._data)
if self._minimum_value <= value <= self._maximum_value:
self._value = value
"""Return the unit this state is expressed in."""
return self.device.unit_of_measurement()
@property
def cache(self):
"""We don't want to cache any Sensor Value."""
return False
def convert_float(raw_value):
"""Conversion for 2 byte floating point values.
2byte Floating Point KNX Telegram.
Defined in KNX 3.7.2 - 3.10
"""
from knxip.conversion import knx2_to_float
from knxip.core import KNXException
try:
return knx2_to_float(raw_value)
except KNXException as exception:
_LOGGER.error("Can't convert %s to float (%s)", raw_value, exception)
def convert_percent(raw_value):
"""Conversion for scaled byte values.
1byte percentage scaled KNX Telegram.
Defined in KNX 3.7.2 - 3.10.
"""
value = 0
try:
value = raw_value[0]
except (IndexError, ValueError):
# pknx returns a non-iterable type for unsuccessful reads
_LOGGER.error("Can't convert %s to percent value", raw_value)
return round(value * 100 / 255)
def device_state_attributes(self):
"""Return the state attributes."""
return None

105
homeassistant/components/switch/knx.py
View file

@ -1,14 +1,16 @@
"""
Support KNX switching actuators.
Support for KNX/IP switches.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/switch.knx/
"""
import asyncio
import voluptuous as vol
from homeassistant.components.knx import (KNXConfig, KNXGroupAddress)
from homeassistant.components.switch import (SwitchDevice, PLATFORM_SCHEMA)
from homeassistant.components.knx import DATA_KNX, ATTR_DISCOVER_DEVICES
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchDevice
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
CONF_ADDRESS = 'address'
@ -24,30 +26,85 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the KNX switch platform."""
add_devices([KNXSwitch(hass, KNXConfig(config))])
@asyncio.coroutine
def async_setup_platform(hass, config, add_devices,
discovery_info=None):
"""Set up switch(es) for KNX platform."""
if DATA_KNX not in hass.data \
or not hass.data[DATA_KNX].initialized:
return False
if discovery_info is not None:
async_add_devices_discovery(hass, discovery_info, add_devices)
else:
async_add_devices_config(hass, config, add_devices)
return True
class KNXSwitch(KNXGroupAddress, SwitchDevice):
"""Representation of a KNX switch device."""
@callback
def async_add_devices_discovery(hass, discovery_info, add_devices):
"""Set up switches for KNX platform configured via xknx.yaml."""
entities = []
for device_name in discovery_info[ATTR_DISCOVER_DEVICES]:
device = hass.data[DATA_KNX].xknx.devices[device_name]
entities.append(KNXSwitch(hass, device))
add_devices(entities)
def turn_on(self, **kwargs):
"""Turn the switch on.
This sends a value 0 to the group address of the device
"""
self.group_write(1)
self._state = [1]
if not self.should_poll:
self.schedule_update_ha_state()
@callback
def async_add_devices_config(hass, config, add_devices):
"""Set up switch for KNX platform configured within plattform."""
import xknx
switch = xknx.devices.Switch(
hass.data[DATA_KNX].xknx,
name=config.get(CONF_NAME),
group_address=config.get(CONF_ADDRESS),
group_address_state=config.get(CONF_STATE_ADDRESS))
hass.data[DATA_KNX].xknx.devices.add(switch)
add_devices([KNXSwitch(hass, switch)])
def turn_off(self, **kwargs):
"""Turn the switch off.
This sends a value 1 to the group address of the device
"""
self.group_write(0)
self._state = [0]
if not self.should_poll:
self.schedule_update_ha_state()
class KNXSwitch(SwitchDevice):
"""Representation of a KNX switch."""
def __init__(self, hass, device):
"""Initialization of KNXSwitch."""
self.device = device
self.hass = hass
self.async_register_callbacks()
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
@asyncio.coroutine
def after_update_callback(device):
"""Callback after device was updated."""
# pylint: disable=unused-argument
yield from self.async_update_ha_state()
self.device.register_device_updated_cb(after_update_callback)
@property
def name(self):
"""Return the name of the KNX device."""
return self.device.name
@property
def should_poll(self):
"""No polling needed within KNX."""
return False
@property
def is_on(self):
"""Return true if device is on."""
return self.device.state
@asyncio.coroutine
def async_turn_on(self, **kwargs):
"""Turn the device on."""
yield from self.device.set_on()
@asyncio.coroutine
def async_turn_off(self, **kwargs):
"""Turn the device off."""
yield from self.device.set_off()

6
requirements_all.txt
View file

@ -362,9 +362,6 @@ jsonrpc-websocket==0.5
# homeassistant.scripts.keyring
keyring>=9.3,<10.0
# homeassistant.components.knx
knxip==0.5
# homeassistant.components.device_tracker.owntracks
libnacl==1.5.2
@ -1012,6 +1009,9 @@ xbee-helper==0.0.7
# homeassistant.components.sensor.xbox_live
xboxapi==0.1.1
# homeassistant.components.knx
xknx==0.7.13
# homeassistant.components.media_player.bluesound
# homeassistant.components.sensor.swiss_hydrological_data
# homeassistant.components.sensor.ted5000