My first attempt : pluging SMARTWATER Gardena/Huskvarna

zicht · Post by **zicht** » Monday 27 January 2025 7:51

import asyncio, ssl, websockets
import websocket
import datetime
from threading import Thread
import time
import sys
import requests
import json

# account specific values
CLIENT_ID = 'Your client ID  huskvarna dev'
CLIENT_SECRET = 'Your client secret huskvarna dev'
API_KEY = 'Your API key huskvarna dev'

# other constants : remember rate limit 700/week or 100/day
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()
		
		# example : store all messages values as text in domoticz, replace IP & IDX
        
		response = requests.get(
            'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue=' + message, 
            verify=False
            )
            # end example
        
        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"]) 

           #status messages (can be deleted0):
		   #IF you want to have diff values for every equipment that can be done like this,
		   #remember this is an example and not working as is, you need to alter IP, PORT,IDx
		   #also you need to find you unique id for every device in you smart water system
		   #this can be found in previous text message. In my case 2 valves and 2 sensors
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue=Sensor: '+ message, 
                       verify=False 
              )
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue=Valve: '+ message, 
                       verify=False 
              )
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue=Valve: '+ message, 
                       verify=False 
              )
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue=Sensor: '+ message, 
                       verify=False 
              )
             #end can be deleted0
           
        if thisdict["type"]=="SENSOR" :
           message = "soilHum: " + str(thisdict["attributes"]["soilHumidity"]["value"]) 
           message = message + " soilTemp:" + str(thisdict["attributes"]["soilTemperature"]["value"]) 

           #Sensor messages (start can be deleted1) :
		   #IF you want to have diff values for every equipment that can be done like this,
		   #remember this is an example and not working as is, you need to alter IP, PORT,ID
		   #also you need to find you unique id for every device in you smart water system
		   #this can be found in previous text message. In my case 2 valves and 2 sensors

           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue='+str(thisdict["attributes"]["soilHumidity"]["value"]), 
                       verify=False
              )
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue='+str(thisdict["attributes"]["soilTemperature"]["value"]), 
                       verify=False 
              )
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
               response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue='+str(thisdict["attributes"]["soilHumidity"]["value"]), 
                       verify=False
              )
              response = requests.get(
                       'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=udevice&idx=IDX&nvalue=0&svalue='+str(thisdict["attributes"]["soilTemperature"]["value"]), 
                       verify=False 
              )
         #end of can be deleted1
           
        if thisdict["type"]=="VALVE" :
           #print("naam " + str(thisdict["attributes"]["name"]["value"]))
           message=str(thisdict["attributes"]["activity"]["value"]) 

           #Valve messages (start can be deleted 2):
		   #IF you want to have diff values for every equipment that can be done like this,
		   #remember this is an example and not working "as is", you need to alter IP, PORT,IDx
		   #also you need to find you unique id for every device in you smart water system
		   #this can be found in previous text message. In my case 2 valves and 2 sensors

           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              if message=="CLOSED":
                    response = requests.get(
                        'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=switchlight&idx=IDX&switchcmd=Off', 
                        verify=False
                    )
              else:
                    response = requests.get(
                        'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=switchlight&idx=IDX&switchcmd=On', 
                        verify=False
                    )
           if thisdict["id"]=="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX":
              if message=="CLOSED":
                    response = requests.get(
                        'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=switchlight&idx=IDX&switchcmd=Off', 
                        verify=False
                    )
              else:
                    response = requests.get(
                        'http://192.168.XXX.XXX:XXXX/json.htm?type=command&param=switchlight&idx=IDX&switchcmd=On', 
                        verify=False
                    )
                #(end can be deleted 2)
           
    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, verify=False)
            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=60, ping_timeout=3, sslopt={"cert_reqs": ssl.CERT_NONE})
        except Exception as e:
            gc.collect()
            print("Websocket connection Error  : {0}".format(e))                    
        print("Reconnecting websocket  after 60 sec")
        time.sleep(60)

I am not good at coding at all. Just took the example from the api discription and started fiddling until i could receive something in domoticz.
It's now running for 2 weeks as expected. Next step transform into a real plugin.

You need to have setup you client-id client-secret and api key on the huskvarna dev site.
The above works with json url to domoticz. If your familiar to python is should not be to complex to understand.
It starts a websocket and the messages get received in domoticz. Added some examples how I put values into domoticz. You can delete or alter those lines.

Edit : there are strickt rate limits on the api, 100/day or 700/week --> if you exceed one of these you only receive errors, Once the websocket is running that counts as one(1) on every start. The websocket will restart when the connection is broken. I hit the rate limit a lot during my experiments due to restarting a lot, now its running constantly far within the rate limits

Now i have to find out how the plugin framework can handle this

My first attempt : pluging SMARTWATER Gardena/Huskvarna

My first attempt : pluging SMARTWATER Gardena/Huskvarna

Who is online