Friedrich A/C platform for home-assistant

__init__.py

from homeassistant.const import Platform

async def async_setup_entry(hass, entry):
    hass.config_entries.setup_platforms(entry, [Platform.CLIMATE])
    return True

friedrich.py

"""
Support for Friedrich's Windows A/Cs with KWIFI module.
Only KWIFI modules bought after Jan 1, 2018.
configuration.yaml
climate:
  - platform: friedrich
    email: [email protected]
    password: hunter2

Limitations:
- does not support heating features, it's easy to add support for, but need a tester to confirm codes for operation modes
- does not support pulling unit state or live MQTT communication, need documentation on Xively MQTT JWT auth
- Xively session expiration is not well tested
- 10 devices max, requires implementation of pagination for more

TODOs:
- need to add a way to customize attributes?
- fuzzy/case match operation modes et al.
"""
import logging
import json
import voluptuous as vol
from threading import Timer

from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.components.climate import ClimateEntity, ENTITY_ID_FORMAT
from homeassistant.components.climate.const import (HVAC_MODE_COOL, HVAC_MODE_FAN_ONLY, HVAC_MODE_OFF,
                                              SUPPORT_TARGET_TEMPERATURE, SUPPORT_FAN_MODE,
                                              FAN_LOW, FAN_MEDIUM, FAN_HIGH)
from homeassistant.const import (CONF_EMAIL, CONF_PASSWORD,
                                 TEMP_FAHRENHEIT, ATTR_TEMPERATURE)
import homeassistant.helpers.config_validation as cv
from homeassistant.util.percentage import (
    percentage_to_ordered_list_item,
    ordered_list_item_to_percentage
)

import requests

_LOGGER = logging.getLogger(__name__)

# see: https://developer.xively.com/docs/using-jwts for expiration etc.
def _xively_jwt_authenticate(email, password):

    XIVELY_JWT_AUTH_URL = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/verifyPassword?key=AIzaSyCgfjr_nYDCpnlUL_ZxG2ayzf7JS5TS0SU'

    payload = {
        'email': email,
        'password': password,
        'returnSecureToken': True
    }

    # Using the json parameter in the request will change the Content-Type in the header to application/json.
    req = requests.post(XIVELY_JWT_AUTH_URL, json=payload)

    if req.status_code != requests.codes.ok:
        _LOGGER.error("Error authenticating to Xively, code %d", req.status_code)
        return None

    return req.json()

def refresh_entities(entities, refresh_token):
    new_jwt = refresh_jwt(refresh_token)
    for entity in entities:
        entity._jwt = new_jwt
    Timer(3500, refresh_entities, (entities, refresh_token)).start()
    

def refresh_jwt(jwt):
    XIVELY_REFRESH_JWT = 'https://securetoken.googleapis.com/v1/token?key=AIzaSyCgfjr_nYDCpnlUL_ZxG2ayzf7JS5TS0SU'
    
    payload = {
        'grant_type': 'refresh_token',
        'refresh_token': jwt
    }

    headers = {
        'Content-Type': 'application/json'
    }

    req = requests.post(XIVELY_REFRESH_JWT, headers=headers, json=payload)

    if req.status_code != requests.codes.ok:
        _LOGGER.error("Error refreshing Xively JWT, code %d", req.status_code)
        return None

    return req.json()['access_token']

def _xively_list_devices(jwt):

    FRIEDRICH_LIST_DEVICES = 'https://us-central1-friedrich-prod.cloudfunctions.net/batch'

    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + jwt
    }

    payload = { 'data': None }

    req = requests.post(FRIEDRICH_LIST_DEVICES, headers=headers, json=payload)

    if req.status_code != requests.codes.ok:
        _LOGGER.error("Error listing Xively devices, code %d", req.status_code)
        return None

    return req.json()['result'][0]['devices']['results']

FRIEDRICH_TEMP_TIMESERIE = 'https://us-central1-friedrich-prod.cloudfunctions.net/getLatestTimeSeriesValues'
FRIEDRICH_PUBLISH_COMMAND = 'https://us-central1-friedrich-prod.cloudfunctions.net/publishToDevice'

