6 使用内置NS网络服务器
6.1 内置NS基础操作
内置NS chirpstack页面登录信息：
- 用户名：
admin - 默认密码：
admin - 页面在
http://192.168.60.1:8080或者http://WAN_IP:8080
6.1.1 切换内置模式
注意事项：
- 网关内置chirpstack，页面在http://192.168.60.1:8080或者http://WAN_IP:8080
- 以网关内部的chirpstack-4.17.0软件为例，
- 示例网关ID：0010502df45635f8
- 到网关页面Network（网络） → LoRa GW（LoRa网关） → Configuration（配置），网关页面配置 ：
- Gateway and server communication protocol（网关和服务器通信协议）：Built-in Network Server
- 在页面右下角，点击 Save & Apply（保存并应用）
6.1.2 添加唯传设备
打开内置NS chirpstack的页面，有三种方式
- 直接通过LAN IP 谷歌浏览器访问
http://192.168.60.1:8080或者http://WAN_IP:8080 - 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态）
- 点击 chirpstack的图标，会进行浏览器页面跳转chirpstack登录页面
- 点击超链接
Go to ChirpStack to Add Device ↗，会进行浏览器页面跳转chirpstack登录页面
- 登录chirpstack 用户名：
admin，默认密码：admin
- 在chirpstack页面，在菜单 Applications，选择
all_sensor，点击进去
- 添加设备,以
AN-303温湿度为例
- 在
Device profile下拉框里选择Vendor→LoRaWAN Sensor→AN-303→选择LoRaWAN频段，示例US915
- 提交后，再填写对应appkey等信息，这样创建的设备，设备的解析脚本都会关联网关内置的js脚本，无需再编写解析脚本
- LoRaWAN设备上电入网以后，查看设备数据，两种方法
- 可以在网关页面Network（网络） → Built-in NS（内置NS） → Status（状态） ，下拉页面找到Device list（设备列表） 查看设备实时数据
- 可以在chirpstack的页面查看也可以
6.1.3 添加第三方设备
- 直接通过LAN IP 谷歌浏览器访问
http://192.168.60.1:8080或者http://WAN_IP:8080
- 登录chirpstack 用户名：
admin，默认密码：admin
- 添加设备需要客户自行编写js脚本
6.1.4 设备管理（删除、编辑、查询）
两个方法，第一种在网关页面单个删除、批量操作，第二种方法是在chirpstack的页面单个删除操作
- 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态），下拉页面找到Device list（设备列表）
6.2 批量设备导入导出
6.2.1 批量导入
- 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态），下拉页面找到LoRaWAN Server Operations（LoRaWAN服务器操作）
- 先下载 导入设备的excel模板
- IMX93-GW8016类型网关最多支持导入2000个LoRaWAN设备
- IMX93-GW8000类型网关最多支持导入1000个LoRaWAN设备
- IMX6-GW8000类型网关最多支持导入100个LoRaWAN设备
6.2.2 批量导出
- 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态），下拉页面找到LoRaWAN Server Operations（LoRaWAN服务器操作）
- 点击 EXPORT（导出）
- 谷歌浏览器会提示文件保留或者保存
6.3 高级开发工具
6.3.1 ChirpStack REST API页面接口
- 直接通过LAN IP 谷歌浏览器访问
http://192.168.60.1:8090或者http://WAN_IP:8090 - 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态），下拉页面找到LoRaWAN Server Operations（LoRaWAN服务器操作） 可以复制调用API的token，token有期限，如果需要长期的token，在chirpstack页面自行创建API key
- API授权token
- 粘贴在网页复制的token
- 尝试api调用
6.3.2 Node-RED页面接口使用
- 进入页面两种方法
- 直接通过LAN IP 谷歌浏览器访问
http://192.168.60.1:1885或者http://WAN_IP:1885 - 在网关页面 Network（网络） → Built-in NS（内置NS） → Status（状态）页面，点击
Node-RED图标跳转，或者超链接
6.3.3 历史数据查询
在网关页面 Network（网络） → Built-in NS（内置NS） → Device historical data query（设备历史数据查询）
- 可以根据最新条数或者时间段查询，也可以导出设备的历史数据到excel表格
6.4 设备数据查询与推送
6.4.1 订阅内置的MQTT获取设备实时数据
6.4.2 作为MQTT客户端推送数据
- 网关支持使用订阅方式，默认内置的mosquito server MQTT用户名：
gateway，MQTT 密码：mqtt88888888，端口：1883，订阅chirpstack的topic主题:application/+/device/+/event/+ - 以网关的WAN IP为
192.168.31.75为例
guo@ubuntu:~$ mosquitto_sub -h 192.168.31.75 -p 1883 -u gateway -P mqtt88888888 -t application/+/device/+/event/+ -d
Client null sending CONNECT
Client null received CONNACK (0)
Client null sending SUBSCRIBE (Mid: 1, Topic: application/+/device/+/event/+, QoS: 0, Options: 0x00)
Client null received SUBACK
Subscribed (mid: 1): 0
Client null received PUBLISH (d0, q0, r0, m0, 'application/3ef9e6b9-ec54-4eda-86b8-a5fb46899f39/device/ffffff200000b702/event/up', ... (1125 bytes))
{"deduplicationId":"3f9215e0-8980-4274-811a-81c853783625","time":"2026-01-30T09:39:29.090891629+00:00","deviceInfo":{"tenantId":"af1374c6-87f5-4986-93cd-57857e412930","tenantName":"ChirpStack","applicationId":"3ef9e6b9-ec54-4eda-86b8-a5fb46899f39","applicationName":"all_sensor","deviceProfileId":"da1387cd-2814-44e5-9d6b-6746f0514b03","deviceProfileName":"AN-303","deviceName":"ffffff200000b702","devEui":"ffffff200000b702","deviceClassEnabled":"CLASS_A","tags":{}},"devAddr":"35f8235a","adr":true,"dr":3,"fCnt":7993,"fPort":210,"confirmed":true,"data":"AAEDBA4VfQB3ARAKCBICPwMB","rawData":"000103040E157D007701100A0812023F0301","object":{"tamper":1.0,"humidity":57.5,"temperature":25.68,"tamperEvent":1.0,"model":"AN-303","batteryVoltage":3.605,"batteryVoltageState":0.0},"rxInfo":[{"gatewayId":"0010502df45635f8","uplinkId":467,"nsTime":"2026-01-30T09:39:28.964630698+00:00","rssi":-79,"snr":13.75,"channel":2,"location":{},"context":"bPfc5g==","crcStatus":"CRC_OK"}],"txInfo":{"frequency":902700000,"modulation":{"lora":{"bandwidth":125000,"spreadingFactor":7,"codeRate":"CR_4_5","preamble":8}}},"regionConfigId":"us915"}- 提供python脚本的代码示例
# ASCII-only ChirpStack v4 MQTT Event Listener
# All printed output and comments use ASCII characters only.
import paho.mqtt.client as mqtt
import json
import base64
import binascii
from datetime import datetime
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT broker successfully")
# Subscribe to application/device events
client.subscribe("application/+/device/+/event/+")
client.subscribe("application/+/device/+/event/up")
client.subscribe("application/+/device/+/event/join")
client.subscribe("application/+/device/+/event/ack")
client.subscribe("application/+/device/+/event/error")
print("Subscribed to topics:")
print(" - application/+/device/+/event/+")
print(" - application/+/device/+/event/up")
print(" - application/+/device/+/event/join")
else:
print("Connection failed with code {}".format(rc))
def on_disconnect(client, userdata, rc):
if rc != 0:
print("Unexpected disconnection from MQTT broker. Reconnecting...")
def decode_base64_to_hex(base64_str):
# Decode a base64 string to a hexadecimal string
try:
decoded_bytes = base64.b64decode(base64_str)
return decoded_bytes.hex()
except:
return None
def decode_hex_to_ascii(hex_str):
# Decode a hex string to ASCII text if possible
try:
bytes_obj = binascii.unhexlify(hex_str)
text = bytes_obj.decode('ascii', errors='ignore')
if any(c.isprintable() for c in text.strip()):
return text.strip()
except:
pass
return None
def process_uplink_event(topic, payload_dict):
# Handle uplink events
print("\n" + "="*80)
print("UPLINK EVENT RECEIVED")
print("="*80)
# Basic information
device_info = payload_dict.get('deviceInfo', {})
print("Basic Information:")
print(" Device EUI: {}".format(device_info.get('devEui', 'N/A')))
print(" Device Name: {}".format(device_info.get('deviceName', 'N/A')))
print(" Application: {}".format(device_info.get('applicationName', 'N/A')))
print(" Device Profile: {}".format(device_info.get('deviceProfileName', 'N/A')))
# Transmission details
print("\nTransmission Details:")
print(" Time: {}".format(payload_dict.get('time', 'N/A')))
print(" DevAddr: {}".format(payload_dict.get('devAddr', 'N/A')))
print(" FCnt: {}".format(payload_dict.get('fCnt', 'N/A')))
print(" FPort: {}".format(payload_dict.get('fPort', 'N/A')))
print(" Data Rate (DR): {}".format(payload_dict.get('dr', 'N/A')))
print(" ADR Enabled: {}".format(payload_dict.get('adr', 'N/A')))
print(" Confirmed: {}".format(payload_dict.get('confirmed', 'N/A')))
# Data analysis
print("\nData Analysis:")
# data field (base64)
data_base64 = payload_dict.get('data')
if data_base64:
print(" 'data' (base64): {}".format(data_base64))
data_hex = decode_base64_to_hex(data_base64)
if data_hex:
print(" 'data' decoded (hex): {}".format(data_hex))
data_ascii = decode_hex_to_ascii(data_hex)
if data_ascii:
print(" 'data' as text: '{}'".format(data_ascii))
# rawData field (hex)
raw_data = payload_dict.get('rawData')
if raw_data:
print(" 'rawData' (hex): {}".format(raw_data))
raw_data_ascii = decode_hex_to_ascii(raw_data)
if raw_data_ascii:
print(" 'rawData' as text: '{}'".format(raw_data_ascii))
# Decoded object field
object_data = payload_dict.get('object')
if object_data:
print("\nDecoded Sensor Data:")
for key, value in object_data.items():
print(" {:25} {}".format(key, value))
# Received gateway information
rx_info = payload_dict.get('rxInfo', [])
if rx_info:
print("\nReceived by {} gateway(s):".format(len(rx_info)))
for i, rx in enumerate(rx_info):
gateway_id = rx.get('gatewayId', 'Unknown')
rssi = rx.get('rssi', 'N/A')
snr = rx.get('snr', 'N/A')
channel = rx.get('channel', 'N/A')
print(" Gateway {}: {}".format(i+1, gateway_id))
print(" RSSI: {} dBm, SNR: {} dB, Channel: {}".format(rssi, snr, channel))
# Transmission info
tx_info = payload_dict.get('txInfo', {})
if tx_info:
print("\nTransmission Info:")
print(" Frequency: {} Hz".format(tx_info.get('frequency', 'N/A')))
modulation = tx_info.get('modulation', {})
if 'lora' in modulation:
lora = modulation['lora']
print(" LoRa Modulation:")
print(" Bandwidth: {} Hz".format(lora.get('bandwidth', 'N/A')))
print(" Spreading: SF{}".format(lora.get('spreadingFactor', 'N/A')))
print(" Code Rate: {}".format(lora.get('codeRate', 'N/A')))
print("\nAdditional Info:")
print(" Deduplication ID: {}".format(payload_dict.get('deduplicationId', 'N/A')))
print(" Region Config ID: {}".format(payload_dict.get('regionConfigId', 'N/A')))
print("="*80)
def process_join_event(topic, payload_dict):
# Handle device join events
print("\n" + "="*80)
print("JOIN EVENT RECEIVED")
print("="*80)
device_info = payload_dict.get('deviceInfo', {})
print("Device Information:")
print(" Device EUI: {}".format(device_info.get('devEui', 'N/A')))
print(" Device Name: {}".format(device_info.get('deviceName', 'N/A')))
print(" Application: {}".format(device_info.get('applicationName', 'N/A')))
print(" Tenant: {}".format(device_info.get('tenantName', 'N/A')))
print("\nJoin Details:")
print(" Time: {}".format(payload_dict.get('time', 'N/A')))
print(" DevAddr: {}".format(payload_dict.get('devAddr', 'N/A')))
# Device capabilities
dev_class = device_info.get('deviceClassEnabled', 'N/A')
print(" Device Class: {}".format(dev_class))
# Tags if present
tags = device_info.get('tags', {})
if tags:
print("\nDevice Tags:")
for key, value in tags.items():
print(" {}: {}".format(key, value))
print("="*80)
def process_ack_event(topic, payload_dict):
# Handle ACK events
print("\n" + "="*80)
print("ACKNOWLEDGEMENT RECEIVED")
print("="*80)
device_info = payload_dict.get('deviceInfo', {})
print("Device: {} ({})".format(device_info.get('devEui', 'N/A'), device_info.get('deviceName', 'N/A')))
print("Details:")
print(" Time: {}".format(payload_dict.get('time', 'N/A')))
print(" FCnt: {}".format(payload_dict.get('fCnt', 'N/A')))
print(" Acknowledged: True")
print("="*80)
def process_error_event(topic, payload_dict):
# Handle error events
print("\n" + "="*80)
print("ERROR EVENT RECEIVED")
print("="*80)
device_info = payload_dict.get('deviceInfo', {})
print("Device: {} ({})".format(device_info.get('devEui', 'N/A'), device_info.get('deviceName', 'N/A')))
print("Error Details:")
print(" Time: {}".format(payload_dict.get('time', 'N/A')))
print(" Type: {}".format(payload_dict.get('type', 'N/A')))
print(" Error: {}".format(payload_dict.get('error', 'N/A')))
print(" FCnt: {}".format(payload_dict.get('fCnt', 'N/A')))
print("="*80)
def on_message(client, userdata, msg):
try:
# Decode message
payload = msg.payload.decode('utf-8')
payload_dict = json.loads(payload)
# Current time
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Parse topic
topic_parts = msg.topic.split('/')
if len(topic_parts) >= 6:
application_id = topic_parts[1]
device_id = topic_parts[3]
event_type = topic_parts[5]
print("\nMessage received at {}".format(current_time))
print(" Topic: {}".format(msg.topic))
print(" Application: {}".format(application_id))
print(" Device: {}".format(device_id))
print(" Event: {}".format(event_type))
print(" QoS: {}".format(msg.qos))
print(" Retain: {}".format(msg.retain))
else:
print("\nMessage received at {}".format(current_time))
print(" Topic: {}".format(msg.topic))
# Dispatch based on event type
if "event/up" in msg.topic:
process_uplink_event(msg.topic, payload_dict)
elif "event/join" in msg.topic:
process_join_event(msg.topic, payload_dict)
elif "event/ack" in msg.topic:
process_ack_event(msg.topic, payload_dict)
elif "event/error" in msg.topic:
process_error_event(msg.topic, payload_dict)
else:
# Unknown event type, print raw JSON with ASCII escaping
print("\nUnknown event type, raw payload:")
print(json.dumps(payload_dict, indent=2, ensure_ascii=True))
except json.JSONDecodeError as e:
print("\nJSON Decode Error: {}".format(e))
print("Raw payload (first 500 bytes): {}".format(msg.payload[:500]))
except UnicodeDecodeError as e:
print("\nUnicode Decode Error: {}".format(e))
print("Raw payload (hex): {}...".format(msg.payload.hex()[:100]))
except Exception as e:
print("\nError processing message: {}".format(e))
import traceback
traceback.print_exc()
def main():
# MQTT client configuration
client_id = "chirpstack-listener-{}".format(datetime.now().strftime('%Y%m%d%H%M%S'))
client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv311)
# Set username and password
client.username_pw_set("gateway", "mqtt88888888")
# Set callbacks
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
# Reconnect options
client.reconnect_delay_set(min_delay=1, max_delay=120)
print("="*80)
print("ChirpStack v4 MQTT Event Listener")
print("="*80)
print("Client ID: {}".format(client_id))
print("Broker: localhost:1883")
print("Username: gateway")
print("Starting connection...")
try:
# Connect to MQTT broker
client.connect("localhost", 1883, 60)
# Start loop
print("\nStarting MQTT loop...")
print("Press Ctrl+C to stop\n")
client.loop_forever()
except KeyboardInterrupt:
print("\n\nReceived interrupt signal. Disconnecting...")
client.disconnect()
print("Disconnected from MQTT broker")
except Exception as e:
print("\nFailed to connect to MQTT broker: {}".format(e))
if __name__ == "__main__":
main()- 执行脚本接收数据如下：
root@Gateway:~# python /root/python-SDK/mqtt_event_listener.py
================================================================================
ChirpStack v4 MQTT Event Listener
================================================================================
Client ID: chirpstack-listener-20260130174156
Broker: localhost:1883
Username: gateway
Starting connection...
Starting MQTT loop...
Press Ctrl+C to stop
Connected to MQTT broker successfully
Subscribed to topics:
- application/+/device/+/event/+
- application/+/device/+/event/up
- application/+/device/+/event/join
Message received at 2026-01-30 17:42:05
Topic: application/3ef9e6b9-ec54-4eda-86b8-a5fb46899f39/device/ffffff200000b702/event/up
Application: 3ef9e6b9-ec54-4eda-86b8-a5fb46899f39
Device: ffffff200000b702
Event: up
QoS: 0
Retain: 0
================================================================================
UPLINK EVENT RECEIVED
================================================================================
Basic Information:
Device EUI: ffffff200000b702
Device Name: ffffff200000b702
Application: all_sensor
Device Profile: AN-303
Transmission Details:
Time: 2026-01-30T09:42:05.675897655+00:00
DevAddr: 35f8235a
FCnt: 8008
FPort: 210
Data Rate (DR): 3
ADR Enabled: True
Confirmed: True
Data Analysis:
'data' (base64): AAEDBA4VfQB3ARAKCBICPwMB
'data' decoded (hex): 000103040e157d007701100a0812023f0301
'data' as text: '}w ?'
'rawData' (hex): 000103040E157D007701100A0812023F0301
'rawData' as text: '}w ?'
Decoded Sensor Data:
batteryVoltage 3.605
temperature 25.68
tamperEvent 1.0
model AN-303
tamper 1.0
batteryVoltageState 0.0
humidity 57.5
Received by 1 gateway(s):
Gateway 1: 0010502df45635f8
RSSI: -83 dBm, SNR: 13.5 dB, Channel: 7
Transmission Info:
Frequency: 903700000 Hz
LoRa Modulation:
Bandwidth: 125000 Hz
Spreading: SF7
Code Rate: CR_4_5
Additional Info:
Deduplication ID: d61b9b59-7998-483e-9432-3901e03b5217
Region Config ID: us915
================================================================================
6.4.2 内置的mqtt作为客户端推送节点设备实时数据
- 网关支持使用推送方式，网关作为mqtt客户端推送数据到客户的MQTT服务器
-到网关页面Network（网络） → LoRa GW（LoRa网关） → Configuration（配置），网关页面配置 ：
- Gateway and server communication protocol（网关和服务器通信协议）：Built-in Network Server
- Push chirpstack parsed data to multiple MQTT servers（推送chirpstack解析后的数据到多个MQTT服务器）：根据需要选择 1个
- Client 1 MQTT Server（客户端 1 MQTT服务器地址）： 根据客户的MQTT broker地址真实填写
- Client 1 MQTT port（客户端 1 MQTT服务器端口）：1883，根据客户的MQTT broker端口真实填写
- Client 1 MQTT Username（客户端 1 MQTT 用户名）：根据客户的MQTT broker 用户名真实填写
- Client 1 MQTT password（客户端 1 MQTT 密码）：根据客户的MQTT broker 密码真实填写
- 其他参数，根据需求填写
6.4.3 HTTP推送数据
- 在内置chirpstack页面里可以配置http推送
- 示例推送http到`http://192.168.31.205:8001/data'
- 提供python脚本的代码接收http推送数据解析示例
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import json
import base64
import binascii
from chirpstack_api import integration
from google.protobuf.json_format import Parse
class Handler(BaseHTTPRequestHandler):
# True - JSON marshaler
# False - Protobuf marshaler (binary)
json = True
def do_POST(self):
self.send_response(200)
self.end_headers()
query_args = parse_qs(urlparse(self.path).query)
content_len = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_len)
# 首先打印原始HTTP请求信息
print(f"\n{'='*80}")
print(f"Received HTTP POST request:")
print(f" Path: {self.path}")
print(f" Event: {query_args.get('event', ['unknown'])[0]}")
print(f" Content-Length: {content_len}")
print(f" Headers: {dict(self.headers.items())}")
# 打印原始JSON数据
try:
raw_json = body.decode('utf-8')
print(f"\nRaw JSON data received:")
print(f"{'-'*40}")
print(raw_json)
print(f"{'-'*40}")
# 格式化打印JSON
parsed_json = json.loads(raw_json)
print(f"\nFormatted JSON data:")
print(json.dumps(parsed_json, indent=2, ensure_ascii=False))
except Exception as e:
print(f"Error parsing JSON: {e}")
print(f"Raw body (hex): {body.hex()}")
if query_args["event"][0] == "up":
self.up(body)
elif query_args["event"][0] == "join":
self.join(body)
else:
print(f"Handler for event {query_args['event'][0]} is not implemented")
def up(self, body):
try:
# 1. 先解析为字典查看结构
data_dict = json.loads(body.decode('utf-8'))
print(f"\n{'='*80}")
print("Processing Uplink Event:")
print(f" Device EUI: {data_dict.get('deviceInfo', {}).get('devEui', 'unknown')}")
print(f" Device Name: {data_dict.get('deviceInfo', {}).get('deviceName', 'unknown')}")
print(f" FCnt: {data_dict.get('fCnt', 'unknown')}")
print(f" FPort: {data_dict.get('fPort', 'unknown')}")
# 2. 检查字段
print(f"\nField Analysis:")
print(f" Has 'data' field: {'data' in data_dict}")
print(f" Has 'rawData' field: {'rawData' in data_dict}")
if 'data' in data_dict:
print(f" 'data' field value: {data_dict['data']}")
try:
# 尝试解码base64的data字段
decoded_data = base64.b64decode(data_dict['data'])
print(f" Decoded 'data' (hex): {decoded_data.hex()}")
print(f" Decoded 'data' (bytes): {decoded_data}")
except:
print(f" 'data' field is not base64 encoded")
if 'rawData' in data_dict:
print(f" 'rawData' field value: {data_dict['rawData']}")
print(f" 'rawData' length: {len(data_dict['rawData'])}")
# 3. 检查object字段
if 'object' in data_dict:
print(f"\nDecoded Object Fields:")
for key, value in data_dict['object'].items():
print(f" {key}: {value}")
# 4. 尝试解析为Protobuf
print(f"\nTrying to parse with Protobuf...")
try:
# 创建一个新的字典，不包含rawData字段
proto_dict = data_dict.copy()
if 'rawData' in proto_dict:
# 如果需要，可以将rawData转换为data字段
# 但根据你的错误，Protobuf期望的是data字段，而不是rawData
del proto_dict['rawData']
# 使用ignore_unknown_fields=True忽略未知字段
up = Parse(json.dumps(proto_dict), integration.UplinkEvent(), ignore_unknown_fields=True)
print(f" Successfully parsed Protobuf message")
print(f" Device EUI from Protobuf: {up.device_info.dev_eui}")
if up.data:
print(f" Data from Protobuf (hex): {up.data.hex()}")
else:
print(f" No data field in Protobuf message")
except Exception as e:
print(f" Error parsing with Protobuf: {e}")
# 5. 处理rxInfo
if 'rxInfo' in data_dict:
print(f"\nReceived by {len(data_dict['rxInfo'])} gateway(s):")
for i, rx in enumerate(data_dict['rxInfo']):
print(f" Gateway {i+1}: {rx.get('gatewayId', 'unknown')} - RSSI: {rx.get('rssi', 'unknown')}, SNR: {rx.get('snr', 'unknown')}")
print(f"{'='*80}\n")
except Exception as e:
print(f"\nError processing uplink: {e}")
import traceback
traceback.print_exc()
def join(self, body):
try:
data_dict = json.loads(body.decode('utf-8'))
print(f"\n{'='*80}")
print("Processing Join Event:")
print(f" Device EUI: {data_dict.get('deviceInfo', {}).get('devEui', 'unknown')}")
print(f" Device Name: {data_dict.get('deviceInfo', {}).get('deviceName', 'unknown')}")
print(f" DevAddr: {data_dict.get('devAddr', 'unknown')}")
print(f" Application: {data_dict.get('deviceInfo', {}).get('applicationName', 'unknown')}")
# 尝试解析为Protobuf
proto_dict = data_dict.copy()
join = Parse(json.dumps(proto_dict), integration.JoinEvent(), ignore_unknown_fields=True)
print(f"\nSuccessfully parsed Protobuf join message")
print(f" Device EUI from Protobuf: {join.device_info.dev_eui}")
print(f" DevAddr from Protobuf: {join.dev_addr}")
print(f"{'='*80}\n")
except Exception as e:
print(f"\nError processing join: {e}")
import traceback
traceback.print_exc()
def unmarshal(self, body, pl):
if self.json:
return Parse(body, pl, ignore_unknown_fields=True)
pl.ParseFromString(body)
return pl
def log_message(self, format, *args):
"""重写日志方法，减少默认日志输出"""
# 你可以取消下面的注释来查看HTTP访问日志
# print(f"HTTP: {format % args}")
pass
if __name__ == "__main__":
httpd = HTTPServer(('0.0.0.0', 8001), Handler)
print("=" * 80)
print("ChirpStack v4 HTTP Event Listener")
print("Listening on port 8001")
print("Ready to receive events...")
print("=" * 80)
httpd.serve_forever()- 执行脚本接收数据如下：
root@Gateway:~# python3 /root/python-SDK/http_event_listener.py
================================================================================
ChirpStack v4 HTTP Event Listener
Listening on port 8001
Ready to receive events...
================================================================================
================================================================================
Received HTTP POST request:
Path: /data?event=up
Event: up
Content-Length: 1124
Headers: {'content-type': 'application/json', 'accept': '*/*', 'host': '192.168.31.205:8001', 'content-length': '1124'}
Raw JSON data received:
----------------------------------------
{"deduplicationId":"d4147289-5dc9-45c7-94b7-14d44f4dc1ce","time":"2026-01-30T10:01:51.329855273+00:00","deviceInfo":{"tenantId":"af1374c6-87f5-4986-93cd-57857e412930","tenantName":"ChirpStack","applicationId":"3ef9e6b9-ec54-4eda-86b8-a5fb46899f39","applicationName":"all_sensor","deviceProfileId":"da1387cd-2814-44e5-9d6b-6746f0514b03","deviceProfileName":"AN-303","deviceName":"ffffff200000b702","devEui":"ffffff200000b702","deviceClassEnabled":"CLASS_A","tags":{}},"devAddr":"35f8235a","adr":true,"dr":3,"fCnt":8123,"fPort":210,"confirmed":true,"data":"AAEDBA4VfQB3ARAKCBICPwMB","rawData":"000103040E157D007701100A0812023F0301","object":{"batteryVoltageState":0.0,"tamper":1.0,"temperature":25.68,"humidity":57.5,"model":"AN-303","batteryVoltage":3.605,"tamperEvent":1.0},"rxInfo":[{"gatewayId":"0010502df45635f8","uplinkId":620,"nsTime":"2026-01-30T10:01:51.204207430+00:00","rssi":-86,"snr":11.5,"channel":1,"location":{},"context":"sAHeNg==","crcStatus":"CRC_OK"}],"txInfo":{"frequency":902500000,"modulation":{"lora":{"bandwidth":125000,"spreadingFactor":7,"codeRate":"CR_4_5","preamble":8}}},"regionConfigId":"us915"}
----------------------------------------
Formatted JSON data:
{
"deduplicationId": "d4147289-5dc9-45c7-94b7-14d44f4dc1ce",
"time": "2026-01-30T10:01:51.329855273+00:00",
"deviceInfo": {
"tenantId": "af1374c6-87f5-4986-93cd-57857e412930",
"tenantName": "ChirpStack",
"applicationId": "3ef9e6b9-ec54-4eda-86b8-a5fb46899f39",
"applicationName": "all_sensor",
"deviceProfileId": "da1387cd-2814-44e5-9d6b-6746f0514b03",
"deviceProfileName": "AN-303",
"deviceName": "ffffff200000b702",
"devEui": "ffffff200000b702",
"deviceClassEnabled": "CLASS_A",
"tags": {}
},
"devAddr": "35f8235a",
"adr": true,
"dr": 3,
"fCnt": 8123,
"fPort": 210,
"confirmed": true,
"data": "AAEDBA4VfQB3ARAKCBICPwMB",
"rawData": "000103040E157D007701100A0812023F0301",
"object": {
"batteryVoltageState": 0.0,
"tamper": 1.0,
"temperature": 25.68,
"humidity": 57.5,
"model": "AN-303",
"batteryVoltage": 3.605,
"tamperEvent": 1.0
},
"rxInfo": [
{
"gatewayId": "0010502df45635f8",
"uplinkId": 620,
"nsTime": "2026-01-30T10:01:51.204207430+00:00",
"rssi": -86,
"snr": 11.5,
"channel": 1,
"location": {},
"context": "sAHeNg==",
"crcStatus": "CRC_OK"
}
],
"txInfo": {
"frequency": 902500000,
"modulation": {
"lora": {
"bandwidth": 125000,
"spreadingFactor": 7,
"codeRate": "CR_4_5",
"preamble": 8
}
}
},
"regionConfigId": "us915"
}
================================================================================
Processing Uplink Event:
Device EUI: ffffff200000b702
Device Name: ffffff200000b702
FCnt: 8123
FPort: 210
Field Analysis:
Has 'data' field: True
Has 'rawData' field: True
'data' field value: AAEDBA4VfQB3ARAKCBICPwMB
Decoded 'data' (hex): 000103040e157d007701100a0812023f0301
Decoded 'data' (bytes): b'\x00\x01\x03\x04\x0e\x15}\x00w\x01\x10\n\x08\x12\x02?\x03\x01'
'rawData' field value: 000103040E157D007701100A0812023F0301
'rawData' length: 36
Decoded Object Fields:
batteryVoltageState: 0.0
tamper: 1.0
temperature: 25.68
humidity: 57.5
model: AN-303
batteryVoltage: 3.605
tamperEvent: 1.0
Trying to parse with Protobuf...
Successfully parsed Protobuf message
Device EUI from Protobuf: ffffff200000b702
Data from Protobuf (hex): 000103040e157d007701100a0812023f0301
Received by 1 gateway(s):
Gateway 1: 0010502df45635f8 - RSSI: -86, SNR: 11.5
================================================================================