IoT Master Library for Arduino and Raspberry Pi

Hello everyone in this Blog I want to tell you about the IoT Master Library. what are the benefits of using this library?

 

It was always challenging interfacing sensor and once you get the sensor working we need to process data in order to upload on a cloud server. in this process, there is a lot of encoding while you want to upload the sensor Data and Decoding while you want. to get the data back

Motivation to build this Module: - It was back in 2014 when I was not a good programmer I used to always struggle when it came to writing code. But I was very good at interfacing sensor we realized that in order to upload data on cloud its very tedious you have to write a lot of codes and also for different cloud use different API it was time-consuming and I was not able to find a single library which has all functionality that was the reason I decided to make one. it is 2019 I am a python developer so decided to contribute to the maker so that they can focus on making project than coding. we do all the hard work for you. Below is the list of cloud server that my library can be used for.

Entire Code can be found on Github.

# ============================================================================                             IOT MASTER LIBRARY==============================================================================
1 .  ThingSpeak                              (Read/ Write)2 .  AdafruitIO                              (Read/Write)3 .  IFTTT                                   (upload get notification etc)4 .  Get Latitude and Longitude              (get Lat and Long)5 .  Get Date and Time                       (get Date and Time in Format)6 .  Get Weather Details                     (Supports oopen Weather API )7 .  Supports Text to Speech                 (ONLY for MAC)8 .  Supports Arduino                        (Read and Write)9 .  Get Youtube Sub Count                   (Get Yout Sub Count)10.  Suports MQTT                            (Unstable Verison ! )
  ============================================================================

Here is a couple of examples of how to use the library.

# ==================== Thingspeak =========================================
#w_key = 'AZRMN9ZP5FKLXYLR'#r_key = '17FSCQ4FTX6V4VWM'#channel_id = 83234
#ob = Thingspeak(write_api_key=w_key, read_api_key=r_key, channel_id=channel_id)##ob.post_cloud(value1=temperature,value2=humidity)#print(ob.read_cloud(result=3))
# ==============================  IFTTT  =====================================
#iftt_k = 'Q-V9NBxTu3-7Xo98chRB_'#feed = 'sensor'
#ob1 = IfTTT(eventname=feed, key=iftt_k)#ob1.iftt_post(data1=temperature,data2=humidity)
# ================== Latitude and Longitude ===================================
#ob2 = Location()#lat,long = ob2.get_locations()#print('{} Latitude '.format(lat))#print("{} Longitude ".format(long))
# ================== Date and Time ==============================================
# b3 = DateandTime()# m_date, m_time = ob3.get_time_date()# print('{} Date'.format(m_date))# print("{} Time ".format(m_time))
# =============== Weather Details  ==============================================
#key ='ce45a4d1079e68c410cd42a3054d00e1'#city = 'Bridgeport'
#ob4 = Weather_details(key=key,city=city)#long, lat, humidity, wind_speed, wind_degree, sunrise, sunset,body = ob4.get_weather_data()#print(long, lat, humidity, wind_speed, wind_degree, sunrise, sunset,body)
# ========================== ADAFRUIT ===================================
# user_name='Scientist1995'# Aio_key = 'eb3673b510f34c2f8f7cda732edea251'# feed_name='temperature'
# ad = Adafruit_cloud(username=user_name, Aio_key=Aio_key)# ad.adafruit_send(feed_name=feed_name, data = 11)# print(ad.adafruit_feed_list)

now the Entire class or Library code can be found below and also on Github 

""""Author : Soumil shahEmail : soushah@my.bridgeport.edu"""
try:                                    # import the important library    from urllib import request
    from urllib.request import urlopen
    import threading                    # import threadding    import json                         # import json    import random                       # import random    import requests                     # import requests    import ssl
    import geocoder
    import datetime
    from twilio.rest import Client
    import serial
    import numpy as np
    from Adafruit_IO import Client
    import os
    from os import  system
    import paho.mqtt.publish as publish
    import sqlite3
    import paho.mqtt.publish as publish
    import urllib

except:
    print("No Library Found")


class Thingspeak(object):                       # define a class called Thingspeak
    def __init__(self, write_api_key = None, read_api_key=None, channel_id=0):

        """
        :param write_key:  takes a string of write api key        :param timer: can take integer values        """
        self.url = 'https://api.thingspeak.com/update?api_key='        self.write_key = write_api_key

        self.channel_id = channel_id
        self.read_api_key = read_api_key
        self.read_url = 'https://api.thingspeak.com/channels/{}/feeds.json?api_key='.format(channel_id)
        self.feild1 = []
        self.feild2 = []

    def post_cloud(self, value1, value2):
        try:
            """            :param value1: can be interger or float            :param value2: can be interger or float            :return: updated to cloud storage            """
            URL = self.url
            KEY = self.write_key

            HEADER = '&field1={}&field2={}'.format(str(value1), str(value2))

            NEW_URL = str(URL) + str(KEY) + str(HEADER)
            print(NEW_URL)

            context = ssl._create_unverified_context()

            data = request.urlopen(NEW_URL,context=context)
            print(data)
        except:
            print('could not post to the cloud server ')

    def read_cloud(self, result=2):
        try:
            """            :param result: how many data you want to fetch accept interger            :return: Two List which contains Sensor data            """
            URL_R = self.read_url
            read_key = self.read_api_key
            header_r ='&results={}'.format(result)

            new_read_url = URL_R + read_key + header_r

            data = requests.get(new_read_url).json()

            field1 = data['feeds']

            for x in field1:
                self.feild1.append(x['field1'])
                self.feild2.append(x['field2'])

            return self.feild1, self.feild2
        except:
            print('read_cloud failed !!!! ')


