#!/usr/bin/env python3 #coding:utf-8 import socketserver import subprocess import pickle offs = 0 #偏移量 p ={} #保存终端信息,ccid与地址对 #p={ccid=addr,} #缓存未能正常发送消息的客户端 pp={} # cache={} from collections import defaultdict cache=defaultdict(list) def save_cache(dic): '''缓存未发送成功的消息 仅在未发送成功时缓存''' # f=open("./cache.txt",mode='w+') # f.write(str(dic)) # f.close pickle.dump(dic,open("./cache.txt",'wb')) def load_cache(): '''程序启动时加载缓存''' try: # f=open("./cache.txt") # c = eval(f.read()) # f.close() c=pickle.load(open("./cache.txt",'rb')) return c except: return None def update_pairs(addr,ccid,clear=0): '''更新终端库 addr:绑定tcp地址 ccid:终端号 clear:清空掉线终端''' if ccid and not clear: #添加或更新客户端 p[ccid] = addr pp[addr] = ccid else: del p[ccid] del pp[addr] # 打印在线终端 for i in p: print(i,"<--->",p[i]) # for i in pp: # print(i,"-",pp[i]) def get_addr(ccid): '''获取ccid对应的地址''' try: # 如果有直接返回 return p[ccid] except Exception: #没有就返回None return None def get_ccid(addr): try: # 如果有直接返回 return pp[addr] except Exception: #没有就返回None return None ''' +--------+--------------+----------+----------+--------------+------------------------+--------+ | ff | 长度 | 地址信息 | 信令 | 内容 | 校验位 | +--------+--------------+----------+----------+--------------+------------------------+--------+ | | 2Bytes | 4Bytes | 心跳 | 0x01 | 时间戳32bit(单位ms) | 2Bytes | + $TTMS +--------------+----------+----------+--------------+------------------------+--------+ | | 2Bytes | 4Bytes | 传输数据 | 0xaa | 载荷 | 2Bytes | +--------+--------------+----------+----------+--------------+------------------------+--------+ ''' def send(fd,data): try: fd.send(data) return 0 except: return -1 def tt_hh(addr,data): '''处理心跳''' # 更新pairs ccid = data[7+offs:11+offs] update_pairs(addr,ccid) # 组帧再返回 # 从缓存中匹配数据并重发 # 仅有新终端上线时才重发 if ccid in cache: msgs = cache[ccid] for i in range(len(msgs)): try: addr.send(msgs[i]) # 发送成功则清除对应缓存数据 del cache[ccid][i] # not msgs[i] except: # 异常时退出循环 break return 0 def tt_trans(addr,data): '''发送数据到指定ccid''' tccid = data[7+offs:11+offs] taddr = get_addr(tccid) # 组帧 msg = bytearray(data[:]) msg[:5]=b'$TTMX' msg[7+offs:11+offs] = get_ccid(addr) # 计算校验和 crc = 0 for i in msg[:-1]: crc = crc ^ i msg[-1]=crc if not taddr: # 未报到匹配的在线终端 #更新缓存 #{tccid1=[msg1,msg2,...],...} cache[tccid].append(msg) save_cache(cache) print("终端未在线或ccid错。") return 0 print("匹配终端:",tccid,"<--->",taddr) # 发送 try: taddr.send(msg) except Exception: # 发送失败 # 更新pairs,清空对应终端 update_pairs(taddr,tccid,1) #更新缓存 #{tccid1=[msg1,msg2,...],...} cache[tccid].append(msg) save_cache(cache) print("发送失败,终端可能已掉线。",taddr) return 0 def err_handle(flag,addr): '''错误处理 默认打印错误信息 1-非法连接,关闭连接 ''' return 0 def tt_decode(addr,data): '''处理收到的数据''' if data[:5] != b'$TTMS' or len(data) <12: # 数据长度不可能小于12,考虑非法接入的可能,主动断开连接 return -1 cmd = data[11+offs] if cmd==0x01: # 处理心跳 tt_hh(addr,data) elif cmd==0xaa: # 处理数据 tt_trans(addr,data) else: return -1 class MyServer(socketserver.BaseRequestHandler): def handle(self):#回调 print ("终端已上线:",self.client_address) conn = self.request # print(type(conn),conn.fd) while True: data = conn.recv(200) # update_pairs(conn,data[7+offs:11+offs]) if not data: break print ("接收到新数据",self.client_address,",长度",len(data),":\r\n", data.hex(" ")) #ack_msg = "got from "+ str(ip) + " to " + str(self.client_address) + data # conn.send(data[:6]) if tt_decode(conn,data) : conn.close() print(self.client_address,"疑是非法连接,已切断。") break if __name__ == '__main__': # load_cache cache = load_cache() server = socketserver.ThreadingTCPServer(('localhost',8005), MyServer) ip, port = server.server_address print ("服务端已建立:",ip, port) server.serve_forever()