本來(lái)想嘗試下如果不使用運(yùn)營(yíng)商網(wǎng)絡(luò)應(yīng)用平臺(tái)情況下，只是在服務(wù)商服務(wù)器上是否可以實(shí)現(xiàn)對(duì)終端完全控制，如果這樣可行，那么物聯(lián)網(wǎng)應(yīng)用服務(wù)端更有靈活性。實(shí)際情況下，很難實(shí)現(xiàn)和運(yùn)營(yíng)商網(wǎng)絡(luò)對(duì)等的處理，用python代碼原型確實(shí)能夠?qū)崿F(xiàn)參數(shù)的變化(如PSM,eDXR等)，但是終端分配的IP地址畢竟屬于接入網(wǎng)部分，更近似一個(gè)局域網(wǎng)，如果采用其他方式訪問(wèn)(如IMSI、IMEI等)，還是需要與運(yùn)營(yíng)商核心網(wǎng)進(jìn)行配合。以下是嘗試遠(yuǎn)程控制的實(shí)現(xiàn)方法。

主要實(shí)現(xiàn)功能

1、使用python pyserial模塊通過(guò)串口發(fā)送AT命令給模組進(jìn)行參數(shù)修改，參考<使用python pyserial模塊串口通信>;

2、通過(guò)inter網(wǎng)進(jìn)行控制命令傳輸，選用UDP進(jìn)行主機(jī)控制，參考；

3、直接通過(guò)NB-IoT無(wú)線網(wǎng)絡(luò)進(jìn)行控制命令的傳輸；

4、python多窗口處理服務(wù)器端程序，實(shí)現(xiàn)接收和發(fā)送同時(shí)進(jìn)行；

遠(yuǎn)程控制主機(jī)腳本

服務(wù)器端程序：監(jiān)測(cè)UDP對(duì)應(yīng)的端口號(hào)，如果接收到register信息則返回allowed，然后進(jìn)入命令輸入狀態(tài)，等待命令輸入完成，發(fā)送給終端，等待終端反饋，并接續(xù)下一個(gè)命令傳送。

#!/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)

#等待結(jié)果返回

data,addrRsv = ss.recvfrom(BUFFER_SIZE)

if not data:

break

else:

print(data)

ss.close()

客戶主機(jī)程序：發(fā)送register并成功接收allowed后，等待控制命令，通過(guò)串口轉(zhuǎn)發(fā)給終端模塊，并接收終端模塊的反饋消息，返回給服務(wù)器側(cè)。

#!/usr/bin/python3.6

import serial

import sys

import os

import re

import socket

#初始化UART端口

ser = serial.Serial("COM5",9600,timeout=30)

#選擇相應(yīng)的協(xié)議類型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 = '開機(jī)命令'.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()

多線程窗口

為了使得服務(wù)器端能夠?qū)崿F(xiàn)同時(shí)實(shí)現(xiàn)接收和發(fā)送，可以在服務(wù)器端開啟兩個(gè)窗口進(jìn)行監(jiān)聽，示例如下：

啟動(dòng)代碼

#!/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)

#設(shè)置creationflags = subprocess.CREATE_NEW_CONSOLE，用來(lái)創(chuàng)建新的控制臺(tái)窗口

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()

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

總結(jié)

以上是生活随笔為你收集整理的python做物联网控制_python实现NB-IoT模块远程控制的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。