Temperature Monitoring System Using Bolt IoT

import requests
from boltiot import Bolt
import json
import time
import statistics

# Replace with your API Key from Bolt Cloud
API_KEY = ''

# Bolt Cloud API URL
BASE_URL = 'https://cloud.boltiot.com/remote/'

# GPIO pin where the temperature sensor is connected
GPIO_PIN = '0'

# Function to read temperature from DS18B20 sensor
def read_temperature():
    try:
        with open(f"/sys/class/gpio/gpio{GPIO_PIN}/value", "r") as file:
            data = file.read()
        return float(data.strip())  # Convert the reading to a floating-point number
    except Exception as e:
        print(f"Error reading temperature: {str(e)}")
        return None

# Function to calculate the Z-score for a given value in a dataset
def calculate_z_score(value, data):
    mean = statistics.mean(data)
    std_dev = statistics.stdev(data)
    if std_dev == 0:
        return 0  # Avoid division by zero
    z_score = (value - mean) / std_dev
    return z_score

# Main loop
temperature_data = []  # Store temperature readings for Z-score analysis

while True:
    temperature = read_temperature()
    if temperature is not None:
        # Append the temperature reading to the data list
        temperature_data.append(temperature)

        # Keep only the last 10 readings for analysis
        if len(temperature_data) > 10:
            temperature_data.pop(0)

        # Calculate Z-score for the latest temperature reading
        z_score = calculate_z_score(temperature, temperature_data)

        # Define a threshold for anomaly detection (adjust as needed)
        anomaly_threshold = 2.0  # You can change this threshold

        # Check if the Z-score exceeds the threshold
        if abs(z_score) > anomaly_threshold:
            print(f"Anomaly detected! Temperature: {temperature}°C, Z-score: {z_score}")

        # Create a JSON payload
        payload = {
            'temperature': temperature,
            'deviceName': 'BoltIoT'
        }
        try:
            # Send the data to Bolt Cloud
            response = requests.post(BASE_URL + 'logData', data=json.dumps(payload), headers={'Authorization': API_KEY})
            print(f"Temperature: {temperature}°C, Response: {response.status_code}")
        except Exception as e:
            print(f"Error sending data to Bolt Cloud: {str(e)}")

    time.sleep(300)  # Wait for 5 minutes before the next reading