class IfTTT(object):

    def __init__(self, eventname='', key=''):

        self.eventname = eventname
        self.Key = key
        self.Url = 'https://maker.ifttt.com/trigger/{}/with/key/'.format(self.eventname)
        self.New_Url =self.Url + self.Key
        print(self.New_Url)

    def iftt_post(self, data1=10, data2=11):
        try:
            """            :param data1:  pass your sensor value only integer            :param data2: pass your data as interger only            :return:      True if it was successful            """
            URl = self.New_Url
            Key = self.Key
            payload = {'value1': data1,                       'value2': data2}


            requests.post(self.New_Url, data=payload)
            print("Done posted on IFTTT")

            return True        except:
            print('failed to post to cloud sever ! ')


class Location(object):

    def __init__(self):
        pass
    def get_locations(self):

        """        :return: Lat and Long        """        try:
            g = geocoder.ip('me')
            my_string=g.latlng
            longitude=my_string[0]
            latitude=my_string[1]

            return longitude,latitude
        except:
            print('Error make sure you have Geo-Coder Installed ')


class DateandTime(object):

    def __init__(self):
        pass    @staticmethod    def get_time_date():
        try:
            """            :return:  date and time            """            my = datetime.datetime.now()
            data_time = '{}:{}:{}'.format(my.hour,my.minute,my.second)
            data_date = '{}/{}/{}'.format(my.day,my.month,my.year)
            return data_date,data_time
        except:
            print('could now get date and time ')


class Weather_details(object):

    def __init__(self,key='', city = ''):
        self.city = city
        self.key = key


    def get_weather_data(self):
        try:
            city = self.city
            key = self.key
            URL='http://api.openweathermap.org/data/2.5/weather?appid={}&q={}'.format(key,city)
            print(URL)

            data = requests.get(URL).json()
            long = data['coord']['lon']
            lat = data['coord']['lat']
            humidity = data['main']['humidity']
            wind_speed = data['wind']['speed']
            wind_degree = data['wind']['deg']
            sunrise = data['sys']['sunrise']
            sunset = data['sys']['sunset']

            m1 = data['weather'][0]['description']
            m2 = data['weather'][0]['main']
            body = '{} {}'.format(m1, m2)

            return long, lat, humidity, wind_speed, wind_degree, sunrise, sunset,body

        except:
            print('Error occured ')


class Arduino(object):

    def __init__(self,comport='com3', baudrate='9600'):
        self.comport=comport
        self.baudrate=baudrate

    def read_data(self):
        try:
            """                :return: data after reading from serial object            """
            arduino = serial.Serial(self.comport, self.baudrate,timeout=1)
            while arduino.inWaiting():
                data = arduino.readline()         # read line by line data                data = data.decode('ascii')
                return data
        except:
            print("please check your com port and baud rate ")

    def write_data(self,data='1'):
        try:

            """                :param data: pass string 1 or 0 or any character and write if statment in Arduino            to compare and trigger events            :return: NONE            """            arduinodata = serial.Serial(self.comport, self.baudrate,timeout=1)
            data_send = data.encode('utf-8')
            arduinodata.write(data_send)
        except:
            print("cannot write data please check com port and baud rate ")


class Adafruit_cloud():

    def __init__(self, username='', Aio_key=''):
        self.username=username
        self.Aio_key = Aio_key
        self.aio = Client(self.username,self.Aio_key)
        self.feed_name = []

    def adafruit_send(self,feed_name='',data=''):
        try:
            sensor = self.aio.feeds(feed_name)
            self.aio.send_data(sensor.key,data)
            print('Data was uploaded ')
        except:
            print("cannot send Data !")


    def adafruit_get(self, feedname=''):
        try:
            data = self.aio.receive(feedname)
            return data.value
        except:
            print('cannot get data ! ')

    def adafruit_feed_list(self):
        try:

            feeds = self.aio.feeds()
            for f in feeds:
                print(f)
                self.feed_name.append(f.name)

            return self.feed_name
        except:
            print("cannot get feed list ! ")


class TextSpeech_Mac(object):

    def __init__(self):
        """        This script works for mac only        """        pass
    def speak(self,text):
        try:

            """            :param text:  Takes String as Input            :return: Return None            """            system("say {}".format(text))
        except:
            print("This works  for mac only ! ")


class Mqtt(object):

    def __init__(self,data):
        self.data= data
        self.hostname="test.mosquitto.org"        self.topic = ''
    def post(self):
        """        POst teh         :return:        """        publish.single(self.topic, "{}".format(self.data), hostname=self.hostname)


class YoutubeSub(object):

    def __init__(self,name ='',google_key=''):
        """
        :param name: Name of Youtube Person , Google Cloud API key        :param google_key:        """        self.name = name
        self.google_key = google_key
        self.context = ssl._create_unverified_context()


    def get_subscriber(self):

        """        :return: Youtube Sub Count        """        self.url = 'https://www.googleapis.com/youtube/v3/channels?part=statistics&forUsername=' \
                   + self.name+'&key='+self.google_key

        data = urllib.request.urlopen(self.url, context=self.context).read()
        subs = json.loads(data)["items"][0]["statistics"]["subscriberCount"]
        subs = int(subs)
        return subs