cube.py

import base64
from datetime import datetime
import json
import logging
import struct
from typing import Callable

from maxcube.device import (
    MAX_CUBE,
    MAX_DEVICE_MODE_AUTOMATIC,
    MAX_DEVICE_MODE_MANUAL,
    MAX_THERMOSTAT,
    MAX_THERMOSTAT_PLUS,
    MAX_WALL_THERMOSTAT,
    MAX_WINDOW_SHUTTER,
    MaxDevice,
)
from maxcube.room import MaxRoom
from maxcube.thermostat import MaxThermostat
from maxcube.wallthermostat import MaxWallThermostat
from maxcube.windowshutter import MaxWindowShutter

from .commander import Commander

logger = logging.getLogger(__name__)

CMD_SET_PROG = "10"
UNKNOWN = "00"
RF_FLAG_IS_ROOM = "04"
RF_FLAG_IS_DEVICE = "00"
RF_NULL_ADDRESS = "000000"
DEFAULT_PORT = 62910
DAYS = [
    "saturday",
    "sunday",
    "monday",
    "tuesday",
    "wednesday",
    "thursday",
    "friday",
    "saturday",
    "sunday",
]


class MaxCube(MaxDevice):
    def __init__(
        self,
        host: str,
        port: int = DEFAULT_PORT,
        now: Callable[[], datetime] = datetime.now,
    ):
        super(MaxCube, self).__init__()
        self.__commander = Commander(host, port)
        self.name = "Cube"
        self.type = MAX_CUBE
        self.firmware_version = None
        self.devices = []
        self.rooms = []
        self._now: Callable[[], datetime] = now
        self.update()
        self.log()

    @property
    def use_persistent_connection(self) -> bool:
        return self.__commander.use_persistent_connection

    @use_persistent_connection.setter
    def use_persistent_connection(self, value: bool) -> None:
        self.__commander.use_persistent_connection = value

    def disconnect(self):
        self.__commander.disconnect()

    def __str__(self):
        return self.describe("CUBE", f"firmware={self.firmware_version}")

    def log(self):
        logger.info(str(self))
        for room in self.rooms:
            logger.info(f" * ROOM {room.name}")
            for device in self.devices_by_room(room):
                logger.info(" --- " + str(device))

    def update(self):
        self.__parse_responses(self.__commander.update())

    def get_devices(self):
        return self.devices

    def device_by_rf(self, rf):
        for device in self.devices:
            if device.rf_address == rf:
                return device
        return None

    def devices_by_room(self, room):
        rooms = []

        for device in self.devices:
            if device.room_id == room.id:
                rooms.append(device)

        return rooms

    def get_rooms(self):
        return self.rooms

    def room_by_id(self, id):
        for room in self.rooms:
            if room.id == id:
                return room
        return None

    def __parse_responses(self, messages):
        for msg in messages:
            try:
                cmd = msg.cmd
                if cmd == "C":
                    self.parse_c_message(msg.arg)
                elif cmd == "H":
                    self.parse_h_message(msg.arg)
                elif cmd == "L":
                    self.parse_l_message(msg.arg)
                elif cmd == "M":
                    self.parse_m_message(msg.arg)
                else:
                    logger.debug("Ignored unsupported message: %s" % (msg))
            except Exception:
                logger.warn(f"Error processing response message {msg}", exc_info=True)

    def parse_c_message(self, message):
        logger.debug("Parsing c_message: " + message)
        params = message.split(",")
        device_rf_address = params[0].upper()
        data = bytearray(base64.b64decode(params[1]))

        device = self.device_by_rf(device_rf_address)
        if device and device.is_thermostat():
            device.comfort_temperature = data[18] / 2.0
            device.eco_temperature = data[19] / 2.0
            device.max_temperature = data[20] / 2.0
            device.min_temperature = data[21] / 2.0
            device.programme = get_programme(data[29:])

        if device and device.is_wallthermostat():
            device.comfort_temperature = data[18] / 2.0
            device.eco_temperature = data[19] / 2.0
            device.max_temperature = data[20] / 2.0
            device.min_temperature = data[21] / 2.0
            device.programme = get_programme(data[22:204])
            
        if device and device.is_windowshutter():
            # Pure Speculation based on this:
            # Before: [17][12][162][178][4][0][20][15]KEQ0839778
            # After:  [17][12][162][178][4][1][20][15]KEQ0839778
            device.initialized = data[5]

    def parse_h_message(self, message):
        logger.debug("Parsing h_message: " + message)
        tokens = message.split(",")
        self.serial = tokens[0]
        self.rf_address = tokens[1]
        self.firmware_version = (tokens[2][0:2]) + "." + (tokens[2][2:4])

    def parse_m_message(self, message):
        logger.debug("Parsing m_message: " + message)
        data = bytearray(base64.b64decode(message.split(",")[2]))
        num_rooms = data[2]

        pos = 3
        for _ in range(0, num_rooms):
            room_id = struct.unpack("bb", data[pos : pos + 2])[0]
            name_length = struct.unpack("bb", data[pos : pos + 2])[1]
            pos += 1 + 1
            name = data[pos : pos + name_length].decode("utf-8")
            pos += name_length
            device_rf_address = self.parse_rf_address(data[pos : pos + 3])
            pos += 3

            room = self.room_by_id(room_id)

            if not room:
                room = MaxRoom()
                room.id = room_id
                room.name = name
                self.rooms.append(room)
            else:
                room.name = name

        num_devices = data[pos]
        pos += 1

        for device_idx in range(0, num_devices):
            device_type = data[pos]
            device_rf_address = self.parse_rf_address(data[pos + 1 : pos + 1 + 3])
            device_serial = data[pos + 4 : pos + 14].decode("utf-8")
            device_name_length = data[pos + 14]
            device_name = data[pos + 15 : pos + 15 + device_name_length].decode("utf-8")
            room_id = data[pos + 15 + device_name_length]

            device = self.device_by_rf(device_rf_address)

            if not device:
                if device_type == MAX_THERMOSTAT or device_type == MAX_THERMOSTAT_PLUS:
                    device = MaxThermostat()

                if device_type == MAX_WINDOW_SHUTTER:
                    device = MaxWindowShutter()

                if device_type == MAX_WALL_THERMOSTAT:
                    device = MaxWallThermostat()

                if device:
                    self.devices.append(device)

            if device:
                device.type = device_type
                device.rf_address = device_rf_address
                device.room_id = room_id
                device.name = device_name
                device.serial = device_serial

            pos += 1 + 3 + 10 + device_name_length + 2

    def parse_l_message(self, message):
        logger.debug("Parsing l_message: " + message)
        data = bytearray(base64.b64decode(message))
        pos = 0

        while pos < len(data):
            length = data[pos]
            device_rf_address = self.parse_rf_address(data[pos + 1 : pos + 4])

            device = self.device_by_rf(device_rf_address)

            if device:
                bits1, bits2 = struct.unpack("BB", bytearray(data[pos + 5 : pos + 7]))
                device.battery = self.resolve_device_battery(bits2)

            # Thermostat or Wall Thermostat
            if device and (device.is_thermostat() or device.is_wallthermostat()):
                device.target_temperature = (data[pos + 8] & 0x7F) / 2.0
                bits1, bits2 = struct.unpack("BB", bytearray(data[pos + 5 : pos + 7]))
                device.mode = self.resolve_device_mode(bits2)

            # Thermostat
            if device and device.is_thermostat():
                device.valve_position = data[pos + 7]
                if (
                    device.mode == MAX_DEVICE_MODE_MANUAL
                    or device.mode == MAX_DEVICE_MODE_AUTOMATIC
                ):
                    actual_temperature = (
                        (data[pos + 9] & 0xFF) * 256 + (data[pos + 10] & 0xFF)
                    ) / 10.0
                    if actual_temperature != 0:
                        device.actual_temperature = actual_temperature
                else:
                    device.actual_temperature = None

            # Wall Thermostat
            if device and device.is_wallthermostat():
                device.actual_temperature = (
                    ((data[pos + 8] & 0x80) << 1) + data[pos + 12]
                ) / 10.0

            # Window Shutter
            if device and device.is_windowshutter():
                status = data[pos + 6] & 0x03
                if status > 0:
                    device.is_open = True
                else:
                    device.is_open = False

            # Advance our pointer to the next submessage
            pos += length + 1

    def set_target_temperature(self, thermostat, temperature):
        return self.set_temperature_mode(thermostat, temperature, None)

    def set_mode(self, thermostat, mode):
        return self.set_temperature_mode(thermostat, None, mode)

    def set_temperature_mode(self, thermostat, temperature, mode):
        logger.debug(
            "Setting temperature %s and mode %s on %s!",
            temperature,
            mode,
            thermostat.rf_address,
        )

        if not thermostat.is_thermostat() and not thermostat.is_wallthermostat():
            logger.error("%s is no (wall-)thermostat!", thermostat.rf_address)
            return

        if mode is None:
            mode = thermostat.mode
        if temperature is None:
            temperature = (
                0
                if mode == MAX_DEVICE_MODE_AUTOMATIC
                else thermostat.target_temperature
            )

        rf_address = thermostat.rf_address
        room = to_hex(thermostat.room_id)
        target_temperature = int(temperature * 2) + (mode << 6)

        byte_cmd = "000440000000" + rf_address + room + to_hex(target_temperature)
        if self.__commander.send_radio_msg(byte_cmd):
            thermostat.mode = mode
            if temperature > 0:
                thermostat.target_temperature = int(temperature * 2) / 2.0
            elif mode == MAX_DEVICE_MODE_AUTOMATIC:
                thermostat.target_temperature = thermostat.get_programmed_temp_at(
                    self._now()
                )
            return True
        return False

    def set_programme(self, thermostat, day, metadata):
        # compare with current programme
        if thermostat.programme[day] == metadata:
            logger.debug("Skipping setting unchanged programme for " + day)
            return

        heat_time_tuples = [(x["temp"], x["until"]) for x in metadata]
        # pad heat_time_tuples so that there are always seven
        for _ in range(7 - len(heat_time_tuples)):
            heat_time_tuples.append((0, "00:00"))
        command = ""
        if thermostat.is_room():
            rf_flag = RF_FLAG_IS_ROOM
            devices = self.devices_by_room(thermostat)
        else:
            rf_flag = RF_FLAG_IS_DEVICE
            devices = [thermostat]
        command += UNKNOWN + rf_flag + CMD_SET_PROG + RF_NULL_ADDRESS
        for device in devices:
            command += device.rf_address
            command += to_hex(device.room_id)
            command += to_hex(n_from_day_of_week(day))
            for heat, time in heat_time_tuples:
                command += temp_and_time(heat, time)
        return self.__commander.send_radio_msg(command)

    def devices_as_json(self):
        devices = []
        for device in self.devices:
            devices.append(device.to_dict())
        return json.dumps(devices, indent=2)

    def set_programmes_from_config(self, config_file):
        config = json.load(config_file)
        for device_config in config:
            device = self.device_by_rf(device_config["rf_address"])
            programme = device_config["programme"]
            if not programme:
                # e.g. a wall thermostat
                continue
            for day, metadata in programme.items():
                self.set_programme(device, day, metadata)

    @classmethod
    def resolve_device_mode(cls, bits):
        return bits & 3

    @classmethod
    def resolve_device_battery(cls, bits):
        return bits >> 7

    @classmethod
    def parse_rf_address(cls, address):
        return "".join("{:02X}".format(x) for x in address)


