Python实现NB-IoT模组远程控制的UDP通信方案
本文介绍利用Python实现NB-IoT模组远程控制的完整方案。通过pyserial模块串口发送AT指令，结合UDP协议进行网络通信，实现服务器与终端的双向交互。方案包含服务器端监听、终端注册机制及多线程并发处理，旨在探索非运营商平台下的物联网灵活控制路径。
本文旨在探讨在不依赖运营商网络应用平台的前提下，仅通过服务商服务器实现对终端设备的完全控制可行性。若该路径可行，将显著提升物联网应用服务端的灵活性。尽管实际环境中难以达到与运营商网络同等的处理能力，但使用Python编写的原型代码已能成功实现参数变更（如PSM、eDRX等配置）。由于终端分配的IP地址属于接入网范畴，其性质更接近局域网，若需通过IMSI、IMEI等标识访问，仍需与运营商核心网协同。以下展示尝试实现远程控制的具体方法。
主要实现功能
1、利用Python pyserial模块通过串口发送AT命令以修改模组参数，具体参考<使用python pyserial模块串口通信>； 2、通过互联网传输控制命令，选用UDP协议进行主机控制，具体参考<python socket网络接口编程>； 3、直接经由NB-IoT无线网络传输控制命令； 4、采用Python多窗口处理服务器端程序，实现接收与发送操作的同步进行。
远程控制主机脚本
服务器端程序逻辑：监测UDP对应端口号，若接收到register信息则返回allowed状态，随后进入命令输入状态，等待命令输入完成后发送给终端，并等待终端反馈，进而继续传送下一个命令。
#!/usr/bin/python3.6
import socket
import sys
import re
BUFFER_SIZE = 1024
TARGET_ADDR = ''
TARGET_PORT = 60000
TARGET = (TARGET_ADDR,TARGET_PORT)
ss = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
ss.bind(TARGET)
print("server online!! wait for register!")
data,addrRsv = ss.recvfrom(BUFFER_SIZE)
if not data:
sys.exit(0)
else:
print(data)
if(re.match(b'register',data)):
ss.sendto(b'allowed',addrRsv)
else:
ss.sendto(b'reject',addrRsv)
while True:
#等待命令输入
aa = input('cmd > ')
if not aa:
break
else:
cmdV = aa+'\r'
ss.sendto(cmdV.encode('utf-8'),addrRsv)
#等待结果返回
data,addrRsv = ss.recvfrom(BUFFER_SIZE)
if not data:
break
else:
print(data)
ss.close()
客户主机程序逻辑：发送register并成功接收allowed后，进入等待控制命令状态，通过串口将命令转发给终端模块，接收终端模块的反馈消息后，将其返回给服务器侧。
#!/usr/bin/python3.6
import serial
import sys
import os
import re
import socket
#初始化UART端口
ser = serial.Serial("COM5",9600,timeout=30)
#选择相应的协议类型UDP
ss = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
BUFFER_SIZE = 1024
TARGET_ADDR = 'IP address'
TARGET_PORT = 60000
TARGET = (TARGET_ADDR,TARGET_PORT)
aa = '开机命令'.encode('utf-8') #convert to bytes type
ser.write(aa)
while True:
line = ser.readline()
if not line:
print("can not get cmd result, release!")
sys.exit(0)
print(line)
if ( re.match(b'OK',line) ):
break
ss.sendto(bytes('register','utf-8'),TARGET)
data,addrRsv = ss.recvfrom(BUFFER_SIZE)
if re.match(b'allowed',data):
print('register successfully!')
pass
else:
print('register failure')
sys.exit(0)
while True:
data,addrRsv = ss.recvfrom(BUFFER_SIZE)
if not data:
print("time out,release now!!")
break
elif re.match(b'end',data):
print("end of process!!")
break;
ser.write(data)
while True:
line = ser.readline()
if not line:
print("can not get cmd result, release!")
break
print(line)
if ( re.match(b'OK',line) ):
ss.sendto(bytes('OK','utf-8'),TARGET)
break
elif(re.match(b'ERROR',line)):
ss.sendto(bytes('ERROR','utf-8'),TARGET)
break
else:
pass
ser.close()
多线程窗口
为使服务器端能够同时执行接收和发送操作，可在服务器端开启两个窗口进行监听，示例如下：
启动代码
#!/usr/bin/python3.6
import threading
import time
import subprocess
import os
import sys
def thread_fun1():
#global vlock
while(1):
print("thread fun1 is running!!!")
time.sleep(1)
#... ...
print(len(sys.argv))
#vlock = threading.Lock()
t1 = threading.Thread(target=thread_fun1,args=())
t1.start()
addr = 'IP address'
port = 60000
cmdStr = "python anotherThread.py %s %d"%(addr,port)
#设置creationflags = subprocess.CREATE_NEW_CONSOLE，用来创建新的控制台窗口
subprocess.Popen(cmdStr,creationflags = subprocess.CREATE_NEW_CONSOLE)
anotherThread.py
#!/usr/bin/python3.6
def thread_fun2():
while(1):
aa = input('cmd > ')
print("thread fun2 is running!!!")
print(aa)
if(aa == 'end'):
break
thread_fun2()
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
