pip install RPi.GPIO

import RPi.GPIO as GPIO
import time

# 設(shè)置GPIO模式為BCM
GPIO.setmode(GPIO.BCM)

# 設(shè)置GPIO17為輸出模式
GPIO.setup(17, GPIO.OUT)

try:
    while True:
        # 打開LED
        GPIO.output(17, GPIO.HIGH)
        time.sleep(1)
        # 關(guān)閉LED
        GPIO.output(17, GPIO.LOW)
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    # 清理GPIO狀態(tài)
    GPIO.cleanup()

pip install Adafruit_DHT

import Adafruit_DHT
import time

# 設(shè)置傳感器類型為DHT11
sensor = Adafruit_DHT.DHT11
# 設(shè)置傳感器引腳為GPIO4
pin = 4

while True:
    # 讀取溫度和濕度數(shù)據(jù)
    humidity, temperature = Adafruit_DHT.read_retry(sensor, pin)
    
    if humidity is not None and temperature is not None:
        print(f'Temperature: {temperature:.1f}°C  Humidity: {humidity:.1f}%')
    else:
        print('Failed to get reading. Try again!')

    # 每隔2秒讀取一次
    time.sleep(2)

pip install requests

import Adafruit_DHT
import time
import requests

# 設(shè)置傳感器類型為DHT11
sensor = Adafruit_DHT.DHT11
# 設(shè)置傳感器引腳為GPIO4
pin = 4

# API端點(diǎn)URL
url = 'http://example.com/api/data'

while True:
    # 讀取溫度和濕度數(shù)據(jù)
    humidity, temperature = Adafruit_DHT.read_retry(sensor, pin)
    
    if humidity is not None and temperature is not None:
        # 準(zhǔn)備數(shù)據(jù)
        data = {
            'temperature': temperature,
            'humidity': humidity
        }
        # 發(fā)送POST請求
        response = requests.post(url, json=data)
        
        if response.status_code == 200:
            print('Data uploaded successfully')
        else:
            print('Failed to upload data')
    else:
        print('Failed to get reading. Try again!')

    # 每隔2秒讀取一次
    time.sleep(2)

pip install pymongo

import Adafruit_DHT
import time
import requests
from pymongo import MongoClient

# 設(shè)置傳感器類型為DHT11
sensor = Adafruit_DHT.DHT11
# 設(shè)置傳感器引腳為GPIO4
pin = 4

# 連接到MongoDB服務(wù)器
client = MongoClient('mongodb://localhost:27017/')
# 選擇數(shù)據(jù)庫
db = client.iot_data
# 選擇集合（相當(dāng)于關(guān)系型數(shù)據(jù)庫中的表）
collection = db.sensor_data

while True:
    # 讀取溫度和濕度數(shù)據(jù)
    humidity, temperature = Adafruit_DHT.read_retry(sensor, pin)
    
    if humidity is not None and temperature is not None:
        # 準(zhǔn)備數(shù)據(jù)
        data = {
            'temperature': temperature,
            'humidity': humidity,
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        # 插入數(shù)據(jù)到MongoDB
        result = collection.insert_one(data)
        
        if result.acknowledged:
            print('Data inserted with id:', result.inserted_id)
        else:
            print('Failed to insert data')
    else:
        print('Failed to get reading. Try again!')

    # 每隔2秒讀取一次
    time.sleep(2)

from pymongo import MongoClient
from datetime import datetime, timedelta

# 連接到MongoDB服務(wù)器
client = MongoClient('mongodb://localhost:27017/')
# 選擇數(shù)據(jù)庫
db = client.iot_data
# 選擇集合
collection = db.sensor_data

# 設(shè)置時(shí)間范圍
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)

# 查詢數(shù)據(jù)
query = {
    'timestamp': {
        '$gte': start_time.strftime('%Y-%m-%d %H:%M:%S'),
        '$lte': end_time.strftime('%Y-%m-%d %H:%M:%S')
    }
}
data = list(collection.find(query))

# 計(jì)算平均溫度和濕度
if data:
    avg_temperature = sum(d['temperature'] for d in data) / len(data)
    avg_humidity = sum(d['humidity'] for d in data) / len(data)
    print(f'Average Temperature: {avg_temperature:.2f}°C')
    print(f'Average Humidity: {avg_humidity:.2f}%')
else:
    print('No data found for the given time range.')

pip install matplotlib

import matplotlib.pyplot as plt
from pymongo import MongoClient
from datetime import datetime, timedelta

