This is a ways off from being completed. My functions can be polished more, I basically wrote each one to work individually and was going to go back and combine some functionality, which I haven't done yet. When it is more polished, I will probably post the whole project on Ignition Exchange.
I have my api tokens and authorization stuff held in tags, which is not very secure. When I'm done with the functionality, I plan on storing those with encryption. Note it has been a while since these have been fully tested. Also note that you have to setup an ecobee dev account before you can use the APIs, ecobee API
def getAuthorization():
import system
apiKey = system.tag.readBlocking('[default]Ecobee/APIKey').value
params = {"response_type":"ecobeePin", "client_id":apiKey, "scope":"smartWrite"}
#url = "https://api.ecobee.com/authorize?response_type=ecobeePin&client_id=%s&scope=smartWrite" % apiKey
url = "https://api.ecobee.com/authorize"
response = http.client.post(url=url, params = params)
json = response.getJson()
tags = ['[default]Ecobee/AuthCode','[default]Ecobee/Pin']
values = [json['code'],json['ecobeePin']]
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def getTokens():
import system
tags = ['[default]Ecobee/AuthCode','[default]Ecobee/APIKey']
values = system.tag.readBlocking(tags)
authCode = values[0].value
apiKey = values[1].value
url = "https://api.ecobee.com/token"
postData = "grant_type=ecobeePin&code=%s&client_id=%s" % (authCode, apiKey)
response = system.net.httpPost(url=url,postData=postData)
json = system.util.jsonDecode(response)
tags = ['[default]Ecobee/Access Timeout','[default]Ecobee/Access Token', '[default]Ecobee/Refresh Token']
values = [json['expires_in'],json['access_token'],json['refresh_token']]
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def refreshTokens():
import system
tags = ['[default]Ecobee/Refresh Token','[default]Ecobee/APIKey']
values = system.tag.readBlocking(tags)
refreshToken = values[0].value
apiKey = values[1].value
params = {"grant_type":"refresh_token", "refresh_token":refreshToken, "client_id":apiKey, "ecobee_type":"jwt"}
url = "https://api.ecobee.com/token"
response = http.client.post(url=url,params=params)
json = response.getJson()
tags = ['[default]Ecobee/Access Timeout','[default]Ecobee/Access Token', '[default]Ecobee/Refresh Token']
values = [json['expires_in'],json['access_token'],json['refresh_token']]
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def getRunTimeInfo():
import system
raw = system.tag.readBlocking('[default]Ecobee/Access Token')
accessToken = raw[0].value
headers = {'Content-Type': 'text/json','Authorization':'Bearer %s' % accessToken}
params = {"format":"json","body":{"selection":{"selectionType":"registered","selectionMatch":"","includeRuntime":"true"}}} #"includeAlerts":true
url = 'https://api.ecobee.com/1/thermostat'
response = http.client.get(url, params = params, headers=headers)
json = response.getJson()
tags = []
values = []
for key in json['status']:
tags.append('[default]Ecobee/API Return %s' % key)
values.append(json['status'][key])
for thermostat in json['thermostatList']:
name = thermostat['name']
baseTagPath = '[default]Ecobee/Thermostats/%s/Runtime/' % name
for key in thermostat['runtime']:
tags.append(baseTagPath + key)
values.append(thermostat['runtime'][key])
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def getSettingsInfo():
import system
raw = system.tag.readBlocking('[default]Ecobee/Access Token')
accessToken = raw[0].value
headers = {"Content-Type": "text/json","Authorization":"Bearer %s" % accessToken}
params = {"format":"json", "body":{"selection":{"selectionType":"registered","selectionMatch":"","includeSettings":"true"}}} #"includeAlerts":true
url = 'https://api.ecobee.com/1/thermostat'
response = http.client.get(url, params=params, headers=headers)
json = response.getJson()
tags = []
values = []
for key in json['status']:
tags.append('[default]Ecobee/API Return %s' % key)
values.append(json['status'][key])
for thermostat in json['thermostatList']:
name = thermostat['name']
baseTagPath = '[default]Ecobee/Thermostats/%s/Settings/' % name
for key in thermostat['settings']:
tags.append(baseTagPath + key)
values.append(thermostat['settings'][key])
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def getStatus():
import system
raw = system.tag.readBlocking('[default]Ecobee/Access Token')
accessToken = raw[0].value
headers = {'Content-Type': 'text/json','Authorization':'Bearer %s' % accessToken}
params = {"format":"json", "body":{"selection":{"selectionType":"registered","selectionMatch":"","includeEquipmentStatus":"true"}}} #"includeAlerts":true
url = 'https://api.ecobee.com/1/thermostat' % body
response = http.client.get(url, params=params, headers=headers)
json = response.getJson()
tags = []
values = []
for key in json['status']:
tags.append('[default]Ecobee/API Return %s' % key)
values.append(json['status'][key])
for thermostat in json['thermostatList']:
name = thermostat['name']
baseTagPath = '[default]Ecobee/Thermostats/%s/' % name
for key in thermostat:
if key != 'name':
tags.append(baseTagPath + key)
values.append(thermostat[key])
system.tag.writeBlocking(tags,values)
##########################################################################################################################################
def setTempHold(newHoldTemp, thermostatName, holdType, **kwargs):
'''
Args: newHoldTemp = temperature to hold at (must be between thermostat temp ranges)
thermostatName = Name of the thermostat to apply hold to
holdType = Type of hold ('dateTime', 'nextTranistion', 'indefinte', 'holdHours')
**kwargs = For optional arguments
optional args:
startDate, startTime, endDate, endTime = required for 'dateTime' hold type, when to start and stop the temp hold
fanSpeed = speed options (LOW, MEDIUM, HIGH, and OPTIMIZED
holdHours = required for 'holdHours' hold type, how long to hold temp
kwargs is a dictionary. To use kwargs the arguments must be defined in the call
example: ecobee.setTempHold(70, 'Main Floor', 'holdHours', holdHours = 2)
'''
import system
if holdType == 'dateTime':
reqKeys = ('startDate','startTime','endDate', 'endTime')
if not all(key in kwargs for key in reqKeys):
# write to status tags
return
elif holdType == 'holdHours':
if 'holdHours' not in kwargs:
# write to status tags
return
newHoldTemp *= 10 # Multiple temp by 10 because 700 = 70.0 deg
keys = ['coolRangeHigh','coolRangeLow','heatRangeHigh','heatRangeLow', 'heatCoolMinDelta', 'hvacMode']
tags = []
for key in keys:
tags.append('[default]Ecobee/Thermostats/%s/Settings/%s' % (thermostatName,key))
tags.append('[default]Ecobee/Access Token')
values = system.tag.readBlocking(tags)
coolHigh = values[0].value
coolLow = values[1].value
heatHigh = values[2].value
heatLow = values[3].value
minDelta = values[4].value
hvacMode = values[5].value
accessToken = values[6].value
coolHold = heatHold = newHoldTemp # Recommended that coolHold = heatHold
if hvacMode == 'auto': # If hvacMode equals auto, cool hold should minDelta higher than heat hold
coolHold += minDelta
if coolLow <= coolHold <= coolHigh and heatLow <= heatHold <= heatHigh: # Hold must be inside range
reqDict = {'holdType':holdType, 'heatHoldTemp':heatHold, 'coolHoldTemp':coolHold}
newDict = reqDict.copy()
newDict.update(kwargs)
dataParams = system.util.jsonEncode(newDict)
data = {
"selection": {
"selectionType":"registered",
"selectionMatch":thermostatName
},
"functions": [
{
"type":"setHold",
"params": dataParams
}
]
}
headers = {"Content-Type": "application/json;charset=UTF-8","Authorization":"Bearer %s" % accessToken}
params = {"format":"json"}
url = 'https://api.ecobee.com/1/thermostat'
response = http.client.post(url, headers=headers, data=data)
print response.getText()
else:
pass
# Add script to write to status tag with error message
##########################################################################################################################################
def resumeProgram(thermostatName):
import system
raw = system.tag.readBlocking('[default]Ecobee/Access Token')
accessToken = raw[0].value
data = {
"selection": {
"selectionType":"registered",
"selectionMatch":thermostatName
},
"functions": [
{
"type":"resumeProgram",
"params":{
"resumeAll":"false"
}
}
]
}
headers = {"Content-Type":"text/json","Authorization":"Bearer %s" % accessToken}
params = {"format":"json"}
url = 'https://api.ecobee.com/1/thermostat'
response = http.client.post(url, params=params, headers=headers, data=data)
##########################################################################################################################################
def getEvents():
import system
raw = system.tag.readBlocking('[default]Ecobee/Access Token')
accessToken = raw[0].value
headers = {'Content-Type': 'text/json','Authorization':'Bearer %s' % accessToken}
params = {"format":"json","body":{"selection":{"selectionType":"registered","selectionMatch":"","includeEvents":"true"}}} #"includeAlerts":true
url = 'https://api.ecobee.com/1/thermostat'
response = http.client.get(url=url, params=params, headers=headers)
print response.getText()