Skip to content

Commit

Permalink
- Added custom exception messages
Browse files Browse the repository at this point in the history
- Cleaned up code
- Added unit tests
- Added a 10 second timeout for requests
- Renamed ET to Etree
  • Loading branch information
laetificat committed Jan 23, 2019
1 parent bbeeec2 commit b5b9546
Show file tree
Hide file tree
Showing 3 changed files with 203 additions and 33 deletions.
133 changes: 101 additions & 32 deletions haanna/haanna.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
Plugwise Anna HomeAssistant component
"""
import requests
import xml.etree.cElementTree as ET
import xml.etree.cElementTree as Etree

USERNAME = ''
PASSWORD = ''
Expand All @@ -19,53 +19,73 @@ def __init__(self, username, password, host):
self.set_credentials(username, password)
self.set_anna_endpoint('http://' + host)

if self.ping_anna_thermostat() is False:
@staticmethod
def ping_anna_thermostat():
"""Ping the thermostat to see if it's online"""
r = requests.get(ANNA_ENDPOINT + ANNA_PING_ENDPOINT, auth=(USERNAME, PASSWORD), timeout=10)

if r.status_code != 404:
raise ConnectionError("Could not connect to the gateway.")

def ping_anna_thermostat(self):
"""Ping the thermostat to see if it's online"""
r = requests.get(ANNA_ENDPOINT + ANNA_PING_ENDPOINT, auth=(USERNAME, PASSWORD))
return r.status_code == 404
return True

def get_domain_objects(self):
r = requests.get(ANNA_ENDPOINT + ANNA_DOMAIN_OBJECTS_ENDPOINT, auth=(USERNAME, PASSWORD))
@staticmethod
def get_domain_objects():
r = requests.get(ANNA_ENDPOINT + ANNA_DOMAIN_OBJECTS_ENDPOINT, auth=(USERNAME, PASSWORD), timeout=10)

if r.status_code != requests.codes.ok:
print("Could not get the domain objects.")
print(r.status_code)
print(r.text)
return
raise ConnectionError("Could not get the domain objects.")

return ET.fromstring(r.text)
return Etree.fromstring(r.text)

def get_presets(self, root):
"""Gets the presets from the thermostat"""
rule_id = self.get_rule_id_by_name(root, 'Thermostat presets')

if rule_id is None:
print("Could not find rule ID")
return
raise RuleIdNotFoundException("Could not find the rule id.")

presets = self.get_preset_dictionary(root, rule_id)
return presets

def set_preset(self, root, preset):
@staticmethod
def set_preset(root, preset):
"""Sets the given preset on the thermostat"""
location_id = root.find("appliance[type='thermostat']/location").attrib['id']

locations_root = ET.fromstring(requests.get(ANNA_ENDPOINT + ANNA_LOCATIONS_ENDPOINT, auth=(USERNAME, PASSWORD)).text)
locations_root = Etree.fromstring(
requests.get(
ANNA_ENDPOINT + ANNA_LOCATIONS_ENDPOINT,
auth=(USERNAME, PASSWORD),
timeout=10
).text
)

current_location = locations_root.find("location[@id='" + location_id + "']")
location_name = current_location.find("name").text
location_type = current_location.find("type").text

r = requests.put(
ANNA_ENDPOINT + ANNA_LOCATIONS_ENDPOINT + ';id=' + location_id,
auth=(USERNAME, PASSWORD),
data='<locations><location id="' + location_id + '"><name>' + location_name + '</name><type>' + location_type + '</type><preset>' + preset + '</preset></location></locations>',
headers={'Content-Type': 'text/xml'}
data='<locations>' +
'<location id="' + location_id + '">' +
'<name>' + location_name + '</name>' +
'<type>' + location_type + '</type>' +
'<preset>' + preset + '</preset>' +
'</location>' +
'</locations>',
headers={'Content-Type': 'text/xml'},
timeout=10
)

return r.status_code == 200
if r.status_code != requests.codes.ok:
raise CouldNotSetPresetException("Could not set the given preset: " + r.text)

return r.text

def get_current_preset(self, root):
@staticmethod
def get_current_preset(root):
"""Gets the current active preset"""
location_id = root.find("appliance[type='thermostat']/location").attrib['id']
return root.find("location[@id='" + location_id + "']/preset").text
Expand All @@ -91,52 +111,101 @@ def get_outdoor_temperature(self, root):

