128 lines
3.1 KiB
Plaintext
128 lines
3.1 KiB
Plaintext
|
#!/usr/bin/python3
|
||
|
|
# get target temperature from AVM/Comet Dect Heizkoerperregler (Heating Device)
|
||
|
|
# need Fritz!OS 6.50+
|
||
|
|
|
||
|
|
import getopt
|
||
|
|
import re
|
||
|
|
import requests
|
||
|
|
import sys
|
||
|
|
import xml.etree.ElementTree as etree
|
||
|
|
import urllib.parse
|
||
|
|
from hashlib import md5
|
||
|
|
|
||
|
|
DEBUG=False
|
||
|
|
|
||
|
|
def OK(text):
|
||
|
|
print("OK: " + text)
|
||
|
|
sys.exit()
|
||
|
|
|
||
|
|
def Warning(text):
|
||
|
|
print("WARNING: " + text)
|
||
|
|
sys.exit()
|
||
|
|
|
||
|
|
def Critical(text):
|
||
|
|
print("CRITICAL: " + text)
|
||
|
|
sys.exit()
|
||
|
|
|
||
|
|
def OptionsError():
|
||
|
|
Critical("Error: not all required options are given.\n -H <hostname>\n -U <username>\n -P <password>\n -A <ID of device>")
|
||
|
|
|
||
|
|
def debug(text):
|
||
|
|
if DEBUG == True:
|
||
|
|
print("DEBUG: " + text)
|
||
|
|
|
||
|
|
def get_sid(loginurl, username, password):
|
||
|
|
if not username:
|
||
|
|
username=''
|
||
|
|
|
||
|
|
# get challenge string
|
||
|
|
response = requests.get(loginurl)
|
||
|
|
|
||
|
|
if response.status_code == 200:
|
||
|
|
tree = etree.fromstring(response.content)
|
||
|
|
challenge = tree.find("Challenge").text
|
||
|
|
sid = tree.find("SID").text
|
||
|
|
debug("challenge: " + challenge)
|
||
|
|
debug("sid: " + sid)
|
||
|
|
else:
|
||
|
|
Critical("Error: GetSid failed")
|
||
|
|
|
||
|
|
if len(sid) > 0 and re.search("^0+$", sid) and challenge:
|
||
|
|
debug("Get new SID")
|
||
|
|
# sid is null, got challenge
|
||
|
|
sid = ""
|
||
|
|
# build password response
|
||
|
|
pwd = challenge + "-" + password
|
||
|
|
# UTF-16LE encoding as required
|
||
|
|
pwd = pwd.encode("UTF-16LE")
|
||
|
|
# md5hash on top
|
||
|
|
m = md5()
|
||
|
|
m.update(pwd)
|
||
|
|
# final answer string
|
||
|
|
challenge_response = challenge + "-" + m.hexdigest()
|
||
|
|
debug("challenge_response: " + challenge_response)
|
||
|
|
# send to box
|
||
|
|
url = loginurl + "?username=" + username + "&response=" + challenge_response
|
||
|
|
response = requests.get(url)
|
||
|
|
#print(response.content)
|
||
|
|
# check answer
|
||
|
|
tree = etree.fromstring(response.content)
|
||
|
|
sid2 = tree.find("SID").text
|
||
|
|
debug("SID: " + sid2)
|
||
|
|
if (len(sid2) > 0) and not re.search("^0+$", sid2):
|
||
|
|
# is not null, bingo!
|
||
|
|
return sid2
|
||
|
|
else:
|
||
|
|
# use existing sid if sid matches an hex string
|
||
|
|
if (len(sid) > 0) and re.search("^[0-9a-f]+$", sid):
|
||
|
|
return sid
|
||
|
|
|
||
|
|
host = None
|
||
|
|
username = None
|
||
|
|
password = None
|
||
|
|
ain = None
|
||
|
|
|
||
|
|
try:
|
||
|
|
opts, args = getopt.gnu_getopt(sys.argv[1:], 'H:U:P:A:')
|
||
|
|
except getopt.GetoptError as err:
|
||
|
|
OptionsError()
|
||
|
|
|
||
|
|
for opt, arg in opts:
|
||
|
|
if opt == "-H":
|
||
|
|
host = arg
|
||
|
|
elif opt == "-U":
|
||
|
|
username = arg
|
||
|
|
elif opt == "-P":
|
||
|
|
password = arg
|
||
|
|
elif opt == "-A":
|
||
|
|
ain = arg
|
||
|
|
|
||
|
|
if not host or not username or not password or not ain:
|
||
|
|
OptionsError()
|
||
|
|
|
||
|
|
# build urls and login
|
||
|
|
# - login_sid.lua for fritzos 5.50+
|
||
|
|
loginurl = "http://" + host + "/login_sid.lua"
|
||
|
|
ahaurl = "http://" + host + "/webservices/homeautoswitch.lua"
|
||
|
|
sid = get_sid(loginurl, username, password);
|
||
|
|
if not sid:
|
||
|
|
Critical("Error: Login invalid or failed")
|
||
|
|
|
||
|
|
#query actual state
|
||
|
|
rain = urllib.parse.quote(ain)
|
||
|
|
url = ahaurl + '?ain=' + rain + '&sid=' + sid + '&switchcmd=gettemperature';
|
||
|
|
|
||
|
|
http_response = requests.get(url)
|
||
|
|
|
||
|
|
if http_response.status_code == 200:
|
||
|
|
# check answer
|
||
|
|
answer = http_response.content.decode().strip()
|
||
|
|
debug("Antwort: " + answer)
|
||
|
|
temperatur = float(answer) / 10
|
||
|
|
temp_str = "{0:<3.1f}".format(temperatur)
|
||
|
|
OK("Temperatur: " + temp_str + " | temp=" + temp_str)
|
||
|
|
else:
|
||
|
|
Critical("Error: Query failed.")
|
||
|
|
|
||
|
|
|