• 11、监测数据采集物联网应用开发步骤(8.2)

    1. 监测数据采集物联网应用开发步骤(8.1)

    新建TCP/IP Client线程类com.zxy.tcp.ClientThread.py

    1. #! python3
    2. # -*- coding: utf-8 -
    3. '''
    4. Created on 2017年05月10日
    5. @author: zxyong 13738196011
    6. '''
    7.  
    8. import datetime
    9. import socket
    10. import threading
    11. import time
    12.  
    13. from com.zxy.adminlog.UsAdmin_Log import UsAdmin_Log
    14. from com.zxy.common import Com_Para
    15. from com.zxy.common.Com_Fun import Com_Fun
    16. from com.zxy.tcp.TcpClient import TcpClient
    17. from com.zxy.z_debug import z_debug
    18.  
    19. #监测数据采集物联网应用--TCP/IP Client线程
    20. class ClientThread(z_debug):
    21.     #断线重连
    22.     idtTimeOut = 60
    23.     sServerIP = "0.0.0.0"
    24.     iPort = 6000
    25.     connectionFlag = False
    26.     
    27.     def __init__(self, timeout, ServerIP, temPort):
    28.         self.idtTimeOut = timeout
    29.         self.sServerIP = ServerIP
    30.         self.iPort = temPort
    31.     
    32.     #inputFlag网络链接是否正常
    33.     def set_dSockList(self,inputFlag):
    34.         key = str(self.sServerIP)+"|"+str(self.iPort) 
    35.         #存在
    36.         if key not in list(Com_Para.dSockList.keys()):
    37.             Com_Para.dSockList[key] = [0,datetime.datetime.now()]
    38.         if inputFlag == False:
    39.             objAry = Com_Para.dSockList[key]
    40.             if objAry[0] <= int(180 / self.idtTimeOut)+12:
    41.                 objAry[0] = objAry[0] + 1
    42.             objAry[1] = datetime.datetime.now()
    43.             Com_Para.dSockList[key] = objAry
    44.         else:
    45.             Com_Para.dSockList[key] = [0,datetime.datetime.now()]    
    46.     
    47.     #判断上次链接时间频率是否合规
    48.     def judge_dSock(self):
    49.         key = str(self.sServerIP)+"|"+str(self.iPort) 
    50.         if key not in list(Com_Para.dSockList.keys()):
    51.             Com_Para.dSockList[key] = [0,datetime.datetime.now()]
    52.         objAry = Com_Para.dSockList[key]
    53.         starttime = datetime.datetime.now()
    54.         if objAry[0] <= int(120 / self.idtTimeOut)+1:
    55.             return True
    56.         elif objAry[0] <= int(120 / self.idtTimeOut)+7 and starttime >= objAry[1] + datetime.timedelta(minutes=10):
    57.             return True
    58.         elif objAry[0] <= int(120 / self.idtTimeOut)+12 and starttime >= objAry[1] + datetime.timedelta(hours=1):
    59.             return True
    60.         elif objAry[0] > int(120 / self.idtTimeOut)+12 and starttime >= objAry[1] + datetime.timedelta(hours=1):
    61.             return True
    62.         else:
    63.             return False
    64.         
    65.     def run(self):
    66.         starttime = datetime.datetime.now()       
    67.         endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)        
    68.         ckeys = self.sServerIP + "|" + str(self.iPort)
    69.         tc = TcpClient(ckeys)
    70.         t = None        
    71.         try:
    72.             #判断上次链接时间频率是否合规
    73.             if not self.judge_dSock():
    74.                 return None
    75.             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    76.             s.connect((self.sServerIP, self.iPort))
    77.             Com_Fun.SetHashTable(Com_Para.dClientThreadList, ckeys, s)            
    78.             self.set_dSockList(True)
    79.             
    80.             temResult = "*"
    81.             Com_Fun.SendSocket(temResult,s)            
    82.             
    83.             t = threading.Thread(target=tc.client_link, name="ClientTh" + ckeys)
    84.             t.start()
    85.         except IOError as e:
    86.             temError = "repeat connect server error:"+self.sServerIP+" "+Com_Fun.GetTime("%Y-%m-%d %H:%M:%S") + repr(e)
    87.             uL = UsAdmin_Log(Com_Para.ApplicationPath, temError)
    88.             uL.WriteLog()
    89.             self.set_dSockList(False);
    90.  
    91.         while True:
    92.             starttime = datetime.datetime.now()
    93.             try:
    94.                 if t is None and starttime <= endtime:
    95.                     if Com_Fun.GetHashTableNone(Com_Para.dClientThreadList, ckeys) is not None:
    96.                         t = threading.Thread(target=tc.client_link, name="ClientTh" + self.sServerIP)
    97.                         t.start()
    98.                     time.sleep(self.idtTimeOut/2)
    99.                 elif starttime >= endtime:
    100.                     endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)                
    101.                     if Com_Fun.GetHashTableNone(Com_Para.dClientThreadList, ckeys) is None and self.judge_dSock():
    102.                         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    103.                         s.connect((self.sServerIP, self.iPort))          
    104.                         self.set_dSockList(True)
    105.                         Com_Fun.SetHashTable(Com_Para.dClientThreadList, s.getpeername()[0]+"|"+str(s.getpeername()[1]), s)
    106.                         t = threading.Thread(target=tc.client_link, name="ClientTh" + self.sServerIP)
    107.                         t.start()
    108.                     time.sleep(self.idtTimeOut/2)
    109.                 #超过10分钟,重新计时
    110.                 elif (endtime - starttime).seconds > 10 * self.idtTimeOut:
    111.                     endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)
    112.                 else:
    113.                     time.sleep(self.idtTimeOut/2)
    114.             except IOError as e:
    115.                 temError = "repeat connect server error:"+self.sServerIP+" "+Com_Fun.GetTime("%Y-%m-%d %H:%M:%S") + repr(e)
    116.                 uL = UsAdmin_Log(Com_Para.ApplicationPath, temError)
    117.                 uL.WriteLog()
    118.                 self.set_dSockList(False);
    119.                 
    120.     def list_run(self):
    121.         starttime = datetime.datetime.now()       
    122.         endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)        
    123.         ckeys = self.sServerIP + "|" + str(self.iPort)
    124.         tc = TcpClient(ckeys)
    125.         t = None        
    126.         try:
    127.             #判断上次链接时间频率是否合规
    128.             if not self.judge_dSock():
    129.                 return None
    130.             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    131.             s.connect((self.sServerIP, self.iPort))
    132.             Com_Fun.SetHashTable(Com_Para.dClientThreadList, ckeys, s)      
    133.             self.set_dSockList(True)
    134.             t = threading.Thread(target=tc.client_link, name="ClientTh" + ckeys)
    135.             t.start()
    136.         except IOError as e:
    137.             temError = "repeat connect server error:"+self.sServerIP+" "+Com_Fun.GetTime("%Y-%m-%d %H:%M:%S") + repr(e)
    138.             uL = UsAdmin_Log(Com_Para.ApplicationPath, temError)
    139.             uL.WriteLog()            
    140.             self.set_dSockList(False)
    141.         while True:
    142.             starttime = datetime.datetime.now()
    143.             try:
    144.                 if t is None and starttime <= endtime:
    145.                     if Com_Fun.GetHashTableNone(Com_Para.dClientThreadList, ckeys) is not None:
    146.                         t = threading.Thread(target=tc.client_link, name="ClientTh" + self.sServerIP)
    147.                         t.start()
    148.                     time.sleep(self.idtTimeOut/2)
    149.                 elif starttime >= endtime:
    150.                     endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)                
    151.                     if Com_Fun.GetHashTableNone(Com_Para.dClientThreadList, ckeys) is None and self.judge_dSock():
    152.                         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    153.                         s.connect((self.sServerIP, self.iPort))
    154.                         Com_Fun.SetHashTable(Com_Para.dClientThreadList, s.getpeername()[0]+"|"+str(s.getpeername()[1]), s)   
    155.                         self.set_dSockList(True)
    156.                         t = threading.Thread(target=tc.client_link, name="ClientTh" + self.sServerIP)
    157.                         t.start()
    158.                     time.sleep(self.idtTimeOut/2)
    159.                 #超过10分钟,重新计时
    160.                 elif (endtime - starttime).seconds > 10 * self.idtTimeOut:
    161.                     endtime = datetime.datetime.now() + datetime.timedelta(seconds=self.idtTimeOut)
    162.                 else:
    163.                     time.sleep(self.idtTimeOut/2)
    164.             except IOError as e:
    165.                 temError = "repeat connect server error:"+self.sServerIP+" "+Com_Fun.GetTime("%Y-%m-%d %H:%M:%S") + repr(e)
    166.                 uL = UsAdmin_Log(Com_Para.ApplicationPath, temError)
    167.                 uL.WriteLog()
    168.                 self.set_dSockList(False);

    新建tcp client数据接收插件类1 com.plugins.Usereflect.testClientReflectClass1.py

    1. #! python3
    2. # -*- coding: utf-8 -
    3. '''
    4. Created on 2017年05月10日
    5. @author: zxyong 13738196011
    6. '''
    7. from com.zxy.z_debug import z_debug
    8.  
    9. #监测数据采集物联网应用--往平台端发送数据协议拦截器,可组装发送平台数据协议程序
    10. class testClientReflectClass1(z_debug):
    11.     #接收到数据
    12.     strResult        = ""
    13.     #需要发送到服务端数据
    14.     strSend          = ""
    15.     #是否继续执行 1:继续执行 0:中断执行
    16.     strContinue      = "1"        
    17.     #所连接的远端IP
    18.     strIP            = ""
    19.     
    20.     def __init__(self):
    21.         pass
    22.  
    23.     def init_start(self):
    24.         #作为客户端接收数据拦截器
    25.         if self.strResult == "A01":
    26.             self.strSend = "#send to server data:server1"
    27.         print("AAA作为客户端接收数据拦截器:"+self.strResult)
    28.         self.strResult = "ok"
    29.         #Com_Para.dClientThreadList  本机作为客户端连接socket list
    30.         #Com_Para.htComPort          串口通讯hashtable 
    31. #         /*示例:此处写业务逻辑,最后给 strResult重新赋值*/
    32. #         //self.strResult = "test"
    33. #         //self.strSend = "发送到服务端数据"
    34. #         /**************************************************/

    新建tcp client数据接收插件类2 com.plugins.Usereflect.testClientReflectClass2.py

    1. #! python3
    2. # -*- coding: utf-8 -
    3. '''
    4. Created on 2017年05月10日
    5. @author: zxyong 13738196011
    6. '''
    7. from com.zxy.z_debug import z_debug
    8.  
    9. #监测数据采集物联网应用--往平台端发送数据协议拦截器,可组装发送平台数据协议程序
    10. class testClientReflectClass2(z_debug):
    11.     #接收到数据
    12.     strResult        = ""
    13.     #需要发送到服务端数据
    14.     strSend          = ""
    15.     #是否继续执行 1:继续执行 0:中断执行
    16.     strContinue      = "1"        
    17.     #所连接的远端IP
    18.     strIP            = ""
    19.     
    20.     def __init__(self):
    21.         pass
    22.  
    23.     def init_start(self):
    24.         #作为客户端接收数据拦截器
    25.         if self.strResult == "A02":
    26.             self.strSend = "#send to server data:server"
    27.         print("BBB作为客户端接收数据拦截器:"+self.strResult)
    28.         self.strResult = "ok"
    29.         #Com_Para.dClientThreadList  本机作为客户端连接socket list
    30.         #Com_Para.htComPort          串口通讯hashtable 
    31. #         /*示例:此处写业务逻辑,最后给 strResult重新赋值*/
    32. #         //self.strResult = "test"
    33. #         //self.strSend = "发送到服务端数据"
    34. #         /**************************************************/

    在com.zxy.main.Init_Page.py中添加代码

    1.     @staticmethod
    2.     def Start_Client():       
    3.             try: 
    4.                 for temSIP in Com_Para.ServerIPList.split(";"):
    5.                     if temSIP != "":
    6.                         sttem = ClientThread(10, temSIP.split(":")[0], int(temSIP.split(":")[1]))
    7.                         # TCP客户端案例
    8.                         t4tem = threading.Thread(target=sttem.list_run, name="ClientMainThread" + temSIP.split(":")[0])
    9.                         t4tem.start()
    10.             except Exception as e:
    11.                 print("TCP client error:" + repr(e)+"=>"+str(e.__traceback__.tb_lineno))
    12.             finally:
    13.                 Pass

    TCP Client测试案例MonitorDataCmd.py主文件中编写：

    在    if __name__ == '__main__':下添加

    1.         #TCP Client配置服务端ip及端口参数
    2.         Com_Para.ServerIPList = "127.0.0.1:5002;127.0.0.1:5003"
    3.         #TCP/IP client连接初始化
    4.         Init_Page.Start_Client()
    5.         
    6.         #暂停3秒钟,tcp client连接需要时间
    7.         time.sleep(3)
    8.         #测试TCP/IP client发送数据
    9.         tcSock = Com_Fun.GetHashTableNone(Com_Para.dClientThreadList,"127.0.0.1|5002")
    10.         #作为客户端接收数据拦截器
    11.         Com_Para.ClientREFLECT_IN_CLASS = "com.plugins.usereflect.testClientReflectClass1"
    12.         if tcSock != None:
    13.             temStr = "TCP client往服务端发送数据:"+Com_Fun.GetTimeDef()+"\r\n"
    14.             print("开始准备发送数据")
    15.             Com_Fun.SendSocket(temStr, tcSock)
    16.             print("TCP client往服务端发送数据")
    17.         else:
    18.             print("TCP Client未准备好")
    19.             
    20.         #改变下作为客户端接收数据拦截器
    21.         #Com_Para.ClientREFLECT_IN_CLASS = "com.plugins.usereflect.testClientReflectClass2"
    22.         #接收数据解析插件已经改变了

    运行测试结果如下图:

    监测数据采集物联网应用开发步骤(9.1)

  • 相关阅读:
    【Vue】修饰符&表单提交方式&自定义组件
    Nginx配置实例-负载均衡
    您需要了解的有关NFT的所有知识
    Dorkish：一款针对OSINT和网络侦查任务的Chrome扩展
    常用眼底图像数据集简介及下载--糖尿病视网膜病变（Eyepacs,APTOS2019,Messdior,Messdior-2,STARE数据集)
    35【Aseprite 作图】苹果——拆解
    前端面试题
    Flink应用案例统计实现TopN的两种方式
    Prometheus+Grafana搭建Jmeter性能监控平台（2）
    【OSPF引入直连路由时巧借静态黑洞路由做汇总】
  • 原文地址：https://blog.csdn.net/yong427/article/details/132580679