def get_programme(bits):
    n = 26
    programme = {}
    days = [bits[i : i + n] for i in range(0, len(bits), n)]
    for j, day in enumerate(days):
        n = 2
        settings = [day[i : i + n] for i in range(0, len(day), n)]
        day_programme = []
        for setting in settings:
            word = format(setting[0], "08b") + format(setting[1], "08b")
            temp = int(int(word[:7], 2) / 2)
            time_mins = int(word[7:], 2) * 5
            mins = time_mins % 60
            hours = int((time_mins - mins) / 60)
            time = "{:02d}:{:02d}".format(hours, mins)
            day_programme.append({"temp": temp, "until": time})
            if time == "24:00":
                # This appears to flag the end of usable set points
                break
        programme[day_of_week_from_n(j)] = day_programme
    return programme


def n_from_day_of_week(day):
    return DAYS.index(day)


def day_of_week_from_n(day):
    return DAYS[day]


def temp_and_time(temp, time):
    temp = float(temp)
    assert temp <= 32, "Temp must be 32 or lower"
    assert temp % 0.5 == 0, "Temp must be increments of 0.5"
    temp = int(temp * 2)
    hours, mins = [int(x) for x in time.split(":")]
    assert mins % 5 == 0, "Time must be a multiple of 5 mins"
    mins = hours * 60 + mins
    bits = format(temp, "07b") + format(int(mins / 5), "09b")
    return to_hex(int(bits, 2))


def to_hex(value):
    "Return value as hex word"
    return format(value, "02X")