Python code to access basic cayenne API

#1

this python code enables you to access the basic cayenne API request.

Requirements:
Install request https://2.python-requests.org//en/master/user/install/#install

on running the code, it will prompt you:

  1. For a new user or an existing user.
    Enter 1 for new and 2 for an existing user.
    if a new user then it will prompt to enter:
  • client_id ----> APP key from the cayenne dashboard.
  • client_secret ----> APP secret from the cayenne dashboard.
  • username ----> Email_id for your account.
  • password ----> Password of your account.
  • mqtt_username ----> Your account MQTT username.
  • mqtt_password ----> Your account MQTT password.
    Next, it will get the access_token and mqtt_token.
    All these credentials will be stored in the credentials.ini.
    If an existing user then this info will be gathered from credentials.ini
  1. Enter the number to get the info.
    1 = Get new Access Token
    2 = Get new Mqtt Token
    3 = Get latest sensor data
    4 = Get sensor history data
    5 = Publish data

Get new Access Token
This function gets the access_token for the https://platform.mydevices.com/ endpoint:

def get_access_token(client_id, client_secret, username, password):
        response = requests.post('https://accounts.mydevices.com/auth/realms/cayenne/protocol/openid-connect/token',
                data={'grant_type':'password',
                'client_id':client_id,
                'client_secret':client_secret,
                'username':username,
                'password':password },
                headers={'content-type':'application/x-www-form-urlencoded'})
        payload = response.json()
        access_token = payload['access_token']
        print('Access_Token = ' + access_token)
        config['DEFAULT']['access_token'] = access_token
        with open('credentials.ini', 'w') as configfile:
                config.write(configfile)
        return (access_token)

Get new MQTT Token
This function gets the mqtt_token for the https://api.mydevices.com

def get_mqtt_token(mqtt_username, mqtt_password ):
        string = mqtt_username + ':' + mqtt_password
        print(string)
        data = base64.b64encode(string.encode())
        mqtt_token = data.decode("utf-8")
        config['DEFAULT']['mqtt_token'] = mqtt_token
        with open('credentials.ini', 'w') as configfile:
                config.write(configfile)
        return (mqtt_token)

Get latest sensor data
This function gets the latest sensor data published for the respective sensor_id.
You will need to add the device_id and the sensor_id for the widget you need to get the latest data.

def get_latest_history(access_token, device_id, sensor_id):
        bearer = 'Bearer '+ access_token
        URL = 'https://platform.mydevices.com/v1.1/telemetry/'+device_id+'/sensors/'+sensor_id+'/summaries?type=latest'
        response = requests.get(URL,
                headers={'authorization': bearer},)
        payload = response.json()
        data = payload[0]['v']
        print ('Data history = %d' %data)

Get sensor history data
This function gets the history data of the respective sensor_id.
You will need to add the device_id and the sensor_id for the widget you need to get the latest data.
Also the duration of the historical period.

def get_history(access_token, device_id, sensor_id):
        list = []
        duration = input("Enter the history duration. example: 1month or 2month or 3month and so on ")
        bearer = 'Bearer '+ access_token
        URL = 'https://platform.mydevices.com/v1.1/telemetry/'+device_id+'/sensors/'+sensor_id+'/summaries?endDateLatest=true&type='+duration
        response = requests.get(URL,
                headers={'authorization': bearer},)
        history_list = response.json()
        for item in history_list:
                #print(item['v'])
                list.append(item['v'])
        print(list)
        print()

Publish data
This function is used to publish data.
You need the add the following when prompted:

  • device_id ----> the device_id to which you want to publish data.
  • channel ----> MQTT channel of the widget to publish data.
  • value ----> data value you need to publish.
  • unit and type ----> Data types for Cayenne MQTT API

def publish_data(mqtt_token):
            device_id  = input ("Enter your Device_ID:  ")
            URL = 'https://api.mydevices.com/things/'+device_id+'/data'
            channel  = int(input("Enter channel:  "))
            value  = int(input("Enter value:  "))
            unit = input ("Enter unit:  ")
            type  = input ("Enter type:  ")
            basic = 'Basic '+mqtt_token
            headers= {'authorization': basic,
    		'content-type': 'application/json'}
            data = [{'value': value,
                     'channel': channel,
                      'unit': unit,
                      'type' : type
                    }]
            response = requests.post(URL, json=data, headers=headers)
            print(response)

You can edit the respective function for your need.
The code needs improvements and will be working on it or someone can contribute to it.

import requests

import base64

import configparser
config = configparser.ConfigParser()


def get_mqtt_token(mqtt_username, mqtt_password ):
        string = mqtt_username + ':' + mqtt_password
        print(string)
        data = base64.b64encode(string.encode())
        mqtt_token = data.decode("utf-8")
        config['DEFAULT']['mqtt_token'] = mqtt_token
        with open('credentials.ini', 'w') as configfile:
                config.write(configfile)
        return (mqtt_token)

