237 lines
5.7 KiB
Python
Executable File
237 lines
5.7 KiB
Python
Executable File
#!/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])
|
||
except:
|
||
# 异常时退出循环，不用再尝试发送后续msgs
|
||
# 退出前清空已发送成功的缓存数据
|
||
# del方法会改变list大小
|
||
while i:
|
||
cache[ccid].pop(0)
|
||
i = i - 1
|
||
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:
|
||
# 发送失败
|
||
taddr.close()
|
||
# 更新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):
|
||
"""处理收到的数据
|
||
数据正常返回0
|
||
数据异常返回-1"""
|
||
|
||
# 依据帧头和长度初判数据有效性
|
||
if data[:5] != b"$TTMS" or len(data) < 12 or len(data) > 200:
|
||
return -1
|
||
|
||
cmd = data[11 + offs]
|
||
# 回退到if-elif，服务器默认版本3.8
|
||
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()
|