import appdaemon.plugins.hass.hassapi as hass import struct import binascii # for PID import time class Climate(hass.Hass): ir_topic = "home/remote/rm2/code/set" target_temp = 24.0 #default pid = PID(P = 0.2, I=0.0, D=0.0) def initialize(self): if "target_temp" not in self.args or "temp_sensor" not in self.args or "ha_panel" not in self.args or "door_window" not in self.args: self.error("Please provide target_temp, temp_sensor, ha_panel, door_window in config!") return self.listen_state(self.target_temp_changed, self.args['target_temp']) self.listen_state(self.temp_sensor_changed, self.args['temp_sensor']) self.listen_state(self.constraint_changed, self.args['constraint']) self.listen_state(self.ha_panel_changed, self.args['ha_panel']) self.listen_state(self.door_window_changed, self.args['door_window']) self.target_temp_changed("", "", "", "", "") def door_window_changed(self, entity, attribute, old, new, kwargs): self.do_action() def ha_panel_changed(self, entity, attribute, old, new, kwargs): self.do_action() def constraint_changed(self, entity, attribute, old, new, kwargs): self.do_action() def target_temp_changed(self, entity, attribute, old, new, kwargs): self.target_temp = float(self.get_state(self.args['target_temp'])) self.pid.SetPoint = self.target_temp self.log('Target temp changed to: ' + str(self.target_temp)) self.do_action() def temp_sensor_changed(self, entity, attribute, old, new, kwargs): self.log('New room temp: ' + str(new)) self.do_action() def split_off(self): remote = Remote() code = remote.set_mode("NONE", "NONE", self.target_temp, "OFF") self.call_service("mqtt/publish", topic = self.ir_topic, payload = code.decode("utf-8")) def do_action(self): if 'constraint' in self.args and not self.constrain_input_boolean(self.args['constraint']): self.log("Temperature control is disabled.") self.split_off() return # Проверить alarm_panel ha_panel_state = self.get_state(self.args['ha_panel']) if (ha_panel_state != 'disarmed'): self.log("Nobody home. Turning off split.") self.split_off() return # Проверить датчики дверей и окон door_window_state = self.get_state(self.args['door_window']) if (door_window_state != 'off'): self.log("Balcony door is opened. Turning off split.") self.split_off() return room_temp = float(self.get_state(self.args['temp_sensor'])) self.pid.update(room_temp) output = self.pid.output self.log("PID: " + str(output) + ".") # Определяем режим mode = "COOLING" room_temp = float(self.get_state(self.args['temp_sensor'])) if (room_temp >= self.target_temp): mode = "COOLING" else: mode = "HEATING" temp_diff = abs(room_temp - self.target_temp) if (output < 0): self.log("Temperature differs on {} degree, turn on split for {}".format(temp_diff, mode)) remote = Remote() code = remote.set_mode(mode, "2", int(self.target_temp), "ON") self.call_service("mqtt/publish", topic = self.ir_topic, payload = code.decode("utf-8")) else: self.log("Temperature is ok, turn off split.") self.split_off() # # App to control climate device (split system) # # Args: # # notify = notification platform to send notifications to # # None # # Release Notes # # Version 1.0: # Initial Version # ------------------------------------------------------------------------------------------------- """ Library that genrates LG air conditioner remote codes """ FIRST_BYTE = 136 # b10001000 STATE = { "ON": 0, "OFF": 24, # b11000 "CHANGE_MODE": 1, } MODE = { "HEATING": 4, # b100 "AUTO": 3, # b011 "DEHUIDIFICATION": 1, # b001 "COOLING": 0, "NONE": 0, } TEMPERATURE_OFFSET = 15 FAN = { "1": 1, "2": 0, "3": 2, "4": 4, # b100 "NONE": 5, # b101 } FIRST_HIGH = 8271 FIRST_LOW = 4298 ZERO_AND_ONE_HIGH = 439 ZERO_LOW = 647 ONE_LOW = 1709 BUFFER_SIZE = 59 def test_bit(num, offset): """ Test the num(int) if at the given offset bit is 1 """ mask = 1 << offset return num & mask def set_bit(num, offset): """ Set bit to at the given offset to 1 """ mask = 1 << offset return num | mask class Remote(object): """ Library that genrates LG air conditioner remote codes """ def __init__(self): self.codes = [0] * BUFFER_SIZE self.crc = 0 def set_mode(self, mode, fan, temperature, state): """ Generate code and put it in the buffer """ self.codes[0] = FIRST_HIGH self.codes[1] = FIRST_LOW self.crc = 0 self.fill_buffer(0, 8, FIRST_BYTE) self.fill_buffer(8, 5, STATE[state]) if state == 'OFF': self.fill_buffer(13, 3, MODE['NONE']) else: self.fill_buffer(13, 3, MODE[mode]) if state == 'OFF': self.fill_buffer(16, 4, 0) else: self.fill_buffer(16, 4, temperature - TEMPERATURE_OFFSET) self.fill_buffer(20, 1, 0) # jet if state == 'OFF': self.fill_buffer(21, 3, FAN['NONE']) else: self.fill_buffer(21, 3, FAN[fan]) self.fill_buffer(24, 4, self.crc) self.codes[BUFFER_SIZE - 1] = ZERO_AND_ONE_HIGH self.codes = binascii.hexlify(self.lirc2broadlink(self.codes)) return self.codes def lirc2broadlink(self, pulses): array = bytearray() for pulse in pulses: pulse = int(pulse * 269 / 8192) # 32.84ms units if pulse < 256: array += bytearray(struct.pack('>B', pulse)) # big endian (1-byte) else: array += bytearray([0x00]) # indicate next number is 2-bytes array += bytearray(struct.pack('>H', pulse)) # big endian (2-bytes) packet = bytearray([0x26, 0x00]) # 0x26 = IR, 0x00 = no repeats packet += bytearray(struct.pack(' 0: index = 2 + 2 * (pos + bits-i) self.codes[index] = ZERO_AND_ONE_HIGH if test_bit(value, i - 1) != 0: self.codes[index + 1] = ONE_LOW else: self.codes[index + 1] = ZERO_LOW if test_bit(value, i - 1) != 0: bitset = 0 bitset = set_bit(bitset, (128 + i - pos - bits - 1) % 4) self.crc = self.crc + bitset i -= 1 # ------------------- PID CONTROLLER -------------------------------------------- class PID: """PID Controller """ def __init__(self, P=0.2, I=0.0, D=0.0): self.Kp = P self.Ki = I self.Kd = D self.sample_time = 0.00 self.current_time = time.time() self.last_time = self.current_time self.clear() def clear(self): """Clears PID computations and coefficients""" self.SetPoint = 0.0 self.PTerm = 0.0 self.ITerm = 0.0 self.DTerm = 0.0 self.last_error = 0.0 # Windup Guard self.int_error = 0.0 self.windup_guard = 20.0 self.output = 0.0 def update(self, feedback_value): """Calculates PID value for given reference feedback .. math:: u(t) = K_p e(t) + K_i \int_{0}^{t} e(t)dt + K_d {de}/{dt} .. figure:: images/pid_1.png :align: center Test PID with Kp=1.2, Ki=1, Kd=0.001 (test_pid.py) """ error = self.SetPoint - feedback_value self.current_time = time.time() delta_time = self.current_time - self.last_time delta_error = error - self.last_error if (delta_time >= self.sample_time): self.PTerm = self.Kp * error self.ITerm += error * delta_time if (self.ITerm < -self.windup_guard): self.ITerm = -self.windup_guard elif (self.ITerm > self.windup_guard): self.ITerm = self.windup_guard self.DTerm = 0.0 if delta_time > 0: self.DTerm = delta_error / delta_time # Remember last time and last error for next calculation self.last_time = self.current_time self.last_error = error self.output = self.PTerm + (self.Ki * self.ITerm) + (self.Kd * self.DTerm) def setKp(self, proportional_gain): """Determines how aggressively the PID reacts to the current error with setting Proportional Gain""" self.Kp = proportional_gain def setKi(self, integral_gain): """Determines how aggressively the PID reacts to the current error with setting Integral Gain""" self.Ki = integral_gain def setKd(self, derivative_gain): """Determines how aggressively the PID reacts to the current error with setting Derivative Gain""" self.Kd = derivative_gain def setWindup(self, windup): """Integral windup, also known as integrator windup or reset windup, refers to the situation in a PID feedback controller where a large change in setpoint occurs (say a positive change) and the integral terms accumulates a significant error during the rise (windup), thus overshooting and continuing to increase as this accumulated error is unwound (offset by errors in the other direction). The specific problem is the excess overshooting. """ self.windup_guard = windup def setSampleTime(self, sample_time): """PID that should be updated at a regular interval. Based on a pre-determined sampe time, the PID decides if it should compute or return immediately. """ self.sample_time = sample_time