Python脚本实现MQTT数据写入InfluxDB报错及优化咨询

1. 立即修复IndexError错误 #

你遇到的IndexError: no such group是因为正则表达式MQTT_REGEX = 'insideroom/([^/]+)'仅定义了1个捕获组，但代码中尝试访问不存在的第2个组match.group(2)。此外，你的消息payload并非单纯数值，而是完整的InfluxDB Line Protocol格式字符串，这意味着你可以直接写入InfluxDB，无需手动解析。

2. 两种可行解决方案 #

方案一：直接写入InfluxDB（推荐） #

由于传感器已经输出标准的InfluxDB Line Protocol格式数据，跳过自定义解析步骤直接写入是最简单高效的方式：

#!/usr/bin/env python3

"""A MQTT to InfluxDB Bridge
This script receives MQTT data and saves those to InfluxDB.
"""

import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient

INFLUXDB_ADDRESS = '*****'
INFLUXDB_USER = '******'
INFLUXDB_PASSWORD = '******'
INFLUXDB_DATABASE = '*******'

MQTT_ADDRESS = '*******'
MQTT_USER = 'pi'
MQTT_PASSWORD = '*******'

influxdb_client = InfluxDBClient(INFLUXDB_ADDRESS, 8086, INFLUXDB_USER, INFLUXDB_PASSWORD)


def on_connect(client, userdata, flags, rc):
    """ The callback for when the client receives a CONNACK response from the server."""
    print('Connected with result code ' + str(rc))
    if rc ==0:
        print("Connected to Broker")
        client.subscribe('insideroom/humidity')
        client.subscribe('insideroom/temperature')


def on_message(client, userdata, msg):
    """The callback for when a PUBLISH message is received from the server."""
    print(msg.topic + ' ' + str(msg.payload))
    # 直接写入Line Protocol格式的payload
    payload = msg.payload.decode('utf-8')
    influxdb_client.write(payload, protocol='line')


def _init_influxdb_database():
    databases = influxdb_client.get_list_database()
    if len(list(filter(lambda x: x['name'] == INFLUXDB_DATABASE, databases))) == 0:
        influxdb_client.create_database(INFLUXDB_DATABASE)
    influxdb_client.switch_database(INFLUXDB_DATABASE)


def main():
    _init_influxdb_database()

    mqtt_client = mqtt.Client()
    mqtt_client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message

    mqtt_client.connect(MQTT_ADDRESS, 1883)
    mqtt_client.loop_forever()


if __name__ == '__main__':
    print('MQTT to InfluxDB bridge')
    main()

方案二：修复自定义解析逻辑（如需手动处理数据） #

如果需要对数据进行额外处理再写入，可修改解析函数从payload中提取测量值、标签和数值：

# 保留原有的SensorData类，修改_parse_mqtt_message函数
def _parse_mqtt_message(payload):
    try:
        # 拆分Line Protocol的测量部分和数值部分
        meas_part, val_part = payload.split(' ', 1)
        # 拆分测量名称和标签
        meas_tags = meas_part.split(',')
        measurement = meas_tags[0]
        tags = {}
        for tag in meas_tags[1:]:
            k, v = tag.split('=')
            tags[k] = v
        # 提取数值
        _, value_str = val_part.split('=')
        value = float(value_str)
        # 返回结构化数据，location从标签site中获取
        return SensorData(location=tags.get('site'), measurement=measurement, value=value)
    except (ValueError, IndexError):
        return None

# 修改on_message函数调用解析
def on_message(client, userdata, msg):
    print(msg.topic + ' ' + str(msg.payload))
    sensor_data = _parse_mqtt_message(msg.payload.decode('utf-8'))
    if sensor_data is not None:
        _send_sensor_data_to_influxdb(sensor_data)

3. 核心问题解答 #

是否必须通过_parse_mqtt_message函数处理消息？
不需要。只要数据符合InfluxDB的写入格式（Line Protocol或JSON），就可以直接写入。你的传感器已经输出标准Line Protocol，直接写入是最优选择。

4. 技术参考资料 #

• InfluxDB Line Protocol 官方文档：了解Line Protocol的格式规范，这是InfluxDB最高效的写入格式。
• InfluxDB Python Client 官方文档：学习如何使用Python客户端写入不同格式的数据到InfluxDB。
• Paho MQTT Python Client 官方文档：掌握MQTT客户端的消息处理、连接管理等核心功能。

内容的提问来源于stack exchange，提问作者StackUser2014