From 363076c18d25fa50ef57a0796d0d14ff34590a37 Mon Sep 17 00:00:00 2001 From: Jeroen van der Hel Date: Sun, 16 Mar 2025 21:43:50 +0100 Subject: [PATCH] Uploaded latest code --- adax_content.py | 101 +++++++++++++++++ adax_openapi.json | 272 ++++++++++++++++++++++++++++++++++++++++++++++ adaxconnector.py | 134 +++++++++++++++++++++++ main.py | 95 ++++++++++++++++ 4 files changed, 602 insertions(+) create mode 100644 adax_content.py create mode 100644 adax_openapi.json create mode 100644 adaxconnector.py create mode 100644 main.py diff --git a/adax_content.py b/adax_content.py new file mode 100644 index 0000000..8f49be2 --- /dev/null +++ b/adax_content.py @@ -0,0 +1,101 @@ +class AdaxRoom: + """ + Represents an Adax room. + """ + + def __init__(self, room_data: dict): + """ + Initializes an AdaxRoom object from a dictionary. + + Args: + room_data: A dictionary containing room information. + """ + self.id:int = room_data.get("id") + self.home_id:int = room_data.get("homeId") + self.name:str = room_data.get("name") + self.heating_enabled:bool = room_data.get("heatingEnabled") + self.target_temperature:int = room_data.get("targetTemperature") + self.temperature:int = room_data.get("temperature")if room_data.get("temperature") else 0 + + def __repr__(self): + """ + Returns a string representation of the AdaxRoom object. + """ + return ( + f"AdaxRoom(" + f"id={self.id}, " + f"home_id={self.home_id}, " + f"name='{self.name}', " + f"heating_enabled={self.heating_enabled}, " + f"target_temperature={self.target_temperature / 100}, " + f"temperature={self.temperature / 100}" + f")" + ) + + def to_payload(self): + """ + Returns a small dict of itself to use for a post request. + """ + return { + "id": self.id, + "heatingEnabled": self.heating_enabled, + "targetTemperature": self.target_temperature, + } + + def to_dict(self): + """ + Returns a dictionary representation of the AdaxRoom object. + """ + return { + "id": self.id, + "homeId": self.home_id, + "name": self.name, + "heatingEnabled": self.heating_enabled, + "targetTemperature": self.target_temperature, + "temperature": self.temperature, + } + + +class AdaxDevice: + """ + Represents an Adax device. + """ + + def __init__(self, device_data: dict): + """ + Initializes an AdaxDevice object from a dictionary. + + Args: + device_data: A dictionary containing device information. + """ + self.id = device_data.get("id") + self.home_id = device_data.get("homeId") + self.room_id = device_data.get("roomId") + self.name = device_data.get("name") + self.type = device_data.get("type") + + def __repr__(self): + """ + Returns a string representation of the AdaxDevice object. + """ + return ( + f"AdaxDevice(" + f"id={self.id}, " + f"home_id={self.home_id}, " + f"room_id={self.room_id}, " + f"name='{self.name}', " + f"type='{self.type}'" + f")" + ) + + def to_dict(self): + """ + Returns a dictionary representation of the AdaxDevice object. + """ + return { + "id": self.id, + "homeId": self.home_id, + "roomId": self.room_id, + "name": self.name, + "type": self.type, + } diff --git a/adax_openapi.json b/adax_openapi.json new file mode 100644 index 0000000..15b86a8 --- /dev/null +++ b/adax_openapi.json @@ -0,0 +1,272 @@ +{ + "openapi": "3.0.1", + "info": { + "title": "Adax API", + "version": "1" + }, + "paths": { + "/v1/content": { + "get": { + "summary": "Get user content", + "operationId": "getContent", + "responses": { + "default": { + "description": "Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ContentResponse" + } + } + } + }, + "400": { + "description": "Invalid parameters" + }, + "401": { + "description": "Invalid authorization" + }, + "402": { + "description": "Request limit exceeded" + }, + "403": { + "description": "Authorization expired" + } + }, + "security": [ + { + "remoteUserServiceSecurity": [] + } + ] + } + }, + "/v1/control": { + "post": { + "summary": "Controls user objects", + "operationId": "control", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ControlRequest" + } + } + }, + "required": true + }, + "responses": { + "default": { + "description": "Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ControlResponse" + } + } + } + }, + "400": { + "description": "Invalid parameters" + }, + "401": { + "description": "Invalid authorization" + }, + "402": { + "description": "Request limit exceeded" + }, + "403": { + "description": "Authorization expired" + } + }, + "security": [ + { + "remoteUserServiceSecurity": [] + } + ] + } + } + }, + "components": { + "schemas": { + "ContentResponse": { + "type": "object", + "properties": { + "homes": { + "type": "array", + "items": { + "$ref": "#/components/schemas/Home" + } + }, + "rooms": { + "type": "array", + "items": { + "$ref": "#/components/schemas/Room" + } + }, + "devices": { + "type": "array", + "items": { + "$ref": "#/components/schemas/Device" + } + } + } + }, + "Device": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "format": "int64" + }, + "homeId": { + "type": "integer", + "format": "int64" + }, + "roomId": { + "type": "integer", + "format": "int64" + }, + "name": { + "type": "string", + "description": "Device Name." + }, + "type": { + "$ref": "#/components/schemas/DeviceType" + } + } + }, + "DeviceType": { + "type": "string", + "description": "Device type.", + "enum": [ + "Heater" + ] + }, + "Home": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "format": "int64" + }, + "name": { + "type": "string", + "description": "Home name." + } + } + }, + "Room": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "format": "int64" + }, + "homeId": { + "type": "integer", + "format": "int64" + }, + "name": { + "type": "string", + "description": "Room name." + }, + "heatingEnabled": { + "type": "boolean", + "description": "Heating is on/off.", + "nullable": true + }, + "targetTemperature": { + "maximum": 3500, + "minimum": 500, + "type": "integer", + "description": "Target temperature. Degrees Celsius x 100 units.", + "format": "int32", + "nullable": true + }, + "temperature": { + "maximum": 3500, + "minimum": 500, + "type": "integer", + "description": "Current temperature. Degrees Celsius x 100 units.", + "format": "int32", + "nullable": true + } + } + }, + "ControlResponse": { + "type": "object", + "properties": { + "rooms": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ControlResponseRoom" + } + } + } + }, + "ControlResponseRoom": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "format": "int64" + }, + "status": { + "$ref": "#/components/schemas/ControlStatus" + } + } + }, + "ControlStatus": { + "type": "string", + "description": "Control status.", + "enum": [ + "OK", + "NoAccess", + "InvalidParams", + "InternalError" + ] + }, + "ControlRequest": { + "type": "object", + "properties": { + "rooms": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ControlRequestRoom" + } + } + } + }, + "ControlRequestRoom": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "format": "int64" + }, + "heatingEnabled": { + "type": "boolean", + "description": "Set heating on/off." + }, + "targetTemperature": { + "maximum": 3500, + "minimum": 500, + "type": "integer", + "description": "Set target temperature. Degrees Celsius x 100 units; Ignored if heatingEnabled is false.", + "format": "int32" + } + } + } + }, + "securitySchemes": { + "remoteUserServiceSecurity": { + "type": "oauth2", + "flows": { + "password": { + "tokenUrl": "https://api-1.adax.no/client-api/auth/token" + } + } + } + } + } + } \ No newline at end of file diff --git a/adaxconnector.py b/adaxconnector.py new file mode 100644 index 0000000..b0d7977 --- /dev/null +++ b/adaxconnector.py @@ -0,0 +1,134 @@ +import requests +import time +import json +from datetime import datetime, timedelta + + +class AdaxConnector: + def __init__(self, id:str, secret:str) -> None: + self.__client_id = id + self.__client_secret = secret + self.__access_token = None + self.__token_expiration = None + self.__refresh_token = None + self._base_url = "https://api-1.adax.no/client-api" + self._endpoints = { + "content":"rest/v1/content/", + "control":"rest/v1/control/", + "energy_log":"rest/v1/energy_log/" + } + self._request_types = self._endpoints.keys() + self.__get_access_token() + + def __send_auth_request(self, data:dict): + req = requests.Request() + req.url = f"{self._base_url}/auth/token" + req.method = 'POST' + req.headers = {'Accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded'} + req.data = data + prepped = req.prepare() + with requests.Session() as s: + response = s.send(prepped) + + return response.json() + + def send_request(self, request_type:str, data:dict={}): + # Check if request_type exists + if not request_type in self._request_types: + return f"Request type [{request_type}] unknown" + else: + # Check if access token is valid + if self.__token_is_expired(): + self.__refresh_access_token() + + req = requests.Request() + req.url = f"{self._base_url}/{self._endpoints[request_type]}" + req.headers = {"Authorization": f"Bearer {self.__access_token}", "Content-Type": "application/json"} + req.method = 'POST' if request_type == 'control' else 'GET' + if request_type == 'energy_log': + req.url += data['room_id'] + if req.method == 'POST': + req.json = data + prepped = req.prepare() + with requests.Session() as s: + response = s.send(prepped) + + return response.json() + + def __get_access_token(self) -> None: + data = { + "grant_type": "password", + "username": self.__client_id, + "password": self.__client_secret + } + response = self.__send_auth_request(data=data) + self.__access_token = response['access_token'] + self.__refresh_token = response['refresh_token'] + self.__token_expiration = datetime.now()+timedelta(seconds=86400) + # print(response) + + def __refresh_access_token(self) -> None: + data = { + "grant_type": "refresh_token", + "refresh_token": self.__refresh_token, + "username": self.__client_id, + "password": self.__client_secret + } + response = self.__send_auth_request(data=data) + self.__access_token = response['access_token'] + self.__refresh_token = response['refresh_token'] + self.__token_expiration = datetime.now()+timedelta(seconds=86400) + # print(response) + + def __token_is_expired(self) -> bool: + now = datetime.now() + return self.__token_expiration < now + + def __time_until_access_token_expires(self) -> dict: + """Converts a timedelta object to a dictionary of days, hours, minutes, and seconds.""" + td = self.__token_expiration-datetime.now() + days = td.days + seconds = td.seconds + microseconds = td.microseconds + + hours, remainder = divmod(seconds, 3600) + minutes, seconds = divmod(remainder, 60) + + # Handle microseconds by adding them to the seconds + seconds += microseconds / 1000000.0 + + return { + "days": days, + "hours": hours, + "minutes": minutes, + "seconds": seconds, + } + + def get_content(self)-> dict: + response = self.send_request(request_type="content") + # r = json.dumps(response, indent=2) + # print(r) + return response + + + + + def disable_heating_for_room(self, room_id:int): + data = {'rooms': [{ 'id': room_id, 'heatingEnabled': False, 'targetTemperature': 1800}] } + response = self.send_request(request_type="control", data=data) + return response + + def _enable_heating_for_room(self, room_id:int): + data = {'rooms': [{ 'id': room_id, 'heatingEnabled': True}] } + response = self.send_request(request_type="control", data=data) + return response['rooms'][0]['status'] + + def _set_room_temperature(self, room_id:int, target_temp:int): + data = {'rooms': [{ 'id': room_id, 'targetTemperature': target_temp}] } + response = self.send_request(request_type="control", data=data) + return response['rooms'][0]['status'] + + def _get_energy_info(self, room_id:int): + data = {'room_id': str(room_id)} + response = self.send_request(request_type='energy_log', data=data) + return response \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..8032a7f --- /dev/null +++ b/main.py @@ -0,0 +1,95 @@ +import requests +import time +import json +from datetime import datetime, timedelta +from adaxconnector import AdaxConnector +from adax_content import AdaxDevice, AdaxRoom + + +class AdaxController: + def __init__(self,id:str="331582", secret:str="7ShZ7YWWruXD7WZA"): + self._conn = AdaxConnector(id=id, secret=secret) + self._rooms = [] + self._devices = [] + self.__get_adax_content() + + def __get_adax_content(self): + data = self._conn.get_content() + for room in data['rooms']: + self._rooms.append(AdaxRoom(room)) + + for device in data['devices']: + self._devices.append(AdaxDevice(device)) + + def print_room_names(self): + for i, room in enumerate(self._rooms): + print(f"{i}. {room.name} [{room.id}]") + + def get_room_by_name(self, room_name: str) -> AdaxRoom|None: + """ + Finds an AdaxRoom object in a list of AdaxRoom objects by its name. + + Args: + rooms: A list of AdaxRoom objects. + room_name: The name of the room to find. + + Returns: + The AdaxRoom object with the matching name, or None if not found. + """ + # found_rooms = [room for room in rooms if room.name == room_name] + + for room in self._rooms: + if room.name == room_name: + return room + return None + + def get_devices_from_room(self, room_name:str)-> list[AdaxDevice]: + room = self.get_room_by_name(room_name) + return [device for device in self._devices if device.room_id == room.id] + + def disable_heating_in_room(self, room_name): + room = self.get_room_by_name(room_name) + room.heating_enabled = False + payload = {'rooms': [room.to_payload()]} + result = self._conn.send_request('control', data=payload) + print(result) + + def enable_heating_in_room(self, room_name): + room = self.get_room_by_name(room_name) + room.heating_enabled = True + payload = {'rooms': [room.to_payload()]} + result = self._conn.send_request('control', data=payload) + print(result) + + def print_room_info(self, room_name): + self.__get_adax_content() + room = self.get_room_by_name(room_name) + print(f"[{room.name}] Kamertemperatuur actueel: {room.temperature / 100}\u00B0C, gewenst: {room.target_temperature / 100}\u00B0C, heater staat {'aan' if room.heating_enabled else 'uit'}.") + + #Todo: adapt until request + def set_current_room_temperature(self, room_name:str, new_temperature:int): + room = self.get_room_by_name(room_name) + if new_temperature < 100: + t = new_temperature * 100 + else: + t = new_temperature + result = self._conn._set_room_temperature(room_id=room.id, target_temp=t) + print(result) + + def get_energy_info(self, room_name): + room = self.get_room_by_name(room_name) + energy_log = self._conn._get_energy_info(room_id=room.id) + return energy_log + + + + +if __name__ == '__main__': + CLIENT_ID = "331582" + CLIENT_SECRET = "7ShZ7YWWruXD7WZA" + ac = AdaxController(CLIENT_ID, CLIENT_SECRET) + ac.print_room_info('Studio') + ac.get_energy_info('Studio') + ac.print_room_info('Ouderslaapkamer') + ac.print_room_info('Kledingkamer') + input() \ No newline at end of file