# 連接到MongoDB服務(wù)器
client = MongoClient('mongodb://localhost:27017/')
# 選擇數(shù)據(jù)庫
db = client.iot_data
# 選擇集合
collection = db.sensor_data

# 設(shè)置時(shí)間范圍
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)

# 查詢數(shù)據(jù)
query = {
    'timestamp': {
        '$gte': start_time.strftime('%Y-%m-%d %H:%M:%S'),
        '$lte': end_time.strftime('%Y-%m-%d %H:%M:%S')
    }
}
data = list(collection.find(query))

# 提取時(shí)間、溫度和濕度數(shù)據(jù)
timestamps = [datetime.strptime(d['timestamp'], '%Y-%m-%d %H:%M:%S') for d in data]
temperatures = [d['temperature'] for d in data]
humidities = [d['humidity'] for d in data]

# 繪制溫度曲線
plt.figure(figsize=(10, 5))
plt.plot(timestamps, temperatures, label='Temperature (°C)', color='tab:red')
plt.xlabel('Time')
plt.ylabel('Temperature (°C)')
plt.title('Temperature Over Time')
plt.legend()
plt.grid(True)

# 顯示圖表
plt.show()

# 繪制濕度曲線
plt.figure(figsize=(10, 5))
plt.plot(timestamps, humidities, label='Humidity (%)', color='tab:blue')
plt.xlabel('Time')
plt.ylabel('Humidity (%)')
plt.title('Humidity Over Time')
plt.legend()
plt.grid(True)

# 顯示圖表
plt.show()

pip install Flask

from flask import Flask, request
import RPi.GPIO as GPIO

app = Flask(__name__)

# 設(shè)置GPIO模式為BCM
GPIO.setmode(GPIO.BCM)
# 設(shè)置GPIO17為輸出模式
GPIO.setup(17, GPIO.OUT)

@app.route('/led', methods=['POST'])
def control_led():
    action = request.json.get('action')
    if action == 'on':
        GPIO.output(17, GPIO.HIGH)
        return 'LED is ON', 200
    elif action == 'off':
        GPIO.output(17, GPIO.LOW)
        return 'LED is OFF', 200
    else:
        return 'Invalid action', 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

import requests

url = 'http://raspberrypi_ip:5000/led'

def turn_led_on():
    response = requests.post(url, json={'action': 'on'})
    print(response.text)

def turn_led_off():
    response = requests.post(url, json={'action': 'off'})
    print(response.text)

# 測試控制LED燈
turn_led_on()
time.sleep(5)
turn_led_off()

pip install Flask-Talisman

from flask_talisman import Talisman

# 初始化Flask-Talisman
Talisman(app)

pip install Flask-HTTPAuth

from flask_httpauth import HTTPBasicAuth

auth = HTTPBasicAuth()

# 用戶認(rèn)證信息
users = {
    "admin": "password"
}

@auth.get_password
def get_pw(username):
    if username in users:
        return users.get(username)
    return None

@app.route('/led', methods=['POST'])
@auth.login_required
def control_led():
    action = request.json.get('action')
    if action == 'on':
        GPIO.output(17, GPIO.HIGH)
        return 'LED is ON',

 200
    elif action == 'off':
        GPIO.output(17, GPIO.LOW)
        return 'LED is OFF', 200
    else:
        return 'Invalid action', 400

pip install fabric

from fabric import Connection

def deploy():
    host = "raspberrypi_ip"
    user = "pi"
    code_dir = "/home/pi/iot_project"

    conn = Connection(host=host, user=user)

    with conn.cd(code_dir):
        conn.run("git pull")
        conn.run("pip install -r requirements.txt")
        conn.run("sudo systemctl restart iot_service")

if __name__ == "__main__":
    deploy()

sudo apt-get update
sudo apt-get install prometheus

sudo apt-get install -y apt-transport-https
sudo apt-get install -y software-properties-common wget
wget -q -O - https://packages.grafana.com/gpg.key | sudo apt-key add -
echo "deb https://packages.grafana.com/oss/deb stable main" | sudo tee /etc/apt/sources.list.d/grafana.list
sudo apt-get update
sudo apt-get install grafana

scrape_configs:
  - job_name: 'iot_devices'
    static_configs:
      - targets: ['localhost:9090']

sudo systemctl start grafana-server
sudo systemctl enable grafana-server

#!/bin/bash

