如何将Raspberry Pi Pico采集的温度数据输出至Zabbix？

问题背景 #

我有一台通过USB连接到Ubuntu 22.04电脑的Raspberry Pi Pico，已经编写了温度读取代码并在Thonny中运行。目前温度数据只能输出到Thonny终端或Pico本地的.txt文件，希望实现将数据直接写入电脑文件，或者推送至Zabbix。

原温度读取代码：

import machine
import utime
 
sensor_temp = machine.ADC(4)
conversion_factor = 3.3 / (65535)

temp = ()

while True:
    reading = sensor_temp.read_u16() * conversion_factor 
    temperature = 27 - (reading - 0.706)/0.001721
#    print(f'Температура: {temperature}°C')
    temp = temperature
    print(temp)
    utime.sleep(1)

方法一：将温度数据写入Ubuntu电脑文件 #

方式1：通过串口捕获输出并重定向到文件 #

Pico通过USB连接后会被识别为串口设备（通常是/dev/ttyACM0或/dev/ttyUSB0），直接在Ubuntu终端执行命令捕获串口输出并写入文件：

1. 安装串口工具（未安装时执行）：

sudo apt install screen

2. 运行命令捕获数据到文件：

screen /dev/ttyACM0 115200 >> ~/pico_temp.log

• 替换/dev/ttyACM0为你的实际串口设备（可用ls /dev/tty*查看）
• 115200是串口波特率，与Pico默认一致
• ~/pico_temp.log是你要保存数据的文件路径

如果需要后台运行，配合nohup：

nohup screen /dev/ttyACM0 115200 >> ~/pico_temp.log 2>&1 &

方式2：用Python脚本监听串口并写入文件 #

编写简单的Python脚本在Ubuntu端运行，监听Pico的串口输出并格式化写入文件：

import serial
import time

# 配置串口参数
ser = serial.Serial('/dev/ttyACM0', 115200, timeout=1)
time.sleep(2)  # 等待串口连接稳定

log_file = open('/home/your_username/pico_temperature.log', 'a')

try:
    while True:
        if ser.in_waiting > 0:
            temp_data = ser.readline().decode('utf-8').strip()
            if temp_data:
                # 添加时间戳后写入
                log_entry = f"{time.strftime('%Y-%m-%d %H:%M:%S')} - 温度: {temp_data}°C\n"
                log_file.write(log_entry)
                log_file.flush()  # 立即写入文件
        time.sleep(0.5)
except KeyboardInterrupt:
    log_file.close()
    ser.close()
    print("程序已停止")

运行脚本前安装pyserial库：

pip install pyserial

方法二：将温度数据推送至Zabbix #

方式1：通过Ubuntu中转推送（推荐） #

先通过上面的方法在Ubuntu端捕获温度数据，再用脚本将数据发送到Zabbix服务器：

1. 安装zabbix-sender：

sudo apt install zabbix-sender

2. 修改Python监听脚本，添加Zabbix推送逻辑：

import serial
import time
import subprocess

# 配置串口参数
ser = serial.Serial('/dev/ttyACM0', 115200, timeout=1)
time.sleep(2)

# Zabbix配置
ZABBIX_SERVER = "你的Zabbix服务器IP"
ZABBIX_HOST = "Pico_Temp"  # Zabbix中创建的主机名
ITEM_KEY = "pico.temperature"  # Zabbix中创建的监控项键值

try:
    while True:
        if ser.in_waiting > 0:
            temp_data = ser.readline().decode('utf-8').strip()
            if temp_data and temp_data.replace('.', '', 1).isdigit():
                # 调用zabbix_sender推送数据
                cmd = [
                    'zabbix_sender',
                    '-z', ZABBIX_SERVER,
                    '-s', ZABBIX_HOST,
                    '-k', ITEM_KEY,
                    '-o', temp_data
                ]
                subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                print(f"已推送温度: {temp_data}°C 到Zabbix")
        time.sleep(1)
except KeyboardInterrupt:
    ser.close()
    print("程序已停止")

方式2：Pico直接推送至Zabbix（需Pico连接网络） #

如果你的Pico通过WiFi模块（如ESP-01）或Pico W连接了网络，可修改Pico的MicroPython代码，直接将温度数据发送到Zabbix：

import machine
import utime
import network
import urequests

# 温度传感器配置
sensor_temp = machine.ADC(4)
conversion_factor = 3.3 / (65535)

# WiFi配置
SSID = "你的WiFi名称"
PASSWORD = "你的WiFi密码"
ZABBIX_SERVER = "http://你的Zabbix服务器IP/zabbix/api_jsonrpc.php"
ZABBIX_USER = "Zabbix用户名"
ZABBIX_PASS = "Zabbix密码"
HOST_ID = "123"  # Zabbix主机ID，可通过API查询
ITEM_KEY = "pico.temperature"

# 连接WiFi
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, PASSWORD)

while not wlan.isconnected():
    utime.sleep(1)
print("WiFi已连接")

# 获取Zabbix认证令牌
auth_data = {
    "jsonrpc": "2.0",
    "method": "user.login",
    "params": {
        "user": ZABBIX_USER,
        "password": ZABBIX_PASS
    },
    "id": 1
}
response = urequests.post(ZABBIX_SERVER, json=auth_data)
auth_token = response.json()["result"]
response.close()

# 循环读取温度并推送
while True:
    reading = sensor_temp.read_u16() * conversion_factor 
    temperature = 27 - (reading - 0.706)/0.001721
    temp_str = "{:.2f}".format(temperature)
    
    # 构造推送数据
    send_data = {
        "jsonrpc": "2.0",
        "method": "item.update",
        "params": {
            "hostid": HOST_ID,
            "key_": ITEM_KEY,
            "value": temp_str
        },
        "auth": auth_token,
        "id": 2
    }
    
    try:
        response = urequests.post(ZABBIX_SERVER, json=send_data)
        print(f"已推送温度: {temp_str}°C")
        response.close()
    except Exception as e:
        print(f"推送失败: {e}")
    
    utime.sleep(10)  # 每10秒推送一次

注意：Pico W需要预先安装包含urequests的MicroPython固件，普通Pico需外接WiFi模块并适配代码。

内容的提问来源于stack exchange，提问作者Леонид