You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Linux环境下LoRa网关Python程序json.dumps()无响应问题求助

问题:LoRa网关Python代码中json.dumps()无响应

运行Linux系统的LoRa网关,使用模拟node-RED的Python应用接收终端上行消息时,出现json.dumps()调用后始终无法返回的问题:

  • 终端消息能正常接收,触发onMessage()回调
  • 若rbPayloadFormatters()直接返回newMsg字典,onMessage()能正常打印内容
  • 调用json.dumps(newMsg)后,print(self.packet)始终无输出,等待数分钟也无结果

核心代码片段:

### formats the payload message from the endpoint
def rbPayloadFormatters(self, msg):
    msgObj = json.loads(msg)
    newMsg = {}
    msgHex = base64.b64decode(msgObj["data"])
    newMsg["payload"] = binascii.hexlify(msgHex)
    newMsg["time"] = msgObj["tmst"]
    newMsg["snr"] = msgObj["lsnr"]
    newMsg["station"] = msgObj["appeui"]
    newMsg["avgsnr"] = msgObj["lsnr"]
    newMsg["lat"] = 0
    newMsg["lng"] = 0
    newMsg["rssi"] = msgObj["rssi"]
    newMsg["seqnumber"] = msgObj["seqn"]
    newMsg["deveui"] = msgObj["deveui"]
    newMsg["authorisation"] = self.rbAuthorization
    return json.dumps(newMsg)

#callback function initiated by on_message
def onMessage(self, mqtt_client, userdata, msg):
    self.packet = self.rbPayloadFormatters(msg.payload)
    pkt = json.loads(self.packet)
    self.devEUI = pkt["deveui"]
    self.payloadData = pkt["payload"]
    
    print(self.packet)

排查思路与解决方案

1. 修复非JSON可序列化的bytes类型

binascii.hexlify(msgHex)返回的是bytes对象,而JSON标准不支持直接序列化bytes类型,这是导致json.dumps()卡住/静默失败的最可能原因。

修改代码,将bytes转成UTF-8字符串:

newMsg["payload"] = binascii.hexlify(msgHex).decode('utf-8')

2. 添加异常捕获定位错误

MQTT回调函数中的异常可能被底层框架静默吞掉,导致无法看到具体错误。在rbPayloadFormatters()中添加异常捕获,直接打印错误信息:

def rbPayloadFormatters(self, msg):
    try:
        msgObj = json.loads(msg)
        newMsg = {}
        msgHex = base64.b64decode(msgObj["data"])
        newMsg["payload"] = binascii.hexlify(msgHex).decode('utf-8')
        newMsg["time"] = msgObj["tmst"]
        newMsg["snr"] = msgObj["lsnr"]
        newMsg["station"] = msgObj["appeui"]
        newMsg["avgsnr"] = msgObj["lsnr"]
        newMsg["lat"] = 0
        newMsg["lng"] = 0
        newMsg["rssi"] = msgObj["rssi"]
        newMsg["seqnumber"] = msgObj["seqn"]
        newMsg["deveui"] = msgObj["deveui"]
        newMsg["authorisation"] = self.rbAuthorization
        return json.dumps(newMsg)
    except Exception as e:
        print(f"rbPayloadFormatters错误: {str(e)}")
        raise  # 抛出异常让上层处理,或返回错误标记

3. 检查self.rbAuthorization的类型

确认self.rbAuthorization是否为JSON可序列化类型(如字符串、数字、列表、字典等)。如果是自定义对象、bytes或其他特殊类型,需转换为可序列化格式。

4. 单独验证序列化逻辑

在调试时,可以将newMsg的内容单独提取出来,在Python控制台手动执行json.dumps(newMsg),直接观察是否报错,快速定位问题点。


内容的提问来源于stack exchange,提问作者u2dragon

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.02 19:10:19