#!/usr/bin/env python
# encoding: utf-8
#
# pmatic - Python API for Homematic. Easy to use.
# Copyright (C) 2016 Lars Michelsen <lm@larsmichelsen.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
# Relevant docs:
# - http://www.eq-3.de/Downloads/PDFs/Dokumentation_und_Tutorials/HM_XmlRpc_V1_502__2_.pdf
# - http://www.eq-3.de/Downloads/eq3/download%20bereich/hm_web_ui_doku/hm_devices_Endkunden.pdf
# Add Python 3.x behaviour to 2.7
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import time
import threading
try:
from builtins import object # pylint:disable=redefined-builtin
except ImportError:
pass
import pmatic.params
import pmatic.utils as utils
from pmatic.exceptions import PMException, PMDeviceOffline
class Entity(object):
_transform_attributes = {}
_skip_attributes = []
_mandatory_attributes = []
def __init__(self, ccu, spec):
assert isinstance(ccu, pmatic.ccu.CCU), "ccu is not of CCU class: %r" % ccu
assert isinstance(spec, dict), "spec is not a dictionary: %r" % spec
self._ccu = ccu
self._set_attributes(spec)
self._verify_mandatory_attributes()
super(Entity, self).__init__()
def _set_attributes(self, obj_dict):
"""Adding provided attributes to this entity.
Transforming and filtering dictionaries containing attributes for this entity
by using the configured transform methods for the individual attributes and also
excluding some attributes which keys are in self._skip_attributes."""
for key, val in obj_dict.items():
if key in self._skip_attributes:
continue
# Optionally convert values using the given transform functions
# for the specific object type
trans_func = self._transform_attributes.get(key)
if trans_func:
func_type = type(trans_func).__name__
if func_type in [ "instancemethod", "function", "method" ]:
args = []
offset = 1 if func_type in [ "instancemethod", "method" ] else 0
argcount = trans_func.__code__.co_argcount
for arg_name in trans_func.__code__.co_varnames[offset:argcount]:
if arg_name == "api":
args.append(self._ccu.api)
elif arg_name == "ccu":
args.append(self._ccu)
elif arg_name == "device":
args.append(self)
elif arg_name == "obj":
args.append(self)
else:
args.append(val)
else:
args = [val]
val = trans_func(*args)
# Transform keys from camel case to our style
key = utils.decamel(key)
setattr(self, key, val)
def _verify_mandatory_attributes(self):
for key in self._mandatory_attributes:
if not hasattr(self, key):
raise PMException("The mandatory attribute \"%s\" is missing." % key)
class Channels(dict):
"""This class has been created to make the dict where the channels are stored
in to have a similar interface like a list.
We want to have a consistent interface when e.g. doing this:
```
for device in ccu.devices:
for channel in device.channels:
...
```
With only a dict for device.channels it would need device.channels.values()
"""
def __iter__(self):
return iter(sorted(self.values(), key=lambda x: x.index))
[docs]class Channel(utils.LogMixin, Entity):
_transform_attributes = {
# ReGa attributes:
"id" : int,
"partner_id" : lambda x: None if x == "" else int(x),
# Low level attributes:
"aes_active" : bool,
"link_source_roles" : lambda v: v if isinstance(v, list) else v.split(" "),
"link_target_roles" : lambda v: v if isinstance(v, list) else v.split(" "),
}
# Don't add these keys to the objects attributes
_skip_attributes = [
# Low level attributes:
"parent",
"parent_type",
]
# These keys have to be set after attribute initialization
_mandatory_attributes = [
# Low level attributes:
# address of channel
"address",
# communication direction of channel:
# 0 = DIRECTION_NONE (Kanal unterstützt keine direkte Verknüpfung)
# 1 = DIRECTION_SENDER
# 2 = DIRECTION_RECEIVER
"direction",
# see device flags (0x01 visible, 0x02 internal, 0x08 can not be deleted)
"flags",
# channel number
"index",
# possible roles as sender
"link_source_roles",
# possible roles as receiver
"link_target_roles",
# list of available parameter sets
"paramsets",
# type of this channel
"type",
# version of the channel description
"version",
]
def __init__(self, device, spec):
if not isinstance(device, Device):
raise PMException("Device object is not a Device derived class: %r" % device)
self.device = device
self._values = {}
self._values_lock = threading.RLock()
self._callbacks_to_register = {
"value_updated": [],
"value_changed": [],
}
super(Channel, self).__init__(device._ccu, spec)
@classmethod
[docs] def from_channel_dicts(cls, device, channel_dicts):
"""Creates channel instances associated with the given *device* instance from the given
attribute dictionaries.
Uses the list of channel attribute dictionaries given with *channel_dicts* to create a
dictionary of specific `Channel` instances (like e.g. :class:`ChannelShutterContact`)
or the generic :class:`Channel` class. Normally each channel should have a specific
class. In case an unknown channel needs to be created a debug message is being logged.
The dictionary uses the index of the channel (the channel id) as key for entries.
The dictionary of the created channels is then returned."""
channel_objects = Channels()
for channel_dict in channel_dicts:
channel_class = channel_classes_by_type_name.get(channel_dict["type"], Channel)
if channel_class == Channel:
cls.cls_logger().debug("Using generic Channel class (Type: %s): %r" %
(channel_dict["type"], channel_dict))
channel_objects[channel_dict["index"]] = channel_class(device, channel_dict)
return channel_objects
@property
def values(self):
"""Provides access to all value objects of this channel.
The values are provided as dictionary where the name of the parameter is used as key
and some kind of specific :class:`.params.Parameter` instance is the value."""
with self._values_lock:
if not self._values:
self._init_value_specs()
if self._value_update_needed():
self._fetch_values()
return self._values
def _init_value_specs(self):
"""Initializes the value objects by fetching the specification from the CCU.
The specification (description) of the VALUES paramset are fetched from
the CCU and Parameter() objects will be created from them. Then they
will be added to self._values.
This method is called on the first access to the values.
"""
self._values.clear()
for value_spec in self._ccu.api.interface_get_paramset_description(interface="BidCos-RF",
address=self.address, paramsetType="VALUES"):
self._init_value_spec(value_spec)
self._register_saved_callbacks()
def _init_value_spec(self, value_spec):
"""Initializes a single value of this channel."""
value_id = value_spec["ID"]
class_name = self._get_class_name_of_param_spec(value_spec)
cls = getattr(pmatic.params, class_name)
if not cls:
self.logger.warning("%s: Skipping unknown parameter %s of type %s, unit %s. "
"Class %s not implemented." %
(self.channel_type, value_id, value_spec["TYPE"],
value_spec["UNIT"], class_name))
else:
self._values[value_id] = cls(self, value_spec)
def _get_class_name_of_param_spec(self, param_spec):
"""Gathers the name of the class to be used for creating a parameter object
from the given parameter specification."""
return "Parameter%s" % param_spec["TYPE"]
def _value_update_needed(self):
"""Tells whether or not the set of values should be fetched from the CCU."""
oldest_value_time = None
for param in self._values.values():
try:
last_updated = param.last_updated
if last_updated == None:
last_updated = 0 # enforce the update
except PMException:
continue # Ignore not readable values
if oldest_value_time == None:
oldest_value_time = last_updated
elif last_updated < oldest_value_time:
oldest_value_time = last_updated
if oldest_value_time == None:
return False # No readable value at all
# FIXME: Make threshold configurable
return oldest_value_time <= time.time() - 60
def _fetch_values(self):
"""Fetches all values of the channel.
Gathers the values of the channel and updates the value parameters in self._values.
The parameter objects need to be initialized before (self._init_value_specs).
"""
if not self._values:
raise PMException("The value parameters are not yet initialized.")
try:
values = self._get_values()
except PMException as e:
# FIXME: Clean this 601 in "%s" up!
if "601" in ("%s" % e) and not self.device.is_online:
raise PMDeviceOffline("Failed to fetch the values. The device is not online.")
else:
raise
for param_id, value in values.items():
if param_id in self._values:
self._values[param_id].set_from_api(value)
else:
self.logger.info("%s (%s - %s): Skipping received value of unknown parameter %s.",
self.address, self.device.name, self.name, param_id)
def _get_values(self):
"""This method returns all values of this channel.
Normally it is using the API call Interface.getParamset() to fetch all values of
the channel at once. But there are devices which have bugged values which are reported
to be readable, but can not be read in fact.
The default way to deal with it is to catch the exectpion from the bulk get and then
fetch the values one by one, again while catching the exceptions of these calls which
which are raised when a value is not readable.
In some cases where it is known to be an issue with specific values the channels in
question have a specific class which overrides this method to fetch the single values
one by one but skips the failing values explicitly."""
try:
return self._get_values_bulk()
except PMException as e:
# Can not check is_online for maintenance channels here (no values yet)
if any(errorcode in ("%s" % e) for errorcode in ["501", "601"]) \
and (isinstance(self, ChannelMaintenance) or self.device.is_online):
self.logger.info("%s (%s - %s): %s. Falling back to single value fetching.",
self.address, self.device.name, self.name, e)
return self._get_values_single(skip_invalid_values=True)
else:
raise
def _get_values_bulk(self):
"""Fetches all values of this channel at once. This is the default method to
fetch the values."""
return self._ccu.api.interface_get_paramset(interface="BidCos-RF",
address=self.address, paramsetKey="VALUES")
def _get_values_single(self, skip_invalid_values=False):
"""Fetches all values known to be readable one by one. One should always
use :meth:`_get_values_bulked` when possible. This is only used for buggy
devices.
The function can be called with the `skip_invalid_values` argument set to `True`
to only log exceptions of single values and continue with the next value."""
values = {}
for param_id, value in self._values.items():
if value.readable:
try:
values[value.id] = self._ccu.api.interface_get_value(
interface="BidCos-RF",
address=self.address,
valueKey=value.internal_name)
except PMException as e:
if not skip_invalid_values:
raise
if not any(errorcode in ("%s" % e) for errorcode in ["501", "601"]):
raise
if isinstance(self, ChannelMaintenance) or self.device.is_online:
self.logger.info("%s (%s - %s - %s): %s",
self.address, self.device.name, self.name, param_id, e)
else:
raise
return values
@property
def summary_state(self):
"""Represents a summary state of the channel.
Formats values and titles of channel values and returns them as string.
Default formating of channel values. Concatenates titles and values of
all channel values except the maintenance channel.
The values are sorted by the titles."""
formated = []
for title, value in sorted([ (v.name, v) for v in self.values.values() if v.readable ]):
formated.append("%s: %s" % (title, value))
return ", ".join(formated)
def set_logic_attributes(self, attrs):
"""Used to update the logic attributes of this channel.
Applying the attributes in the dictionary to this object. Special handling
for some attributes which are already set by the low level attributes."""
#import pprint
#pprint.pprint(self.__dict__)
#pprint.pprint(attrs)
#sys.exit(1)
# Skip non needed attributes (already set by low level data)
# FIXME: 'direction': 1, from low level API might be duplicate of
# u'category': u'CATEGORY_SENDER',
# FIXME: 'aes_active': True, from low level API might be duplicate
# of u'mode': u'MODE_AES',
attrs = attrs.copy()
for a in [ "address", "device_id" ]:
del attrs[a]
self._set_attributes(attrs)
[docs] def on_value_changed(self, func):
"""Register a function to be called each time a value of this channel parameters
has changed."""
try:
values = self.values.values()
except PMDeviceOffline:
# Unable to register with parameters right now. Save for later registration
# when values are available one day.
self._save_callback_to_register("value_changed", func)
return
for value in values:
value.register_callback("value_changed", func)
[docs] def on_value_updated(self, func):
"""Register a function to be called each time a value of this channel parameters has
been updated."""
try:
values = self.values.values()
except PMDeviceOffline:
# Unable to register with parameters right now. Save for later registration
# when values are available one day.
self._save_callback_to_register("value_updated", func)
return
for value in values:
value.register_callback("value_updated", func)
def _save_callback_to_register(self, cb_name, func):
"""Stores a callback function for attaching it later to the parameters."""
self._callbacks_to_register[cb_name].append(func)
def _register_saved_callbacks(self):
"""If there are saved callbacks to register to new fetched parameters, register them!"""
values = self._values.values()
for cb_name, callbacks in self._callbacks_to_register.items():
if not callbacks:
continue
for func in callbacks:
for value in values:
value.register_callback(cb_name, func)
# FIXME: Implement this. The Device() object already has a lot of methods
# related to the maintenance channel. Shouldn't they be moved here and evantually
# only be available as shortcut in the Device object?
[docs]class ChannelMaintenance(Channel):
type_name = "MAINTENANCE"
name = "Maintenance"
id = 0
@property
def summary_state(self):
"""The maintenance channel does not provide a summary state.
If you want to get a formated maintenance state, you need to use the property
:attr:`maintenance_state`."""
return None
@property
def maintenance_state(self):
"""Provides the formated maintenance state of the associated device."""
return super(ChannelMaintenance, self).summary_state
# FIXME: Handle LOWBAT/ERROR
# FIXME: Handle STOP, INHIBIT, INSTALL_TEST
[docs]class ChannelBlind(Channel):
type_name = "BLIND"
@property
def level(self):
"""Look up the level at which the shutter is set."""
return self.values["LEVEL"].value
[docs] def set_level(self, level):
"""Set the level at which the shutter is to be set."""
return self.values["LEVEL"].set(level)
@property
def working(self):
"""Look up the WORKING value."""
return self.values["WORKING"].value
# FIXME: Handle INHIBIT, WORKING
[docs]class ChannelSwitch(Channel):
type_name = "SWITCH"
@property
def is_on(self):
"""``True`` when the power is on, otherwise ``False``."""
return self.values["STATE"].value
@property
def summary_state(self):
"""Provides the current state as well formated string."""
return "%s: %s" % (self.values["STATE"].name, self.is_on and "on" or "off")
[docs] def toggle(self):
"""Use this to toggle the switch."""
if self.is_on:
return self.switch_off()
else:
return self.switch_on()
[docs] def switch_off(self):
"""Power off!"""
return self.values["STATE"].set(False)
[docs] def switch_on(self):
"""Lights on!"""
return self.values["STATE"].set(True)
# FIXME: Handle LED_STATUS, ALL_LEDS, LED_SLEEP_MODE, INSTALL_TEST
[docs]class ChannelKey(Channel):
type_name = "KEY"
[docs] def press_short(self):
"""Call this to trigger a short press."""
return self.values["PRESS_SHORT"].set(True)
[docs] def press_long(self):
"""Triggers a long press."""
return self.values["PRESS_LONG"].set(True)
# Not verified working
[docs] def press_long_release(self):
"""Triggers the release of a long press."""
return self.values["PRESS_LONG_RELEASE"].set(True)
# Not verified
[docs] def press_cont(self):
"""Unknown. Untested. Please let me know what this is."""
return self.values["PRESS_CONT"].set(True)
@property
def summary_state(self):
"""Has no state info as it's a toggle button. This is only to override the
default summary_state property."""
return None
[docs]class ChannelVirtualKey(ChannelKey):
type_name = "VIRTUAL_KEY"
# FIXME: Handle all values:
# {u'POWER': u'3.520000', u'ENERGY_COUNTER': u'501.400000', u'BOOT': u'1',
# u'CURRENT': u'26.000000', u'FREQUENCY': u'50.010000', u'VOLTAGE': u'228.900000'}
[docs]class ChannelPowermeter(Channel):
type_name = "POWERMETER"
# FIXME: To be implemented.
[docs]class ChannelConditionPower(Channel):
type_name = "CONDITION_POWER"
# FIXME: To be implemented.
[docs]class ChannelConditionCurrent(Channel):
type_name = "CONDITION_CURRENT"
# FIXME: To be implemented.
[docs]class ChannelConditionVoltage(Channel):
type_name = "CONDITION_VOLTAGE"
# FIXME: To be implemented.
[docs]class ChannelConditionFrequency(Channel):
type_name = "CONDITION_FREQUENCY"
# FIXME: To be implemented.
# Devices:
# HM-Sen-LI-O
class ChannelLuxmeter(Channel):
type_name = "LUXMETER"
# FIXME: To be implemented.
# Devices:
# HM-WDS10-TH-O
[docs]class ChannelWeather(Channel):
type_name = "WEATHER"
# FIXME: Handle ERROR
[docs]class ChannelClimaVentDrive(Channel):
type_name = "CLIMATECONTROL_VENT_DRIVE"
# FIXME: Handle ADJUSTING_COMMAND, ADJUSTING_DATA
[docs]class ChannelClimaRegulator(Channel):
type_name = "CLIMATECONTROL_REGULATOR"
@property
def summary_state(self):
"""Provides the ventil state."""
val = self.values["SETPOINT"]
if val == 0.0:
return "Ventil closed"
elif val == 100.0:
return "Ventil open"
else:
return "Ventil: %s" % self.values["SETPOINT"]
# Devices:
# HM-CC-RT-DN
# FIXME: Values:
# {u'SET_TEMPERATURE': u'21.500000', u'PARTY_START_MONTH': u'1', u'BATTERY_STATE': u'2.400000',
# u'PARTY_START_DAY': u'1', u'PARTY_STOP_DAY': u'1', u'PARTY_START_YEAR': u'0',
# u'FAULT_REPORTING': u'0', u'PARTY_STOP_TIME': u'0', u'ACTUAL_TEMPERATURE': u'23.100000',
# u'BOOST_STATE': u'15', u'PARTY_STOP_YEAR': u'0', u'PARTY_STOP_MONTH': u'1',
# u'VALVE_STATE': u'10', u'PARTY_START_TIME': u'450', u'PARTY_TEMPERATURE': u'5.000000',
# u'CONTROL_MODE': u'1'}
[docs]class ChannelClimaRTTransceiver(Channel):
type_name = "CLIMATECONTROL_RT_TRANSCEIVER"
@property
def summary_state(self):
"""Provides the actual and target temperature together with the valve state in
some readable format."""
return "Temperature: %s (Target: %s, Valve: %s)" % \
(self.values["ACTUAL_TEMPERATURE"],
self.values["SET_TEMPERATURE"],
self.values["VALVE_STATE"])
def _get_class_name_of_param_spec(self, param_spec):
if param_spec["ID"] == "CONTROL_MODE":
return "ParameterControlMode"
else:
return super(ChannelClimaRTTransceiver, self)._get_class_name_of_param_spec(param_spec)
[docs]class ChannelWindowSwitchReceiver(Channel):
type_name = "WINDOW_SWITCH_RECEIVER"
@property
def summary_state(self):
"""Provides ``None`` since the channel has not any values"""
return None
[docs]class ChannelWeatherReceiver(Channel):
type_name = "WEATHER_RECEIVER"
@property
def summary_state(self):
"""Provides ``None`` since the channel has not any values"""
return None
# Devices:
# HM-CC-RT-DN
[docs]class ChannelClimateControlReceiver(Channel):
type_name = "CLIMATECONTROL_RECEIVER"
@property
def summary_state(self):
"""Provides ``None`` since the channel has not any values"""
return None
# Devices:
# HM-CC-RT-DN
[docs]class ChannelClimateControlRTReceiver(Channel):
type_name = "CLIMATECONTROL_RT_RECEIVER"
@property
def summary_state(self):
"""Provides ``None`` since the channel has not any values"""
return None
# Devices:
# HM-CC-RT-DN
[docs]class ChannelRemoteControlReceiver(Channel):
type_name = "REMOTECONTROL_RECEIVER"
@property
def summary_state(self):
"""Provides ``None`` since the channel has not any values"""
return None
# Devices:
# HM-TC-IT-WM-W-EU
[docs]class ChannelWeatherTransmit(Channel):
type_name = "WEATHER_TRANSMIT"
@property
def summary_state(self):
"""Provides the temperature and humidity in readable format."""
return "Temperature: %s, Humidity: %s" % \
(self.values["TEMPERATURE"],
self.values["HUMIDITY"])
# Devices:
# HM-TC-IT-WM-W-EU
[docs]class ChannelThermalControlTransmit(Channel):
type_name = "THERMALCONTROL_TRANSMIT"
def _init_value_spec(self, value_spec):
# The value PARTY_MODE_SUBMIT seems to be declared to be readable by
# the CCU which is wrong. This value can not be read.
# See <https://github.com/LarsMichelsen/pmatic/issues/7>.
if value_spec["ID"] == "PARTY_MODE_SUBMIT":
value_spec["OPERATIONS"] = "2"
super(ChannelThermalControlTransmit, self)._init_value_spec(value_spec)
def _get_values(self):
# This is needed to not let the CCU decide which values to be read from
# the device because of the bug mentioned above.
return self._get_values_single()
# Devices:
# HM-TC-IT-WM-W-EU
[docs]class ChannelSwitchTransmit(Channel):
type_name = "SWITCH_TRANSMIT"
def _init_value_spec(self, value_spec):
# The value SWITCH_TRANSMIT seems to be declared to be readable by
# the CCU which is wrong. This value can not be read.
# See <https://github.com/LarsMichelsen/pmatic/issues/7>.
if value_spec["ID"] == "DECISION_VALUE":
value_spec["OPERATIONS"] = "4" # only supports events
super(ChannelSwitchTransmit, self)._init_value_spec(value_spec)
def _get_values(self):
# This is needed to not let the CCU decide which values to be read from
# the device because of the bug mentioned above.
return self._get_values_single()
[docs]class Devices(object):
"""Manages a collection of CCU devices."""
def __init__(self, ccu):
super(Devices, self).__init__()
if not isinstance(ccu, pmatic.ccu.CCU):
raise PMException("Invalid ccu object provided: %r" % ccu)
self._ccu = ccu
self._device_dict = {}
@property
def _devices(self):
"""Optional initializer of the devices data structure, called on first access."""
return self._device_dict
[docs] def get(self, address, deflt=None):
"""Returns the device matching the given device address.
If there is none matching the given address either None or the value
specified by the optional attribute *deflt* is returned."""
return self._devices.get(address, deflt)
[docs] def add(self, device):
"""Add a :class:`.Device` object to the collection."""
if not isinstance(device, Device):
raise PMException("You can only add device objects.")
self._devices[device.address] = device
[docs] def exists(self, address):
"""Check whether or not a device with the given address is in this collection."""
return address in self._devices
[docs] def addresses(self):
"""Returns a list of all addresses of all initialized devices."""
return self._devices.keys()
[docs] def delete(self, address):
"""Deletes the device with the given address from the pmatic runtime.
The device is not deleted from the CCU.
When the device is not known, the method is tollerating that."""
try:
del self._devices[address]
except KeyError:
pass
[docs] def clear(self):
"""Remove all objects from this devices collection."""
self._devices.clear()
[docs] def get_device_or_channel_by_address(self, address):
"""Returns the device or channel object of the given address.
Raises a KeyError exception when no device exists for this
address in the already fetched objects."""
if ":" in address:
device_address = address.split(":", 1)[0]
return self._devices[device_address].channel_by_address(address)
else:
return self._devices[address]
[docs] def on_value_changed(self, func):
"""Register a function to be called each time a value of a device in this
collection changed."""
for device in self._devices.values():
device.on_value_changed(func)
[docs] def on_value_updated(self, func):
"""Register a function to be called each time a value of a device in this
collection updated."""
for device in self._devices.values():
device.on_value_updated(func)
[docs] def __iter__(self):
"""Provides an iterator over the devices of this collection."""
for value in self._devices.values():
yield value
[docs] def __len__(self):
"""Is e.g. used by :func:`len`. Returns the number of devices in this collection."""
return len(self._devices)
# FIXME: self.channels[0]: Provide better access to the channels. e.g. by names or ids or similar
[docs]class Device(Entity):
_transform_attributes = {
# ReGa attributes:
#"id" : int,
#"deviceId" : int,
#"operateGroupOnly" : lambda v: v != "false",
# Low level attributes:
"flags" : int,
"roaming" : bool,
"updateable" : bool,
"channels" : Channel.from_channel_dicts,
}
# Don't add these keys to the objects attributes
_skip_attributes = [
# Low level attributes:
"children", # not needed
"parent", # not needed
"rf_address", # not available through XML-RPC and API, so exclude at all
"rx_mode", # not available through XML-RPC and API, so exclude at all
]
# These keys have to be set after attribute initialization
_mandatory_attributes = [
# Low level attributes:
# Address of the device
"address",
# Firmware version string
"firmware",
# 0x01: show to user, 0x02 hide from user, 0x08 can not be deleted
"flags",
# serial number of the device
"interface",
# true when the device assignment is automatically adjusted
"roaming",
# device type
"type",
# true when an update is available
"updatable",
# version of the device description
"version",
# list of channel objects
"channels",
]
def __init__(self, ccu, spec):
super(Device, self).__init__(ccu, spec)
@classmethod
[docs] def from_dict(self, ccu, spec):
"""Creates a new device object from the attributes given in the *spec* dictionary.
The *spec* dictionary needs to contain the mandatory attributes with values of the correct
format. Depending on the device type specified by the *spec* dictionary, either a specific
device class or the generic :class:`Device` class is used to create the object."""
device_class = device_classes_by_type_name.get(spec["type"], Device)
return device_class(ccu, spec)
# {u'UNREACH': u'1', u'AES_KEY': u'1', u'UPDATE_PENDING': u'1', u'RSSI_PEER': u'-65535',
# u'LOWBAT': u'0', u'STICKY_UNREACH': u'1', u'DEVICE_IN_BOOTLOADER': u'0',
# u'CONFIG_PENDING': u'0', u'RSSI_DEVICE': u'-65535', u'DUTYCYCLE': u'0'}
@property
def maintenance(self):
"""Returns the :class:`ChannelMaintenance` object of this device. It provides
access to generic maintenance information available on this device."""
return self.channels[0]
[docs] def set_logic_attributes(self, attrs):
"""Used to update the logic attributes of this device.
Applying the attributes in the dictionary to this object. Special handling
for some attributes which are already set by the low level attributes and
for the channel attributes which are also part of attrs."""
for channel_attrs in attrs["channels"]:
self.channels[channel_attrs["index"]].set_logic_attributes(channel_attrs)
# Skip non needed attributes (already set by low level data)
attrs = attrs.copy()
del attrs["channels"]
del attrs["address"]
del attrs["interface"]
del attrs["type"]
self._set_attributes(attrs)
@property
def is_online(self):
"""Is ``True`` when the device is currently reachable. Otherwise it is ``False``."""
if self.type == "HM-RCV-50":
return True # CCU is always assumed to be online
else:
return not self.maintenance.values["UNREACH"].value
@property
def is_battery_low(self):
"""Is ``True`` when the battery is reported to be low.
When the battery is in normal state, it is ``False``. It might be a
non battery powered device, then it is ``None``."""
try:
return self.maintenance.values["LOWBAT"].value
except KeyError:
return None # not battery powered
@property
def has_pending_config(self):
"""Is ``True`` when the CCU has pending configuration changes for this device.
Otherwise it is ``False``."""
if self.type == "HM-RCV-50":
return False
else:
return self.maintenance.values["CONFIG_PENDING"].value
@property
def has_pending_update(self):
"""Is ``True`` when the CCU has a pending firmware update for this device.
Otherwise it is ``False``."""
try:
return self.maintenance.values["UPDATE_PENDING"].value
except KeyError:
return False
@property
def rssi(self):
"""Is a two element tuple of the devices current RSSI (Received Signal Strength Indication).
The first element is the devices RSSI, the second one the CCUs RSSI.
In case of the CCU itself or a non radio device it is set to ``(None, None)``."""
try:
return self.maintenance.values["RSSI_DEVICE"].value, \
self.maintenance.values["RSSI_PEER"].value
except KeyError:
return None, None
#{u'CONTROL': u'NONE', u'OPERATIONS': u'7', u'NAME': u'INHIBIT', u'MIN': u'0',
# u'DEFAULT': u'0', u'MAX': u'1', u'TAB_ORDER': u'6', u'FLAGS': u'1', u'TYPE': u'BOOL',
# u'ID': u'INHIBIT', u'UNIT': u''}
@property
def inhibit(self):
"""The actual inhibit state of the device.
:getter: Whether or not the device is currently locked, provided as
:class:`params.ParameterBOOL`.
:setter: Specify the new inhibit state as boolean.
:type: :class:`params.ParameterBOOL`/bool
"""
return self.maintenance.values["INHIBIT"]
@inhibit.setter
def inhibit(self, state):
self.maintenance.values["INHIBIT"].value = state
@property
def summary_state(self):
"""Provides a textual summary state of the device.
Gives you a string representing some kind of summary state of the device. This
string does not necessarly contain all state information of the devices.
When a device is unreachable, it does only contain this information.
This default method concatenates values and titles of channel values and
provides them as string. The values are sorted by the titles."""
return self._get_summary_state()
def _get_summary_state(self, skip_channel_types=None):
"""Internal helper for :prop:`summary_state`.
It is possible to exclude the states of specific channels by listing the
names of the channel classes in the optional *skip_channel_types* argument."""
formated = []
if not self.is_online:
return "The device is unreachable"
if self.is_battery_low:
formated.append("The battery is low")
if self.has_pending_config:
formated.append("Config pending")
if self.has_pending_update:
formated.append("Update pending")
# FIXME: Add bad rssi?
for channel in self.channels:
if skip_channel_types == None or type(channel).__name__ not in skip_channel_types:
txt = channel.summary_state
if txt != None:
formated.append(txt)
if formated:
return ", ".join(formated)
else:
return "Device reports no operational state"
[docs] def channel_by_address(self, address):
"""Returns the channel object having the requested address.
When the device has no such channel, a KeyError() is raised.
"""
for channel in self.channels:
if address == channel.address:
return channel
raise KeyError("The channel could not be found on this device.")
[docs] def on_value_changed(self, func):
"""Register a function to be called each time a value of this device has changed."""
for channel in self.channels:
channel.on_value_changed(func)
[docs] def on_value_updated(self, func):
"""Register a function to be called each time a value of this device has updated."""
for channel in self.channels:
channel.on_value_updated(func)
# Funk-Heizkörperthermostat
# TODO:
#{u'CONTROL': u'NONE', u'OPERATIONS': u'5', u'NAME': u'FAULT_REPORTING', u'MIN': u'0',
# u'DEFAULT': u'0', u'MAX': u'7', u'VALUE_LIST': u'NO_FAULT VALVE_TIGHT
# ADJUSTING_RANGE_TOO_LARGE ADJUSTING_RANGE_TOO_SMALL COMMUNICATION_ERROR {}
# LOWBAT VALVE_ERROR_POSITION', u'TAB_ORDER': u'1', u'FLAGS': u'9',
# u'TYPE': u'ENUM', u'ID': u'FAULT_REPORTING', u'UNIT': u''}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_TEMP', u'OPERATIONS': u'3', u'NAME': u'PARTY_TEMPERATURE',
# u'MIN': u'5.000000', u'DEFAULT': u'20.000000', u'MAX': u'30.000000', u'TAB_ORDER': u'13',
# u'FLAGS': u'1', u'TYPE': u'FLOAT', u'ID': u'PARTY_TEMPERATURE', u'UNIT': u'\xb0C'}
#{u'CONTROL': u'NONE', u'OPERATIONS': u'2', u'NAME': u'PARTY_MODE_SUBMIT', u'MIN': u'',
# u'DEFAULT': u'', u'MAX': u'', u'TAB_ORDER': u'12', u'FLAGS': u'1', u'TYPE': u'STRING',
# u'ID': u'PARTY_MODE_SUBMIT', u'UNIT': u''}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_START_TIME', u'OPERATIONS': u'3',
# u'NAME': u'PARTY_START_TIME', u'MIN': u'0', u'DEFAULT': u'0', u'MAX': u'1410',
# u'TAB_ORDER': u'14', u'FLAGS': u'1', u'TYPE': u'INTEGER', u'ID': u'PARTY_START_TIME',
# u'UNIT': u'minutes'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_START_DAY', u'OPERATIONS': u'3', u'NAME': u'PARTY_START_DAY',
# u'MIN': u'1', u'DEFAULT': u'1', u'MAX': u'31', u'TAB_ORDER': u'15', u'FLAGS': u'1',
# u'TYPE': u'INTEGER', u'ID': u'PARTY_START_DAY', u'UNIT': u'day'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_START_MONTH', u'OPERATIONS': u'3',
# u'NAME': u'PARTY_START_MONTH', u'MIN': u'1', u'DEFAULT': u'1', u'MAX': u'12',
# u'TAB_ORDER': u'16', u'FLAGS': u'1', u'TYPE': u'INTEGER', u'ID':
# u'PARTY_START_MONTH', u'UNIT': u'month'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_START_YEAR', u'OPERATIONS': u'3',
# u'NAME': u'PARTY_START_YEAR', u'MIN': u'0', u'DEFAULT': u'12', u'MAX': u'99',
# u'TAB_ORDER': u'17', u'FLAGS': u'1', u'TYPE': u'INTEGER', u'ID': u'PARTY_START_YEAR',
# u'UNIT': u'year'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_STOP_TIME', u'OPERATIONS': u'3',
# u'NAME': u'PARTY_STOP_TIME', u'MIN': u'0', u'DEFAULT': u'0', u'MAX': u'1410',
# u'TAB_ORDER': u'18', u'FLAGS': u'1', u'TYPE': u'INTEGER', u'ID': u'PARTY_STOP_TIME',
# u'UNIT': u'minutes'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_STOP_DAY', u'OPERATIONS': u'3', u'NAME': u'PARTY_STOP_DAY',
# u'MIN': u'1', u'DEFAULT': u'1', u'MAX': u'31', u'TAB_ORDER': u'19', u'FLAGS': u'1',
# u'TYPE': u'INTEGER', u'ID': u'PARTY_STOP_DAY', u'UNIT': u'day'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_STOP_MONTH', u'OPERATIONS': u'3', u'NAME': u'PARTY_STOP_MONTH',
# u'MIN': u'1', u'DEFAULT': u'1', u'MAX': u'12', u'TAB_ORDER': u'20', u'FLAGS': u'1',
# u'TYPE': u'INTEGER', u'ID': u'PARTY_STOP_MONTH', u'UNIT': u'month'}
#{u'CONTROL': u'HEATING_CONTROL.PARTY_STOP_YEAR', u'OPERATIONS': u'3', u'NAME': u'PARTY_STOP_YEAR',
# u'MIN': u'0', u'DEFAULT': u'12', u'MAX': u'99', u'TAB_ORDER': u'21', u'FLAGS': u'1',
# u'TYPE': u'INTEGER', u'ID': u'PARTY_STOP_YEAR', u'UNIT': u'year'}
[docs]class HM_CC_RT_DN(Device):
type_name = "HM-CC-RT-DN"
@property
def temperature(self):
"""Provides the current temperature.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[4].values["ACTUAL_TEMPERATURE"]
#{u'CONTROL': u'NONE', u'OPERATIONS': u'5', u'NAME': u'VALVE_STATE', u'MIN': u'0',
# u'DEFAULT': u'0', u'MAX': u'99', u'TAB_ORDER': u'3', u'FLAGS': u'1', u'TYPE': u'INTEGER',
# u'ID': u'VALVE_STATE', u'UNIT': u'%'}
@property
def valve_state(self):
"""Provides the current valve state in percentage.
Returns an instance of :class:`ParameterINTEGER`.
"""
return self.channels[4].values["VALVE_STATE"]
@property
def set_temperature(self):
"""The actual set temperature of the device.
:getter: Provides the actual target temperature as :class:`ParameterFLOAT`.
:setter: Specify the new set temperature as float. Please note that the CCU rounds
this values to
.0 or .5 after the comma. So if you provide .e.g 22.1 as new set temperature,
the CCU will convert this to 22.0. This is totally equal to the control on the
device.
:type: ParameterFloat/float
"""
return self.channels[4].values["SET_TEMPERATURE"]
@set_temperature.setter
def set_temperature(self, target):
self.channels[4].values["SET_TEMPERATURE"].value = target
# {u'CONTROL': u'HEATING_CONTROL.COMFORT', u'OPERATIONS': u'2', u'NAME': u'COMFORT_MODE',
# u'MIN': u'0', u'DEFAULT': u'0', u'MAX': u'1', u'TAB_ORDER': u'10', u'FLAGS': u'1',
# u'TYPE': u'ACTION', u'ID': u'COMFORT_MODE', u'UNIT': u''}
[docs] def set_temperature_comfort(self):
"""Sets the :attr:`set_temperature` to the configured comfort temperature"""
self.channels[4].values["COMFORT_MODE"].value = True
#{u'CONTROL': u'HEATING_CONTROL.LOWERING', u'OPERATIONS': u'2', u'NAME': u'LOWERING_MODE',
# u'MIN': u'0', u'DEFAULT': u'0', u'MAX': u'1', u'TAB_ORDER': u'11', u'FLAGS': u'1',
# u'TYPE': u'ACTION', u'ID': u'LOWERING_MODE', u'UNIT': u''}
[docs] def set_temperature_lowering(self):
"""Sets the :attr:`set_temperature` to the configured lowering temperature"""
self.channels[4].values["LOWERING_MODE"].value = True
@property
def is_off(self):
"""Is set to `True` when the device is not enabled to heat."""
return self.channels[4].values["SET_TEMPERATURE"].value == 4.5
[docs] def turn_off(self):
"""Call this method to tell the thermostat that it should not heat."""
self.set_temperature = 4.5
@property
def control_mode(self):
"""
The actual control mode of the device. This is either ``AUTO``, ``MANUAL``,
``PARTY`` or ``BOOST``.
:getter: Provides the current control mode as :class:`ParameterENUM`.
:setter: Set the control mode by the name of the mode (see above). When setting
to ``MANUAL`` it uses either the current set temperature as target
temperature or the default temperature when the
device is currently turned off.
:type: ParameterENUM/string
"""
return self.channels[4].values["CONTROL_MODE"]
@control_mode.setter
def control_mode(self, mode):
modes = ["AUTO", "MANUAL", "PARTY", "BOOST"]
if mode not in modes:
raise PMException("The control mode must be one of: %s" % ", ".join(modes))
if mode == "MANUAL":
mode = "MANU"
value = True
# In manual mode the set temperature needs to be provided. Set it to the
# current set temperature. When the set temperature is "off", use the default
# value.
if mode == "MANU":
if self.is_off:
value = self.set_temperature.default
# Also set the set_temperature attribute
self.set_temperature = value
else:
value = self.set_temperature.value
self.channels[4].values["%s_MODE" % mode].value = value
@property
def is_battery_low(self):
"""Is ``True`` when the battery is reported to be low, otherwise ``False``.
If you want more details about the current battery, use :meth:`battery_state` to get
the current reported voltage."""
return self.channels[4].values["FAULT_REPORTING"].formated() == "LOWBAT"
@property
def battery_state(self):
"""Provides the actual battery voltage reported by the device."""
return self.channels[4].values["BATTERY_STATE"]
# {u'CONTROL': u'NONE', u'OPERATIONS': u'5', u'NAME': u'BOOST_STATE', u'MIN': u'0',
# u'DEFAULT': u'0', u'MAX': u'30', u'TAB_ORDER': u'4', u'FLAGS': u'1',
# u'TYPE': u'INTEGER', u'ID': u'BOOST_STATE', u'UNIT': u'min'}
@property
def boost_duration(self):
"""When boost mode is currently active this returns the number of minutes left
in boost mode. Otherwise it returns ``None``.
Provides the configured boost duration as :class:`ParameterINTEGER`.
"""
if self.control_mode == "BOOST":
return self.channels[4].values["BOOST_STATE"]
# Funk-Temperatur-/Luftfeuchtesensor OTH
[docs]class HM_WDS10_TH_O(Device):
type_name = "HM-WDS10-TH-O"
@property
def temperature(self):
"""Provides the current temperature.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[1].values["TEMPERATURE"]
@property
def humidity(self):
"""Provides the current humidity.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[1].values["HUMIDITY"]
# Funk-Temperatur-/Luftfeuchtesensor ITH
[docs]class HM_WDS40_TH_I_2(Device):
type_name = "HM-WDS40-TH-I-2"
@property
def temperature(self):
"""Provides the current temperature.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[1].values["TEMPERATURE"]
@property
def humidity(self):
"""Provides the current humidity.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[1].values["HUMIDITY"]
# Funk-Außen-Helligkeitssensor OLI
[docs]class HM_Sen_LI_O(Device):
type_name = "HM-Sen-LI-O"
@property
def brightness(self):
"""Provides the current brightness.
Returns an instance of :class:`ParameterFLOAT`.
"""
return self.channels[1].values["LUX"]
# Virtuelle Fernbedienung der CCU
class HM_RCV_50(Device):
type_name = "HM-RCV-50"
# Funk-Tür-/ Fensterkontakt
[docs]class HM_Sec_SC(Device):
type_name = "HM-Sec-SC"
# Make methods of ChannelShutterContact() available
def __getattr__(self, attr):
return getattr(self.channels[1], attr)
# Optischer Funk-Tür-/ Fensterkontakt
[docs]class HM_Sec_SCo(HM_Sec_SC):
type_name = "HM-Sec-SCo"
# Funk-Schaltaktor mit Leistungsmessung
[docs]class HM_ES_PMSw1_Pl(Device):
type_name = "HM-ES-PMSw1-Pl"
# Make methods of ChannelSwitch() available
def __getattr__(self, attr):
return getattr(self.channels[1], attr)
@property
def summary_state(self):
return super(HM_ES_PMSw1_Pl, self)._get_summary_state(
skip_channel_types=["ChannelConditionPower", "ChannelConditionCurrent",
"ChannelConditionVoltage", "ChannelConditionFrequency"])
# Funk-Schaltaktor ohne Leistungsmessung
[docs]class HM_LC_Sw1_Pl_DN_R1(Device):
type_name = "HM-LC-Sw1-Pl-DN-R1"
# Make methods of ChannelSwitch() available
def __getattr__(self, attr):
return getattr(self.channels[1], attr)
@property
def summary_state(self):
return super(HM_LC_Sw1_Pl_DN_R1, self)._get_summary_state()
@property
def switch(self):
"""Provides to the :class:`.ChannelKey` object of the switch.
You can do something like ``self.switch.switch_on()`` with this. For details take
a look at the methods provided by the :class:``.ChannelKey`` class."""
return self.channels[1]
# Funk-Rolladenaktor
[docs]class HM_LC_Bl1PBU_FM(Device):
type_name = "HM-LC-Bl1PBU-FM"
# Make methods of ChannelBlind() available
def __getattr__(self, attr):
return getattr(self.channels[1], attr)
@property
def blind(self):
"""Provides to the :class:`.ChannelKey` object of the blind channel.
You can do something like ``self.blind.set_level(0.6)`` with this. For details take
a look at the methods provided by the :class:``.ChannelKey`` class."""
return self.channels[1]
[docs]class HM_PBI_4_FM(Device):
type_name = "HM-PBI-4-FM"
@property
def switch1(self):
"""Provides to the :class:`.ChannelKey` object of the first switch.
You can do something like ``self.switch1.press_short()`` with this. For details take
a look at the methods provided by the :class:``.ChannelKey`` class."""
return self.channels[1]
@property
def switch2(self):
"""Provides to the :class:`.ChannelKey` object of the second switch."""
return self.channels[2]
@property
def switch3(self):
"""Provides to the :class:`.ChannelKey` object of the third switch."""
return self.channels[3]
@property
def switch4(self):
"""Provides to the :class:`.ChannelKey` object of the fourth switch."""
return self.channels[4]
[docs]class Rooms(object):
"""Manages a collection of rooms."""
def __init__(self, ccu):
super(Rooms, self).__init__()
if not isinstance(ccu, pmatic.ccu.CCU):
raise PMException("Invalid ccu object provided: %r" % ccu)
self._ccu = ccu
self._room_dict = {}
@property
def _rooms(self):
"""Optional initializer of the rooms data structure, called on first access."""
return self._room_dict
[docs] def get(self, room_id, deflt=None):
"""Returns the :class:`Room` matching the given room id.
If there is none matching the given ID either None or the value
specified by the optional attribute *deflt* is returned."""
return self._rooms.get(room_id, deflt)
@property
def ids(self):
"""Provides a sorted list of all ids of all initialized room."""
return sorted(self._rooms.keys())
[docs] def add(self, room):
"""Add a :class:`Room` to the collection."""
if not isinstance(room, Room):
raise PMException("You can only add Room objects.")
self._rooms[room.id] = room
[docs] def exists(self, room_id):
"""Check whether or not a :class:`Room` with the given id is in this collection."""
return room_id in self._rooms
[docs] def delete(self, room_id):
"""Deletes the :class:`Room` with the given id from the pmatic runtime.
The room is not deleted from the CCU. When the room is not known, the method is
tollerating that."""
try:
del self._rooms[room_id]
except KeyError:
pass
[docs] def clear(self):
"""Remove all :class:`Room` objects from this collection."""
self._rooms.clear()
[docs] def __iter__(self):
"""Provides an iterator over the rooms of this collection."""
for value in self._rooms.values():
yield value
[docs] def __len__(self):
"""Is e.g. used by :func:`len`. Returns the number of rooms in this collection."""
return len(self._rooms)
[docs]class Room(Entity):
_transform_attributes = {
"id" : int,
"channelIds" : lambda x: list(map(int, x)),
}
def __init__(self, ccu, spec):
self._values = {}
self._devices = None
super(Room, self).__init__(ccu, spec)
#@classmethod
#def get_rooms(self, api):
# """Returns a list of all currently configured :class:`.Room` instances."""
# rooms = []
# for room_dict in api.room_get_all():
# rooms.append(Room(api, room_dict))
# return rooms
@property
def devices(self):
"""Provides access to a collection of :class:`.Device` objects which have at least one
channel associated with this room.
The collections is a :class:`.Devices` instance."""
if not self._devices:
self._devices = self._ccu.devices.query(has_channel_ids=self.channel_ids)
return self._devices
@property
def channels(self):
"""Holds a list of channel objects associated with this room."""
# FIXME: Cache this?
room_channels = []
for device in self.devices:
for channel in device.channels:
if channel.id in self.channel_ids:
room_channels.append(channel)
return room_channels
# @property
# def programs(self):
# """Returns list of program objects which use at least one channel associated
# with this room."""
# # FIXME: Implement!
# # FIXME: Cache this?
# return []
#
#
# def add(self, channel):
# """Adds a channel to this room."""
# # FIXME: Implement!
#
#
# def remove(self, channel):
# """Removes a channel to this room."""
# # FIXME: Implement!
# Build a list of all specific product classes. If a device is initialized
# Device() checks whether or not a specific class or the generic Device()
# class should be used to initialize an object.
device_classes_by_type_name = {}
for key, val in list(globals().items()):
if isinstance(val, type):
if issubclass(val, Device) and key != "Device":
device_classes_by_type_name[val.type_name] = val
channel_classes_by_type_name = {}
for key, val in list(globals().items()):
if isinstance(val, type):
if issubclass(val, Channel) and val != Channel:
channel_classes_by_type_name[val.type_name] = val