# pylint: disable=unused-argument
def setup_platform(hass, config, add_devices, discovery_info=None):
    """Setup the friedrich platform."""
    # import requirements for connection

    # Assign configuration variables.
    email = config.get(CONF_EMAIL)
    password = config.get(CONF_PASSWORD)

    # Authenticate to Xively
    auth_data = _xively_jwt_authenticate(email, password)
    jwt = auth_data['idToken']
    refresh_token = auth_data['refreshToken']
    if jwt is None:
        _LOGGER.error("Could not connect to Friedrich account")
        return

    # Fetch devices
    devices = _xively_list_devices(jwt)
    if devices is None:
        _LOGGER.error("Could not list Friedrich devices")
        return

    # Add devices
    entities = list(FriedrichHVAC(device['name'], jwt, device['id']) for device in devices)
    entities += list(FriedrichFan(devices[i]['name'], entities[i]) for i in range(len(devices)))
    Timer(3500, refresh_entities, (entities, refresh_token)).start()
    add_devices(entities)

class FriedrichFan(FanEntity):
    def __init__(self, name, hvac):
        self._name = name + ' Fan'
        self._hvac = hvac
        self._percentage = 0
        self._is_on = False

    @property
    def supported_features(self):
        return FanEntityFeature.SET_SPEED

    @property
    def speed_count(self):
        return 3

    @property
    def preset_modes(self):
        return []

    @property
    def name(self):
        return self._name

    @property
    def percentage(self):
        return self._percentage

    @property
    def is_on(self):
        return self._is_on

    @property
    def unique_id(self):
        return 'fan-' + self._hvac.unique_id

    def turn_on(self, speed = None, percentage = None, preset_mode = None, **kwargs):
        self._is_on = True
        if self._hvac.hvac_mode == HVAC_MODE_OFF:
            self._hvac.set_hvac_mode(HVAC_MODE_FAN_ONLY)
        if percentage is not None:
            self.set_percentage(percentage)
    
    def turn_off(self, **kwargs):
        self._is_on = False
        if self._hvac.hvac_mode != HVAC_MODE_OFF:
            self._hvac.set_hvac_mode(HVAC_MODE_OFF)

    def toggle(self, **kwargs):
        if self._is_on:
            self.turn_off()
        else:
            self.turn_on()

    def update(self):
        self._is_on = self._hvac.hvac_mode != HVAC_MODE_OFF
        speeds = [FAN_LOW, FAN_MEDIUM, FAN_HIGH]
        self._percentage = ordered_list_item_to_percentage(speeds, self._hvac.fan_mode)

    def set_percentage(self, percentage):
        self._percentage = percentage
        if percentage == 0:
            self.turn_off()
        else:
            self.turn_on()
        fan_speed = percentage_to_ordered_list_item([FAN_LOW, FAN_MEDIUM, FAN_HIGH], percentage)
        self._hvac.set_fan_mode(fan_speed)