def get_access_token(client_id, client_secret, username, password):
        response = requests.post('https://accounts.mydevices.com/auth/realms/cayenne/protocol/openid-connect/token',
                data={'grant_type':'password',
                'client_id':client_id,
                'client_secret':client_secret,
                'username':username,
                'password':password },
                headers={'content-type':'application/x-www-form-urlencoded'})
        payload = response.json()
        access_token = payload['access_token']
        print('Access_Token = ' + access_token)
        config['DEFAULT']['access_token'] = access_token
        with open('credentials.ini', 'w') as configfile:
                config.write(configfile)
        return (access_token)

def get_latest_history(access_token, device_id, sensor_id):
        bearer = 'Bearer '+ access_token
        URL = 'https://platform.mydevices.com/v1.1/telemetry/'+device_id+'/sensors/'+sensor_id+'/summaries?type=latest'
        response = requests.get(URL,
                headers={'authorization': bearer},)
        payload = response.json()
        data = payload[0]['v']
        print ('Data history = %d' %data)

def get_user_credentials():
        client_id  = input ("Enter your client_id:  ")
        client_secret  = input ("Enter your client_Secret:  ")
        username = input ("Enter your Username:  ")
        password  = input ("Enter your Password:  ")
        mqtt_username = input("Enter your MQTT Username")
        mqtt_password = input("Enter your MQTT Password")
        config['DEFAULT']['client_id'] = client_id
        config['DEFAULT']['client_secret'] = client_secret
        config['DEFAULT']['username'] = username
        config['DEFAULT']['password'] = password
        config['DEFAULT']['mqtt_username'] = mqtt_username
        config['DEFAULT']['mqtt_password'] = mqtt_password

        with open('credentials.ini', 'w') as configfile:
                config.write(configfile)
        return client_id, client_secret, username, password, mqtt_username, mqtt_password;


def get_device_info():
        device_id  = input ("Enter your Device_ID:  ")
        sensor_id  = input ("Enter your Sensor_ID:  ")
        return device_id, sensor_id;

def get_client_data():
        config.read('credentials.ini')
        client_id = config['DEFAULT']['client_id']
        client_secret = config['DEFAULT']['client_secret']
        username = config['DEFAULT']['username']
        password = config['DEFAULT']['password']
        access_token = config['DEFAULT']['access_token']
        mqtt_token = config['DEFAULT']['mqtt_token']
        mqtt_username = config['DEFAULT']['mqtt_username']
        mqtt_password = config['DEFAULT']['mqtt_password']
        return client_id, client_secret, username, password, access_token, mqtt_token, mqtt_username, mqtt_password;

def get_history(access_token, device_id, sensor_id):
        list = []
        duration = input("Enter the history duration. example: 1month or 2month or 3month and so on ")
        bearer = 'Bearer '+ access_token
        URL = 'https://platform.mydevices.com/v1.1/telemetry/'+device_id+'/sensors/'+sensor_id+'/summaries?endDateLatest=true&type='+duration
        response = requests.get(URL,
                headers={'authorization': bearer},)
        history_list = response.json()
        for item in history_list:
                #print(item['v'])
                list.append(item['v'])
        print(list)
        print()

def publish_data(mqtt_token):
        device_id  = input ("Enter your Device_ID:  ")
        URL = 'https://api.mydevices.com/things/'+device_id+'/data'
        channel  = int(input("Enter channel:  "))
        value  = int(input("Enter value:  "))
        unit = input ("Enter unit:  ")
        type  = input ("Enter type:  ")
        basic = 'Basic '+mqtt_token
        headers= {'authorization': basic,
		'content-type': 'application/json'}
        data = [{'value': value,
                 'channel': channel,
                  'unit': unit,
                  'type' : type
                }]
        response = requests.post(URL, json=data, headers=headers)
        print(response)
x = 0

while True:
        if x == 0:
                print ("/////////////////////////////////////////////////")
                print ("")
                print ("WELCOME TO CAYENNE API")
                print (" 1 = For new number ")
                print (" 2 = Existing USer")
                user_input = int(input("Enter the appropriate number "))
                if user_input == 1 :
                        client_id, client_secret, username, password = get_user_credentials()
                        print ("Welcome" + username)
                        access_token = get_access_token(client_id, client_secret, username, password)
#                        mqtt_token = get_mqtt_token(username, password)
                elif  user_input == 2:
                        client_id, client_secret, username, password, access_token, mqtt_token, mqtt_username, mqtt_password = get_client_data()
                        print ("Welcome back " + username)
                x = 1

        if x == 1:
                print ("/////////////////////////////////////////////////")
                print (" 1 = Get new Access Token")
                print (" 2 = Get new Mqtt Token")
                print (" 3 = Get latest sensor data")
                print (" 4 = Get sensor history data")
                print (" 5 = Publish data")
                get_input = int(input("Enter the appropriate number "))
                if get_input == 1:
                        access_token = get_access_token(client_id, client_secret, username, password)
                elif get_input == 2:
                        mqtt_token = get_mqtt_token(mqtt_username, mqtt_password)
                elif get_input == 3:
                        device_id, sensor_id = get_device_info()
                        get_latest_history(access_token, device_id, sensor_id)
                elif get_input == 4:
                        device_id, sensor_id = get_device_info()
                        get_history(access_token, device_id, sensor_id)
                elif get_input == 5:
                        publish_data(mqtt_token)
2 Likes
How to send synchronized data to mysql from cayenne