怎么用脚本读取传感器数据

本文目录导读：

1. Python 读取串口传感器数据
2. 读取 I2C 传感器（使用 smbus2）
3. 读取 GPIO 传感器（使用 RPi.GPIO）
4. 使用 MQTT 接收传感器数据
5. 从 Web API 读取传感器数据
6. 使用 Redis 接收传感器数据
7. 安装依赖
8. 关键注意事项：

Python 读取串口传感器数据

import serial
import time
import json
# 连接串口（根据实际情况修改）
ser = serial.Serial(
    port='COM3',  # Windows
    # port='/dev/ttyUSB0',  # Linux/Mac
    baudrate=9600,
    timeout=1
)
try:
    while True:
        if ser.in_waiting:
            # 读取一行数据
            line = ser.readline().decode('utf-8').strip()
            # 按逗号分隔解析数据
            data = line.split(',')
            # 示例：假设数据格式是 "温度,湿度,光照"
            if len(data) >= 3:
                temperature = float(data[0])
                humidity = float(data[1])
                light = float(data[2])
                print(f"温度: {temperature}°C, 湿度: {humidity}%, 光照: {light} lux")
                # 保存到文件
                with open('sensor_data.csv', 'a') as f:
                    f.write(f"{time.time()},{temperature},{humidity},{light}\n")
        time.sleep(0.1)
except KeyboardInterrupt:
    print("程序退出")
    ser.close()

读取 I2C 传感器（使用 smbus2）

from smbus2 import SMBus
import time
# 创建 I2C 总线对象（通常地址是 /dev/i2c-1）
bus = SMBus(1)
# 传感器地址
SENSOR_ADDR = 0x40  # 根据实际传感器修改
# 寄存器地址
TEMP_REG = 0x00
HUMI_REG = 0x01
def read_sensor_data():
    """读取 I2C 传感器数据"""
    try:
        # 读取温度寄存器的 2 字节数据
        temp_raw = bus.read_word_data(SENSOR_ADDR, TEMP_REG)
        # 转换字节序（小端到大端）
        temp = (temp_raw & 0xFF) << 8 | (temp_raw >> 8)
        # 转换为实际温度（根据传感器datasheet）
        temperature = temp / 100.0
        # 读取湿度
        humi_raw = bus.read_word_data(SENSOR_ADDR, HUMI_REG)
        humi = (humi_raw & 0xFF) << 8 | (humi_raw >> 8)
        humidity = humi / 100.0
        return temperature, humidity
    except Exception as e:
        print(f"读取失败: {e}")
        return None, None
# 主循环
while True:
    temp, humi = read_sensor_data()
    if temp is not None:
        print(f"温度: {temp:.1f}°C, 湿度: {humi:.1f}%")
    time.sleep(1)

读取 GPIO 传感器（使用 RPi.GPIO）

import RPi.GPIO as GPIO
import time
import Adafruit_DHT
# 设置 GPIO 模式
GPIO.setmode(GPIO.BCM)
# 定义传感器类型和引脚
DHT_SENSOR = Adafruit_DHT.DHT11
DHT_PIN = 4  # GPIO引脚号
# 数字传感器读取（如按钮）
BUTTON_PIN = 17
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
# 模拟传感器读取（ADC - 需要额外的ADC芯片）
# 这里使用 MCP3008 作为示例
# from gpiozero import MCP3008
# adc = MCP3008(channel=0)
def read_dht_sensor():
    """读取温湿度传感器"""
    humidity, temperature = Adafruit_DHT.read_retry(DHT_SENSOR, DHT_PIN)
    if humidity is not None and temperature is not None:
        return temperature, humidity
    else:
        return None, None
def read_button():
    """读取按钮状态"""
    if GPIO.input(BUTTON_PIN) == GPIO.LOW:
        return True
    return False
try:
    while True:
        # 读取温湿度
        temp, humi = read_dht_sensor()
        if temp is not None:
            print(f"温度: {temp:.1f}°C, 湿度: {humi:.1f}%")
        # 读取按钮
        if read_button():
            print("按钮被按下!")
        time.sleep(1)
