import requests import json import threading import websocket import time # import rel import datetime data = "username=1202283%40uad.ac.uk&password=24e76d8e4&grant_type=password" url = "https://api-tevolve.termoweb.net/client/token" headers = {'content-type': 'application/x-www-form-urlencoded', 'authorization': 'Basic NWM0OWRjZTk3NzUxMDM1MTUwNmM0MmRiOnRldm9sdmU='} dev_id = "082e858131ff012c51" API_URL = "https://api-tevolve.termoweb.net/api/v2" DEV_URL = "/grouped_devs" class Tevolve: token = "" def __init__(self): self.need_token = True self.dev_id = "" self.token_primary = None self.token_refresh = "" self.sid = "" self.devices = [] self.latest_message = "" self.time_token_start = 0 setmode = "" setTemperature = "" sid = "" websocket_url = "" mode = "" set_point = "" websocket_message = "" def token_manager(self, event): while 1: current_time = time.time() time_since_refresh = current_time - self.time_token_start print(time_since_refresh) if time_since_refresh > 1000: x = requests.post(url, headers=headers, data=data) if x.status_code == 200: self.token_primary = x.json()["access_token"] self.time_token_start = time.time() print("GOT TOKEN") event.set() else: time.sleep(5) else: time.sleep(60) def get_dev(self): if self.token_primary != "": headers = {'content-type': 'application/json', 'Authorization': "Bearer " + self.token_primary} dev_request = requests.get(str(API_URL+DEV_URL), headers=headers).json() self.dev_id = dev_request[0]["devs"][0]["dev_id"] def get_token(self): if self.token_primary == "": headers = {'content-type': 'application/x-www-form-urlencoded', 'authorization': 'Basic NWM0OWRjZTk3NzUxMDM1MTUwNmM0MmRiOnRldm9sdmU='} data = "username=1202283%40uad.ac.uk&password=24e76d8e4&grant_type=password" token_request = requests.post("https://api-tevolve.termoweb.net/client/token", headers=headers, data=data) if token_request.status_code == 200: request_json = json.loads(token_request.text) self.token_primary = request_json["access_token"] self.token_refresh = request_json["refresh_token"] else: self.token_primary = "" self.token_refresh = "" print("Could not get token") def get_sid(self): try: url = "https://api-tevolve.termoweb.net/socket.io/?token=" + self.token_primary + "&dev_id=" + str(self.dev_id) + "&EIO" \ "=3&transport=polling" response = requests.request("GET", url) if response.status_code == 200: self.sid = json.loads(response.text[5:])["sid"] print("got SID" + str(self.sid)) else: print("NO SID") print(response.status_code) # else: # self.token_primary = "" # Tevolve.token_manager() except Exception as e: print("SID EXCEPTION") raise e def get_devices(self): try: endpoint = "/devs/"+self.dev_id+"/mgr/nodes" header = {'content-type': 'application/json', "Authorization": "Bearer " + self.token_primary} url = str(API_URL + endpoint) devices_request = requests.get(str(API_URL + endpoint), headers=header).json() self.devices = devices_request except: raise def post_websocket(self): # Tevolve.token_manager() try: print("PREVIOUS SID: " + str(self.sid)) self.get_sid() print("NEW SID: " + str(self.sid)) url = "https://api-tevolve.termoweb.net/socket.io/?token=" + self.token_primary + "&dev_id=" + str(self.dev_id) + "&EIO" \ "=3&transport=polling&t=Ntb6xXn" \ "&sid=" + self.sid payload = "401:40/api/v2/socket_io?token=" + self.token_primary + "&dev_id=" + str(self.dev_id) headers = { 'Cookie': "io=" + self.sid, 'Content-Type': 'text/plain', 'Connection': 'keep-alive' } r = requests.request("POST", url, headers=headers, data=payload) print("post_websocket " + str(r.status_code)) except Exception as e: raise e def set_mode(self, heater_id, mode): heater_id = heater_id # mode = Tevolve.setmode headers = { 'host': 'api-tevolve.termoweb.net', 'origin': 'https://tevolve.termoweb.net', 'content-type': 'application/json', 'accept': 'application/json, text/plain, */*', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) ' 'Version/15.4 Safari/605.1.15', 'authorization': 'Bearer ' + self.token_primary, 'referer': 'https://tevolve.termoweb.net/' } url = "https://api-tevolve.termoweb.net/api/v2/devs/082e858131ff012c51/htr/" + str(heater_id) + "/status" if mode == "heat": mode = "manual" payload = json.dumps({ "mode": mode }) response = requests.post(url, headers=headers, data=payload) if response.status_code == 201 or response.status_code == 200: pass # def get_mode(self): # headers = { # 'host': 'api-tevolve.termoweb.net', # 'origin': 'https://tevolve.termoweb.net', # 'content-type': 'application/json', # 'accept': 'application/json, text/plain, */*', # 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) ' # 'Version/15.4 Safari/605.1.15', # 'authorization': 'Bearer ' + self.token_primary, # 'referer': 'https://tevolve.termoweb.net/', # # } # url = "https://api-tevolve.termoweb.net/api/v2/devs/082e858131ff012c51/htr/" + self.heater_id + "/status" # # response = requests.get(url, headers=headers) # if response.status_code == 201 or response.status_code == 200: # mode = response.json()["mode"] # if mode == "manual": # return "heat" def get_status(self): result = {} try: for index ,i in enumerate(self.devices["nodes"]): headers = { 'host': 'api-tevolve.termoweb.net', 'origin': 'https://tevolve.termoweb.net', 'content-type': 'application/json', 'accept': 'application/json, text/plain, */*', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) ' 'Version/15.4 Safari/605.1.15', 'authorization': 'Bearer ' + self.token_primary, 'referer': 'https://tevolve.termoweb.net/', } url = "https://api-tevolve.termoweb.net/api/v2/devs/"+self.dev_id+"/htr/"+str(i["addr"])+"/status" response = requests.get(url, headers=headers) if response.status_code == 201 or response.status_code == 200: result.update({i["name"]: response.json()}) except Exception as e: raise e return result def set_temperature(self, heater_id, set_temperature): headers = { 'host': 'api-tevolve.termoweb.net', 'origin': 'https://tevolve.termoweb.net', 'content-type': 'application/json', 'accept': 'application/json, text/plain, */*', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) ' 'Version/15.4 Safari/605.1.15', 'authorization': 'Bearer ' + self.token_primary, 'referer': 'https://tevolve.termoweb.net/' } url = "https://api-tevolve.termoweb.net/api/v2/devs/082e858131ff012c51/htr/" + str(heater_id) + "/status" payload = json.dumps({ "stemp": str(float(set_temperature)), "units": "C", "mode": "manual" }) response = requests.post(url, headers=headers, data=payload) if response.status_code == 201 or response.status_code == 200: pass def create_websocket(self, mqtt_manager): global start_time global end_time websocket_url = "wss://api-tevolve.termoweb.net/socket.io/?token=" + self.token_primary + \ "&dev_id=" +str(self.dev_id)+ "&EIO=3&transport=websocket&sid=" + self.sid + "" def on_message(ws, message): print("MESSAGE") self.websocket_message = message print(message) if len(message) > 20: a = json.loads(message[20:]) if a[0] == "update": self.last_update = a[1] mqtt_manager.update_homeassistant_entity(self.last_update) def on_open(ws): try: self.post_websocket() websocket_url = "wss://api-tevolve.termoweb.net/socket.io/?token=" + self.token_primary + \ "&dev_id=" + str(self.dev_id) + "&EIO=3&transport=websocket&sid=" + self.sid + "" print("postwebsocket") print("websocket (run)") counter = 0 ws.send("2probe") time.sleep(2) ws.send(str("5")) # while 1: # ws.send("2") # counter = counter + 1 # if counter >= 360: # break # time.sleep(10) # self.create_websocket(mqtt_connection) except Exception as e: print(e) ws.close() def on_close(ws, close_status_code, close_msg): print("WEB SOCKET ClOSED") self.r.join() self.r = threading.Thread(target=run, args=(mqtt_manager,)) self.r.start() # global end_time # end_time = datetime.datetime.now() # print("starttime: " + str(start_time)) # print("endtime: " + str(end_time)) # breakpoint() # # # mqtt_connection.publish_heater() # time.sleep(2) # try: # self.post_websocket() # # self.create_websocket(mqtt_connection) # except Exception as e: # print(e) def on_error(ws, err): print("WEB SOCKET ERROR") breakpoint() # self.post_websocket() # self.create_websocket(mqtt_connection def on_pong(ws, message): ws.send("2") print(message) def on_reconnect(ws): try: self.post_websocket() print("ReconnectHERE") except Exception as e: print(e) def run(): print("websocket (run)") counter = 0 ws.send("2probe") time.sleep(2) ws.send(str("5")) while 1: ws.send("2") counter = counter + 1 if counter >= 360: break time.sleep(10) ws = websocket.WebSocketApp(websocket_url, on_open=on_open, on_close=on_close, on_error=on_error, on_message=on_message, on_pong=on_pong) websocket.enableTrace(True) # ws.on_message = on_message # ws.on_open = on_open # ws.on_close = on_close # ws.on_error = on_error # ws.on_reconnect = on_reconnect ws.run_forever(ping_interval=10, ping_payload='42/api/v2/socket_io,["message","ping"]') # rel.signal(2, rel.abort) # rel.dispatch() self.r = threading.Thread(target=run, args=(mqtt_manager,)) self.r.start()