diff --git a/Makefile b/Makefile index 4f989da..b9d00b7 100644 --- a/Makefile +++ b/Makefile @@ -22,15 +22,21 @@ deploy: haaska.zip --zip-file fileb://$< TEST_PAYLOAD:=' \ -{ \ - "header": { \ - "payloadVersion": "2", \ - "namespace": "Alexa.ConnectedHome.System", \ - "name": "HealthCheckRequest" \ - }, \ - "payload": { \ - "accessToken": "..." \ - } \ +{ \ + "directive": { \ + "header": { \ + "namespace": "Alexa.Discovery", \ + "name": "Discover", \ + "payloadVersion": "3", \ + "messageId": "1bd5d003-31b9-476f-ad03-71d471922820" \ + }, \ + "payload": { \ + "scope": { \ + "type": "BearerToken", \ + "token": "access-token-from-skill" \ + } \ + } \ + } \ }' .PHONY: test @@ -38,7 +44,7 @@ test: @aws lambda invoke \ --function-name $(FUNCTION_NAME) \ --payload ${TEST_PAYLOAD} \ - /dev/fd/3 3>&1 >/dev/null | jq -e '., .payload.isHealthy' + /dev/fd/3 3>&1 >/dev/null DISCOVERY_PAYLOAD:=' \ { \ diff --git a/haaska.py b/haaska.py index f1e3f43..5fc7ca0 100644 --- a/haaska.py +++ b/haaska.py @@ -24,20 +24,11 @@ import os import json import logging -import operator import requests -import colorsys -from hashlib import sha1 -from uuid import uuid4 logger = logging.getLogger() -LIGHT_SUPPORT_COLOR_TEMP = 2 -LIGHT_SUPPORT_RGB_COLOR = 16 -LIGHT_SUPPORT_XY_COLOR = 64 - - class HomeAssistant(object): def __init__(self, config): self.config = config @@ -75,540 +66,6 @@ def post(self, relurl, d, wait=False): return r -class ConnectedHomeCall(object): - def __init__(self, namespace, name, ha, payload): - self.namespace = namespace - self.name = name - self.response_name = self.name.replace('Request', 'Response') - self.ha = ha - self.payload = payload - self.entity = None - if 'appliance' in self.payload: - details = payload['appliance']['additionalApplianceDetails'] - self.entity = mk_entity(ha, details['entity_id']) - - class ConnectedHomeException(Exception): - def __init__(self, name="DriverInternalError", payload={}): - self.error_name = name - self.payload = payload - - class ValueOutOfRangeError(ConnectedHomeException): - def __init__(self, minValue, maxValue): - self.error_name = 'ValueOutOfRangeError' - self.payload = {'minimumValue': minValue, 'maximumValue': maxValue} - - def invoke(self, name): - logger.debug('invoking %s %s', self.namespace, name) - r = {} - try: - payload = operator.attrgetter(name)(self)() - if payload: - r['payload'] = payload - else: - r['payload'] = {'success': True} - logger.debug('response payload: %s', str(r['payload'])) - except ConnectedHomeCall.ConnectedHomeException as e: - logger.exception('handler failed: %s, %s', e.error_name, e.payload) - self.response_name = e.error_name - r['payload'] = e.payload - except Exception: - logger.exception('handler failed unexpectedly') - self.response_name = 'DriverInternalError' - r['payload'] = {} - - r['header'] = {'namespace': self.namespace, - 'messageId': str(uuid4()), - 'name': self.response_name, - 'payloadVersion': '2'} - return r - - -class Alexa(object): - class ConnectedHome(object): - class System(ConnectedHomeCall): - def HealthCheckRequest(self): - try: - self.ha.get('states') - return {'isHealthy': True} - except Exception as e: - logger.exception('HealthCheckRequest failed') - return {'isHealthy': False, 'description': str(e)} - - class Discovery(ConnectedHomeCall): - def DiscoverAppliancesRequest(self): - try: - return {'discoveredAppliances': - discover_appliances(self.ha)} - except Exception: - logger.exception('DiscoverAppliancesRequest failed') - # v2 documentation is unclear as to what should be returned - # here if discovery fails, so in the mean-time, just return - # 0 devices and log the error - return {'discoveredAppliances': {}} - - class Control(ConnectedHomeCall): - def __init__(self, namespace, name, ha, payload): - super(Alexa.ConnectedHome.Control, self).__init__( - namespace, name, ha, payload) - self.response_name = name.replace('Request', 'Confirmation') - - def TurnOnRequest(self): - self.entity.turn_on() - - def TurnOffRequest(self): - self.entity.turn_off() - - def SetPercentageRequest(self): - percentage = self.payload['percentageState']['value'] - self.entity.set_percentage(percentage) - - def handle_percentage_adj(self, deltaValue): - current = self.entity.get_percentage() - new = current + deltaValue - - # So this looks weird, but the relative adjustments seem to - # always be +/- 25%, which means depending on the current - # brightness we could over-/undershoot the acceptable range. - # Instead, if we're not currently saturated, clamp the desired - # brightness to the allowed brightness. - if current != 100 and current != 0: - if new < 0: - new = 0 - elif new > 100: - new = 100 - - if new > 100 or new < 0: - raise ConnectedHomeCall.ValueOutOfRangeError(0, 100) - - self.entity.set_percentage(new) - - def IncrementPercentageRequest(self): - deltaValue = self.payload['deltaPercentage']['value'] - return self.handle_percentage_adj(deltaValue) - - def DecrementPercentageRequest(self): - deltaValue = -self.payload['deltaPercentage']['value'] - return self.handle_percentage_adj(deltaValue) - - def handle_color_temperature_adj(self, op): - current = self.entity.get_color_temperature() - new = op(current, 500) - self.entity.set_color_temperature(new) - return {'achievedState': {'colorTemperature': {'value': new}}} - - def IncrementColorTemperatureRequest(self): - return self.handle_color_temperature_adj(operator.add) - - def DecrementColorTemperatureRequest(self): - return self.handle_color_temperature_adj(operator.sub) - - def SetColorTemperatureRequest(self): - temp = self.payload['colorTemperature']['value'] - self.entity.set_color_temperature(temp) - return {'achievedState': {'colorTemperature': {'value': temp}}} - - def handle_temperature_adj(self, op=None): - state = self.ha.get('states/' + self.entity.entity_id) - unit = state['attributes']['unit_of_measurement'] - min_temp = convert_temp(state['attributes']['min_temp'], unit) - max_temp = convert_temp(state['attributes']['max_temp'], unit) - - temperature, mode = self.entity.get_temperature(state) - - if op is not None and 'deltaTemperature' in self.payload: - new = op(temperature, - float(self.payload['deltaTemperature']['value'])) - # Clamp the allowed temperature for relative adjustments - if temperature != max_temp and temperature != min_temp: - if new < min_temp: - new = min_temp - elif new > max_temp: - new = max_temp - else: - new = float(self.payload['targetTemperature']['value']) - - if new > max_temp or new < min_temp: - raise ConnectedHomeCall.ValueOutOfRangeError(min_temp, - max_temp) - - # Only 3 allowed values for mode in this response - if mode not in ['AUTO', 'COOL', 'HEAT']: - current = self.entity.get_current_temperature(state) - mode = 'COOL' if current >= new else 'HEAT' - - self.entity.set_temperature(new, mode.lower(), state) - - return {'targetTemperature': {'value': new}, - 'temperatureMode': {'value': mode}, - 'previousState': { - 'targetTemperature': {'value': temperature}, - 'mode': {'value': mode}}} - - def SetTargetTemperatureRequest(self): - return self.handle_temperature_adj() - - def IncrementTargetTemperatureRequest(self): - return self.handle_temperature_adj(operator.add) - - def DecrementTargetTemperatureRequest(self): - return self.handle_temperature_adj(operator.sub) - - def SetLockStateRequest(self): - self.entity.set_lock_state(self.payload["lockState"]) - return {'lockState': self.payload["lockState"]} - - def SetColorRequest(self): - self.entity.set_color(self.payload['color']['hue'], - self.payload['color']['saturation'], - self.payload['color']['brightness']) - return {'achievedState': {'color': self.payload['color']}} - - class Query(ConnectedHomeCall): - def __init__(self, namespace, name, ha, payload): - super(Alexa.ConnectedHome.Query, self).__init__( - namespace, name, ha, payload) - - def GetTemperatureReadingRequest(self): - temperature = self.entity.get_current_temperature() - return {'temperatureReading': {'value': temperature}} - - def GetTargetTemperatureRequest(self): - temperature, mode = self.entity.get_temperature() - payload = {'targetTemperature': {'value': temperature}, - 'temperatureMode': {'value': mode}} - if mode not in ['AUTO', 'COOL', 'HEAT', 'OFF']: - payload['temperatureMode'] = { - 'value': 'CUSTOM', - 'friendlyName': mode.replace('_', ' ').title()} - return payload - - def GetLockStateRequest(self): - lock_state = self.entity.get_lock_state().upper() - return {'lockState': lock_state} - - -def invoke(namespace, name, ha, context): - class allowed(object): - Alexa = Alexa - make_class = operator.attrgetter(namespace) - obj = make_class(allowed)(namespace, name, ha, context) - return obj.invoke(name) - - -def discover_appliances(ha): - def entity_domain(x): - return x['entity_id'].split('.', 1)[0] - - def is_supported_entity(x): - return entity_domain(x) in ha.config.exposed_domains - - def is_exposed_entity(x): - attr = x['attributes'] - if 'haaska_hidden' in attr: - return not attr['haaska_hidden'] - elif 'hidden' in attr: - return not attr['hidden'] - else: - return ha.config.expose_by_default - - def mk_appliance(x): - features = 0 - if 'supported_features' in x['attributes']: - features = x['attributes']['supported_features'] - entity = mk_entity(ha, x['entity_id'], features) - o = {} - # this needs to be unique and has limitations on allowed characters: - o['applianceId'] = sha1(x['entity_id'].encode('utf-8')).hexdigest() - o['manufacturerName'] = 'Unknown' - o['modelName'] = 'Unknown' - o['version'] = 'Unknown' - if 'haaska_name' in x['attributes']: - o['friendlyName'] = x['attributes']['haaska_name'] - else: - o['friendlyName'] = x['attributes']['friendly_name'] - suffix = ha.config.entity_suffixes[entity_domain(x)] - if suffix != '': - o['friendlyName'] += ' ' + suffix - if 'haaska_desc' in x['attributes']: - o['friendlyDescription'] = x['attributes']['haaska_desc'] - else: - o['friendlyDescription'] = 'Home Assistant ' + \ - entity_domain(x).replace('_', ' ').title() - o['isReachable'] = True - o['actions'] = entity.get_actions() - o['additionalApplianceDetails'] = {'entity_id': x['entity_id'], - 'supported_features': features} - return o - - states = ha.get('states') - return [mk_appliance(x) for x in states if is_supported_entity(x) and - is_exposed_entity(x)] - - -def supported_features(payload): - try: - details = 'additionalApplianceDetails' - return payload['appliance'][details]['supported_features'] - except: - return 0 - - -def convert_temp(temp, from_unit=u'°C', to_unit=u'°C'): - if temp is None or from_unit == to_unit: - return temp - if from_unit == u'°C': - return temp * 1.8 + 32 - else: - return (temp - 32) / 1.8 - - -class Entity(object): - def __init__(self, ha, entity_id, supported_features): - self.ha = ha - self.entity_id = entity_id - self.supported_features = supported_features - self.entity_domain = self.entity_id.split('.', 1)[0] - - def _call_service(self, service, data={}): - data['entity_id'] = self.entity_id - self.ha.post('services/' + service, data) - - def get_actions(self): - actions = [] - - if hasattr(self, 'turn_on'): - actions.append('turnOn') - if hasattr(self, 'turn_off'): - actions.append('turnOff') - - if hasattr(self, 'set_percentage'): - actions.append('setPercentage') - if hasattr(self, 'get_percentage'): - actions.append('incrementPercentage') - actions.append('decrementPercentage') - - if hasattr(self, 'get_current_temperature'): - actions.append('getTemperatureReading') - if hasattr(self, 'set_temperature'): - actions.append('setTargetTemperature') - if hasattr(self, 'get_temperature'): - actions.append('getTargetTemperature') - actions.append('incrementTargetTemperature') - actions.append('decrementTargetTemperature') - - if hasattr(self, 'get_lock_state'): - actions.append('getLockState') - if hasattr(self, 'set_lock_state'): - actions.append('setLockState') - - if self.entity_domain == "light": - if self.supported_features & LIGHT_SUPPORT_RGB_COLOR: - actions.append('setColor') - if self.supported_features & LIGHT_SUPPORT_COLOR_TEMP: - actions.append('setColorTemperature') - actions.append('incrementColorTemperature') - actions.append('decrementColorTemperature') - - return actions - - -class ToggleEntity(Entity): - def turn_on(self): - self._call_service('homeassistant/turn_on') - - def turn_off(self): - self._call_service('homeassistant/turn_off') - - -class InputSliderEntity(Entity): - def get_percentage(self): - state = self.ha.get('states/' + self.entity_id) - value = float(state['state']) - minimum = state['attributes']['min'] - maximum = state['attributes']['max'] - adjusted = value - minimum - - return (adjusted * 100.0 / (maximum - minimum)) - - def set_percentage(self, val): - state = self.ha.get('states/' + self.entity_id) - minimum = state['attributes']['min'] - maximum = state['attributes']['max'] - step = state['attributes']['step'] - scaled = val * (maximum - minimum) / 100.0 - rounded = step * round(scaled / step) - adjusted = rounded + minimum - - self._call_service('input_slider/select_value', {'value': adjusted}) - - -class GarageDoorEntity(ToggleEntity): - def turn_on(self): - self._call_service('garage_door/open') - - def turn_off(self): - self._call_service('garage_door/close') - - -class CoverEntity(ToggleEntity): - def turn_on(self): - self._call_service('cover/open_cover') - - def turn_off(self): - self._call_service('cover/close_cover') - - -class LockEntity(Entity): - def set_lock_state(self, state): - if state == "LOCKED": - self._call_service('lock/lock') - elif state == "UNLOCKED": - self._call_service('lock/unlock') - - def get_lock_state(self): - state = self.ha.get('states/' + self.entity_id) - return state['state'] - - -class ScriptEntity(ToggleEntity): - def turn_off(self): - self.turn_on() - - -class SceneEntity(ToggleEntity): - def turn_off(self): - self.turn_on() - - -class LightEntity(ToggleEntity): - def get_percentage(self): - state = self.ha.get('states/' + self.entity_id) - current_brightness = state['attributes']['brightness'] - return (current_brightness / 255.0) * 100.0 - - def set_percentage(self, val): - brightness = (val / 100.0) * 255.0 - self._call_service('light/turn_on', {'brightness': brightness}) - - def get_color_temperature(self): - state = self.ha.get('states/' + self.entity_id) - current_temperature = state['attributes']['color_temp'] - return (1000000 / current_temperature) - - def set_color(self, hue, saturation, brightness): - rgb = [int(round(i * 255)) for i in colorsys.hsv_to_rgb(hue / 360.0, - saturation, - brightness)] - self._call_service('light/turn_on', {'rgb_color': rgb}) - - def set_color_temperature(self, val): - self._call_service('light/turn_on', - {'color_temp': (1000000 / val)}) - - -class MediaPlayerEntity(ToggleEntity): - def get_percentage(self): - state = self.ha.get('states/' + self.entity_id) - vol = state['attributes']['volume_level'] - return vol * 100.0 - - def set_percentage(self, val): - vol = val / 100.0 - self._call_service('media_player/volume_set', {'volume_level': vol}) - - -class ClimateEntity(Entity): - def turn_on(self): - state = self.ha.get('states/' + self.entity_id) - current = self.get_current_temperature(state) - temperature, mode = self.get_temperature(state) - if temperature is None: - mode = 'auto' - else: - mode = 'cool' if current >= temperature else 'heat' - self._call_service('climate/set_operation_mode', - {'operation_mode': mode}) - - def turn_off(self): - self._call_service('climate/set_operation_mode', - {'operation_mode': 'off'}) - - def get_current_temperature(self, state=None): - if not state: - state = self.ha.get('states/' + self.entity_id) - return convert_temp( - state['attributes']['current_temperature'], - state['attributes']['unit_of_measurement']) - - def get_temperature(self, state=None): - if not state: - state = self.ha.get('states/' + self.entity_id) - temperature = convert_temp( - state['attributes']['temperature'], - state['attributes']['unit_of_measurement']) - mode = state['state'].replace('idle', 'off').upper() - return (temperature, mode) - - def set_temperature(self, val, mode=None, state=None): - if not state: - state = self.ha.get('states/' + self.entity_id) - temperature = convert_temp( - val, - to_unit=state['attributes']['unit_of_measurement']) - data = {'temperature': temperature} - if mode: - data['operation_mode'] = mode - self._call_service('climate/set_temperature', data) - - -class FanEntity(ToggleEntity): - def get_percentage(self): - state = self.ha.get('states/' + self.entity_id) - speed = state['attributes']['speed'] - if speed == "off": - return 0 - elif speed == "low": - return 33 - elif speed == "medium": - return 66 - elif speed == "high": - return 100 - - def set_percentage(self, val): - speed = "off" - if val <= 33: - speed = "low" - elif val <= 66: - speed = "medium" - elif val <= 100: - speed = "high" - self._call_service('fan/set_speed', {'speed': speed}) - - -DOMAINS = { - 'garage_door': GarageDoorEntity, - 'group': ToggleEntity, - 'input_boolean': ToggleEntity, - 'input_slider': InputSliderEntity, - 'switch': ToggleEntity, - 'fan': FanEntity, - 'cover': CoverEntity, - 'lock': LockEntity, - 'script': ScriptEntity, - 'scene': SceneEntity, - 'light': LightEntity, - 'media_player': MediaPlayerEntity, - 'climate': ClimateEntity, - 'alert': ToggleEntity, - 'automation': ToggleEntity -} - - -def mk_entity(ha, entity_id, supported_features=0): - entity_domain = entity_id.split('.', 1)[0] - return DOMAINS[entity_domain](ha, entity_id, supported_features) - - class Configuration(object): def __init__(self, filename=None, optsDict=None): self._json = {} @@ -624,17 +81,7 @@ def __init__(self, filename=None, optsDict=None): default='http://localhost:8123/api') opts['ssl_verify'] = self.get(['ssl_verify', 'ha_cert'], default=True) opts['password'] = self.get(['password', 'ha_passwd'], default='') - opts['exposed_domains'] = \ - sorted(self.get(['exposed_domains', 'ha_allowed_entities'], - default=DOMAINS.keys())) - default_entity_suffixes = {'group': 'Group', 'scene': 'Scene'} - opts['entity_suffixes'] = {domain: '' for domain in DOMAINS.keys()} - opts['entity_suffixes'].update(self.get(['entity_suffixes'], - default=default_entity_suffixes)) - - opts['expose_by_default'] = self.get(['expose_by_default'], - default=True) opts['debug'] = self.get(['debug'], default=False) self.opts = opts @@ -657,12 +104,4 @@ def event_handler(event, context): logger.setLevel(logging.DEBUG) ha = HomeAssistant(config) - name = event['header']['name'] - payload = event['payload'] - - logger.debug('calling event handler for %s, payload: %s', name, - str({k: v for k, v in payload.items() - if k != u'accessToken'})) - - return invoke(event['header']['namespace'], event['header']['name'], - ha, payload) + return ha.post('alexa/smart_home', event, wait=True).json() diff --git a/test/config.json b/test/config.json deleted file mode 100644 index d2f7fff..0000000 --- a/test/config.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "url": "http://localhost:8123/api", - "password": "", - "debug": true -} diff --git a/test/test.py b/test/test.py deleted file mode 100644 index cc3f7e2..0000000 --- a/test/test.py +++ /dev/null @@ -1,546 +0,0 @@ -#!/usr/bin/env python2.7 -# coding: utf-8 - -# Basic tests meant to be run against a demo instance of Home-Assistant -# $ hass --demo - -import sys -import unittest -from nose.tools import assert_raises -sys.path.insert(0, '..') -import haaska # noqa: E402 - - -def discover_appliance_request(): - return { - "header": { - "messageId": "6d6d6e14-8aee-473e-8c24-0d31ff9c17a2", - "name": "DiscoverAppliancesRequest", - "namespace": "Alexa.ConnectedHome.Discovery", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "" - } - } - - -discovery = haaska.event_handler(discover_appliance_request(), None) -appliances = discovery['payload']['discoveredAppliances'] - - -class DiscoveryTests(unittest.TestCase): - def test_discovery_header(self): - self.assertEqual(discovery['header']['namespace'], - 'Alexa.ConnectedHome.Discovery') - self.assertEqual(discovery['header']['name'], - 'DiscoverAppliancesResponse') - - def test_reachable(self): - for ap in appliances: - self.assertTrue(ap['isReachable']) - - -def find_appliance(entity_id): - for ap in appliances: - if ap['additionalApplianceDetails']['entity_id'] == entity_id: - return ap - return None - - -def get_state(ap): - entity_id = ap['additionalApplianceDetails']['entity_id'] - ha = haaska.HomeAssistant(haaska.Configuration('config.json')) - return ha.get('states/' + entity_id) - - -def entity_domain(ap): - entity_id = ap['additionalApplianceDetails']['entity_id'] - return entity_id.split('.')[0] - - -def to_appliance(ap): - return {"additionalApplianceDetails": ap['additionalApplianceDetails'], - "applianceId": ap['applianceId']} - - -class UnexpectedResponseException(Exception): - pass - - -def turn_off(ap): - req = { - "header": { - "messageId": "01ebf625-0b89-4c4d-b3aa-32340e894688", - "name": "TurnOffRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "", - "appliance": to_appliance(ap) - } - } - resp = haaska.event_handler(req, None) - if resp['header']['name'] != 'TurnOffConfirmation': - raise UnexpectedResponseException - return resp - - -def turn_on(ap): - req = { - "header": { - "messageId": "01ebf625-0b89-4c4d-b3aa-32340e894688", - "name": "TurnOnRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "", - "appliance": to_appliance(ap) - } - } - resp = haaska.event_handler(req, None) - if resp['header']['name'] != 'TurnOnConfirmation': - raise UnexpectedResponseException - return resp - - -def assert_state_is(ap, state): - assert get_state(ap)['state'] == state - - -class OnOffTests(unittest.TestCase): - def assertStateIs(self, ap, state): - self.assertEqual(get_state(ap)['state'], state) - - def test_switch_on_off_on(self): - sw = find_appliance(u'switch.decorative_lights') - turn_on(sw) - self.assertStateIs(sw, 'on') - turn_off(sw) - self.assertStateIs(sw, 'off') - turn_on(sw) - self.assertStateIs(sw, 'on') - - def test_switch_turn_on_twice(self): - sw = find_appliance(u'switch.decorative_lights') - turn_on(sw) - self.assertStateIs(sw, 'on') - turn_on(sw) - self.assertStateIs(sw, 'on') - - def test_switch_turn_off_twice(self): - sw = find_appliance(u'switch.decorative_lights') - turn_off(sw) - self.assertStateIs(sw, 'off') - turn_off(sw) - self.assertStateIs(sw, 'off') - - def test_light_on_off_on(self): - light = find_appliance(u'light.bed_light') - turn_on(light) - self.assertStateIs(light, 'on') - turn_off(light) - self.assertStateIs(light, 'off') - turn_on(light) - self.assertStateIs(light, 'on') - - def test_input_boolean_on_off_on(self): - ib = find_appliance('input_boolean.notify') - turn_on(ib) - self.assertStateIs(ib, 'on') - turn_off(ib) - self.assertStateIs(ib, 'off') - turn_on(ib) - self.assertStateIs(ib, 'on') - - def test_media_player_on_off_on(self): - player = find_appliance('media_player.bedroom') - self.assertStateIs(player, 'playing') - turn_off(player) - self.assertStateIs(player, 'off') - turn_on(player) - self.assertStateIs(player, 'playing') - - def test_climate_on_off_on(self): - climate = find_appliance('climate.ecobee') - self.assertIn(get_state(climate)['state'], ['auto', 'cool', 'heat']) - turn_off(climate) - self.assertStateIs(climate, 'off') - turn_on(climate) - self.assertIn(get_state(climate)['state'], ['auto', 'cool', 'heat']) - - def test_lock_off_on_fails(self): - lock = find_appliance('lock.kitchen_door') - assert_raises(UnexpectedResponseException, turn_off, lock) - assert_raises(UnexpectedResponseException, turn_on, lock) - - def test_cover_on_off_on(self): - cover = find_appliance('cover.garage_door') - turn_on(cover) - self.assertStateIs(cover, 'open') - turn_off(cover) - self.assertStateIs(cover, 'closed') - turn_on(cover) - self.assertStateIs(cover, 'open') - - def test_turn_off(self): - for ap in appliances: - if 'turnOff' not in ap['actions']: - continue - resp = turn_off(ap) - self.assertEqual(resp['header']['name'], 'TurnOffConfirmation') - self.assertTrue(resp['payload']['success']) - if entity_domain(ap) == 'light' or \ - entity_domain(ap) == 'input_boolean': - self.assertStateIs(ap, 'off') - elif entity_domain(ap) == 'media_player': - self.assertStateIs(ap, 'off') - elif entity_domain(ap) == 'climate': - self.assertStateIs(ap, 'off') - elif entity_domain(ap) == 'garage_door': - self.assertStateIs(ap, 'closed') - elif entity_domain(ap) == 'lock': - self.assertStateIs(ap, 'unlocked') - - def test_turn_on(self): - for ap in appliances: - if 'turnOn' not in ap['actions']: - continue - resp = turn_on(ap) - self.assertEqual(resp['header']['name'], 'TurnOnConfirmation') - self.assertTrue(resp['payload']['success']) - if entity_domain(ap) == 'light' or \ - entity_domain(ap) == 'input_boolean': - self.assertStateIs(ap, 'on') - elif entity_domain(ap) == 'media_player': - self.assertStateIs(ap, 'playing') - elif entity_domain(ap) == 'climate': - self.assertIn(get_state(ap)['state'], ['auto', 'cool', 'heat']) - elif entity_domain(ap) == 'garage_door': - self.assertStateIs(ap, 'open') - elif entity_domain(ap) == 'lock': - self.assertStateIs(ap, 'locked') - - -def set_lock_state(ap, locked): - req = { - "header": { - "messageId": "01ebf625-0b89-4c4d-b3aa-32340e894688", - "name": "SetLockStateRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - "lockState": "LOCKED" if locked else "UNLOCKED" - } - } - - resp = haaska.event_handler(req, None) - if resp['header']['name'] != 'SetLockStateConfirmation': - raise UnexpectedResponseException - return resp - - -def get_lock_state(ap): - req = { - "header": { - "messageId": "01ebf625-0b89-4c4d-b3aa-32340e894688", - "name": "GetLockStateRequest", - "namespace": "Alexa.ConnectedHome.Query", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - } - } - - resp = haaska.event_handler(req, None) - if resp['header']['name'] != 'GetLockStateResponse': - raise UnexpectedResponseException - return resp - - -class LockTests(unittest.TestCase): - # TODO: don't copy this all over - def assertStateIs(self, ap, state): - self.assertEqual(get_state(ap)['state'], state) - - def test_lock_unlock(self): - lock = find_appliance('lock.kitchen_door') - set_lock_state(lock, True) - self.assertStateIs(lock, 'locked') - set_lock_state(lock, False) - self.assertStateIs(lock, 'unlocked') - set_lock_state(lock, True) - self.assertStateIs(lock, 'locked') - - def test_get_lock_state(self): - lock = find_appliance('lock.kitchen_door') - - set_lock_state(lock, True) - self.assertStateIs(lock, 'locked') - r = get_lock_state(lock) - self.assertEqual(r['payload']['lockState'], 'LOCKED') - - set_lock_state(lock, False) - self.assertStateIs(lock, 'unlocked') - r = get_lock_state(lock) - self.assertEqual(r['payload']['lockState'], 'UNLOCKED') - - -def set_percentage(ap, percentage): - req = { - "header": { - "messageId": "95872301-4ff6-4146-b3a4-ae84c760c13e", - "name": "SetPercentageRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "", - "appliance": to_appliance(ap), - "percentageState": { - "value": percentage - } - } - } - return haaska.event_handler(req, None) - - -def get_brightness(ap): - b = (get_state(ap)['attributes']['brightness'] * 100) / 255.0 - b = int(b + 0.5) - return b - - -class PercentageTests(unittest.TestCase): - def test_set_light_percentage(self): - for ap in appliances: - if 'setPercentage' not in ap['actions']: - continue - if entity_domain(ap) != 'light': - continue - turn_on(ap) - if 'brightness' not in get_state(ap)['attributes']: - continue - for v in [10, 50, 75, 100]: - resp = set_percentage(ap, v) - self.assertEqual(resp['header']['name'], - 'SetPercentageConfirmation') - self.assertEqual(get_state(ap)['state'], 'on') - self.assertEqual(get_brightness(ap), v) - - def test_set_volume(self): - for ap in appliances: - if 'setPercentage' not in ap['actions']: - continue - if entity_domain(ap) != 'media_player': - continue - features = get_state(ap)['attributes']['supported_features'] - if int(features) & 4 == 0: - continue - for v in [10, 50, 75, 100]: - resp = set_percentage(ap, v) - self.assertEqual(resp['header']['name'], - 'SetPercentageConfirmation') - level = get_state(ap)['attributes']['volume_level'] * 100.0 - self.assertEqual(level, v) - - -def convert_temp(temp, from_unit=u'°C', to_unit=u'°C'): - if temp is None or from_unit == to_unit: - return temp - if from_unit == u'°C': - return temp * 1.8 + 32 - else: - return (temp - 32) / 1.8 - - -def get_temperature_reading(ap): - req = { - "header": { - "messageId": "01ebf625-0b89-4c4d-b3aa-32340e894689", - "name": "GetTemperatureReadingRequest", - "namespace": "Alexa.ConnectedHome.Query", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - } - } - - resp = haaska.event_handler(req, None) - if resp['header']['name'] != 'GetTemperatureReadingResponse': - raise UnexpectedResponseException - return resp - - -def set_target_temperature(ap, temperature): - req = { - "header": { - "messageId": "95872301-4ff6-4146-b3a4-ae84c760c13f", - "name": "SetTargetTemperatureRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - "targetTemperature": { - "value": temperature - } - } - } - return haaska.event_handler(req, None) - - -def lower_target_temperature(ap, temperature): - req = { - "header": { - "messageId": "95872301-4ff6-4146-b3a4-ae84c760c140", - "name": "DecrementTargetTemperatureRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - "deltaTemperature": { - "value": temperature - } - } - } - return haaska.event_handler(req, None) - - -class ClimateTests(unittest.TestCase): - def test_get_current_temperature(self): - climate = find_appliance('climate.heatpump') - r = get_temperature_reading(climate) - self.assertEqual(r['payload']['temperatureReading']['value'], 25) - climate = find_appliance('climate.hvac') - r = get_temperature_reading(climate) - self.assertEqual(r['payload']['temperatureReading']['value'], 22) - climate = find_appliance('climate.ecobee') - r = get_temperature_reading(climate) - self.assertEqual(r['payload']['temperatureReading']['value'], 23) - - def test_set_temperature(self): - for ap in appliances: - if 'setTargetTemperature' not in ap['actions']: - continue - if entity_domain(ap) != 'climate': - continue - turn_on(ap) - if 'temperature' not in get_state(ap)['attributes']: - continue - for t in [10, 15, 20, 25]: - r = set_target_temperature(ap, t) - self.assertEqual(r['header']['name'], - 'SetTargetTemperatureConfirmation') - self.assertEqual(t, convert_temp( - get_state(ap)['attributes']['temperature'], - get_state(ap)['attributes']['unit_of_measurement'])) - - def test_lower_temperature(self): - for ap in appliances: - if 'decrementTargetTemperature' not in ap['actions']: - continue - if entity_domain(ap) != 'climate': - continue - turn_on(ap) - if 'temperature' not in get_state(ap)['attributes']: - continue - t = 20 - set_target_temperature(ap, t) - self.assertEqual(t, convert_temp( - get_state(ap)['attributes']['temperature'], - get_state(ap)['attributes']['unit_of_measurement'])) - - r = lower_target_temperature(ap, 5) - self.assertEqual(r['header']['name'], - 'DecrementTargetTemperatureConfirmation') - t = 15 - self.assertEqual(t, convert_temp( - get_state(ap)['attributes']['temperature'], - get_state(ap)['attributes']['unit_of_measurement'])) - - -def dim_light(ap, val): - req = { - "header": { - "messageId": "7048c18d-4141-4871-bf0e-da3e54dee3f7", - "name": "DecrementPercentageRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - "deltaPercentage": { - "value": val - } - } - } - return haaska.event_handler(req, None) - - -def brighten_light(ap, val): - req = { - "header": { - "messageId": "7048c18d-4141-4871-bf0e-da3e54dee3f7", - "name": "IncrementPercentageRequest", - "namespace": "Alexa.ConnectedHome.Control", - "payloadVersion": "2" - }, - "payload": { - "accessToken": "[OAuth Token here]", - "appliance": to_appliance(ap), - "deltaPercentage": { - "value": val - } - } - } - return haaska.event_handler(req, None) - - -class LightTests(unittest.TestCase): - def test_brighten_light(self): - for ap in appliances: - if 'incrementPercentage' not in ap['actions']: - continue - if entity_domain(ap) != 'light': - continue - turn_on(ap) - if 'brightness' not in get_state(ap)['attributes']: - continue - set_percentage(ap, 50) - self.assertEqual(get_brightness(ap), 50) - resp = brighten_light(ap, 20) - self.assertEqual(resp['header']['name'], - 'IncrementPercentageConfirmation') - self.assertEqual(get_state(ap)['state'], 'on') - self.assertEqual(get_brightness(ap), 70) - - def test_dim_light(self): - for ap in appliances: - if 'decrementPercentage' not in ap['actions']: - continue - if entity_domain(ap) != 'light': - continue - turn_on(ap) - if 'brightness' not in get_state(ap)['attributes']: - continue - set_percentage(ap, 50) - self.assertEqual(get_brightness(ap), 50) - resp = dim_light(ap, 20) - self.assertEqual(resp['header']['name'], - 'DecrementPercentageConfirmation') - self.assertEqual(get_state(ap)['state'], 'on') - self.assertEqual(get_brightness(ap), 30)