finally:
    GPIO.cleanup()

使用 MQTT 接收传感器数据

import paho.mqtt.client as mqtt
import json
import time
# MQTT 服务器配置
BROKER = "127.0.0.1"
PORT = 1883
TOPIC = "sensors/room1"
def on_connect(client, userdata, flags, rc):
    """连接成功的回调"""
    print(f"连接成功，返回码: {rc}")
    client.subscribe(TOPIC)
def on_message(client, userdata, msg):
    """接收到消息的回调"""
    try:
        # 解析JSON数据
        data = json.loads(msg.payload.decode())
        # 处理传感器数据
        print(f"收到数据: {data}")
        # 可以在这里添加数据存储逻辑
        with open('mqtt_sensor_data.log', 'a') as f:
            f.write(f"{time.time()}: {data}\n")
    except Exception as e:
        print(f"解析错误: {e}")
# 创建客户端
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
# 连接服务器
client.connect(BROKER, PORT, 60)
# 持续监听
client.loop_forever()

从 Web API 读取传感器数据

import requests
import json
import time
def read_from_api(api_url, api_key=None):
    """从 REST API 读取传感器数据"""
    headers = {}
    if api_key:
        headers['Authorization'] = f"Bearer {api_key}"
    try:
        response = requests.get(api_url, headers=headers)
        response.raise_for_status()  # 检查HTTP错误
        data = response.json()
        return data
    except requests.exceptions.RequestException as e:
        print(f"API请求失败: {e}")
        return None
# 示例：读取天气API
def read_weather():
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": 31.2304,
        "longitude": 121.4737,
        "current_weather": True
    }
    try:
        response = requests.get(url, params=params)
        data = response.json()
        if 'current_weather' in data:
            weather = data['current_weather']
            return {
                'temperature': weather['temperature'],
                'wind_speed': weather['windspeed']
            }
    except Exception as e:
        print(f"读取天气失败: {e}")
    return None
# 主程序
if __name__ == "__main__":
    while True:
        weather = read_weather()
        if weather:
            print(f"当前温度: {weather['temperature']}°C, 风速: {weather['wind_speed']} km/h")
        time.sleep(10)  # 每10秒读取一次

使用 Redis 接收传感器数据

import redis
import json
import time
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def subscribe_to_sensors():
    """订阅传感器主题"""
    pubsub = r.pubsub()
    pubsub.subscribe('sensor_data')
    for message in pubsub.listen():
        if message['type'] == 'message':
            try:
                data = json.loads(message['data'])
                print(f"收到传感器数据: {data}")
                # 将数据存储
                r.lpush('sensor_history', json.dumps(data))
                r.ltrim('sensor_history', 0, 999)  # 保留最近1000条
            except json.JSONDecodeError:
                print(f"无效的JSON数据: {message['data']}")
def get_latest_data():
    """获取最新数据"""
    data = r.lindex('sensor_history', 0)
    if data:
        return json.loads(data)
    return None
if __name__ == "__main__":
    # 订阅和持续监听
    try:
        subscribe_to_sensors()
    except KeyboardInterrupt:
        print("程序退出")

安装依赖

# 基础依赖
pip install pyserial smbus2
# GPIO 相关（树莓派）
pip install RPi.GPIO Adafruit-DHT
# MQTT 协议
pip install paho-mqtt
# HTTP 请求
pip install requests
# Redis
pip install redis
pip install aioredis  # 异步版本

关键注意事项：

1. 权限问题：访问GPIO或串口可能需要sudo权限
2. 错误处理：传感器可能因各种原因读取失败
3. 数据格式：确保理解传感器输出的数据格式
4. 采样频率：不要过于频繁读取，根据传感器规格设置
5. 日志记录：建议记录数据以备后续分析
6. 线程安全：如果多线程读取，需要注意并发控制

请根据你的具体传感器类型和硬件环境,选择合适的方案进行修改。