193 lines
7.0 KiB
Python
193 lines
7.0 KiB
Python
from builtins import staticmethod
|
|
import logging
|
|
|
|
from mqtt import MqttMessage, MqttConfigMessage
|
|
|
|
from workers.base import BaseWorker
|
|
import logger
|
|
|
|
REQUIREMENTS = ['python-eq3bt']
|
|
monitoredAttrs = ["low_battery", "valve_state", "target_temperature", "window_open", "locked"]
|
|
_LOGGER = logger.get(__name__)
|
|
|
|
STATE_AWAY = 'away'
|
|
STATE_ECO = 'eco'
|
|
STATE_HEAT = 'heat'
|
|
STATE_AUTO = 'auto'
|
|
STATE_MANUAL = 'manual'
|
|
STATE_ON = 'on'
|
|
STATE_OFF = 'off'
|
|
|
|
|
|
class ThermostatWorker(BaseWorker):
|
|
class ModesMapper():
|
|
def __init__(self):
|
|
from eq3bt import Mode
|
|
|
|
self._mapped_modes = {
|
|
Mode.Closed: STATE_OFF,
|
|
Mode.Open: STATE_ON,
|
|
Mode.Auto: STATE_AUTO,
|
|
Mode.Manual: STATE_MANUAL,
|
|
Mode.Away: STATE_ECO,
|
|
Mode.Boost: STATE_HEAT,
|
|
}
|
|
|
|
self._reverse_modes = {v: k for k, v in self._mapped_modes.items()}
|
|
|
|
def get_mapping(self, mode):
|
|
if mode < 0:
|
|
return None
|
|
return self._mapped_modes[mode]
|
|
|
|
def get_reverse_mapping(self, mode):
|
|
return self._reverse_modes[mode]
|
|
|
|
@staticmethod
|
|
def away_mode_on_off(mode):
|
|
if mode == STATE_ECO:
|
|
return STATE_ON
|
|
else:
|
|
return STATE_OFF
|
|
|
|
@staticmethod
|
|
def on_off_to_mode(on_off):
|
|
if on_off == STATE_ON:
|
|
return STATE_ECO
|
|
else:
|
|
return STATE_HEAT
|
|
|
|
def _setup(self):
|
|
from eq3bt import Thermostat
|
|
|
|
_LOGGER.info("Adding %d %s devices", len(self.devices), repr(self))
|
|
for name, mac in self.devices.items():
|
|
_LOGGER.debug("Adding %s device '%s' (%s)", repr(self), name, mac)
|
|
self.devices[name] = {"mac": mac, "thermostat": Thermostat(mac)}
|
|
|
|
self._modes_mapper = self.ModesMapper()
|
|
|
|
def config(self):
|
|
ret = []
|
|
for name, data in self.devices.items():
|
|
ret += self.config_device(name, data["mac"])
|
|
return ret
|
|
|
|
def config_device(self, name, mac):
|
|
ret = []
|
|
device={"identifiers": [mac, self.format_id(name, separator="_")],
|
|
"manufacturer": "eQ-3",
|
|
"model": "Smart Radiator Thermostat",
|
|
"name": self.format_topic(name, separator=" ").title()}
|
|
|
|
payload = {"unique_id": self.format_id(name, 'climate', separator="_"),
|
|
"qos": 1,
|
|
"temperature_state_topic": self.format_topic(name, 'target_temperature'),
|
|
"temperature_command_topic": self.format_topic(name, 'target_temperature', 'set'),
|
|
"mode_state_topic": self.format_topic(name, 'mode'),
|
|
"mode_command_topic": self.format_topic(name, 'mode', 'set'),
|
|
"away_mode_state_topic": self.format_topic(name, 'away'),
|
|
"away_mode_command_topic": self.format_topic(name, 'away', 'set'),
|
|
"min_temp": 5.0,
|
|
"max_temp": 29.5,
|
|
"temp_step": 0.5,
|
|
"payload_on": "'on'",
|
|
"payload_off": "'off'",
|
|
"modes": ["heat", "auto", "manual", "eco", 'off'],
|
|
"device": device}
|
|
ret.append(MqttConfigMessage(MqttConfigMessage.CLIMATE, self.format_topic(name, 'climate', separator="_"), payload=payload))
|
|
|
|
payload = {"unique_id": self.format_id(name, 'window_open', separator="_"),
|
|
"state_topic": self.format_topic(name, 'window_open'),
|
|
"device_class": 'window',
|
|
"payload_on": "True",
|
|
"payload_off": "False",
|
|
"device": device}
|
|
ret.append(MqttConfigMessage(MqttConfigMessage.BINARY_SENSOR, self.format_topic(name, 'window_open', separator="_"), payload=payload))
|
|
|
|
payload = {"unique_id": self.format_id(name, 'low_battery', separator="_"),
|
|
"state_topic": self.format_topic(name, 'low_battery'),
|
|
"device_class": 'battery',
|
|
"payload_on": "True",
|
|
"payload_off": "False",
|
|
"device": device}
|
|
ret.append(MqttConfigMessage(MqttConfigMessage.BINARY_SENSOR, self.format_topic(name, 'low_battery', separator="_"), payload=payload))
|
|
|
|
payload = {"unique_id": self.format_id(name, 'locked', separator="_"),
|
|
"state_topic": self.format_topic(name, 'locked'),
|
|
"device_class": 'lock',
|
|
"payload_on": "False",
|
|
"payload_off": "True",
|
|
"device": device}
|
|
ret.append(MqttConfigMessage(MqttConfigMessage.BINARY_SENSOR, self.format_topic(name, 'locked', separator="_"),
|
|
payload=payload))
|
|
|
|
payload = {"unique_id": self.format_id(name, "valve_state", separator="_"),
|
|
"state_topic": self.format_topic(name, "valve_state"),
|
|
"unit_of_measurement": "%",
|
|
"device": device}
|
|
ret.append(MqttConfigMessage(MqttConfigMessage.SENSOR, self.format_topic(name, 'valve_state', separator="_"), payload=payload))
|
|
|
|
return ret
|
|
|
|
|
|
def status_update(self):
|
|
from bluepy import btle
|
|
|
|
ret = []
|
|
_LOGGER.info("Updating %d %s devices", len(self.devices), repr(self))
|
|
for name, data in self.devices.items():
|
|
_LOGGER.debug("Updating %s device '%s' (%s)", repr(self), name, data["mac"])
|
|
try:
|
|
ret += self.update_device_state(name, data["thermostat"])
|
|
except btle.BTLEException as e:
|
|
logger.log_exception(_LOGGER, "Error during update of %s device '%s' (%s): %s", repr(self), name, data["mac"], type(e).__name__, suppress=True)
|
|
return ret
|
|
|
|
def on_command(self, topic, value):
|
|
from bluepy import btle
|
|
_, device_name, method, _ = topic.split('/')
|
|
|
|
data = self.devices[device_name]
|
|
|
|
value = value.decode('utf-8')
|
|
|
|
if method == STATE_AWAY:
|
|
method = "mode"
|
|
value = self.ModesMapper.on_off_to_mode(value)
|
|
|
|
# It needs to be on separate if because first if can change method
|
|
if method == "mode":
|
|
value = self._modes_mapper.get_reverse_mapping(value)
|
|
elif method == "target_temperature":
|
|
value = float(value)
|
|
|
|
_LOGGER.info("Setting %s to %s on %s device '%s' (%s)", method, value, repr(self), device_name, data["mac"])
|
|
try:
|
|
setattr(data["thermostat"], method, value)
|
|
except btle.BTLEException as e:
|
|
logger.log_exception(_LOGGER, "Error setting %s to %s on %s device '%s' (%s): %s", method, value, repr(self), device_name, data["mac"], type(e).__name__)
|
|
return []
|
|
|
|
try:
|
|
return self.update_device_state(device_name, data["thermostat"])
|
|
except btle.BTLEException as e:
|
|
logger.log_exception(_LOGGER, "Error during update of %s device '%s' (%s): %s", repr(self), device_name, data["mac"], type(e).__name__, suppress=True)
|
|
return []
|
|
|
|
def update_device_state(self, name, thermostat):
|
|
thermostat.update()
|
|
|
|
ret = []
|
|
for attr in monitoredAttrs:
|
|
ret.append(MqttMessage(topic=self.format_topic(name, attr), payload=getattr(thermostat, attr)))
|
|
|
|
ha_mode = self._modes_mapper.get_mapping(thermostat.mode)
|
|
ret.append(MqttMessage(topic=self.format_topic(name, 'mode'), payload=ha_mode))
|
|
ret.append(MqttMessage(topic=self.format_topic(name, STATE_AWAY), payload=self.ModesMapper.away_mode_on_off(ha_mode)))
|
|
|
|
return ret
|
|
|
|
def device_for(self, mac):
|
|
return eq3bt.Thermostat(mac)
|