# coding=utf-8
import json
from collections import Counter
from typing import Iterable
from homematicip.base.enums import *
from homematicip.base.functionalChannels import FunctionalChannel
from homematicip.base.helpers import get_functional_channel, get_functional_channels
from homematicip.base.homematicip_object import HomeMaticIPObject
from homematicip.group import Group
LOGGER = logging.getLogger(__name__)
[docs]
class BaseDevice(HomeMaticIPObject):
"""Base device class. This is the foundation for homematicip and external (hue) devices"""
_supportedFeatureAttributeMap = {}
def __init__(self, connection):
super().__init__(connection)
self.id = None
self.homeId = None
self.label = None
self.connectionType = ConnectionType.HMIP_LAN
self.deviceArchetype = DeviceArchetype.HMIP
self.lastStatusUpdate = None
self.firmwareVersion = None
self.modelType = ""
self.permanentlyReachable = False
self.functionalChannels = []
self.functionalChannelCount = Counter()
self.deviceType = None
# must be imported in init. otherwise we have cross import issues
from homematicip.class_maps import TYPE_FUNCTIONALCHANNEL_MAP
self._typeFunctionalChannelMap = TYPE_FUNCTIONALCHANNEL_MAP
[docs]
def from_json(self, js):
super().from_json(js)
self.id = js["id"]
self.homeId = js["homeId"]
self.label = js["label"]
self.lastStatusUpdate = self.fromtimestamp(js["lastStatusUpdate"])
self.firmwareVersion = js["firmwareVersion"]
self.modelType = js["modelType"]
self.permanentlyReachable = js["permanentlyReachable"]
self.deviceType = js["type"]
self.connectionType = ConnectionType.from_str(js["connectionType"])
if "deviceArchetype" in js:
self.deviceArchetype = DeviceArchetype.from_str(js["deviceArchetype"])
[docs]
def load_functionalChannels(
self, groups: Iterable[Group], channels: Iterable[FunctionalChannel]
):
"""this function will load the functionalChannels into the device"""
for channel in self._rawJSONData["functionalChannels"].values():
items = [
ch for ch in self.functionalChannels if ch.index == channel["index"]
]
fc = items[0] if items else None
if fc is not None:
fc.from_json(channel, groups)
else:
fc = self._parse_functionalChannel(channel, groups)
channels.append(fc)
self.functionalChannels.append(fc)
self.functionalChannelCount = Counter(
x.functionalChannelType for x in self.functionalChannels
)
def _parse_functionalChannel(self, json_state, groups: Iterable[Group]):
fc = None
try:
channelType = FunctionalChannelType.from_str(
json_state["functionalChannelType"]
)
fc = self._typeFunctionalChannelMap[channelType](self, self._connection)
fc.from_json(json_state, groups)
except:
fc = self._typeFunctionalChannelMap[
FunctionalChannelType.FUNCTIONAL_CHANNEL
](self, self._connection)
fc.from_json(json_state, groups)
fc.device = self
LOGGER.warning(
"There is no class for functionalChannel '%s' yet",
json_state["functionalChannelType"],
)
return fc
[docs]
class Device(BaseDevice):
"""this class represents a generic homematic ip device"""
_supportedFeatureAttributeMap = {
"IFeatureBusConfigMismatch": ["busConfigMismatch"],
"IFeatureDeviceCoProError": ["coProFaulty"],
"IFeatureDeviceCoProRestart": ["coProRestartNeeded"],
"IFeatureDeviceCoProUpdate": ["coProUpdateFailure"],
"IFeatureDeviceIdentify": [],
"IFeatureDeviceOverheated": ["deviceOverheated"],
"IFeatureDeviceOverloaded": ["deviceOverloaded"],
"IFeatureDeviceParticulateMatterSensorCommunicationError": "particulateMatterSensorCommunicationError",
"IFeatureDeviceParticulateMatterSensorError": "particulateMatterSensorError",
"IFeatureDevicePowerFailure": ["devicePowerFailureDetected"],
"IFeatureDeviceSensorCommunicationError": ["sensorCommunicationError"],
"IFeatureDeviceSensorError": ["sensorError"],
"IFeatureDeviceTemperatureHumiditySensorCommunicationError": "temperatureHumiditySensorCommunicationError",
"IFeatureDeviceTemperatureHumiditySensorError": "temperatureHumiditySensorError",
"IFeatureDeviceTemperatureOutOfRange": ["temperatureOutOfRange"],
"IFeatureDeviceUndervoltage": ["deviceUndervoltage"],
"IFeatureMinimumFloorHeatingValvePosition": ["minimumFloorHeatingValvePosition"],
"IFeatureMulticastRouter": ["multicastRoutingEnabled"],
"IFeaturePowerShortCircuit": ["powerShortCircuit"],
"IFeatureProfilePeriodLimit": [],
"IFeaturePulseWidthModulationAtLowFloorHeatingValvePosition": [
"pulseWidthModulationAtLowFloorHeatingValvePositionEnabled"],
"IFeatureRssiValue": ["rssiDeviceValue"],
"IFeatureShortCircuitDataLine": ["shortCircuitDataLine"],
"IOptionalFeatureColorTemperature": "colorTemperature",
# "IOptionalFeatureColorTemperatureDim2Warm": false,
# "IOptionalFeatureColorTemperatureDynamicDaylight": false,
"IOptionalFeatureControlsMountingOrientation": ["controlsMountingOrientation"],
"IOptionalFeatureDeviceErrorLockJammed": ["lockJammed"],
"IOptionalFeatureDisplayContrast": [],
"IOptionalFeatureDutyCycle": ["dutyCycle"],
"IOptionalFeatureFilteredMulticastRouter": ["filteredMulticastRoutingEnabled"],
# "IOptionalFeatureHardwareColorTemperature": false,
# "IOptionalFeatureHueSaturationValue": false,
# "IOptionalFeatureLightScene": false,
# "IOptionalFeatureLightSceneWithShortTimes": false,
"IOptionalFeatureLowBat": ["lowBat"],
"IOptionalFeatureMountingOrientation": ["mountingOrientation"],
}
def __init__(self, connection):
super().__init__(connection)
self.liveUpdateState = None
self.updateState = DeviceUpdateState.UP_TO_DATE
self.availableFirmwareVersion = None
self.firmwareVersionInteger = (
0 # firmwareVersion = A.B.C -> firmwareVersionInteger ((A<<16)|(B<<8)|C)
)
self.unreach = False
self.lowBat = False
self.routerModuleSupported = False
self.routerModuleEnabled = False
self.modelId = 0
self.oem = ""
self.manufacturerCode = 0
self.serializedGlobalTradeItemNumber = ""
self.rssiDeviceValue = 0
self.rssiPeerValue = 0
self.dutyCycle = False
self.configPending = False
self._baseChannel = "DEVICE_BASE"
self.deviceOverheated = False
self.deviceOverloaded = False
self.deviceUndervoltage = False
self.temperatureOutOfRange = False
self.coProFaulty = False
self.coProRestartNeeded = False
self.coProUpdateFailure = False
self.busConfigMismatch = False
self.shortCircuitDataLine = False
self.powerShortCircuit = False
self.deviceUndervoltage = False
self.devicePowerFailureDetected = False
self.deviceIdentifySupported = (
False # just placeholder at the moment the feature doesn't set any values
)
[docs]
def from_json(self, js):
super().from_json(js)
self.updateState = DeviceUpdateState.from_str(js["updateState"])
self.firmwareVersionInteger = js["firmwareVersionInteger"]
self.availableFirmwareVersion = js["availableFirmwareVersion"]
self.modelId = js["modelId"]
self.oem = js["oem"]
self.manufacturerCode = js["manufacturerCode"]
self.serializedGlobalTradeItemNumber = js["serializedGlobalTradeItemNumber"]
self.liveUpdateState = LiveUpdateState.from_str(js["liveUpdateState"])
c = get_functional_channel(self._baseChannel, js)
if c:
self.set_attr_from_dict("lowBat", c)
self.set_attr_from_dict("unreach", c)
self.set_attr_from_dict("rssiDeviceValue", c)
self.set_attr_from_dict("rssiPeerValue", c)
self.set_attr_from_dict("configPending", c)
self.set_attr_from_dict("dutyCycle", c)
self.routerModuleSupported = c["routerModuleSupported"]
self.routerModuleEnabled = c["routerModuleEnabled"]
sof = c.get("supportedOptionalFeatures")
if sof:
for k, v in sof.items():
if v:
if k in Device._supportedFeatureAttributeMap:
for attribute in Device._supportedFeatureAttributeMap[k]:
self.set_attr_from_dict(attribute, c)
else: # pragma: no cover
LOGGER.warning(
"Optional Device Feature '%s' is not yet supported",
k,
)
def __str__(self):
return f"{self.modelType} {self.label} {self.str_from_attr_map()}"
[docs]
def set_label(self, label):
return self._run_non_async(lambda: self.set_label_async(label))
[docs]
async def set_label_async(self, label):
data = {"deviceId": self.id, "label": label}
return await self._rest_call_async("device/setDeviceLabel", data)
[docs]
def is_update_applicable(self):
return self._run_non_async(self.is_update_applicable_async)
[docs]
async def is_update_applicable_async(self):
data = {"deviceId": self.id}
return await self._rest_call_async("device/isUpdateApplicable", data)
[docs]
def authorizeUpdate(self):
return self._run_non_async(self.authorizeUpdate_async)
[docs]
async def authorizeUpdate_async(self):
data = {"deviceId": self.id}
return await self._rest_call_async("device/authorizeUpdate", data)
[docs]
def delete(self):
return self._run_non_async(self.delete_async)
[docs]
async def delete_async(self):
data = {"deviceId": self.id}
return await self._rest_call_async("device/deleteDevice", data)
[docs]
def set_router_module_enabled(self, enabled=True):
return self._run_non_async(lambda: self.set_router_module_enabled_async(enabled))
[docs]
async def set_router_module_enabled_async(self, enabled=True):
if not self.routerModuleSupported:
return False
data = {"deviceId": self.id, "channelIndex": 0, "routerModuleEnabled": enabled}
return await self._rest_call_async(
"device/configuration/setRouterModuleEnabled", data
)
[docs]
class ExternalDevice(BaseDevice):
"""Represents devices with archtetype EXTERNAL"""
def __init__(self, connection):
super().__init__(connection)
self.hasCustomLabel = None
self.externalService = ""
self.supported = None
self._baseChannel = "EXTERNAL_BASE_CHANNEL"
[docs]
def from_json(self, js):
super().from_json(js)
self.hasCustomLabel = js["hasCustomLabel"]
self.externalService = js["externalService"]
self.supported = js["supported"]
[docs]
class HomeControlAccessPoint(Device):
def __init__(self, connection):
super().__init__(connection)
self.dutyCycleLevel = 0.0
self.accessPointPriority = 0
self.signalBrightness = 0
self._baseChannel = "ACCESS_CONTROLLER_CHANNEL"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel(self._baseChannel, js)
if c:
self.set_attr_from_dict("dutyCycleLevel", c)
self.set_attr_from_dict("accessPointPriority", c)
self.set_attr_from_dict("signalBrightness", c)
[docs]
class WiredDinRailAccessPoint(Device):
def __init__(self, connection):
super().__init__(connection)
self.accessPointPriority = 0
self.signalBrightness = 0
self._baseChannel = "ACCESS_CONTROLLER_WIRED_CHANNEL"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel(self._baseChannel, js)
if c:
self.set_attr_from_dict("accessPointPriority", c)
self.set_attr_from_dict("signalBrightness", c)
[docs]
class SabotageDevice(Device):
def __init__(self, connection):
super().__init__(connection)
self.sabotage = None
self._baseChannel = "DEVICE_SABOTAGE"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel(self._baseChannel, js)
if c:
self.set_attr_from_dict("sabotage", c)
[docs]
class OperationLockableDevice(Device):
def __init__(self, connection):
super().__init__(connection)
self.operationLockActive = None
self._baseChannel = "DEVICE_OPERATIONLOCK"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel(self._baseChannel, js)
if c:
self.set_attr_from_dict("operationLockActive", c)
[docs]
def set_operation_lock(self, operationLock=True):
return self._run_non_async(lambda: self.set_operation_lock_async(operationLock))
[docs]
async def set_operation_lock_async(self, operationLock=True):
data = {"channelIndex": 0, "deviceId": self.id, "operationLock": operationLock}
return await self._rest_call_async("device/configuration/setOperationLock", data)
[docs]
class HeatingThermostat(OperationLockableDevice):
"""HMIP-eTRV (Radiator Thermostat)"""
def __init__(self, connection):
super().__init__(connection)
#:float: the offset temperature for the thermostat (+/- 3.5)
self.temperatureOffset = 0.0
#:float: the current position of the valve 0.0 = closed, 1.0 max opened
self.valvePosition = 0.0
#:ValveState: the current state of the valve
self.valveState = ValveState.ERROR_POSITION
#:float: the current temperature which should be reached in the room
self.setPointTemperature = 0.0
#:float: the current measured temperature at the valve
self.valveActualTemperature = 0.0
#:bool: must the adaption re-run?
self.automaticValveAdaptionNeeded = False
[docs]
def from_json(self, js):
super().from_json(js)
automaticValveAdaptionNeeded = js["automaticValveAdaptionNeeded"]
c = get_functional_channel("HEATING_THERMOSTAT_CHANNEL", js)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.valvePosition = c["valvePosition"]
self.valveState = ValveState.from_str(c["valveState"])
self.setPointTemperature = c["setPointTemperature"]
self.valveActualTemperature = c["valveActualTemperature"]
def __str__(self):
return "{} valvePosition({}) valveState({}) temperatureOffset({}) setPointTemperature({}) valveActualTemperature({})".format(
super().__str__(),
self.valvePosition,
self.valveState,
self.temperatureOffset,
self.setPointTemperature,
self.valveActualTemperature,
)
[docs]
class HeatingThermostatCompact(SabotageDevice):
"""HMIP-eTRV-C (Heating-thermostat compact without display)"""
def __init__(self, connection):
super().__init__(connection)
#:float: the offset temperature for the thermostat (+/- 3.5)
self.temperatureOffset = 0.0
#:float: the current position of the valve 0.0 = closed, 1.0 max opened
self.valvePosition = 0.0
#:ValveState: the current state of the valve
self.valveState = ValveState.ERROR_POSITION
#:float: the current temperature which should be reached in the room
self.setPointTemperature = 0.0
#:float: the current measured temperature at the valve
self.valveActualTemperature = 0.0
#:bool: must the adaption re-run?
self.automaticValveAdaptionNeeded = False
[docs]
def from_json(self, js):
super().from_json(js)
automaticValveAdaptionNeeded = js["automaticValveAdaptionNeeded"]
c = get_functional_channel("HEATING_THERMOSTAT_CHANNEL", js)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.valvePosition = c["valvePosition"]
self.valveState = ValveState.from_str(c["valveState"])
self.setPointTemperature = c["setPointTemperature"]
self.valveActualTemperature = c["valveActualTemperature"]
def __str__(self):
return "{} valvePosition({}) valveState({}) temperatureOffset({}) setPointTemperature({}) valveActualTemperature({})".format(
super().__str__(),
self.valvePosition,
self.valveState,
self.temperatureOffset,
self.setPointTemperature,
self.valveActualTemperature,
)
[docs]
class HeatingThermostatEvo(OperationLockableDevice):
"""HMIP-eTRV-E (Heating-thermostat new evo version)"""
def __init__(self, connection):
super().__init__(connection)
#:float: the offset temperature for the thermostat (+/- 3.5)
self.temperatureOffset = 0.0
#:float: the current position of the valve 0.0 = closed, 1.0 max opened
self.valvePosition = 0.0
#:ValveState: the current state of the valve
self.valveState = ValveState.ERROR_POSITION
#:float: the current temperature which should be reached in the room
self.setPointTemperature = 0.0
#:float: the current measured temperature at the valve
self.valveActualTemperature = 0.0
#:bool: must the adaption re-run?
self.automaticValveAdaptionNeeded = False
[docs]
def from_json(self, js):
super().from_json(js)
automaticValveAdaptionNeeded = js["automaticValveAdaptionNeeded"]
c = get_functional_channel("HEATING_THERMOSTAT_CHANNEL", js)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.valvePosition = c["valvePosition"]
self.valveState = ValveState.from_str(c["valveState"])
self.setPointTemperature = c["setPointTemperature"]
self.valveActualTemperature = c["valveActualTemperature"]
def __str__(self):
return "{} valvePosition({}) valveState({}) temperatureOffset({}) setPointTemperature({}) valveActualTemperature({})".format(
super().__str__(),
self.valvePosition,
self.valveState,
self.temperatureOffset,
self.setPointTemperature,
self.valveActualTemperature,
)
[docs]
class RotaryHandleSensor(SabotageDevice):
"""HMIP-SRH"""
def __init__(self, connection):
super().__init__(connection)
self.windowState = WindowState.CLOSED
self.eventDelay = None
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("ROTARY_HANDLE_CHANNEL", js)
if c:
self.windowState = WindowState.from_str(c["windowState"])
self.eventDelay = c["eventDelay"]
def __str__(self):
return "{} windowState({})".format(super().__str__(), self.windowState)
[docs]
class TemperatureHumiditySensorOutdoor(Device):
"""HMIP-STHO (Temperature and Humidity Sensor outdoor)"""
def __init__(self, connection):
super().__init__(connection)
self.actualTemperature = 0
self.humidity = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("CLIMATE_SENSOR_CHANNEL", js)
if c:
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.vaporAmount = c["vaporAmount"]
def __str__(self):
return "{} actualTemperature({}) humidity({}) vaporAmount({})".format(
super().__str__(), self.actualTemperature, self.humidity, self.vaporAmount
)
[docs]
class TemperatureHumiditySensorWithoutDisplay(Device):
"""HMIP-STH (Temperature and Humidity Sensor without display - indoor)"""
def __init__(self, connection):
super().__init__(connection)
self.temperatureOffset = 0
self.actualTemperature = 0
self.humidity = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel(
"WALL_MOUNTED_THERMOSTAT_WITHOUT_DISPLAY_CHANNEL", js
)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.vaporAmount = c["vaporAmount"]
def __str__(self):
return "{} actualTemperature({}) humidity({}) vaporAmount({})".format(
super().__str__(), self.actualTemperature, self.humidity, self.vaporAmount
)
[docs]
class TemperatureHumiditySensorDisplay(Device):
"""HMIP-STHD (Temperature and Humidity Sensor with display - indoor)"""
def __init__(self, connection):
super().__init__(connection)
self.temperatureOffset = 0
self.display = ClimateControlDisplay.ACTUAL
self.actualTemperature = 0
self.humidity = 0
self.setPointTemperature = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("WALL_MOUNTED_THERMOSTAT_PRO_CHANNEL", js)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.display = ClimateControlDisplay.from_str(c["display"])
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.setPointTemperature = c["setPointTemperature"]
self.vaporAmount = c["vaporAmount"]
[docs]
def set_display(self, display: ClimateControlDisplay = ClimateControlDisplay.ACTUAL):
return self._run_non_async(lambda: self.set_display_async(display))
[docs]
async def set_display_async(self, display: ClimateControlDisplay = ClimateControlDisplay.ACTUAL):
data = {"channelIndex": 1, "deviceId": self.id, "display": str(display)}
return await self._rest_call_async(
"device/configuration/setClimateControlDisplay", data
)
def __str__(self):
return "{} actualTemperature({}) humidity({}) vaporAmount({}) setPointTemperature({})".format(
super().__str__(),
self.actualTemperature,
self.humidity,
self.vaporAmount,
self.setPointTemperature,
)
[docs]
class WallMountedThermostatPro(
TemperatureHumiditySensorDisplay, OperationLockableDevice
):
"""HMIP-WTH, HMIP-WTH-2 (Wall Thermostat with Humidity Sensor) / HMIP-BWTH (Brand Wall Thermostat with Humidity Sensor)"""
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("WALL_MOUNTED_THERMOSTAT_PRO_CHANNEL", js)
if c:
self.temperatureOffset = c["temperatureOffset"]
self.display = ClimateControlDisplay.from_str(c["display"])
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.setPointTemperature = c["setPointTemperature"]
[docs]
class RoomControlDevice(WallMountedThermostatPro):
"""ALPHA-IP-RBG (Alpha IP Wall Thermostat Display)"""
pass
[docs]
class RoomControlDeviceAnalog(Device):
"""ALPHA-IP-RBGa (ALpha IP Wall Thermostat Display analog)"""
def __init__(self, connection):
super().__init__(connection)
self.actualTemperature = 0.0
self.setPointTemperature = 0.0
self.temperatureOffset = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("ANALOG_ROOM_CONTROL_CHANNEL", js)
if c:
self.set_attr_from_dict("actualTemperature", c)
self.set_attr_from_dict("setPointTemperature", c)
self.set_attr_from_dict("temperatureOffset", c)
[docs]
class WallMountedThermostatBasicHumidity(WallMountedThermostatPro):
"""HMIP-WTH-B (Wall Thermostat – basic)"""
pass
[docs]
class SmokeDetector(Device):
"""HMIP-SWSD (Smoke Alarm with Q label)"""
def __init__(self, connection):
super().__init__(connection)
self.smokeDetectorAlarmType = SmokeDetectorAlarmType.IDLE_OFF
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("SMOKE_DETECTOR_CHANNEL", js)
if c:
self.smokeDetectorAlarmType = SmokeDetectorAlarmType.from_str(
c["smokeDetectorAlarmType"]
)
def __str__(self):
return "{} smokeDetectorAlarmType({})".format(
super().__str__(), self.smokeDetectorAlarmType
)
[docs]
class FloorTerminalBlock6(Device):
"""HMIP-FAL230-C6 (Floor Heating Actuator - 6 channels, 230 V)"""
def __init__(self, connection):
super().__init__(connection)
self.globalPumpControl = False
self.heatingValveType = HeatingValveType.NORMALLY_CLOSE
self.heatingLoadType = HeatingLoadType.LOAD_BALANCING
self.frostProtectionTemperature = 0.0
self.heatingEmergencyValue = 0.0
self.valveProtectionDuration = 0
self.valveProtectionSwitchingInterval = 20
self.coolingEmergencyValue = 0
self.pumpFollowUpTime = 0
self.pumpLeadTime = 0
self.pumpProtectionDuration = 0
self.pumpProtectionSwitchingInterval = 20
self._baseChannel = "DEVICE_GLOBAL_PUMP_CONTROL"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DEVICE_GLOBAL_PUMP_CONTROL", js)
if c:
self.globalPumpControl = c["globalPumpControl"]
self.heatingValveType = HeatingValveType.from_str(c["heatingValveType"])
self.heatingLoadType = HeatingLoadType.from_str(c["heatingLoadType"])
self.coolingEmergencyValue = c["coolingEmergencyValue"]
self.frostProtectionTemperature = c["frostProtectionTemperature"]
self.heatingEmergencyValue = c["heatingEmergencyValue"]
self.valveProtectionDuration = c["valveProtectionDuration"]
self.valveProtectionSwitchingInterval = c[
"valveProtectionSwitchingInterval"
]
c = get_functional_channel("FLOOR_TERMINAL_BLOCK_LOCAL_PUMP_CHANNEL", js)
if c:
self.pumpFollowUpTime = c["pumpFollowUpTime"]
self.pumpLeadTime = c["pumpLeadTime"]
self.pumpProtectionDuration = c["pumpProtectionDuration"]
self.pumpProtectionSwitchingInterval = c["pumpProtectionSwitchingInterval"]
def __str__(self):
return (
"{} globalPumpControl({}) heatingValveType({}) heatingLoadType({}) coolingEmergencyValue({}) frostProtectionTemperature({}) heatingEmergencyValue({}) "
"valveProtectionDuration({}) valveProtectionSwitchingInterval({}) pumpFollowUpTime({}) pumpLeadTime({}) pumpProtectionDuration({}) "
"pumpProtectionSwitchingInterval({})"
).format(
super().__str__(),
self.globalPumpControl,
self.heatingValveType,
self.heatingLoadType,
self.coolingEmergencyValue,
self.frostProtectionTemperature,
self.heatingEmergencyValue,
self.valveProtectionDuration,
self.valveProtectionSwitchingInterval,
self.pumpFollowUpTime,
self.pumpLeadTime,
self.pumpProtectionDuration,
self.pumpProtectionSwitchingInterval,
)
[docs]
class FloorTerminalBlock10(FloorTerminalBlock6):
"""HMIP-FAL24-C10 (Floor Heating Actuator – 10x channels, 24V)"""
[docs]
class FloorTerminalBlock12(Device):
"""HMIP-FALMOT-C12 (Floor Heating Actuator – 12x channels, motorised)"""
def __init__(self, connection):
super().__init__(connection)
self.frostProtectionTemperature = 0.0
self.valveProtectionDuration = 0
self.valveProtectionSwitchingInterval = 20
self.coolingEmergencyValue = 0
self.minimumFloorHeatingValvePosition = 0.0
self.pulseWidthModulationAtLowFloorHeatingValvePositionEnabled = False
self._baseChannel = "DEVICE_BASE_FLOOR_HEATING"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DEVICE_BASE_FLOOR_HEATING", js)
if c:
self.set_attr_from_dict("coolingEmergencyValue", c)
self.set_attr_from_dict("frostProtectionTemperature", c)
self.set_attr_from_dict("valveProtectionDuration", c)
self.set_attr_from_dict("valveProtectionSwitchingInterval", c)
[docs]
def set_minimum_floor_heating_valve_position(self, minimumFloorHeatingValvePosition: float):
"""sets the minimum floot heating valve position
Args:
minimumFloorHeatingValvePosition(float): the minimum valve position. must be between 0.0 and 1.0
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda:
self.set_minimum_floor_heating_valve_position_async(
minimumFloorHeatingValvePosition
)
)
[docs]
async def set_minimum_floor_heating_valve_position_async(self, minimumFloorHeatingValvePosition: float):
"""sets the minimum floot heating valve position
Args:
minimumFloorHeatingValvePosition(float): the minimum valve position. must be between 0.0 and 1.0
Returns:
the result of the _restCall
"""
data = {
"channelIndex": 0,
"deviceId": self.id,
"minimumFloorHeatingValvePosition": minimumFloorHeatingValvePosition,
}
return await self._rest_call_async(
"device/configuration/setMinimumFloorHeatingValvePosition",
body=data,
)
[docs]
class WiredFloorTerminalBlock12(FloorTerminalBlock12):
"""Implementation of HmIPW-FALMOT-C12"""
[docs]
class Switch(Device):
"""Generic Switch class"""
def __init__(self, connection):
super().__init__(connection)
self.on = None
self.profileMode = None
self.userDesiredProfileMode = None
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("SWITCH_CHANNEL", js)
if c:
self.on = c["on"]
self.profileMode = c["profileMode"]
self.userDesiredProfileMode = c["userDesiredProfileMode"]
def __str__(self):
return "{} on({}) profileMode({}) userDesiredProfileMode({})".format(
super().__str__(), self.on, self.profileMode, self.userDesiredProfileMode
)
[docs]
def set_switch_state(self, on=True, channelIndex=1):
return self._run_non_async(self.set_switch_state_async, on, channelIndex)
[docs]
async def set_switch_state_async(self, on=True, channelIndex=1):
data = {"channelIndex": channelIndex, "deviceId": self.id, "on": on}
return await self._rest_call_async("device/control/setSwitchState", body=data)
[docs]
def turn_on(self, channelIndex=1):
return self._run_non_async(self.turn_on_async, channelIndex)
[docs]
async def turn_on_async(self, channelIndex=1):
return await self.set_switch_state_async(True, channelIndex)
[docs]
def turn_off(self, channelIndex=1):
return self._run_non_async(self.turn_off_async, channelIndex)
[docs]
async def turn_off_async(self, channelIndex=1):
return await self.set_switch_state_async(False, channelIndex)
[docs]
class CarbonDioxideSensor(Switch):
"""HmIP-SCTH230"""
[docs]
class PlugableSwitch(Switch):
"""HMIP-PS (Pluggable Switch), HMIP-PCBS (Switch Circuit Board - 1 channel)"""
[docs]
class PrintedCircuitBoardSwitchBattery(Switch):
"""HMIP-PCBS-BAT (Printed Circuit Board Switch Battery)"""
[docs]
class PrintedCircuitBoardSwitch2(Switch):
"""HMIP-PCBS2 (Switch Circuit Board - 2x channels)"""
[docs]
class OpenCollector8Module(Switch):
"""HMIP-MOD-OC8 ( Open Collector Module )"""
[docs]
class HeatingSwitch2(Switch):
"""HMIP-WHS2 (Switch Actuator for heating systems – 2x channels)"""
[docs]
class WiredSwitch8(Switch):
"""HMIPW-DRS8 (Homematic IP Wired Switch Actuator – 8x channels)"""
[docs]
class WiredSwitch4(Switch):
"""HMIPW-DRS4 (Homematic IP Wired Switch Actuator – 4x channels)"""
[docs]
class DinRailSwitch4(Switch):
"""HMIP-DRSI4 (Homematic IP Switch Actuator for DIN rail mount – 4x channels)"""
[docs]
class BrandSwitch2(Switch):
"""ELV-SH-BS2 (ELV Smart Home ARR-Bausatz Schaltaktor für Markenschalter – 2-fach powered by Homematic IP)"""
[docs]
class SwitchMeasuring(Switch):
"""Generic class for Switch and Meter"""
def __init__(self, connection):
super().__init__(connection)
self.energyCounter = 0
self.currentPowerConsumption = 0
[docs]
def reset_energy_counter(self):
return self._run_non_async(self.reset_energy_counter_async)
[docs]
async def reset_energy_counter_async(self):
data = {"channelIndex": 1, "deviceId": self.id}
return await self._rest_call_async("device/control/resetEnergyCounter", body=data)
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("SWITCH_MEASURING_CHANNEL", js)
if c:
self.on = c["on"]
self.energyCounter = c["energyCounter"]
self.currentPowerConsumption = c["currentPowerConsumption"]
self.profileMode = c["profileMode"]
self.userDesiredProfileMode = c["userDesiredProfileMode"]
def __str__(self):
return "{} energyCounter({}) currentPowerConsumption({}W)".format(
super().__str__(), self.energyCounter, self.currentPowerConsumption
)
[docs]
class MultiIOBox(Switch):
"""HMIP-MIOB (Multi IO Box for floor heating & cooling)"""
def __init__(self, connection):
super().__init__(connection)
self.analogOutputLevel = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("ANALOG_OUTPUT_CHANNEL", js)
if c:
self.analogOutputLevel = c["analogOutputLevel"]
def __str__(self):
return "{} analogOutputLevel({})".format(
super().__str__(), self.analogOutputLevel
)
[docs]
class BrandSwitchNotificationLight(Switch):
"""HMIP-BSL (Switch Actuator for brand switches – with signal lamp)"""
def __init__(self, connection):
super().__init__(connection)
#:int:the channel number for the top light
self.topLightChannelIndex = 2
#:int:the channel number for the bottom light
self.bottomLightChannelIndex = 3
def __str__(self):
top = self.functionalChannels[self.topLightChannelIndex]
bottom = self.functionalChannels[self.bottomLightChannelIndex]
return (
"{} topDimLevel({}) topColor({}) bottomDimLevel({}) bottomColor({})".format(
super().__str__(),
top.dimLevel,
top.simpleRGBColorState,
bottom.dimLevel,
bottom.simpleRGBColorState,
)
)
[docs]
def set_rgb_dim_level(self, channelIndex: int, rgb: RGBColorState, dimLevel: float):
"""sets the color and dimlevel of the lamp
Args:
channelIndex(int): the channelIndex of the lamp. Use self.topLightChannelIndex or self.bottomLightChannelIndex
rgb(RGBColorState): the color of the lamp
dimLevel(float): the dimLevel of the lamp. 0.0 = off, 1.0 = MAX
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda: self.set_rgb_dim_level_async(channelIndex, rgb, dimLevel))
[docs]
async def set_rgb_dim_level_async(self, channelIndex: int, rgb: RGBColorState, dimLevel: float):
"""sets the color and dimlevel of the lamp
Args:
channelIndex(int): the channelIndex of the lamp. Use self.topLightChannelIndex or self.bottomLightChannelIndex
rgb(RGBColorState): the color of the lamp
dimLevel(float): the dimLevel of the lamp. 0.0 = off, 1.0 = MAX
Returns:
the result of the _restCall
"""
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"simpleRGBColorState": rgb,
"dimLevel": dimLevel,
}
return await self._rest_call_async(
"device/control/setSimpleRGBColorDimLevel", body=data
)
[docs]
def set_rgb_dim_level_with_time(
self,
channelIndex: int,
rgb: RGBColorState,
dimLevel: float,
onTime: float,
rampTime: float,
):
"""sets the color and dimlevel of the lamp
Args:
channelIndex(int): the channelIndex of the lamp. Use self.topLightChannelIndex or self.bottomLightChannelIndex
rgb(RGBColorState): the color of the lamp
dimLevel(float): the dimLevel of the lamp. 0.0 = off, 1.0 = MAX
onTime(float):
rampTime(float):
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda:
self.set_rgb_dim_level_with_time_async(
channelIndex, rgb, dimLevel, onTime, rampTime
))
[docs]
async def set_rgb_dim_level_with_time_async(
self,
channelIndex: int,
rgb: RGBColorState,
dimLevel: float,
onTime: float,
rampTime: float,
):
"""sets the color and dimlevel of the lamp
Args:
channelIndex(int): the channelIndex of the lamp. Use self.topLightChannelIndex or self.bottomLightChannelIndex
rgb(RGBColorState): the color of the lamp
dimLevel(float): the dimLevel of the lamp. 0.0 = off, 1.0 = MAX
onTime(float):
rampTime(float):
Returns:
the result of the _restCall
"""
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"simpleRGBColorState": rgb,
"dimLevel": dimLevel,
"onTime": onTime,
"rampTime": rampTime,
}
return await self._rest_call_async(
"device/control/setSimpleRGBColorDimLevelWithTime", body=data
)
[docs]
class PlugableSwitchMeasuring(SwitchMeasuring):
"""HMIP-PSM (Pluggable Switch and Meter)"""
[docs]
class BrandSwitchMeasuring(SwitchMeasuring):
"""HMIP-BSM (Brand Switch and Meter)"""
[docs]
class FullFlushSwitchMeasuring(SwitchMeasuring):
"""HMIP-FSM, HMIP-FSM16 (Full flush Switch and Meter)"""
[docs]
class KeyRemoteControl4(PushButton):
"""HMIP-KRC4 (Key Ring Remote Control - 4 buttons)"""
[docs]
class RemoteControl8(PushButton):
"""HMIP-RC8 (Remote Control - 8 buttons)"""
[docs]
class RemoteControl8Module(RemoteControl8):
"""HMIP-MOD-RC8 (Open Collector Module Sender - 8x)"""
[docs]
class RgbwDimmer(Device):
"""HmIP-RGBW"""
fastColorChangeSupported: bool = False
[docs]
def from_json(self, js):
super().from_json(js)
self.set_attr_from_dict("fastColorChangeSupported", js)
[docs]
class AlarmSirenIndoor(SabotageDevice):
"""HMIP-ASIR (Alarm Siren)"""
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("ALARM_SIREN_CHANNEL", js)
if c:
# The ALARM_SIREN_CHANNEL doesn't have any values yet.
pass
[docs]
class AlarmSirenOutdoor(AlarmSirenIndoor):
"""HMIP-ASIR-O (Alarm Siren Outdoor)"""
def __init__(self, connection):
super().__init__(connection)
self.badBatteryHealth = False
self._baseChannel = "DEVICE_RECHARGEABLE_WITH_SABOTAGE"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DEVICE_RECHARGEABLE_WITH_SABOTAGE", js)
if c:
self.set_attr_from_dict("badBatteryHealth", c)
[docs]
class MotionDetectorIndoor(SabotageDevice):
"""HMIP-SMI (Motion Detector with Brightness Sensor - indoor)"""
def __init__(self, connection):
super().__init__(connection)
self.currentIllumination = None
self.motionDetected = None
self.illumination = None
self.motionBufferActive = False
self.motionDetectionSendInterval = MotionDetectionSendInterval.SECONDS_30
self.numberOfBrightnessMeasurements = 0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("MOTION_DETECTION_CHANNEL", js)
if c:
self.motionDetected = c["motionDetected"]
self.illumination = c["illumination"]
self.motionBufferActive = c["motionBufferActive"]
self.motionDetectionSendInterval = MotionDetectionSendInterval.from_str(
c["motionDetectionSendInterval"]
)
self.numberOfBrightnessMeasurements = c["numberOfBrightnessMeasurements"]
self.currentIllumination = c["currentIllumination"]
def __str__(self):
return "{} motionDetected({}) illumination({}) motionBufferActive({}) motionDetectionSendInterval({}) numberOfBrightnessMeasurements({})".format(
super().__str__(),
self.motionDetected,
self.illumination,
self.motionBufferActive,
self.motionDetectionSendInterval,
self.numberOfBrightnessMeasurements,
)
[docs]
class MotionDetectorOutdoor(Device):
"""HMIP-SMO-A (Motion Detector with Brightness Sensor - outdoor)"""
def __init__(self, connection):
super().__init__(connection)
self.currentIllumination = None
self.motionDetected = None
self.illumination = None
self.motionBufferActive = False
self.motionDetectionSendInterval = MotionDetectionSendInterval.SECONDS_30
self.numberOfBrightnessMeasurements = 0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("MOTION_DETECTION_CHANNEL", js)
if c:
self.set_attr_from_dict("motionDetected", c)
self.set_attr_from_dict("illumination", c)
self.set_attr_from_dict("motionBufferActive", c)
self.set_attr_from_dict("motionDetectionSendInterval", c)
self.set_attr_from_dict("numberOfBrightnessMeasurements", c)
self.set_attr_from_dict("currentIllumination", c)
[docs]
class PresenceDetectorIndoor(SabotageDevice):
"""HMIP-SPI (Presence Sensor - indoor)"""
def __init__(self, connection):
super().__init__(connection)
self.presenceDetected = False
self.currentIllumination = None
self.illumination = 0
self.motionBufferActive = False
self.motionDetectionSendInterval = MotionDetectionSendInterval.SECONDS_30
self.numberOfBrightnessMeasurements = 0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("PRESENCE_DETECTION_CHANNEL", js)
if c:
self.presenceDetected = c["presenceDetected"]
self.currentIllumination = c["currentIllumination"]
self.illumination = c["illumination"]
self.motionBufferActive = c["motionBufferActive"]
self.motionDetectionSendInterval = MotionDetectionSendInterval.from_str(
c["motionDetectionSendInterval"]
)
self.numberOfBrightnessMeasurements = c["numberOfBrightnessMeasurements"]
def __str__(self):
return "{} presenceDetected({}) illumination({}) motionBufferActive({}) motionDetectionSendInterval({}) numberOfBrightnessMeasurements({})".format(
super().__str__(),
self.presenceDetected,
self.illumination,
self.motionBufferActive,
self.motionDetectionSendInterval,
self.numberOfBrightnessMeasurements,
)
[docs]
class PassageDetector(SabotageDevice):
"""HMIP-SPDR (Passage Detector)"""
def __init__(self, connection):
super().__init__(connection)
self.leftCounter = 0
self.leftRightCounterDelta = 0
self.passageBlindtime = 0.0
self.passageDirection = PassageDirection.RIGHT
self.passageSensorSensitivity = 0.0
self.passageTimeout = 0.0
self.rightCounter = 0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("PASSAGE_DETECTOR_CHANNEL", js)
if c:
self.leftCounter = c["leftCounter"]
self.leftRightCounterDelta = c["leftRightCounterDelta"]
self.passageBlindtime = c["passageBlindtime"]
self.passageDirection = PassageDirection.from_str(c["passageDirection"])
self.passageSensorSensitivity = c["passageSensorSensitivity"]
self.passageTimeout = c["passageTimeout"]
self.rightCounter = c["rightCounter"]
def __str__(self):
return "{} leftCounter({}) leftRightCounterDelta({}) passageBlindtime({}) passageDirection({}) passageSensorSensitivity({}) passageTimeout({}) rightCounter({})".format(
super().__str__(),
self.leftCounter,
self.leftRightCounterDelta,
self.passageBlindtime,
self.passageDirection,
self.passageSensorSensitivity,
self.passageTimeout,
self.rightCounter,
)
[docs]
class KeyRemoteControlAlarm(Device):
"""HMIP-KRCA (Key Ring Remote Control - alarm)"""
[docs]
class Shutter(Device):
"""Base class for shutter devices"""
[docs]
def set_shutter_level(self, level=0.0, channelIndex=1):
"""sets the shutter level
Args:
level(float): the new level of the shutter. 0.0 = open, 1.0 = closed
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda: self.set_shutter_level_async(level, channelIndex))
[docs]
async def set_shutter_level_async(self, level=0.0, channelIndex=1):
"""sets the shutter level
Args:
level(float): the new level of the shutter. 0.0 = open, 1.0 = closed
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"shutterLevel": level,
}
return await self._rest_call_async("device/control/setShutterLevel", body=data)
[docs]
def set_shutter_stop(self, channelIndex=1):
"""stops the current shutter operation
Args:
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda: self.set_shutter_stop_async(channelIndex))
[docs]
async def set_shutter_stop_async(self, channelIndex=1):
"""stops the current shutter operation
Args:
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
data = {"channelIndex": channelIndex, "deviceId": self.id}
return await self._rest_call_async("device/control/stop", body=data)
[docs]
class Blind(Shutter):
"""Base class for blind devices"""
[docs]
def set_slats_level(self, slatsLevel=0.0, shutterLevel=None, channelIndex=1):
"""sets the slats and shutter level
Args:
slatsLevel(float): the new level of the slats. 0.0 = open, 1.0 = closed
shutterLevel(float): the new level of the shutter. 0.0 = open, 1.0 = closed, None = use the current value
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
return self._run_non_async(lambda:
self.set_slats_level_async(slatsLevel, shutterLevel, channelIndex)
)
[docs]
async def set_slats_level_async(self, slatsLevel=0.0, shutterLevel=None, channelIndex=1):
"""sets the slats and shutter level
Args:
slatsLevel(float): the new level of the slats. 0.0 = open, 1.0 = closed,
shutterLevel(float): the new level of the shutter. 0.0 = open, 1.0 = closed, None = use the current value
channelIndex(int): the channel to control
Returns:
the result of the _restCall
"""
if shutterLevel is None:
shutterLevel = self.functionalChannels[channelIndex].shutterLevel
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"slatsLevel": slatsLevel,
"shutterLevel": shutterLevel,
}
return await self._rest_call_async("device/control/setSlatsLevel", data)
[docs]
class FullFlushShutter(Shutter):
"""HMIP-FROLL (Shutter Actuator - flush-mount) / HMIP-BROLL (Shutter Actuator - Brand-mount)"""
def __init__(self, connection):
super().__init__(connection)
self.shutterLevel = 0
self.changeOverDelay = 0.0
self.bottomToTopReferenceTime = 0.0
self.topToBottomReferenceTime = 0.0
self.delayCompensationValue = 0
self.endpositionAutoDetectionEnabled = False
self.previousShutterLevel = None
self.processing = False
self.profileMode = "AUTOMATIC"
self.selfCalibrationInProgress = None
self.supportingDelayCompensation = False
self.supportingEndpositionAutoDetection = False
self.supportingSelfCalibration = False
self.userDesiredProfileMode = "AUTOMATIC"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("SHUTTER_CHANNEL", js)
if c:
self.shutterLevel = c["shutterLevel"]
self.changeOverDelay = c["changeOverDelay"]
self.delayCompensationValue = c["delayCompensationValue"]
self.bottomToTopReferenceTime = c["bottomToTopReferenceTime"]
self.topToBottomReferenceTime = c["topToBottomReferenceTime"]
self.endpositionAutoDetectionEnabled = c["endpositionAutoDetectionEnabled"]
self.previousShutterLevel = c["previousShutterLevel"]
self.processing = c["processing"]
self.profileMode = c["profileMode"]
self.selfCalibrationInProgress = c["selfCalibrationInProgress"]
self.supportingDelayCompensation = c["supportingDelayCompensation"]
self.supportingEndpositionAutoDetection = c[
"supportingEndpositionAutoDetection"
]
self.supportingSelfCalibration = c["supportingSelfCalibration"]
self.userDesiredProfileMode = c["userDesiredProfileMode"]
def __str__(self):
return "{} shutterLevel({}) topToBottom({}) bottomToTop({})".format(
super().__str__(),
self.shutterLevel,
self.topToBottomReferenceTime,
self.bottomToTopReferenceTime,
)
[docs]
class FullFlushBlind(FullFlushShutter, Blind):
"""HMIP-FBL (Blind Actuator - flush-mount)"""
def __init__(self, connection):
super().__init__(connection)
self.slatsLevel = 0
self.slatsReferenceTime = 0.0
self.previousSlatsLevel = 0
self.blindModeActive = False
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("BLIND_CHANNEL", js)
if c:
self.shutterLevel = c["shutterLevel"]
self.changeOverDelay = c["changeOverDelay"]
self.delayCompensationValue = c["delayCompensationValue"]
self.bottomToTopReferenceTime = c["bottomToTopReferenceTime"]
self.topToBottomReferenceTime = c["topToBottomReferenceTime"]
self.endpositionAutoDetectionEnabled = c["endpositionAutoDetectionEnabled"]
self.previousShutterLevel = c["previousShutterLevel"]
self.processing = c["processing"]
self.profileMode = c["profileMode"]
self.selfCalibrationInProgress = c["selfCalibrationInProgress"]
self.supportingDelayCompensation = c["supportingDelayCompensation"]
self.supportingEndpositionAutoDetection = c[
"supportingEndpositionAutoDetection"
]
self.supportingSelfCalibration = c["supportingSelfCalibration"]
self.userDesiredProfileMode = c["userDesiredProfileMode"]
self.slatsLevel = c["slatsLevel"]
self.slatsReferenceTime = c["slatsReferenceTime"]
self.previousSlatsLevel = c["previousSlatsLevel"]
self.blindModeActive = c["blindModeActive"]
def __str__(self):
return "{} slatsLevel({}) blindModeActive({})".format(
super().__str__(), self.slatsLevel, self.blindModeActive
)
[docs]
class BrandBlind(FullFlushBlind):
"""HMIP-BBL (Blind Actuator for brand switches)"""
[docs]
class DinRailBlind4(Blind):
"""HmIP-DRBLI4 (Blind Actuator for DIN rail mount – 4 channels)"""
[docs]
class WiredDinRailBlind4(Blind):
"""HmIPW-DRBL4"""
[docs]
class BlindModule(Device):
"""HMIP-HDM1 (Hunter Douglas & erfal window blinds)"""
def __init__(self, connection):
super().__init__(connection)
self.automationDriveSpeed = DriveSpeed.CREEP_SPEED
self.manualDriveSpeed = DriveSpeed.CREEP_SPEED
self.favoritePrimaryShadingPosition = 0.0
self.favoriteSecondaryShadingPosition = 0.0
self.primaryShadingLevel = 0.0
self.secondaryShadingLevel = 0.0
self.previousPrimaryShadingLevel = 0.0
self.previousSecondaryShadingLevel = 0.0
self.identifyOemSupported = False
self.productId = 0
self.primaryCloseAdjustable = False
self.primaryOpenAdjustable = False
self.primaryShadingStateType = ShadingStateType.NOT_EXISTENT
self.primaryCloseAdjustable = False
self.primaryOpenAdjustable = False
self.primaryShadingStateType = ShadingStateType.NOT_EXISTENT
self.profileMode = ProfileMode.MANUAL
self.userDesiredProfileMode = ProfileMode.MANUAL
self.processing = False
self.shadingDriveVersion = None
self.shadingPackagePosition = ShadingPackagePosition.NOT_USED
self.shadingPositionAdjustmentActive = None
self.shadingPositionAdjustmentClientId = None
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("SHADING_CHANNEL", js)
if c:
self.set_attr_from_dict("automationDriveSpeed", c, DriveSpeed)
self.set_attr_from_dict("manualDriveSpeed", c, DriveSpeed)
self.set_attr_from_dict("favoritePrimaryShadingPosition", c)
self.set_attr_from_dict("favoriteSecondaryShadingPosition", c)
self.set_attr_from_dict("primaryCloseAdjustable", c)
self.set_attr_from_dict("primaryOpenAdjustable", c)
self.set_attr_from_dict("primaryShadingStateType", c, ShadingStateType)
self.set_attr_from_dict("secondaryCloseAdjustable", c)
self.set_attr_from_dict("secondaryOpenAdjustable", c)
self.set_attr_from_dict("secondaryShadingStateType", c, ShadingStateType)
self.set_attr_from_dict("primaryShadingLevel", c)
self.set_attr_from_dict("secondaryShadingLevel", c)
self.set_attr_from_dict("previousPrimaryShadingLevel", c)
self.set_attr_from_dict("previousSecondaryShadingLevel", c)
self.set_attr_from_dict("identifyOemSupported", c)
self.set_attr_from_dict("productId", c)
self.set_attr_from_dict("profileMode", c, ProfileMode)
self.set_attr_from_dict("userDesiredProfileMode", c, ProfileMode)
self.set_attr_from_dict("shadingDriveVersion", c)
self.set_attr_from_dict("shadingPackagePosition", c, ShadingPackagePosition)
self.set_attr_from_dict("shadingPositionAdjustmentActive", c)
self.set_attr_from_dict("shadingPositionAdjustmentClientId", c)
[docs]
def set_primary_shading_level(self, primaryShadingLevel: float):
return self._run_non_async(lambda: self.set_primary_shading_level_async(primaryShadingLevel))
[docs]
async def set_primary_shading_level_async(self, primaryShadingLevel: float):
data = {
"channelIndex": 1,
"deviceId": self.id,
"primaryShadingLevel": primaryShadingLevel,
}
return await self._rest_call_async("device/control/setPrimaryShadingLevel", data)
[docs]
def set_secondary_shading_level(
self, primaryShadingLevel: float, secondaryShadingLevel: float
):
return self._run_non_async(lambda:
self.set_secondary_shading_level_async(primaryShadingLevel, secondaryShadingLevel)
)
[docs]
async def set_secondary_shading_level_async(
self, primaryShadingLevel: float, secondaryShadingLevel: float):
data = {
"channelIndex": 1,
"deviceId": self.id,
"primaryShadingLevel": primaryShadingLevel,
"secondaryShadingLevel": secondaryShadingLevel,
}
return await self._rest_call_async(
"device/control/setSecondaryShadingLevel", data
)
[docs]
async def stop(self):
"""stops the current operation
Returns:
the result of the _restCall
"""
return self._run_non_async(self.stop_async)
[docs]
async def stop_async(self):
"""stops the current operation
Returns:
the result of the _restCall
"""
data = {"channelIndex": 1, "deviceId": self.id}
return await self._rest_call_async("device/control/stop", body=data)
[docs]
class LightSensor(Device):
"""HMIP-SLO (Light Sensor outdoor)"""
def __init__(self, connection):
super().__init__(connection)
#:float:the average illumination value
self.averageIllumination = 0.0
#:float:the current illumination value
self.currentIllumination = 0.0
#:float:the highest illumination value
self.highestIllumination = 0.0
#:float:the lowest illumination value
self.lowestIllumination = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("LIGHT_SENSOR_CHANNEL", js)
if c:
self.averageIllumination = c["averageIllumination"]
self.currentIllumination = c["currentIllumination"]
self.highestIllumination = c["highestIllumination"]
self.lowestIllumination = c["lowestIllumination"]
def __str__(self):
return "{} averageIllumination({}) currentIllumination({}) highestIllumination({}) lowestIllumination({})".format(
super().__str__(),
self.averageIllumination,
self.currentIllumination,
self.highestIllumination,
self.lowestIllumination,
)
[docs]
class Dimmer(Device):
"""Base dimmer device class"""
def __init__(self, connection):
super().__init__(connection)
self.dimLevel = 0.0
self.profileMode = ""
self.userDesiredProfileMode = ""
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DIMMER_CHANNEL", js)
if c:
self.dimLevel = c["dimLevel"]
self.profileMode = c["profileMode"]
self.userDesiredProfileMode = c["userDesiredProfileMode"]
def __str__(self):
return "{} dimLevel({}) profileMode({}) userDesiredProfileMode({})".format(
super().__str__(),
self.dimLevel,
self.profileMode,
self.userDesiredProfileMode,
)
[docs]
def set_dim_level(self, dimLevel=0.0, channelIndex=1):
return self._run_non_async(lambda: self.set_dim_level_async(dimLevel, channelIndex))
[docs]
async def set_dim_level_async(self, dimLevel=0.0, channelIndex=1):
data = {"channelIndex": channelIndex, "deviceId": self.id, "dimLevel": dimLevel}
return await self._rest_call_async("device/control/setDimLevel", data)
[docs]
class PluggableDimmer(Dimmer):
"""HMIP-PDT Pluggable Dimmer"""
[docs]
class BrandDimmer(Dimmer):
"""HMIP-BDT Brand Dimmer"""
[docs]
class FullFlushDimmer(Dimmer):
"""HMIP-FDT Dimming Actuator flush-mount"""
[docs]
class WiredDimmer3(Dimmer):
"""HMIPW-DRD3 (Homematic IP Wired Dimming Actuator – 3x channels)"""
[docs]
class DinRailDimmer3(Dimmer):
"""HMIP-DRDI3 (Dimming Actuator Inbound 230V – 3x channels, 200W per channel) electrical DIN rail"""
def __init__(self, connection):
super().__init__(connection)
self.c1dimLevel = 0.0
self.c2dimLevel = 0.0
self.c3dimLevel = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
channels = get_functional_channels("MULTI_MODE_INPUT_DIMMER_CHANNEL", js)
if channels:
self.c1dimLevel = channels[0]["dimLevel"]
self.dimLevel = self.c1dimLevel
self.c2dimLevel = channels[1]["dimLevel"]
self.c3dimLevel = channels[2]["dimLevel"]
def __str__(self):
return "{} c1DimLevel({}) c2DimLevel({}) c3DimLevel({})".format(
super().__str__(), self.c1dimLevel, self.c2dimLevel, self.c3dimLevel
)
[docs]
class WeatherSensor(Device):
"""HMIP-SWO-B"""
def __init__(self, connection):
super().__init__(connection)
self.actualTemperature = 0
self.humidity = 0
self.illumination = 0
self.illuminationThresholdSunshine = 0
self.storm = False
self.sunshine = False
self.todaySunshineDuration = 0
self.totalSunshineDuration = 0
self.windSpeed = 0
self.windValueType = WindValueType.AVERAGE_VALUE
self.yesterdaySunshineDuration = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("WEATHER_SENSOR_CHANNEL", js)
if c:
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.illumination = c["illumination"]
self.illuminationThresholdSunshine = c["illuminationThresholdSunshine"]
self.storm = c["storm"]
self.sunshine = c["sunshine"]
self.todaySunshineDuration = c["todaySunshineDuration"]
self.totalSunshineDuration = c["totalSunshineDuration"]
self.windSpeed = c["windSpeed"]
self.windValueType = WindValueType.from_str(c["windValueType"])
self.yesterdaySunshineDuration = c["yesterdaySunshineDuration"]
self.vaporAmount = c["vaporAmount"]
def __str__(self):
return (
"{} actualTemperature({}) humidity({}) vaporAmount({}) illumination({}) illuminationThresholdSunshine({}) storm({}) sunshine({}) "
"todaySunshineDuration({}) totalSunshineDuration({}) "
"windSpeed({}) windValueType({}) "
"yesterdaySunshineDuration({})"
).format(
super().__str__(),
self.actualTemperature,
self.humidity,
self.vaporAmount,
self.illumination,
self.illuminationThresholdSunshine,
self.storm,
self.sunshine,
self.todaySunshineDuration,
self.totalSunshineDuration,
self.windSpeed,
self.windValueType,
self.yesterdaySunshineDuration,
)
[docs]
class WeatherSensorPlus(Device):
"""HMIP-SWO-PL"""
def __init__(self, connection):
super().__init__(connection)
self.actualTemperature = 0
self.humidity = 0
self.illumination = 0
self.illuminationThresholdSunshine = 0
self.raining = False
self.storm = False
self.sunshine = False
self.todayRainCounter = 0
self.todaySunshineDuration = 0
self.totalRainCounter = 0
self.totalSunshineDuration = 0
self.windSpeed = 0
self.windValueType = WindValueType.AVERAGE_VALUE
self.yesterdayRainCounter = 0
self.yesterdaySunshineDuration = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("WEATHER_SENSOR_PLUS_CHANNEL", js)
if c:
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.illumination = c["illumination"]
self.illuminationThresholdSunshine = c["illuminationThresholdSunshine"]
self.raining = c["raining"]
self.storm = c["storm"]
self.sunshine = c["sunshine"]
self.todayRainCounter = c["todayRainCounter"]
self.todaySunshineDuration = c["todaySunshineDuration"]
self.totalRainCounter = c["totalRainCounter"]
self.totalSunshineDuration = c["totalSunshineDuration"]
self.windSpeed = c["windSpeed"]
self.windValueType = WindValueType.from_str(c["windValueType"])
self.yesterdayRainCounter = c["yesterdayRainCounter"]
self.yesterdaySunshineDuration = c["yesterdaySunshineDuration"]
self.vaporAmount = c["vaporAmount"]
def __str__(self):
return (
"{} actualTemperature({}) humidity({}) vaporAmount({}) illumination({}) illuminationThresholdSunshine({}) raining({}) storm({}) sunshine({}) "
"todayRainCounter({}) todaySunshineDuration({}) totalRainCounter({}) totalSunshineDuration({}) "
"windSpeed({}) windValueType({}) yesterdayRainCounter({}) yesterdaySunshineDuration({})"
).format(
super().__str__(),
self.actualTemperature,
self.humidity,
self.vaporAmount,
self.illumination,
self.illuminationThresholdSunshine,
self.raining,
self.storm,
self.sunshine,
self.todayRainCounter,
self.todaySunshineDuration,
self.totalRainCounter,
self.totalSunshineDuration,
self.windSpeed,
self.windValueType,
self.yesterdayRainCounter,
self.yesterdaySunshineDuration,
)
[docs]
class WeatherSensorPro(Device):
"""HMIP-SWO-PR"""
def __init__(self, connection):
super().__init__(connection)
self.actualTemperature = 0
self.humidity = 0
self.illumination = 0
self.illuminationThresholdSunshine = 0
self.raining = False
self.storm = False
self.sunshine = False
self.todayRainCounter = 0
self.todaySunshineDuration = 0
self.totalRainCounter = 0
self.totalSunshineDuration = 0
self.weathervaneAlignmentNeeded = False
self.windDirection = 0
self.windDirectionVariation = 0
self.windSpeed = 0
self.windValueType = WindValueType.AVERAGE_VALUE
self.yesterdayRainCounter = 0
self.yesterdaySunshineDuration = 0
self.vaporAmount = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("WEATHER_SENSOR_PRO_CHANNEL", js)
if c:
self.actualTemperature = c["actualTemperature"]
self.humidity = c["humidity"]
self.illumination = c["illumination"]
self.illuminationThresholdSunshine = c["illuminationThresholdSunshine"]
self.raining = c["raining"]
self.storm = c["storm"]
self.sunshine = c["sunshine"]
self.todayRainCounter = c["todayRainCounter"]
self.todaySunshineDuration = c["todaySunshineDuration"]
self.totalRainCounter = c["totalRainCounter"]
self.totalSunshineDuration = c["totalSunshineDuration"]
self.weathervaneAlignmentNeeded = c["weathervaneAlignmentNeeded"]
self.windDirection = c["windDirection"]
self.windDirectionVariation = c["windDirectionVariation"]
self.windSpeed = c["windSpeed"]
self.windValueType = WindValueType.from_str(c["windValueType"])
self.yesterdayRainCounter = c["yesterdayRainCounter"]
self.yesterdaySunshineDuration = c["yesterdaySunshineDuration"]
self.vaporAmount = c["vaporAmount"]
def __str__(self):
return (
"{} actualTemperature({}) humidity({}) vaporAmount({}) illumination({}) illuminationThresholdSunshine({}) raining({}) storm({}) sunshine({}) "
"todayRainCounter({}) todaySunshineDuration({}) totalRainCounter({}) totalSunshineDuration({}) "
"weathervaneAlignmentNeeded({}) windDirection({}) windDirectionVariation({}) windSpeed({}) windValueType({}) "
"yesterdayRainCounter({}) yesterdaySunshineDuration({})"
).format(
super().__str__(),
self.actualTemperature,
self.humidity,
self.vaporAmount,
self.illumination,
self.illuminationThresholdSunshine,
self.raining,
self.storm,
self.sunshine,
self.todayRainCounter,
self.todaySunshineDuration,
self.totalRainCounter,
self.totalSunshineDuration,
self.weathervaneAlignmentNeeded,
self.windDirection,
self.windDirectionVariation,
self.windSpeed,
self.windValueType,
self.yesterdayRainCounter,
self.yesterdaySunshineDuration,
)
# Any set/calibration functions?
[docs]
class WaterSensor(Device):
"""HMIP-SWD ( Water Sensor )"""
def __init__(self, connection):
super().__init__(connection)
self.incorrectPositioned = False
self.acousticAlarmSignal = AcousticAlarmSignal.DISABLE_ACOUSTIC_SIGNAL
self.acousticAlarmTiming = AcousticAlarmTiming.PERMANENT
self.acousticWaterAlarmTrigger = WaterAlarmTrigger.NO_ALARM
self.inAppWaterAlarmTrigger = WaterAlarmTrigger.NO_ALARM
self.moistureDetected = False
self.sirenWaterAlarmTrigger = WaterAlarmTrigger.NO_ALARM
self.waterlevelDetected = False
self._baseChannel = "DEVICE_INCORRECT_POSITIONED"
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DEVICE_INCORRECT_POSITIONED", js)
if c:
self.incorrectPositioned = c["incorrectPositioned"]
c = get_functional_channel("WATER_SENSOR_CHANNEL", js)
if c:
self.acousticAlarmSignal = AcousticAlarmSignal.from_str(
c["acousticAlarmSignal"]
)
self.acousticAlarmTiming = AcousticAlarmTiming.from_str(
c["acousticAlarmTiming"]
)
self.acousticWaterAlarmTrigger = WaterAlarmTrigger.from_str(
c["acousticWaterAlarmTrigger"]
)
self.inAppWaterAlarmTrigger = WaterAlarmTrigger.from_str(
c["inAppWaterAlarmTrigger"]
)
self.moistureDetected = c["moistureDetected"]
self.sirenWaterAlarmTrigger = WaterAlarmTrigger.from_str(
c["sirenWaterAlarmTrigger"]
)
self.waterlevelDetected = c["waterlevelDetected"]
def __str__(self):
return (
"{} incorrectPositioned({}) acousticAlarmSignal({}) acousticAlarmTiming({}) acousticWaterAlarmTrigger({})"
" inAppWaterAlarmTrigger({}) moistureDetected({}) sirenWaterAlarmTrigger({}) waterlevelDetected({})"
).format(
super().__str__(),
self.incorrectPositioned,
self.acousticAlarmSignal,
self.acousticAlarmTiming,
self.acousticWaterAlarmTrigger,
self.inAppWaterAlarmTrigger,
self.moistureDetected,
self.sirenWaterAlarmTrigger,
self.waterlevelDetected,
)
[docs]
def set_acoustic_alarm_signal(self, acousticAlarmSignal: AcousticAlarmSignal):
return self._run_non_async(lambda:
self.set_acoustic_alarm_signal_async(acousticAlarmSignal)
)
[docs]
async def set_acoustic_alarm_signal_async(self, acousticAlarmSignal: AcousticAlarmSignal):
data = {
"channelIndex": 1,
"deviceId": self.id,
"acousticAlarmSignal": str(acousticAlarmSignal),
}
return await self._rest_call_async(
"device/configuration/setAcousticAlarmSignal", data
)
[docs]
def set_acoustic_alarm_timing(self, acousticAlarmTiming: AcousticAlarmTiming):
return self._run_non_async(lambda:
self.set_acoustic_alarm_timing_async(acousticAlarmTiming)
)
[docs]
async def set_acoustic_alarm_timing_async(self, acousticAlarmTiming: AcousticAlarmTiming):
data = {
"channelIndex": 1,
"deviceId": self.id,
"acousticAlarmTiming": str(acousticAlarmTiming),
}
return await self._rest_call_async(
"device/configuration/setAcousticAlarmTiming", data
)
[docs]
def set_acoustic_water_alarm_trigger(
self, acousticWaterAlarmTrigger: WaterAlarmTrigger
):
return self._run_non_async(lambda:
self.set_acoustic_water_alarm_trigger_async(acousticWaterAlarmTrigger)
)
[docs]
async def set_acoustic_water_alarm_trigger_async(
self, acousticWaterAlarmTrigger: WaterAlarmTrigger
):
data = {
"channelIndex": 1,
"deviceId": self.id,
"acousticWaterAlarmTrigger": str(acousticWaterAlarmTrigger),
}
return await self._rest_call_async("device/configuration/setAcousticWaterAlarmTrigger", data)
[docs]
def set_inapp_water_alarm_trigger(self, inAppWaterAlarmTrigger: WaterAlarmTrigger):
return self._run_non_async(lambda:
self.set_inapp_water_alarm_trigger_async(inAppWaterAlarmTrigger)
)
[docs]
async def set_inapp_water_alarm_trigger_async(self, inAppWaterAlarmTrigger: WaterAlarmTrigger):
data = {
"channelIndex": 1,
"deviceId": self.id,
"inAppWaterAlarmTrigger": str(inAppWaterAlarmTrigger),
}
return await self._rest_call_async(
"device/configuration/setInAppWaterAlarmTrigger", data
)
[docs]
def set_siren_water_alarm_trigger(self, sirenWaterAlarmTrigger: WaterAlarmTrigger):
return self._run_non_async(lambda:
self.set_siren_water_alarm_trigger_async(sirenWaterAlarmTrigger)
)
[docs]
async def set_siren_water_alarm_trigger_async(self, sirenWaterAlarmTrigger: WaterAlarmTrigger):
LOGGER.warning(
"set_siren_water_alarm_trigger is currently not available in the HMIP App. It might not be available in the cloud yet"
)
data = {
"channelIndex": 1,
"deviceId": self.id,
"sirenWaterAlarmTrigger": str(sirenWaterAlarmTrigger),
}
return await self._rest_call_async(
"device/configuration/setSirenWaterAlarmTrigger", data
)
[docs]
class DinRailSwitch(FullFlushInputSwitch):
"""HMIP-DRSI1 (Switch Actuator for DIN rail mount – 1x channel)"""
[docs]
class AccelerationSensor(Device):
"""HMIP-SAM (Contact Interface flush-mount – 1 channel)"""
def __init__(self, connection):
super().__init__(connection)
#:float:
self.accelerationSensorEventFilterPeriod = 100.0
#:AccelerationSensorMode:
self.accelerationSensorMode = AccelerationSensorMode.ANY_MOTION
#:AccelerationSensorNeutralPosition:
self.accelerationSensorNeutralPosition = (
AccelerationSensorNeutralPosition.HORIZONTAL
)
#:AccelerationSensorSensitivity:
self.accelerationSensorSensitivity = (
AccelerationSensorSensitivity.SENSOR_RANGE_2G
)
#:int:
self.accelerationSensorTriggerAngle = 0
#:bool:
self.accelerationSensorTriggered = False
#:NotificationSoundType:
self.notificationSoundTypeHighToLow = NotificationSoundType.SOUND_NO_SOUND
#:NotificationSoundType:
self.notificationSoundTypeLowToHigh = NotificationSoundType.SOUND_NO_SOUND
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("ACCELERATION_SENSOR_CHANNEL", js)
if c:
self.set_attr_from_dict("accelerationSensorEventFilterPeriod", c)
self.set_attr_from_dict("accelerationSensorMode", c, AccelerationSensorMode)
self.set_attr_from_dict(
"accelerationSensorNeutralPosition",
c,
AccelerationSensorNeutralPosition,
)
self.set_attr_from_dict(
"accelerationSensorSensitivity", c, AccelerationSensorSensitivity
)
self.set_attr_from_dict("accelerationSensorTriggerAngle", c)
self.set_attr_from_dict("accelerationSensorTriggered", c)
self.set_attr_from_dict(
"notificationSoundTypeHighToLow", c, NotificationSoundType
)
self.set_attr_from_dict(
"notificationSoundTypeLowToHigh", c, NotificationSoundType
)
[docs]
def set_acceleration_sensor_mode(
self, mode: AccelerationSensorMode, channelIndex=1
):
return self._run_non_async(lambda:
self.set_acceleration_sensor_mode_async(mode, channelIndex)
)
[docs]
async def set_acceleration_sensor_mode_async(
self, mode: AccelerationSensorMode, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorMode": str(mode),
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorMode", data
)
[docs]
def set_acceleration_sensor_neutral_position(
self, neutralPosition: AccelerationSensorNeutralPosition, channelIndex=1
):
return self._run_non_async(lambda:
self.set_acceleration_sensor_neutral_position_async(neutralPosition, channelIndex)
)
[docs]
async def set_acceleration_sensor_neutral_position_async(
self, neutralPosition: AccelerationSensorNeutralPosition, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorNeutralPosition": str(neutralPosition),
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorNeutralPosition",
data,
)
[docs]
def set_acceleration_sensor_sensitivity(
self, sensitivity: AccelerationSensorSensitivity, channelIndex=1
):
return self._run_non_async(lambda:
self.set_acceleration_sensor_sensitivity_async(sensitivity, channelIndex)
)
[docs]
async def set_acceleration_sensor_sensitivity_async(
self, sensitivity: AccelerationSensorSensitivity, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorSensitivity": str(sensitivity),
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorSensitivity", data
)
[docs]
def set_acceleration_sensor_trigger_angle(self, angle: int, channelIndex=1):
return self._run_non_async(lambda:
self.set_acceleration_sensor_trigger_angle_async(angle, channelIndex)
)
[docs]
async def set_acceleration_sensor_trigger_angle_async(self, angle: int, channelIndex=1):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorTriggerAngle": angle,
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorTriggerAngle", data
)
[docs]
def set_acceleration_sensor_event_filter_period(
self, period: float, channelIndex=1
):
return self._run_non_async(lambda:
self.set_acceleration_sensor_event_filter_period_async(period, channelIndex)
)
[docs]
async def set_acceleration_sensor_event_filter_period_async(
self, period: float, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorEventFilterPeriod": period,
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorEventFilterPeriod",
data,
)
[docs]
def set_notification_sound_type(
self, soundType: NotificationSoundType, isHighToLow: bool, channelIndex=1
):
return self._run_non_async(lambda:
self.set_notification_sound_type_async(soundType, isHighToLow, channelIndex)
)
[docs]
async def set_notification_sound_type_async(
self, soundType: NotificationSoundType, isHighToLow: bool, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"notificationSoundType": str(soundType),
"isHighToLow": isHighToLow,
}
return await self._rest_call_async(
"device/configuration/setNotificationSoundType", data
)
[docs]
class DoorModule(Device):
"""Generic class for a door module"""
def __init__(self, connection):
super().__init__(connection)
self.doorState = DoorState.POSITION_UNKNOWN
self.on = False
self.processing = False
self.ventilationPositionSupported = False
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DOOR_CHANNEL", js)
if c:
self.set_attr_from_dict("doorState", c, DoorState)
self.set_attr_from_dict("on", c)
self.set_attr_from_dict("processing", c)
self.set_attr_from_dict("ventilationPositionSupported", c)
[docs]
def send_door_command(self, doorCommand=DoorCommand.STOP):
return self._run_non_async(lambda: self.send_door_command_async(doorCommand))
[docs]
async def send_door_command_async(self, doorCommand=DoorCommand.STOP):
data = {"channelIndex": 1, "deviceId": self.id, "doorCommand": doorCommand}
return await self._rest_call_async("device/control/sendDoorCommand", data)
[docs]
class GarageDoorModuleTormatic(DoorModule):
"""HMIP-MOD-TM (Garage Door Module Tormatic)"""
[docs]
class HoermannDrivesModule(DoorModule):
"""HMIP-MOD-HO (Garage Door Module for Hörmann)"""
[docs]
class PluggableMainsFailureSurveillance(Device):
"""HMIP-PMFS (Plugable Power Supply Monitoring)"""
def __init__(self, connection):
super().__init__(connection)
self.powerMainsFailure = False
self.genericAlarmSignal = AlarmSignalType.NO_ALARM
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("MAINS_FAILURE_CHANNEL", js)
if c:
self.set_attr_from_dict("powerMainsFailure", c)
self.set_attr_from_dict("genericAlarmSignal", c, AlarmSignalType)
[docs]
class WallMountedGarageDoorController(Device):
"""HmIP-WGC Wall mounted Garage Door Controller"""
def __init__(self, connection):
super().__init__(connection)
self.impulseDuration = 0
self.processing = False
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("IMPULSE_OUTPUT_CHANNEL", js)
if c:
self.set_attr_from_dict("impulseDuration", c)
self.set_attr_from_dict("processing", c)
[docs]
def send_start_impulse(self, channelIndex=2):
"""Toggle Wall mounted Garage Door Controller."""
return self._run_non_async(lambda: self.send_start_impulse_async(channelIndex))
[docs]
async def send_start_impulse_async(self, channelIndex=2):
"""Toggle Wall mounted Garage Door Controller."""
data = {"channelIndex": channelIndex, "deviceId": self.id}
return await self._rest_call_async("device/control/startImpulse", body=data)
[docs]
class TiltVibrationSensor(Device):
"""HMIP-STV (Inclination and vibration Sensor)"""
def __init__(self, connection):
super().__init__(connection)
#:float:
self.accelerationSensorEventFilterPeriod = 100.0
#:AccelerationSensorMode:
self.accelerationSensorMode = AccelerationSensorMode.ANY_MOTION
#:AccelerationSensorSensitivity:
self.accelerationSensorSensitivity = (
AccelerationSensorSensitivity.SENSOR_RANGE_2G
)
#:int:
self.accelerationSensorTriggerAngle = 0
#:bool:
self.accelerationSensorTriggered = False
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("TILT_VIBRATION_SENSOR_CHANNEL", js)
if c:
self.set_attr_from_dict("accelerationSensorEventFilterPeriod", c)
self.set_attr_from_dict("accelerationSensorMode", c, AccelerationSensorMode)
self.set_attr_from_dict(
"accelerationSensorSensitivity", c, AccelerationSensorSensitivity
)
self.set_attr_from_dict("accelerationSensorTriggerAngle", c)
self.set_attr_from_dict("accelerationSensorTriggered", c)
[docs]
def set_acceleration_sensor_mode(
self, mode: AccelerationSensorMode, channelIndex=1
):
return self._run_non_async(
self.set_acceleration_sensor_mode_async, mode, channelIndex
)
[docs]
async def set_acceleration_sensor_mode_async(
self, mode: AccelerationSensorMode, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorMode": str(mode),
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorMode", data
)
[docs]
def set_acceleration_sensor_sensitivity(
self, sensitivity: AccelerationSensorSensitivity, channelIndex=1
):
return self._run_non_async(
self.set_acceleration_sensor_sensitivity_async, sensitivity, channelIndex
)
[docs]
async def set_acceleration_sensor_sensitivity_async(
self, sensitivity: AccelerationSensorSensitivity, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorSensitivity": str(sensitivity),
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorSensitivity", data
)
[docs]
def set_acceleration_sensor_trigger_angle(self, angle: int, channelIndex=1):
return self._run_non_async(
self.set_acceleration_sensor_trigger_angle_async, angle, channelIndex
)
[docs]
async def set_acceleration_sensor_trigger_angle_async(self, angle: int, channelIndex=1):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorTriggerAngle": angle,
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorTriggerAngle", data
)
[docs]
def set_acceleration_sensor_event_filter_period(
self, period: float, channelIndex=1
):
return self._run_non_async(
self.set_acceleration_sensor_event_filter_period_async, period, channelIndex
)
[docs]
async def set_acceleration_sensor_event_filter_period_async(
self, period: float, channelIndex=1
):
data = {
"channelIndex": channelIndex,
"deviceId": self.id,
"accelerationSensorEventFilterPeriod": period,
}
return await self._rest_call_async(
"device/configuration/setAccelerationSensorEventFilterPeriod",
data,
)
[docs]
class RainSensor(Device):
"""HMIP-SRD (Rain Sensor)"""
def __init__(self, connection):
super().__init__(connection)
#:bool:
self.raining = False
#:float:
self.rainSensorSensitivity = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("RAIN_DETECTION_CHANNEL", js)
if c:
self.set_attr_from_dict("rainSensorSensitivity", c)
self.set_attr_from_dict("raining", c)
[docs]
class TemperatureDifferenceSensor2(Device):
"""HmIP-STE2-PCB (Temperature Difference Sensors - 2x sensors)"""
def __init__(self, connection):
super().__init__(connection)
#:float:
self.temperatureExternalDelta = 0.0
#:float:
self.temperatureExternalOne = 0.0
#:float:
self.temperatureExternalTwo = 0.0
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("TEMPERATURE_SENSOR_2_EXTERNAL_DELTA_CHANNEL", js)
if c:
self.set_attr_from_dict("temperatureExternalDelta", c)
self.set_attr_from_dict("temperatureExternalOne", c)
self.set_attr_from_dict("temperatureExternalTwo", c)
[docs]
class DoorLockDrive(OperationLockableDevice):
"""HmIP-DLD"""
def __init__(self, connection):
super().__init__(connection)
self.autoRelockDelay = False
self.doorHandleType = "UNKNOWN"
self.doorLockDirection = False
self.doorLockNeutralPosition = False
self.doorLockTurns = False
self.lockState = LockState.UNLOCKED
self.motorState = MotorState.STOPPED
self.door_lock_channel = 1
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DOOR_LOCK_CHANNEL", js)
if c:
self.set_attr_from_dict("autoRelockDelay", c)
self.set_attr_from_dict("doorHandleType", c)
self.set_attr_from_dict("doorLockDirection", c)
self.set_attr_from_dict("doorLockNeutralPosition", c)
self.set_attr_from_dict("doorLockTurns", c)
self.set_attr_from_dict("lockState", c, LockState)
self.set_attr_from_dict("motorState", c, MotorState)
self.door_lock_channel = c["index"]
[docs]
def set_lock_state(self, doorLockState: LockState, pin="", channelIndex=1):
"""sets the door lock state
Args:
doorLockState(float): the state of the door. See LockState from base/enums.py
pin(string): Pin, if specified.
channelIndex(int): the channel to control. Normally the channel from DOOR_LOCK_CHANNEL is used.
Returns:
the result of the _restCall
"""
self._run_non_async(self.set_lock_state_async, doorLockState, pin, channelIndex)
[docs]
async def set_lock_state_async(self, doorLockState: LockState, pin="", channelIndex=1):
"""sets the door lock state
Args:
doorLockState(float): the state of the door. See LockState from base/enums.py
pin(string): Pin, if specified.
channelIndex(int): the channel to control. Normally the channel from DOOR_LOCK_CHANNEL is used.
Returns:
the result of the _restCall
"""
if channelIndex == 1:
channelIndex = self.door_lock_channel
data = {
"deviceId": self.id,
"channelIndex": channelIndex,
"authorizationPin": pin,
"targetLockState": doorLockState,
}
return await self._rest_call_async("device/control/setLockState", data)
[docs]
class DoorLockSensor(Device):
"""HmIP-DLS"""
def __init__(self, connection):
super().__init__(connection)
self.door_lock_channel = None
self.doorLockDirection = ""
self.doorLockNeutralPosition = ""
self.doorLockTurns = 0
self.lockState = LockState.OPEN
[docs]
def from_json(self, js):
super().from_json(js)
c = get_functional_channel("DOOR_LOCK_SENSOR_CHANNEL", js)
if c:
self.set_attr_from_dict("doorLockDirection", c)
self.set_attr_from_dict("doorLockNeutralPosition", c)
self.set_attr_from_dict("doorLockTurns", c)
self.set_attr_from_dict("lockState", c, LockState)
self.door_lock_channel = c["index"]
[docs]
class EnergySensorsInterface(Device):
"""HmIP-ESI"""
def __init__(self, connection):
super().__init__(connection)
[docs]
class DaliGateway(Device):
"""HmIP-DRG-DALI Dali Gateway device."""
pass