Devesh Kumar Rai
Published © GPL3+

Anomaly Detection and Temperature Prediction Using Bolt IoT

Use Bolt IoT Platform and LM35 temperature sensor to make your own anomaly detection device at a very low cost.

BeginnerFull instructions provided909
Anomaly Detection and Temperature Prediction Using Bolt IoT

Things used in this project

Hardware components

Bolt WiFi Module
Bolt IoT Bolt WiFi Module
×1
Temperature Sensor
Temperature Sensor
×1
Jumper wires (generic)
Jumper wires (generic)
×1
Micro-USB to USB Cable (Generic)
Micro-USB to USB Cable (Generic)
×1

Software apps and online services

Bolt Cloud
Bolt IoT Bolt Cloud
Digital Ocean

Story

Read more

Code

Code for threshold checking and anomaly detection

Python
Run this code on your server for threshold checking and anomaly detection
import conf, email_conf, json, time, math, statistics
from boltiot import Sms, Email, Bolt



def compute_bounds(history_data,frame_size,factor):
    if len(history_data)<frame_size :
        return None

    if len(history_data)>frame_size :
        del history_data[0:len(history_data)-frame_size]
    Mn=statistics.mean(history_data)
    Variance=0
    for data in history_data :
        Variance += math.pow((data-Mn),2)
    Zn = factor * math.sqrt(Variance / frame_size)
    High_bound = history_data[frame_size-1]+Zn
    Low_Bound = history_data[frame_size-1]-Zn
    return [High_bound,Low_Bound]

minimum_limit = 4  
maximum_limit = 7

mybolt = Bolt(email_conf.API_KEY, email_conf.DEVICE_ID)
mailer = Email(email_conf.MAILGUN_API_KEY, email_conf.SANDBOX_URL, email_conf.SENDER_EMAIL, email_conf.RECIPIENT_EMAIL)
sms = Sms(conf.SSID, conf.AUTH_TOKEN, conf.TO_NUMBER, conf.FROM_NUMBER)

history_data=[]

while True:
    response = mybolt.analogRead('A0')
    data = json.loads(response)
    
    if data['success'] != '1':
        print("There was an error while retriving the data.")
        print("This is the error:"+data['value'])
        time.sleep(10)
        continue

    
    print ("The sensor value is "+(data['value']))
    sensor_value=0
    try:
        sensor_value = int(data['value'])
        
    
    except Exception as e:
        print("There was an error while parsing the response: ",e)
        continue

    bound = compute_bounds(history_data,email_conf.FRAME_SIZE,email_conf.MUL_FACTOR)
    if not bound:
        required_data_count=email_conf.FRAME_SIZE-len(history_data)
        print("Not enough data to compute Z-score. Need ",required_data_count," more data points")
        history_data.append(int(data['value']))
        time.sleep(10)
        continue

    try:
        if sensor_value > maximum_limit or sensor_value < minimum_limit:
            response = mailer.send_email("Alert", "The Current temperature is beyond the threshold ")
        if sensor_value > bound[0] :
            print ("The temperature increased suddenly. Sending Sms.")
            response = sms.send_sms("Someone opened the chamber")
            print("This is the response ",(response))
        elif sensor_value < bound[1]:
            print ("The temperature decreased suddenly. Sending an email.")
            response = mailer.send_email("Someone opened the chamber")
            print("This is the response ",response)
        history_data.append(sensor_value);
    except Exception as e:
        print ("Error",e)
    time.sleep(10)

conf.py

Python
include this file (import conf) in main code for sms sending
SSID = 'You can find SSID in your Twilio Dashboard' 
AUTH_TOKEN = 'You can find  on your Twilio Dashboard' 
FROM_NUMBER = 'This is the no. generated by Twilio. You can find this on your Twilio Dashboard'
TO_NUMBER = 'This is your number. Make sure you are adding +91 in beginning'
API_KEY = 'This is your Bolt Cloud account API key'
DEVICE_ID = 'This is the ID of your Bolt device'
FRAME_SIZE = 10
MUL_FACTOR = 6

email_conf.py

Python
include this file (import email_conf) for recieving emails
MAILGUN_API_KEY = 'This is the private API key which you can find on your Mailgun Dashboard' 
SANDBOX_URL= 'You can find this on your Mailgun Dashboard' 
SENDER_EMAIL = 'This would be test@your SANDBOX_URL'
RECIPIENT_EMAIL = 'Enter your Email ID Here'
API_KEY = 'This is your Bolt Cloud accout API key'
DEVICE_ID = 'This is the ID of your Bolt device'
FRAME_SIZE = 10
MUL_FACTOR = 6

Code for Tweet Posting

Python
Run the code on the server to get tweet regarding the temperature change
import tweepy
import json, time
from boltiot import Bolt

consumer_key = "Consumer Key (API Key) for Twitter App"
consumer_secret = "Consumer Secret (API Secret) for Twitter App"
access_token = "Access Token for Twitter App"
access_token_secret = "Access Token Secret for Twitter App"
bolt_cloud_api_key = "API key of your Bolt Cloud. You can find in API section on Bolt Cloud"
device_id = "ID of your Bolt Device"

# Dictionary to store credentials as key-value pairs.
config = {
"consumer_key"    : conf.consumer_key,
"consumer_secret"     : conf.consumer_secret,
"access_token"        : conf.access_token,
"access_token_secret" : conf.access_token_secret
}

# Method to authenticate user via Tweepy and return API object
def get_api_object(cfg):
    auth =tweepy.OAuthHandler(cfg['consumer_key'],
                                  cfg['consumer_secret'])
    auth.set_access_token(cfg['access_token'],
                              cfg['access_token_secret'])
    return tweepy.API(auth)

mybolt = Bolt(conf.bolt_cloud_api_key, conf.device_id)
temperature_threshold = 7
while True:
    response = mybolt.analogRead('A0')
    data = json.loads(response)
    print (data['value'])
    try:
        sensor_value = int(data['value'])
        if sensor_value > temperature_threshold:
            print ("Check out !!! Temperature has crossed the threshold.")
            # Call get_api_object to authenticate user and get the API object
            api_object = get_api_object(config)
            # Store the tweet message in the variable
            tweet = ("Check out !!! Temperature has crossed the threshold.")
            # Post the tweet on your Twitter account using the update_status method.
            status = api_object.update_status(status=tweet)
    except Exception as e:
        print ("An error occurred ", e)
    time.sleep(10)

Credits

Devesh Kumar Rai

Devesh Kumar Rai

1 project • 4 followers
I am an undergraduate student highly enthusiast in Machine Learning and IoT. Great passion for building projects.

Comments