本文目录导读：
- Python 读取串口传感器数据
- 读取 I2C 传感器（使用 smbus2）
- 读取 GPIO 传感器（使用 RPi.GPIO）
- 使用 MQTT 接收传感器数据
- 从 Web API 读取传感器数据
- 使用 Redis 接收传感器数据
- 安装依赖
- 关键注意事项：
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 # 异步版本
关键注意事项：
- 权限问题：访问GPIO或串口可能需要sudo权限
- 错误处理：传感器可能因各种原因读取失败
- 数据格式：确保理解传感器输出的数据格式
- 采样频率：不要过于频繁读取，根据传感器规格设置
- 日志记录：建议记录数据以备后续分析
- 线程安全：如果多线程读取，需要注意并发控制
请根据你的具体传感器类型和硬件环境,选择合适的方案进行修改。
