Help needed Husqvara/Gardena Smart Water
Posted: Wednesday 12 July 2023 8:30
Hi,
I am trying to control a Gardena Smartwater valve via SmartWater API, but I am stuck at some Python code and i do not understand my mistake.
Currently i am able via a webhook to get reallife status :
The above code works and i get status for my sensors and valves in domoticz uservars. Part1 : get status of devices is finished, but needs cleanup to make it interact nicely with domoticz. So all works on this part for now.
Problem is on controlling the water valves :
Again i get logged in but no control happens, expected behavior is activate all valves for 60 seconds. But all i get is 400, incorrect request .
I cannot find where my mistake is made....
What am i missing ?
<EDIT> My domoticz is not conncted to internet, so this code is running on another machine Using python 3.7.9 32bit, the plan is when all works to combine it to a plugin, but i am not a coder at all, so thats another project, let first make it work this way.
(Many thanks for reading so far...)
I am trying to control a Gardena Smartwater valve via SmartWater API, but I am stuck at some Python code and i do not understand my mistake.
Currently i am able via a webhook to get reallife status :
Code: Select all
import websocket
import datetime
from threading import Thread
import time
import sys
import requests
import json
# account specific values
CLIENT_ID = 'f48eae05-3d76-4x3b-ay90-<FICTIEF>'
CLIENT_SECRET = '4d4ce773-cx5b-4yeb-9781-<FICTIEF>'
API_KEY = 'f48eae05-3d76-4x3b-ay90-<FICTIEF>'
# other constants
AUTHENTICATION_HOST = 'https://api.authentication.husqvarnagroup.dev'
SMART_HOST = 'https://api.smart.gardena.dev'
class Client:
def restart():
import sys
print("argv was",sys.argv)
print("sys.executable was", sys.executable)
print("restart now")
import os
os.execv(sys.executable, ['python'] + sys.argv)
def on_message(self, ws, message):
x = datetime.datetime.now()
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=udevice&idx=6204&nvalue=0&svalue=' + message,
)
ws.messages_count+=1
print("--")
print("msg ", x.strftime("%H:%M:%S") + " " + str(ws.messages_count))
print(message)
sys.stdout.flush()
thisdict = dict(json.loads(message))
if thisdict["type"]=="COMMON" :
message = "bat:" + str(thisdict["attributes"]["batteryLevel"]["value"])
message = message + " lev:" + str(thisdict["attributes"]["rfLinkLevel"]["value"])
message = message + " stat:" + str(thisdict["attributes"]["rfLinkState"]["value"])
if thisdict["id"]=="fe59933d-9864-4711-90a8-ad3ef78a1af0":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_sensor_achter_2&vtype=2&vvalue=' + message,
)
if thisdict["id"]=="d25221cc-bca6-401c-8143-75c67a2efdba":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_valve_achter_1&vtype=2&vvalue=' + message,
)
if thisdict["id"]=="5d84502e-7d3c-4f6b-bb9d-<FICTIEF>'":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_valve_voor_1&vtype=2&vvalue=' + message,
)
if thisdict["id"]=="97d5a2fa-9de3-404a-85ba-<FICTIEF>'":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_sensor_voor_2&vtype=2&vvalue=' + message,
)
if thisdict["type"]=="SENSOR" :
message = "soilHum: " + str(thisdict["attributes"]["soilHumidity"]["value"])
message = message + " soilTemp:" + str(thisdict["attributes"]["soilTemperature"]["value"])
if thisdict["id"]=="fe59933d-9864-4711-<FICTIEF>':
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_sensor_achter_1&vtype=2&vvalue=' + message,
)
if thisdict["id"]=="97d5a2fa-9de3-404a-85ba-<FICTIEF>'":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_sensor_voor_1&vtype=2&vvalue=' + message,
)
if thisdict["type"]=="VALVE" :
#print("naam " + str(thisdict["attributes"]["name"]["value"]))
message=str(thisdict["attributes"]["activity"]["value"])
if thisdict["id"]=="d25221cc-bca6-401c-8143-<FICTIEF>'":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_valve_achter_2&vtype=2&vvalue=' + message,
)
if thisdict["id"]=="5d84502e-7d3c-4f6b-bb9d-<FICTIEF>'":
response = requests.get(
'http://192.168.0.200:8080/json.htm?type=command¶m=updateuservariable&vname=Gardena_valve_voor_2&vtype=2&vvalue=' + message,
)
def on_error(self, ws, error):
x = datetime.datetime.now()
print("error ", x.strftime("%H:%M:%S"))
print(error)
def on_close(self, ws, close_status_code, close_msg):
self.live = False
x = datetime.datetime.now()
print("closed ", x.strftime("%H:%M:%S,%f"))
print("### closed ###")
if close_status_code:
print("status code: "+str(close_status_code))
if close_msg:
print("status message: "+str(close_msg))
client.restart
def on_open(self, ws):
x = datetime.datetime.now()
print("connected ", x.strftime("%H:%M:%S,%f"))
print("### connected ###")
self.live = True
def run(*args):
while self.live:
time.sleep(1)
Thread(target=run).start()
def format(response):
formatted = [response.url, "%s %s" % (response.status_code, response.reason)]
for k,v in response.headers.items():
formatted.append("%s: %s" % (k, v))
formatted.append("")
formatted.append(r.text)
return "\n".join(formatted)
if __name__ == "__main__":
while True:
try:
payload = {'grant_type': 'client_credentials', 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET}
print("Logging into authentication system...")
r = requests.post(f'{AUTHENTICATION_HOST}/v1/oauth2/token', data=payload)
assert r.status_code == 200, format(r)
auth_token = r.json()["access_token"]
#print("\nLogged in auth_token=(%s)" % auth_token)
headers = {
"Content-Type": "application/vnd.api+json",
"x-api-key": API_KEY,
"Authorization": "Bearer " + auth_token
}
print("\n### get locations ###")
r = requests.get(f'{SMART_HOST}/v1/locations', headers=headers)
assert r.status_code == 200, format(r)
assert len(r.json()["data"]) > 0, 'location missing - user has not setup system'
location_id = r.json()["data"][0]["id"]
#print("\nLocationId=(%s)" % location_id)
payload = {
"data": {
"type": "WEBSOCKET",
"attributes": {
"locationId": location_id
},
"id": "does-not-matter"
}
}
print("\ngetting websocket ID...")
r = requests.post(f'{SMART_HOST}/v1/websocket', json=payload, headers=headers)
assert r.status_code == 201, format(r)
print("\nWebsocket ID obtained, connecting...")
response = r.json()
websocket_url = response["data"]["attributes"]["url"]
websocket.enableTrace(False) #True om raw output in display te hebben
client = Client()
ws = websocket.WebSocketApp(
websocket_url,
on_message=client.on_message,
on_error=client.on_error,
on_close=client.on_close)
ws.messages_count = 0 # Create a counter and an empty list to store messages
ws.on_open = client.on_open
ws.run_forever(ping_interval=120, ping_timeout=3)
except Exception as e:
gc.collect()
print("Websocket connection Error : {0}".format(e))
print("Reconnecting websocket after 60 sec")
time.sleep(60)
Problem is on controlling the water valves :
Code: Select all
import websocket
import datetime
from threading import Thread
import time
import sys
import requests
import json
# account specific values
CLIENT_ID = 'f48eae05-3d76-4x3b-ay90-<FICTIEF>'
CLIENT_SECRET = '4d4ce773-cx5b-4yeb-9781-<FICTIEF>'
# other constants
AUTHENTICATION_HOST = 'https://api.authentication.husqvarnagroup.dev'
SMART_HOST = 'https://api.smart.gardena.dev'
def format(response):
formatted = [response.url, "%s %s" % (response.status_code, response.reason)]
for k,v in response.headers.items():
formatted.append("%s: %s" % (k, v))
formatted.append("")
formatted.append(r.text)
return "\n".join(formatted)
if __name__ == "__main__":
payload = {'grant_type': 'client_credentials', 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET}
print("Logging into authentication system...")
r = requests.post(f'{AUTHENTICATION_HOST}/v1/oauth2/token', data=payload)
assert r.status_code == 200, format(r)
auth_token = r.json()["access_token"]
print("\nLogged in auth_token=(%s)" % auth_token)
headers = {
"Content-Type": "application/vnd.api+json",
"x-api-key": API_KEY,
"Authorization": "Bearer " + auth_token
}
print("\n### get locations ###")
r = requests.get(f'{SMART_HOST}/v1/locations', headers=headers)
assert r.status_code == 200, format(r)
assert len(r.json()["data"]) > 0, 'location missing - user has not setup system'
location_id = r.json()["data"][0]["id"]
print("\nLocationId=(%s)" % location_id)
r = requests.get('{}/v1/locations/{}'.format(SMART_HOST, location_id), headers=headers)
assert r.status_code == 200, r
assert len(r.json()["included"][0]['relationships']['services']['data']) > 0, 'no devices setup'
devices = r.json()["included"][0]['relationships']['services']['data']
for device in devices:
print(device['type'])
if(device['type'] == "VALVE"):
payload = {
"data": {
"type": "VALVE_CONTROL",
"attributes": {
"command": "START_SECONDS_TO_OVERRIDE",
"seconds": "60"
},
"id": "does-not-matter"
}
}
r = requests.put('{}/v1/command/{}'.format(SMART_HOST, device['id']), data=payload, headers=headers)
format(r)
print(r)
print("Done")
I cannot find where my mistake is made....
What am i missing ?
<EDIT> My domoticz is not conncted to internet, so this code is running on another machine Using python 3.7.9 32bit, the plan is when all works to combine it to a plugin, but i am not a coder at all, so thats another project, let first make it work this way.
(Many thanks for reading so far...)