# 定義備份目錄和目標(biāo)目錄
BACKUP_DIR="/home/pi/backups"
TARGET_DIR="/mnt/external_drive/backups"

# 創(chuàng)建備份目錄
mkdir -p $BACKUP_DIR

# 備份MongoDB數(shù)據(jù)
mongodump --out $BACKUP_DIR/mongodb_backup_$(date +%Y%m%d)

# 使用rsync同步到目標(biāo)目錄
rsync -av --delete $BACKUP_DIR $TARGET_DIR

crontab -e

0 2 * * * /home/pi/backup.sh

sudo apt-get update && sudo apt-get upgrade -y
pip install --upgrade pip setuptools wheel
pip list --outdated | cut -d ' ' -f1 | xargs -n1 pip install -U

pip install scikit-learn

from sklearn.linear_model import LinearRegression
from pymongo import MongoClient
import numpy as np

# 連接到MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client.iot_data
collection = db.sensor_data

# 獲取數(shù)據(jù)
data = list(collection.find().sort('timestamp', -1).limit(100))

# 提取特征和標(biāo)簽
temperatures = np.array([d['temperature'] for d in data]).reshape(-1, 1)
timestamps = np.array([i for i in range(len(data))]).reshape(-1, 1)

# 創(chuàng)建并訓(xùn)練模型
model = LinearRegression()
model.fit(timestamps, temperatures)

# 預(yù)測未來溫度
future_timestamps = np.array([i for i in range(len(data), len(data) + 10)]).reshape(-1, 1)
predictions = model.predict(future_timestamps)

print('Predicted future temperatures:', predictions)

import Adafruit_DHT
import time
import paho.mqtt.client as mqtt

# MQTT配置
MQTT_BROKER = "broker_ip"
MQTT_PORT = 1883
MQTT_TOPIC = "home/environment"

# 設(shè)置傳感器類型和引腳
sensor = Adafruit_DHT.DHT11
pin = 4

# MQTT客戶端配置
client = mqtt.Client()
client.connect(MQTT_BROKER, MQTT_PORT, 60)

while True:
    # 讀取溫濕度數(shù)據(jù)
    humidity, temperature = Adafruit_DHT.read_retry(sensor, pin)
    
    if humidity is not None and temperature is not None:
        # 發(fā)布數(shù)據(jù)到MQTT
        payload = f"{{'temperature': {temperature}, 'humidity': {humidity}}}"
        client.publish(MQTT_TOPIC, payload)
        print(f"Published: {payload}")
    else:
        print("Failed to get reading. Try again!")
    
    time.sleep(2)

import paho.mqtt.client as mqtt
import RPi.GPIO as GPIO

# MQTT配置
MQTT_BROKER = "broker_ip"
MQTT_PORT = 1883
MQTT_TOPIC = "home/light"

# 設(shè)置GPIO模式和引腳
GPIO.setmode(GPIO.BCM)
GPIO.setup(17, GPIO.OUT)

# MQTT回調(diào)函數(shù)
def on_message(client, userdata, msg):
    if msg.topic == MQTT_TOPIC:
        action = msg.payload.decode()
        if action == "on":
            GPIO.output(17, GPIO.HIGH)
            print("Light ON")
        elif action == "off":
            GPIO.output(17, GPIO.LOW)
            print("Light OFF")

# MQTT客戶端配置
client = mqtt.Client()
client.connect(MQTT_BROKER, MQTT_PORT, 60)
client.subscribe(MQTT_TOPIC)
client.on_message = on_message

# 啟動MQTT客戶端
client.loop_forever()

import RPi.GPIO as GPIO
from mfrc522 import SimpleMFRC522
import paho.mqtt.client as mqtt

# MQTT配置
MQTT_BROKER = "broker_ip"
MQTT_PORT = 1883
MQTT_TOPIC = "home/access"

# 初始化RFID閱讀器
reader = SimpleMFRC522()

# MQTT客戶端配置
client = mqtt.Client()
client.connect(MQTT_BROKER, MQTT_PORT, 60)

try:
    while True:
        print("Place your card to scan")
        id, text = reader.read()
        print(f"Card ID: {id}, Text: {text.strip()}")
        
        # 發(fā)布門禁數(shù)據(jù)到MQTT
        payload = f"{{'card_id': {id}, 'text': '{text.strip()}'}}"
        client.publish(MQTT_TOPIC, payload)
        print(f"Published: {payload}")

finally:
    GPIO.cleanup()