return float(measurement)

def set_temperature(self, root, temperature):
@staticmethod
def set_temperature(root, temperature):
"""Sends a set request to the temperature with the given temperature"""
location_id = root.find("appliance[type='thermostat']/location").attrib['id']
thermostat_functionality_id = root.find("location[@id='" + location_id + "']/actuator_functionalities/thermostat_functionality").attrib['id']

thermostat_functionality_id = root.find(
"location[@id='" + location_id + "']/actuator_functionalities/thermostat_functionality"
).attrib['id']

temperature = str(temperature)

r = requests.put(
ANNA_ENDPOINT + ANNA_LOCATIONS_ENDPOINT + ';id=' + location_id + '/thermostat;id=' + thermostat_functionality_id,
ANNA_ENDPOINT +
ANNA_LOCATIONS_ENDPOINT +
';id=' + location_id +
'/thermostat;id=' + thermostat_functionality_id,
auth=(USERNAME, PASSWORD),
data='<thermostat_functionality><setpoint>' + temperature + '</setpoint></thermostat_functionality>',
headers={'Content-Type': 'text/xml'}
headers={'Content-Type': 'text/xml'},
timeout=10
)

return r.status_code == 202
if r.status_code != requests.codes.ok:
CouldNotSetTemperatureException("Could not set the temperature." + r.text)

def set_credentials(self, username, password):
return r.text

@staticmethod
def set_credentials(username, password):
"""Sets the username and password variables"""
global USERNAME
global PASSWORD
USERNAME = username
PASSWORD = password

def set_anna_endpoint(self, endpoint):
@staticmethod
def get_credentials():
return {'username': USERNAME, 'password': PASSWORD}

@staticmethod
def set_anna_endpoint(endpoint):
"""Sets the endpoint where the Anna resides on the network"""
global ANNA_ENDPOINT
ANNA_ENDPOINT = endpoint

def get_point_log_id(self, root, log_type):
@staticmethod
def get_anna_endpoint():
return ANNA_ENDPOINT

@staticmethod
def get_point_log_id(root, log_type):
"""Gets the point log ID based on log type"""
return root.find("module/services/*[@log_type='" + log_type + "']/functionalities/point_log").attrib['id']

def get_measurement_from_point_log(self, root, point_log_id):
@staticmethod
def get_measurement_from_point_log(root, point_log_id):
"""Gets the measurement from a point log based on point log ID"""
return root.find("*/logs/point_log[@id='" + point_log_id + "']/period/measurement").text

def get_rule_id_by_name(self, root, rule_name):
@staticmethod
def get_rule_id_by_name(root, rule_name):
"""Gets the rule ID based on name"""
rules = root.findall("rule")
for rule in rules:
if rule.find("name").text == rule_name:
return rule.attrib['id']

def get_preset_dictionary(self, root, rule_id):
@staticmethod
def get_preset_dictionary(root, rule_id):
"""Gets the presets from a rule based on rule ID and returns a dictionary with all the key-value pairs"""
preset_dictionary = {}
directives = root.find("rule[@id='" + rule_id + "']/directives")
for directive in directives:
preset_dictionary[directive.attrib['preset']] = float(directive.find("then").attrib['setpoint'])
return preset_dictionary


class AnnaException(Exception):
def __init__(self, arg1, arg2=None):
"""Base exception for interaction with the Anna gateway"""
self.arg1 = arg1
self.arg2 = arg2
super(AnnaException, self).__init__(arg1)


class RuleIdNotFoundException(AnnaException):
"""Raise an exception for when the rule id is not found in the direct objects"""
pass


class CouldNotSetPresetException(AnnaException):
"""Raise an exception for when the preset can not be set"""
pass


class CouldNotSetTemperatureException(AnnaException):
"""Raise an exception for when the temperature could not be set"""
pass
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ def readme():

