Jeroenvdhel/AdaxController
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Uploaded latest code

Browse Source
This commit is contained in:
Jeroen van der Hel 2025-03-16 21:43:50 +01:00
parent 3b2d000cd9
commit 363076c18d
4 changed files with 602 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
adax_content.py Normal file
View File

@ -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,
}

272
adax_openapi.json Normal file
View File

@ -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"
}
}
}
}
}
}

134
adaxconnector.py Normal file
View File

@ -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

95
main.py Normal file
View File

@ -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()