72 lines
1.7 KiB
Python
72 lines
1.7 KiB
Python
|
import json
|
||
|
|
import time
|
||
|
|
|
||
|
|
def simulate_json_chunk_sending(data, chunk_size=15):
|
||
|
|
"""
|
||
|
|
模拟JSON数据分包发送
|
||
|
|
"""
|
||
|
|
json_string = json.dumps(data)
|
||
|
|
print(f"原始JSON数据: {json_string}")
|
||
|
|
print(f"数据总长度: {len(json_string)} 字符")
|
||
|
|
|
||
|
|
chunks = []
|
||
|
|
# 分包
|
||
|
|
for i in range(0, len(json_string), chunk_size):
|
||
|
|
chunk = json_string[i:i+chunk_size]
|
||
|
|
chunks.append(chunk)
|
||
|
|
|
||
|
|
print(f"\n分包结果 ({len(chunks)} 个包):")
|
||
|
|
for i, chunk in enumerate(chunks):
|
||
|
|
print(f" 包 {i+1}: {chunk}")
|
||
|
|
|
||
|
|
return chunks
|
||
|
|
|
||
|
|
def create_sample_json_data():
|
||
|
|
"""
|
||
|
|
创建示例JSON数据
|
||
|
|
"""
|
||
|
|
return {
|
||
|
|
"type": "radarData",
|
||
|
|
"deviceId": 1001,
|
||
|
|
"timestamp": int(time.time() * 1000),
|
||
|
|
"presence": 1,
|
||
|
|
"heartRate": 72.5,
|
||
|
|
"breathRate": 16.2,
|
||
|
|
"motion": 0,
|
||
|
|
"rssi": -45,
|
||
|
|
"heartbeatWaveform": 120,
|
||
|
|
"breathingWaveform": 45,
|
||
|
|
"rawSignal": -25
|
||
|
|
}
|
||
|
|
|
||
|
|
def create_status_json_data():
|
||
|
|
"""
|
||
|
|
创建状态JSON数据
|
||
|
|
"""
|
||
|
|
return {
|
||
|
|
"type": "status",
|
||
|
|
"wifiConfigured": True,
|
||
|
|
"wifiConnected": True,
|
||
|
|
"ipAddress": "192.168.1.100",
|
||
|
|
"deviceId": 1001
|
||
|
|
}
|
||
|
|
|
||
|
|
def main():
|
||
|
|
print("=== JSON分包发送模拟测试 ===\n")
|
||
|
|
|
||
|
|
# 测试雷达数据
|
||
|
|
print("1. 雷达数据分包测试:")
|
||
|
|
radar_data = create_sample_json_data()
|
||
|
|
simulate_json_chunk_sending(radar_data, 15)
|
||
|
|
|
||
|
|
print("\n" + "="*50 + "\n")
|
||
|
|
|
||
|
|
# 测试状态数据
|
||
|
|
print("2. 状态数据分包测试:")
|
||
|
|
status_data = create_status_json_data()
|
||
|
|
simulate_json_chunk_sending(status_data, 15)
|
||
|
|
|
||
|
|
print("\n=== 测试完成 ===")
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|