setup(
name='haanna',
version='0.5.1',
version='0.6.1',
description='Plugwise Anna API to use in conjunction with Home Assistant.',
long_description='Plugwise Anna API to use in conjunction with Home Assistant, but it can also be used without Home Assistant.',
keywords='HomeAssistant HA Home Assistant Anna Plugwise',
Expand Down
101 changes: 101 additions & 0 deletions tests/test_haanna.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
import time
import unittest
import xml.etree.cElementTree as Et

from haanna import Haanna


class TestHaannaMethods(unittest.TestCase):

def setUp(self):
self.haanna = Haanna('smile', 'short_id', 'ip_address')

def test_ping_anna_thermostat(self):
"""Ping the thermostst"""
self.assertTrue(self.haanna.ping_anna_thermostat())

def test_get_domain_objects(self):
"""Get the domain objects"""
self.assertTrue(len(Et.tostring(self.haanna.get_domain_objects(), encoding='utf8', method='xml')) > 0)
time.sleep(3)

def test_get_presets(self):
"""Get the available presets"""
domain_objects = self.haanna.get_domain_objects()
self.assertTrue(len(self.haanna.get_presets(domain_objects)) > 0)
time.sleep(3)

def test_set_preset(self):
"""Set preset to 'Away'"""
domain_objects = self.haanna.get_domain_objects()
self.assertTrue(len(self.haanna.set_preset(domain_objects, 'away')) > 0)
self.haanna.set_preset(domain_objects, 'home')
time.sleep(3)

def test_get_current_preset(self):
"""Get the current active preset"""
domain_objects = self.haanna.get_domain_objects()
self.assertTrue(len(self.haanna.get_current_preset(domain_objects)) > 0)
time.sleep(3)

def test_get_temperature(self):
"""Get the current temperature"""
domain_objects = self.haanna.get_domain_objects()
self.assertIsInstance(self.haanna.get_temperature(domain_objects), float)
time.sleep(3)

def test_get_target_temperature(self):
"""Get the target temperature"""
domain_objects = self.haanna.get_domain_objects()
self.assertIsInstance(self.haanna.get_target_temperature(domain_objects), float)
time.sleep(3)

def test_get_outdoor_temperature(self):
"""Get the outdoor temperature"""
domain_objects = self.haanna.get_domain_objects()
self.assertIsInstance(self.haanna.get_outdoor_temperature(domain_objects), float)
time.sleep(3)

def test_set_temperature(self):
"""Set a new target temperature"""
domain_objects = self.haanna.get_domain_objects()
self.haanna.set_temperature(domain_objects, 22.00)
domain_objects = self.haanna.get_domain_objects()
self.assertEqual(22.00, self.haanna.get_target_temperature(domain_objects))
time.sleep(3)

def test_set_credentials(self):
"""Sets the credentials used to connect to the gateway"""
self.haanna.set_credentials('username', 'password')
self.assertEqual({'username': 'username', 'password': 'password'}, self.haanna.get_credentials())

def test_set_anna_endpoint(self):
"""Sets the endpoint to connect to"""
self.haanna.set_anna_endpoint('http://example.com')
self.assertEqual('http://example.com', self.haanna.get_anna_endpoint())

def test_get_point_log_id(self):
"""Gets the point log id by log type"""
domain_objects = self.haanna.get_domain_objects()
self.assertTrue(len(self.haanna.get_point_log_id(domain_objects, 'temperature')) > 0)
time.sleep(3)

def test_get_measurement_from_point_log(self):
"""Gets the measurement from the given point log"""
domain_objects = self.haanna.get_domain_objects()
temperature_point_log_id = self.haanna.get_point_log_id(domain_objects, 'temperature')
self.assertTrue(float(self.haanna.get_measurement_from_point_log(domain_objects, temperature_point_log_id)) > 0)
time.sleep(3)

def test_get_rule_id_by_name(self):
"""Gets the rule id based on rule name"""
domain_objects = self.haanna.get_domain_objects()
self.assertTrue(len(self.haanna.get_rule_id_by_name(domain_objects, 'Thermostat presets')) > 0)
time.sleep(3)

def test_get_preset_dictionary(self):
"""Gets the available presets based on rule name"""
domain_objects = self.haanna.get_domain_objects()
rule_id = self.haanna.get_rule_id_by_name(domain_objects, 'Thermostat presets')
self.assertTrue(len(self.haanna.get_preset_dictionary(domain_objects, rule_id)) > 0)
time.sleep(3)

0 comments on commit b5b9546

Please sign in to comment.