4.1. Remote Client API User Guide
Overview
Adax API is provided as a REST service and that allows you to:
- Get your rooms target and current temperatures,
- Change your rooms heating target temperatures.
Getting Started
For using Adax API you will need an authentication token for secure access and some REST services usage knowledge. The API is provided at the https://
api-1.adax.no/client-api/ address. The services description can be accessed at the https://api-1.adax.no/client-api/openapi in an OpenAPI format. You
could also use the provided Python and Java samples as a starting point.
Obtaining Authentication Credentials Required for Adax API
Using the Adax WiFi application you can create and get the authentication token used for accessing the Adax API. You should use at least the Adax WiFi
app version 3.7. The authentication token can be accessed following those steps:
- Navigate to Account Tab,
- Select “Remote user client API”
- Select “Add Credential”
- Give some name to the created credential and copy the generated password.
Using Adax API
The API provides 3 end points:
- /auth/ for authentication to the REST service.
- /rest/v1/control/ for controlling your rooms
- /rest/v1/content/ for getting rooms statuses. (add ?withEnergy=1 to get heaters power consumption for past hour)
- /rest/v1/energy_log/{roomId} – returns hourly energy log from current time back to 1 week for the specified room in 1h intervals.
Control and status endpoins accepts and returns json documents.
Code Samples
Python Sample
# install dependencies with: pip install requests sanction # install dependencies with: pip install requests sanction
# replace with your client ID (see Adax WiFi app,Account Section)
CLIENT_SECRET = "xd8TdSCendw2bc6h"
# replace with your client secret (see Adax WiFi app,Account Section)
API_URL = "https://api-1.adax.no/client-api"
oauthClinet = sanction.Client(token_endpoint = API_URL + '/auth/token')
# Authenticate and obtain JWT token
oauthClinet.request_token(grant_type = 'password', username = CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
oauthClinet.request_token(grant_type='refresh_token', refresh_token = oauthClinet.refresh_token, username =
CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
def set_room_target_temperature(roomId, temperature, token):
# Sets target temperature of the room
headers = { "Authorization": "Bearer " + token }
json = { 'rooms': [{ 'id': roomId, 'targetTemperature': str(temperature) }] }
response = requests.post(API_URL + '/rest/v1/control/', json = json, headers = headers)
def get_energy_info(token, roomId):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/energy_log/" + str(roomId), headers = headers)
for log in json['points']:
fromTime = datetime.datetime.utcfromtimestamp(int(log['fromTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
toTime = datetime.datetime.utcfromtimestamp(int(log['toTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
energy = log['energyWh'];
print("From: %15s, To: %15s, %5dwh" % (fromTime, toTime, energy))
def get_homes_info(token):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/content/?withEnergy=1", headers = headers)
for room in json['rooms']:
if ('targetTemperature'in room):
targetTemperature = room['targetTemperature'] / 100.0
if ('temperature' in room):
currentTemperature = room['temperature'] / 100.0
print("Room: %15s, Target: %5.2fC, Temperature: %5.2fC, id: %5d" % (roomName, targetTemperature, currentTemperature, room['id']))
for device in jason['devices']:
deviceName = device['name']
energy = device['energyWh'];
energyTime = datetime.datetime.utcfromtimestamp(int(device['energyTime']) / 1000).strftime('%Y-%m-% d %H:%M:%S')
print(Device: %15s, Time: %15s, Energy: %5dwh, id: %5d" % (deviceName, energyTime, energy, device
# Change the temperature to 24 C in the room with an Id of 196342
set_room_target_temperature(735362, 2400, token)
# Replace the 196342 with the room id from the
set_room_target_temperature(750269, 2400, token)
# Replace the 196342 with the room id from the
get_energy_info(token, 735362)
# install dependencies with: pip install requests sanction # install dependencies with: pip install requests sanction
import requests
import sanction
import time
import datetime
CLIENT_ID = "300288"
# replace with your client ID (see Adax WiFi app,Account Section)
CLIENT_SECRET = "xd8TdSCendw2bc6h"
# replace with your client secret (see Adax WiFi app,Account Section)
API_URL = "https://api-1.adax.no/client-api"
oauthClinet = sanction.Client(token_endpoint = API_URL + '/auth/token')
def get_token():
# Authenticate and obtain JWT token
oauthClinet.request_token(grant_type = 'password', username = CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
def refresh_token():
oauthClinet.request_token(grant_type='refresh_token', refresh_token = oauthClinet.refresh_token, username =
CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
def set_room_target_temperature(roomId, temperature, token):
# Sets target temperature of the room
headers = { "Authorization": "Bearer " + token }
json = { 'rooms': [{ 'id': roomId, 'targetTemperature': str(temperature) }] }
response = requests.post(API_URL + '/rest/v1/control/', json = json, headers = headers)
def get_energy_info(token, roomId):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/energy_log/" + str(roomId), headers = headers)
json = response.json()
for log in json['points']:
fromTime = datetime.datetime.utcfromtimestamp(int(log['fromTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
toTime = datetime.datetime.utcfromtimestamp(int(log['toTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
energy = log['energyWh'];
print("From: %15s, To: %15s, %5dwh" % (fromTime, toTime, energy))
def get_homes_info(token):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/content/?withEnergy=1", headers = headers)
print(response)
json = response.json()
for room in json['rooms']:
roomName = room['name']
if ('targetTemperature'in room):
targetTemperature = room['targetTemperature'] / 100.0
else:
targetTemperature = 0
if ('temperature' in room):
currentTemperature = room['temperature'] / 100.0
else:
currentTemperature = 0
print("Room: %15s, Target: %5.2fC, Temperature: %5.2fC, id: %5d" % (roomName, targetTemperature, currentTemperature, room['id']))
if ('devices' in json):
for device in jason['devices']:
deviceName = device['name']
energy = device['energyWh'];
energyTime = datetime.datetime.utcfromtimestamp(int(device['energyTime']) / 1000).strftime('%Y-%m-% d %H:%M:%S')
print(Device: %15s, Time: %15s, Energy: %5dwh, id: %5d" % (deviceName, energyTime, energy, device
['id']))
token = get_token()
while True:
time.sleep(10)
# Change the temperature to 24 C in the room with an Id of 196342
set_room_target_temperature(735362, 2400, token)
# Replace the 196342 with the room id from the
get_homes_info output
time.sleep(10)
set_room_target_temperature(750269, 2400, token)
# Replace the 196342 with the room id from the
get_homes_info output
time.sleep(10)
get_homes_info(token)
time.sleep(10)
get_energy_info(token, 735362)
token = refresh_token()
# install dependencies with: pip install requests sanction # install dependencies with: pip install requests sanction
import requests
import sanction
import time
import datetime
CLIENT_ID = "300288"
# replace with your client ID (see Adax WiFi app,Account Section)
CLIENT_SECRET = "xd8TdSCendw2bc6h"
# replace with your client secret (see Adax WiFi app,Account Section)
API_URL = "https://api-1.adax.no/client-api"
oauthClinet = sanction.Client(token_endpoint = API_URL + '/auth/token')
def get_token():
# Authenticate and obtain JWT token
oauthClinet.request_token(grant_type = 'password', username = CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
def refresh_token():
oauthClinet.request_token(grant_type='refresh_token', refresh_token = oauthClinet.refresh_token, username =
CLIENT_ID, password = CLIENT_SECRET)
return oauthClinet.access_token
def set_room_target_temperature(roomId, temperature, token):
# Sets target temperature of the room
headers = { "Authorization": "Bearer " + token }
json = { 'rooms': [{ 'id': roomId, 'targetTemperature': str(temperature) }] }
response = requests.post(API_URL + '/rest/v1/control/', json = json, headers = headers)
def get_energy_info(token, roomId):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/energy_log/" + str(roomId), headers = headers)
json = response.json()
for log in json['points']:
fromTime = datetime.datetime.utcfromtimestamp(int(log['fromTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
toTime = datetime.datetime.utcfromtimestamp(int(log['toTime']) / 1000).strftime('%Y-%m-%d %H:%M:%S')
energy = log['energyWh'];
print("From: %15s, To: %15s, %5dwh" % (fromTime, toTime, energy))
def get_homes_info(token):
headers = { "Authorization": "Bearer " + token }
response = requests.get(API_URL + "/rest/v1/content/?withEnergy=1", headers = headers)
print(response)
json = response.json()
for room in json['rooms']:
roomName = room['name']
if ('targetTemperature'in room):
targetTemperature = room['targetTemperature'] / 100.0
else:
targetTemperature = 0
if ('temperature' in room):
currentTemperature = room['temperature'] / 100.0
else:
currentTemperature = 0
print("Room: %15s, Target: %5.2fC, Temperature: %5.2fC, id: %5d" % (roomName, targetTemperature, currentTemperature, room['id']))
if ('devices' in json):
for device in jason['devices']:
deviceName = device['name']
energy = device['energyWh'];
energyTime = datetime.datetime.utcfromtimestamp(int(device['energyTime']) / 1000).strftime('%Y-%m-% d %H:%M:%S')
print(Device: %15s, Time: %15s, Energy: %5dwh, id: %5d" % (deviceName, energyTime, energy, device
['id']))
token = get_token()
while True:
time.sleep(10)
# Change the temperature to 24 C in the room with an Id of 196342
set_room_target_temperature(735362, 2400, token)
# Replace the 196342 with the room id from the
get_homes_info output
time.sleep(10)
set_room_target_temperature(750269, 2400, token)
# Replace the 196342 with the room id from the
get_homes_info output
time.sleep(10)
get_homes_info(token)
time.sleep(10)
get_energy_info(token, 735362)
token = refresh_token()
Local API User Guide
Features
Local API allows controlling Adax WiFi generation 2 heaters ( thermostat that have both WiFi and Bluetooth) directly over local network. Local API
provides the following functions:
- Connect heater to local WiFi network using Bluetooth Low Energy command.
- Get heater status that reports measured ambient temperature, and target temperature.
- Change target temperature.
Supported Devices
1. Adax WiFi gen. 2 thermostat (that have both WiFi and Bluetooth) with firmware 2.0.1.7+ version
Communication
Heaters listen on HTTPS (443 SSL port) and accepts REST API interface.
Authorisation
Each REST API call is authorised by passing Authorisation header with value “Basic ” + heater token. Token is assigned during heater registration.
Heaters use self-signed, self-generated SSL certificates.
Commands
No. |
Command |
HTTP GET Request |
HTTP POST Request Header |
Heater Response |
1. |
Get Status |
/api/?command=stat&time=<<current_unix_time>> |
Authorization: 'Basic ' + <<heater token >> |
{ 'targTemp': <<target>>, 'currTemp':<<current>> } |
2. |
Change Temperature |
/api/?command=set_target&time=<<current_unix_time>>&value<<target _temperature>> |
Authorization: 'Basic ' + <<heater token >> |
HTTP 200 |
Registration Procedure
Heater is registered on local network using a Bluetooth Low Energy communication. During the registration, device is provided with WiFi SSID, PSK, and secured heater token.
Steps:
- Heaters OK button is held until blue led starts blinking to initiated registration procedure.
- Generate/choose some random “Heater token” string value.
- Build heater registration command string “command=join&ssid=<<ssid>>&psk=<<psk>>&token=<<heater token>>
- Encode registration command as byte array (use ascii encoding)
- Scan for Bluetooth Low Energy device that advertises “3885cc10-7c18-4ad4-a48d-bf11abf7cb92” service.
- Get manufacturer data from Advertisement packet
- Check if manufacturer data byte 0 is equal to 5 (Heater device Type)
- Check if manufacturer data byte 1 two first bits are 0 (registered and managed flags are false)
- Check if manufacturer data bytes from 2 to 10 are not 0 (device_id > 0)
- If manufacturer data contains required values
- Connect to Device using Bluetooth Low Energy
- Register for “0000cc11-0000-1000-8000-00805f9b34fb” characteristics notifications.
- Split registration command bytes into multiple 17 byte chunks.
- Write each chunk to “0000cc11-0000-1000-8000-00805f9b34fb” characteristic ,with 1 byte header where 1 marks last chunk, 0 marks all other chunks.
- Wait for notification on “0000cc11-0000-1000-8000-00805f9b34fb” characteristic
- Get Byte array from notification.
- Check if notification data byte 0 contains 0 value (registration successful), or 1 value (WiFi connection failed)
- Get device ip address from notification data: bytes 1 to 4.
Code samples:
Controlling heater:
# install dependencies with: pip install bleak requests
HEATER_IP = '172.16.14.137' # replace with your heater IP from the prepare_demo
HEATER_TOKEN = '12345678' # replace with your secure password from the prepare_demo
def set_target_temperature(target_temperature):
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'set_target', 'time': int(time.time()), 'value': target_temperature }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
return requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'stat', 'time': int(time.time()) }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
response = requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
response_json = response.json()
target_temperature = response_json["targTemp"]
current_temperature = response_json["currTemp"]
print('Current temperature %d, target temperature: %d \n' % (current_temperature, target_temperature))
urllib3.disable_warnings() # heater uses self signed certificate
set_target_temperature(2300)
# controls heater
# install dependencies with: pip install bleak requests
import time
import requests
import urllib3
HEATER_IP = '172.16.14.137' # replace with your heater IP from the prepare_demo
HEATER_TOKEN = '12345678' # replace with your secure password from the prepare_demo
def set_target_temperature(target_temperature):
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'set_target', 'time': int(time.time()), 'value': target_temperature }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
return requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
def get_status():
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'stat', 'time': int(time.time()) }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
response = requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
response_json = response.json()
target_temperature = response_json["targTemp"]
current_temperature = response_json["currTemp"]
print('Current temperature %d, target temperature: %d \n' % (current_temperature, target_temperature))
urllib3.disable_warnings() # heater uses self signed certificate
set_target_temperature(2300)
get_status()
# controls heater
# install dependencies with: pip install bleak requests
import time
import requests
import urllib3
HEATER_IP = '172.16.14.137' # replace with your heater IP from the prepare_demo
HEATER_TOKEN = '12345678' # replace with your secure password from the prepare_demo
def set_target_temperature(target_temperature):
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'set_target', 'time': int(time.time()), 'value': target_temperature }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
return requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
def get_status():
url = 'https://' + HEATER_IP + '/api'
payload = { 'command': 'stat', 'time': int(time.time()) }
headers = { 'Authorization': 'Basic ' + HEATER_TOKEN }
response = requests.get(url, params = payload, headers = headers, timeout = 15, verify = False)
response_json = response.json()
target_temperature = response_json["targTemp"]
current_temperature = response_json["currTemp"]
print('Current temperature %d, target temperature: %d \n' % (current_temperature, target_temperature))
urllib3.disable_warnings() # heater uses self signed certificate
set_target_temperature(2300)
get_status()
Heater Registration:
# registers heater to your wifi network
# install dependencies with: pip install bleak requests
wifi_ssid = 'HEATER-TEST' # replace with your wifi name
wifi_psk = 'HEATER-PASS' # replace with your wifi psk
access_token = '12345678' # generate some strong password for your heater
UUID_ADAX_BLE_SERVICE = '3885cc10-7c18-4ad4-a48d-bf11abf7cb92'
UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND = '0000cc11-0000-1000-8000-00805f9b34fb'
# ignore registered, non heater devices
def is_available_device(manufacturer_data):
ADAX_DEVICE_TYPE_HEATER_BLE = 5
if manufacturer_data and len(manufacturer_data) >= 10:
type_id = manufacturer_data[0]
status_byte = manufacturer_data[1]
for byte in manufacturer_data[2:10]:
mac_id = mac_id * 256 + byte
registered = status_byte & (0x1 << 0)
managed = status_byte & (0x1 << 1)
return mac_id and type_id == ADAX_DEVICE_TYPE_HEATER_BLE and not registered and not managed
# scan and return available heater
async def scan_for_any_available_ble_device():
discovered = await bleak.discover(timeout = 60)
for discovered_item in discovered:
metadata = discovered_item.metadata
uuids = metadata['uuids'] if 'uuids' in metadata else None
if uuids and UUID_ADAX_BLE_SERVICE in uuids:
manufacturer_data_list = None
manufacturer_data = metadata['manufacturer_data'] if 'manufacturer_data' in metadata else None
first_bytes = next(iter(manufacturer_data))
other_bytes = manufacturer_data[first_bytes]
manufacturer_data_list = [first_bytes % 256, operator.floordiv(first_bytes, 256)] + list(other_bytes)
if is_available_device(manufacturer_data_list):
return discovered_item.address
# handle notifications for heater
def command_notification_handler(uuid, data):
BLE_COMMAND_STATUS_OK = 0
BLE_COMMAND_STATUS_INVALID_WIFI = 1
byte_list = None if not data else list(data)
status = None if not byte_list else byte_list[0]
if status == BLE_COMMAND_STATUS_OK:
ip = None if not byte_list or len(byte_list) < 5 else '%d.%d.%d.%d' % (byte_list[1], byte_list[2], byte_list[3], byte_list[4])
print('Heater Registered, use client_control_demo with IP: %s and token: %s' % (ip, access_token))
elif status == BLE_COMMAND_STATUS_INVALID_WIFI:
print('Invalid WiFi crendentials')
async def write_command(command_byte_list, client):
MAX_BYTES_IN_COMMAND_CHUNK = 17
byte_count = len(command_byte_list)
chunk_count = operator.floordiv(byte_count, MAX_BYTES_IN_COMMAND_CHUNK)
if chunk_count * MAX_BYTES_IN_COMMAND_CHUNK < byte_count:
while chunk_nr < chunk_count:
is_last = chunk_nr == (chunk_count - 1)
chunk_data_length = byte_count - sent_byte_count if is_last else MAX_BYTES_IN_COMMAND_CHUNK
chunk = [chunk_nr, 1 if is_last else 0] + command_byte_list[sent_byte_count:(sent_byte_count + chunk_data_length)]
await client.write_gatt_char(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, bytearray(chunk))
sent_byte_count += chunk_data_length
async def register_devices(loop):
print('Press and hold OK button of the heater until the blue led starts blinking')
device = await scan_for_any_available_ble_device()
client = bleak.BleakClient(device, loop = loop)
await client.start_notify(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, command_notification_handler)
ssid_encoded = urllib.parse.quote(wifi_ssid)
psk_encoded = urllib.parse.quote(wifi_psk)
access_token_encoded = urllib.parse.quote(access_token)
byte_list = list(bytearray('command=join&ssid=' + ssid_encoded + '&psk=' + psk_encoded + '&token=' + access_token_encoded, 'ascii'))
await write_command(byte_list, client)
loop = asyncio.get_event_loop()
loop.create_task(register_devices(loop))
# registers heater to your wifi network
# install dependencies with: pip install bleak requests
import bleak
import asyncio
import operator
import urllib
wifi_ssid = 'HEATER-TEST' # replace with your wifi name
wifi_psk = 'HEATER-PASS' # replace with your wifi psk
access_token = '12345678' # generate some strong password for your heater
UUID_ADAX_BLE_SERVICE = '3885cc10-7c18-4ad4-a48d-bf11abf7cb92'
UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND = '0000cc11-0000-1000-8000-00805f9b34fb'
# ignore registered, non heater devices
def is_available_device(manufacturer_data):
ADAX_DEVICE_TYPE_HEATER_BLE = 5
if manufacturer_data and len(manufacturer_data) >= 10:
type_id = manufacturer_data[0]
status_byte = manufacturer_data[1]
mac_id = 0
for byte in manufacturer_data[2:10]:
mac_id = mac_id * 256 + byte
registered = status_byte & (0x1 << 0)
managed = status_byte & (0x1 << 1)
return mac_id and type_id == ADAX_DEVICE_TYPE_HEATER_BLE and not registered and not managed
return False
# scan and return available heater
async def scan_for_any_available_ble_device():
discovered = await bleak.discover(timeout = 60)
if discovered:
for discovered_item in discovered:
metadata = discovered_item.metadata
uuids = metadata['uuids'] if 'uuids' in metadata else None
if uuids and UUID_ADAX_BLE_SERVICE in uuids:
manufacturer_data_list = None
manufacturer_data = metadata['manufacturer_data'] if 'manufacturer_data' in metadata else None
if manufacturer_data:
first_bytes = next(iter(manufacturer_data))
if first_bytes:
other_bytes = manufacturer_data[first_bytes]
manufacturer_data_list = [first_bytes % 256, operator.floordiv(first_bytes, 256)] + list(other_bytes)
if is_available_device(manufacturer_data_list):
return discovered_item.address
return None
# handle notifications for heater
def command_notification_handler(uuid, data):
BLE_COMMAND_STATUS_OK = 0
BLE_COMMAND_STATUS_INVALID_WIFI = 1
byte_list = None if not data else list(data)
status = None if not byte_list else byte_list[0]
if status == BLE_COMMAND_STATUS_OK:
ip = None if not byte_list or len(byte_list) < 5 else '%d.%d.%d.%d' % (byte_list[1], byte_list[2], byte_list[3], byte_list[4])
print('Heater Registered, use client_control_demo with IP: %s and token: %s' % (ip, access_token))
elif status == BLE_COMMAND_STATUS_INVALID_WIFI:
print('Invalid WiFi crendentials')
# send command to heater
async def write_command(command_byte_list, client):
MAX_BYTES_IN_COMMAND_CHUNK = 17
byte_count = len(command_byte_list)
chunk_count = operator.floordiv(byte_count, MAX_BYTES_IN_COMMAND_CHUNK)
if chunk_count * MAX_BYTES_IN_COMMAND_CHUNK < byte_count:
chunk_count += 1
sent_byte_count = 0
chunk_nr = 0
while chunk_nr < chunk_count:
is_last = chunk_nr == (chunk_count - 1)
chunk_data_length = byte_count - sent_byte_count if is_last else MAX_BYTES_IN_COMMAND_CHUNK
chunk = [chunk_nr, 1 if is_last else 0] + command_byte_list[sent_byte_count:(sent_byte_count + chunk_data_length)]
await client.write_gatt_char(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, bytearray(chunk))
sent_byte_count += chunk_data_length
chunk_nr += 1
# register heater
async def register_devices(loop):
print('Press and hold OK button of the heater until the blue led starts blinking')
device = await scan_for_any_available_ble_device()
if device:
client = bleak.BleakClient(device, loop = loop)
await client.connect()
await client.start_notify(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, command_notification_handler)
ssid_encoded = urllib.parse.quote(wifi_ssid)
psk_encoded = urllib.parse.quote(wifi_psk)
access_token_encoded = urllib.parse.quote(access_token)
byte_list = list(bytearray('command=join&ssid=' + ssid_encoded + '&psk=' + psk_encoded + '&token=' + access_token_encoded, 'ascii'))
await write_command(byte_list, client)
loop = asyncio.get_event_loop()
loop.create_task(register_devices(loop))
loop.run_forever()
# registers heater to your wifi network
# install dependencies with: pip install bleak requests
import bleak
import asyncio
import operator
import urllib
wifi_ssid = 'HEATER-TEST' # replace with your wifi name
wifi_psk = 'HEATER-PASS' # replace with your wifi psk
access_token = '12345678' # generate some strong password for your heater
UUID_ADAX_BLE_SERVICE = '3885cc10-7c18-4ad4-a48d-bf11abf7cb92'
UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND = '0000cc11-0000-1000-8000-00805f9b34fb'
# ignore registered, non heater devices
def is_available_device(manufacturer_data):
ADAX_DEVICE_TYPE_HEATER_BLE = 5
if manufacturer_data and len(manufacturer_data) >= 10:
type_id = manufacturer_data[0]
status_byte = manufacturer_data[1]
mac_id = 0
for byte in manufacturer_data[2:10]:
mac_id = mac_id * 256 + byte
registered = status_byte & (0x1 << 0)
managed = status_byte & (0x1 << 1)
return mac_id and type_id == ADAX_DEVICE_TYPE_HEATER_BLE and not registered and not managed
return False
# scan and return available heater
async def scan_for_any_available_ble_device():
discovered = await bleak.discover(timeout = 60)
if discovered:
for discovered_item in discovered:
metadata = discovered_item.metadata
uuids = metadata['uuids'] if 'uuids' in metadata else None
if uuids and UUID_ADAX_BLE_SERVICE in uuids:
manufacturer_data_list = None
manufacturer_data = metadata['manufacturer_data'] if 'manufacturer_data' in metadata else None
if manufacturer_data:
first_bytes = next(iter(manufacturer_data))
if first_bytes:
other_bytes = manufacturer_data[first_bytes]
manufacturer_data_list = [first_bytes % 256, operator.floordiv(first_bytes, 256)] + list(other_bytes)
if is_available_device(manufacturer_data_list):
return discovered_item.address
return None
# handle notifications for heater
def command_notification_handler(uuid, data):
BLE_COMMAND_STATUS_OK = 0
BLE_COMMAND_STATUS_INVALID_WIFI = 1
byte_list = None if not data else list(data)
status = None if not byte_list else byte_list[0]
if status == BLE_COMMAND_STATUS_OK:
ip = None if not byte_list or len(byte_list) < 5 else '%d.%d.%d.%d' % (byte_list[1], byte_list[2], byte_list[3], byte_list[4])
print('Heater Registered, use client_control_demo with IP: %s and token: %s' % (ip, access_token))
elif status == BLE_COMMAND_STATUS_INVALID_WIFI:
print('Invalid WiFi crendentials')
# send command to heater
async def write_command(command_byte_list, client):
MAX_BYTES_IN_COMMAND_CHUNK = 17
byte_count = len(command_byte_list)
chunk_count = operator.floordiv(byte_count, MAX_BYTES_IN_COMMAND_CHUNK)
if chunk_count * MAX_BYTES_IN_COMMAND_CHUNK < byte_count:
chunk_count += 1
sent_byte_count = 0
chunk_nr = 0
while chunk_nr < chunk_count:
is_last = chunk_nr == (chunk_count - 1)
chunk_data_length = byte_count - sent_byte_count if is_last else MAX_BYTES_IN_COMMAND_CHUNK
chunk = [chunk_nr, 1 if is_last else 0] + command_byte_list[sent_byte_count:(sent_byte_count + chunk_data_length)]
await client.write_gatt_char(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, bytearray(chunk))
sent_byte_count += chunk_data_length
chunk_nr += 1
# register heater
async def register_devices(loop):
print('Press and hold OK button of the heater until the blue led starts blinking')
device = await scan_for_any_available_ble_device()
if device:
client = bleak.BleakClient(device, loop = loop)
await client.connect()
await client.start_notify(UUID_ADAX_BLE_SERVICE_CHARACTERISTIC_COMMAND, command_notification_handler)
ssid_encoded = urllib.parse.quote(wifi_ssid)
psk_encoded = urllib.parse.quote(wifi_psk)
access_token_encoded = urllib.parse.quote(access_token)
byte_list = list(bytearray('command=join&ssid=' + ssid_encoded + '&psk=' + psk_encoded + '&token=' + access_token_encoded, 'ascii'))
await write_command(byte_list, client)
loop = asyncio.get_event_loop()
loop.create_task(register_devices(loop))
loop.run_forever()