From 9c781359e1a9949303ff772159feea4cc1532cbd Mon Sep 17 00:00:00 2001 From: connorroy Date: Sun, 30 Mar 2025 18:08:40 +0100 Subject: [PATCH] Begin refactor --- src/pytevolve.py | 150 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 112 insertions(+), 38 deletions(-) diff --git a/src/pytevolve.py b/src/pytevolve.py index 1c68640..ce65f9a 100644 --- a/src/pytevolve.py +++ b/src/pytevolve.py @@ -12,7 +12,18 @@ headers = {'content-type': 'application/x-www-form-urlencoded', dev_id = "082e858131ff012c51" + + + class Tevolve: + def __init__(self): + self.api_url = "https://api-tevolve.termoweb.net/api/v2" + self.dev_url = "/grouped_devs" + self.dev_id = "" + self.token_primary = "" + self.token_refresh = "" + self.sid = "" + setmode = "" setTemperature = "" @@ -33,6 +44,45 @@ class Tevolve: livingroom_set_temp = "" livingroom_active = "" + + def get_dev(self): + + if self.token_primary != "": + + headers = {'content-type': 'application/json', + 'Authorization': "Bearer " + self.token_primary} + + + dev_request = requests.get(str(self.api_url+self.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") + + + + + + + @staticmethod def token_manager(): @@ -47,42 +97,47 @@ class Tevolve: else: time.sleep(30) - @staticmethod - def get_sid(): - time.sleep(5) - url = "https://api-tevolve.termoweb.net/socket.io/?token=" + Tevolve.token + "&dev_id=082e858131ff012c51&EIO" \ + def get_sid(self): + + url = "https://api-tevolve.termoweb.net/socket.io/?token=" + self.token_primary + "&dev_id=" + str(self.dev_id) + "&EIO" \ "=3&transport=polling" - - payload = {} - - response = requests.request("GET", url, data=payload) + response = requests.request("GET", url) if response.status_code == 200: - - temp = response.text[5:] - temp = json.loads(temp) - - Tevolve.sid = temp['sid'] + self.sid = json.loads(response.text[5:])["sid"] else: - Tevolve.token = None - Tevolve.token_manager() - - @staticmethod - def post_websocket(): - - Tevolve.token_manager() - Tevolve.get_sid() + self.token_primary = "" + # Tevolve.token_manager() - url = "https://api-tevolve.termoweb.net/socket.io/?token=" + Tevolve.token + "&dev_id=082e858131ff012c51&EIO" \ + def get_devices(self): + endpoint = "/devs/082e858131ff012c51/mgr/nodes" + + header = {"Authorization": "Bearer " + self.token_primary} + + devices_request = requests.get(str(self.api_url + endpoint), headers=headers).json() + + print(devices_request) + + + + + + def post_websocket(self): + + # Tevolve.token_manager() + # Tevolve.get_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=" + Tevolve.sid + "&sid=" + self.sid - payload = "401:40/api/v2/socket_io?token=" + Tevolve.token + "&dev_id=082e858131ff012c51" + payload = "401:40/api/v2/socket_io?token=" + self.token_primary + "&dev_id=" + str(self.dev_id) headers = { - 'Cookie': "io=" + Tevolve.sid, + 'Cookie': "io=" + self.sid, 'Content-Type': 'text/plain', 'Connection': 'keep-alive' @@ -90,14 +145,15 @@ class Tevolve: requests.request("POST", url, headers=headers, data=payload) - @staticmethod - def create_websocket(): - time.sleep(5) - websocket_url = "wss://api-tevolve.termoweb.net/socket.io/?token=" + Tevolve.token + "&dev_id=082e858131ff012c51&EIO=3&transport=websocket&sid=" + Tevolve.sid + "" + def create_websocket(self): + + + 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): - Tevolve.websocket_message = message + self.websocket_message = message + print(message) def on_open(ws): @@ -119,14 +175,14 @@ class Tevolve: def on_close(ws, close_status_code, close_msg): print("WEB SOCKET ClOSED") - Tevolve.post_websocket() - Tevolve.create_websocket() + self.post_websocket() + self.create_websocket() def on_error(ws, err): print("WEB SOCKET ERROR") - Tevolve.post_websocket() - Tevolve.create_websocket() + self.post_websocket() + self.create_websocket() ws = websocket.WebSocketApp(websocket_url) websocket.enableTrace(False) @@ -187,8 +243,8 @@ class Tevolve: if mode == "manual": return "heat" - @staticmethod - def get_status(heater_id): + + def get_status(self): headers = { 'host': 'api-tevolve.termoweb.net', @@ -197,11 +253,11 @@ class Tevolve: '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 ' + Tevolve.token, + 'authorization': 'Bearer ' + self.token_primary, 'referer': 'https://tevolve.termoweb.net/', } - url = "https://api-tevolve.termoweb.net/api/v2/devs/082e858131ff012c51/htr/" + heater_id + "/status" + url = "https://api-tevolve.termoweb.net/api/v2/devs/082e858131ff012c51/status" response = requests.get(url, headers=headers) if response.status_code == 201 or response.status_code == 200: @@ -233,3 +289,21 @@ class Tevolve: response = requests.post(url, headers=headers, data=payload) if response.status_code == 201 or response.status_code == 200: pass + +if __name__ == '__main__': + test = Tevolve() + test.get_token() + test.get_dev() + + test.get_devices() + + test.get_sid() + test.get_status() + test.post_websocket() + test.create_websocket() + + + while 1: + if test.websocket_message != "": + print(test.websocket_message) + test.websocket_message = "" \ No newline at end of file