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