# pylint: disable=abstract-method
# pylint: disable=too-many-instance-attributes
class FriedrichHVAC(ClimateEntity):
    """Representation of a Friedrich HVAC unit."""

    def __init__(self, name, jwt, device_id):
        """Initialize a Friedrich HVAC unit."""

        # initialization attributes
        self._name = name + ' AC'
        self._jwt = jwt
        self._device_id = device_id
        # self.entity_id = ENTITY_ID_FORMAT.format(device_id.replace("-", ""))
        # this attribute can be obtained from the Xively API
        self._current_temperature = None

        # these attributes cannot be obtained and have to be assumed (=start defaults)
        self._target_temperature = 65
        self._current_operation = HVAC_MODE_OFF
        self._current_fan_mode = FAN_LOW

        # fixed settings
        self._operation_list = [HVAC_MODE_OFF, HVAC_MODE_COOL, HVAC_MODE_FAN_ONLY]
        self.operations = self._operation_list
        self._fan_list = [FAN_LOW, FAN_MEDIUM, FAN_HIGH]
        self._unit_of_measurement = TEMP_FAHRENHEIT
        self._target_temperature_step = 1

    @property
    def unique_id(self):
        return self._device_id

    @property
    def should_poll(self):
        """Polling needed for room temperature reading."""
        return True

    def update(self):
        """Read the room temperature from Fridrich HVAC unit."""
        headers = {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer {0}'.format(self._jwt)
        }
        payload = { 'data': self._device_id }
        req = requests.post(FRIEDRICH_TEMP_TIMESERIE, headers=headers, json=payload)

        if req.status_code != requests.codes.ok:
            _LOGGER.error("Error obtaining Fridrich HVAC unit ambiant temperature reading, code %d", req.status_code)
            return

        temp_points = reversed(req.json()['result']['result'])
        self._current_temperature = next(filter(lambda p: p['category'] == 'amb', temp_points))['numericValue']

    def _command(self, target_temperature=None, operation_mode=None, fan_mode=None):
        """Device publish command method which simultaneously update all commands' state"""
        COMMAND_OPERATION = {
            HVAC_MODE_OFF: 5,
            HVAC_MODE_COOL: 1,
            HVAC_MODE_FAN_ONLY: 3,
        }

        COMMAND_FAN_MODE = {
            FAN_LOW: 0,
            FAN_MEDIUM: 1,
            FAN_HIGH: 2,
        }

        target_temperature = target_temperature or self._target_temperature
        operation_mode = operation_mode or self._current_operation
        fan_mode = fan_mode or self._current_fan_mode

        headers = {
            'Authorization': 'Bearer ' + self._jwt
        }
        payload = {
            'data': {
                'data': json.dumps({
                    'csp': target_temperature,
                    'hsp': 60, # heat set point
                    'asp': 70, # auto mode set point
                    'mode': COMMAND_OPERATION[operation_mode],
                    'fanMode': 0, # 0=auto operation, 1=continuous operation
                    'fanSpeed': COMMAND_FAN_MODE[fan_mode], # the device also has an auto speed in Cool mode, unknown code
                }),
                'device_ids': [self._device_id],
                'queue': 'command'
            }
        }

        req = requests.post(FRIEDRICH_PUBLISH_COMMAND, headers=headers, json=payload)

        if req.status_code != requests.codes.ok:
            _LOGGER.error("Error publishing Fridrich HVAC unit command, code %d", req.status_code)
            return None

        return True

    @property
    def assumed_state(self):
        return True # we are assuming command state attributes at all times as we do not get updates
                    # from the device except for ambiant room temperature

    @property
    def supported_features(self):
        """Return the list of supported features."""
        return SUPPORT_TARGET_TEMPERATURE | SUPPORT_FAN_MODE

    @property
    def name(self):
        """Return the name of the thermostat."""
        return self._name

    @property
    def temperature_unit(self):
        """Return the unit of measurement."""
        return self._unit_of_measurement

    @property
    def current_temperature(self):
        """Return the current temperature."""
        return self._current_temperature

    @property
    def target_temperature(self):
        """Return the temperature we try to reach."""
        return self._target_temperature

    @property
    def hvac_mode(self):
        """Return the current mode of the HVAC."""
        return self._current_operation

    @property
    def fan_mode(self):
        """Return the fan setting."""
        return self._current_fan_mode

    @property
    def hvac_modes(self):
        """List of available operation modes."""
        return self._operation_list

    @property
    def fan_modes(self):
        """Return the list of available fan modes."""
        return self._fan_list

    @property
    def target_temperature_step(self):
        """Return the supported step of target temperature."""
        return self._target_temperature_step

    def set_hvac_mode(self, operation_mode):
        """Set Fridrich HVAC operation mode (off, fan only, cool)."""
        if self._command(operation_mode=operation_mode) is None:
            _LOGGER.error("Could not set Friedrich HVAC unit operation mode")
            return

        self._current_operation = operation_mode

    def turn_on(self):
        self.set_hvac_mode(HVAC_MODE_COOL)

    def turn_off(self):
        self.set_hvac_mode(HVAC_MODE_OFF)

    def set_fan_mode(self, fan_mode):
        """Set Friedrich HVAC fan speed mode (auto, low, medium, high, max)."""
        if self._command(fan_mode=fan_mode) is None:
            _LOGGER.error("Could not set Friedrich HVAC unit fan mode")
            return

        self._current_fan_mode = fan_mode

    def set_temperature(self, **kwargs):
        """Set Friedrich HVAC cooling target temperature."""
        temperature = kwargs.get(ATTR_TEMPERATURE)

        if self._command(target_temperature=temperature) is None:
            _LOGGER.error("Could not set Friedrich HVAC target temperature")
            return

        self._target_temperature = temperature

manifest.json

{
  "codeowners": ["@tarik", "@alexhulbert"],
  "domain": "friedrich",
  "name": "Friedrich",
  "version": "1.0.0",
  "config_flow": true,
  "documentation": "",
  "issue_tracker": "",
  "requirements": [],
  "iot_class": "cloud_polling",
  "ssdp": [],
  "zeroconf": [],
  "homekit": {},
  "dependencies": []
}