【初始服务代码】

code/tx1_box_serv/apps/api/http_serv.py

+#!/usr/bin/evn python
+# coding=utf-8
+from flask import Blueprint,request,redirect,url_for
+from werkzeug import secure_filename
+import json,re,time,io
+import sys
+import os
+from apps.commons.util import *
+import apps.gbvar.ws_serv_global  as sv_global
+import apps.gbvar.database_global as database
+import apps.gbvar.dict_properties as dict_p
+import apps.commons.editXml as editXml
+
+
+tx1_box = Blueprint('tx1_box', __name__)
+#数据接收接口
+@tx1_box.route('/events', methods=["POST"])
+def dataRecv():
+
+    data = request.get_data()
+    print data
+    send_all_data(sv_global.get_value(),data)
+
+    return data
+
+# 给所有用户发送信息
+def send_all_data(sv, data):
+    if sv.users is not None and len ( sv.users ) > 0:
+        for addr in sv.users:
+            print 'addr=%s' % addr
+            sv.send_data (data, addr )
+
+
+
+#系统初始化：读取配置文件获取初始化任务列表
+def get_task_info():
+    dictProperties = dict_p.get_value()
+    result_list = []
+    if dictProperties:
+        for i in range(4):
+            result_data = {}
+            task_id = 'task_%s' % i
+            ds = dictProperties[task_id]
+            if ds.find ( "@" ) != -1:
+                task_inf = re.split ( '[@]', ds )
+                result_data.setdefault ( "task_id",i+1 )
+                result_data.setdefault ( "task_name", task_inf[1] )
+                result_data.setdefault ( "task_info", readerXml(task_inf[0]) )
+                result_list.append ( result_data )
+            else:
+                continue
+    return json.dumps ( result_list, ensure_ascii=False )
+
+#初始化任务详情信息：包含algo_args信息、calibration标定参数信息以及roi信息的组合
+def creat_mtasks():
+
+    '''任务配置参数'''
+    mtasks = []
+    '''场景'''
+    scenes = []
+    '''标定信息'''
+    callibration = {}
+    '''rois信息'''
+    rois = []
+    '''最新设置的算法配置参数'''
+    config = {}
+    '''basicConfig系统默认算法参数'''
+    basicConfig = {}
+    '''play_urls待分析的原始视频源的输入地址'''
+    play_urls = {}
+    return mtasks
+#读取任务信息参数
+def readerXml(file_path):
+    return editXml.readerXml ( file_path )
\ No newline at end of file

code/tx1_box_serv/apps/api/ws_serv.py

+# coding: utf-8
+
+from socket import *
+import json, time, threading
+import apps.gbvar.ws_serv_global  as sv_global
+config = {
+    'LISTEN_CLIENT': 50,
+    'KEY': '391f10fadc339e9ec5fa15af60030ac1',
+    'SIZE': 2048,
+    'TIME_OUT': 1000,
+    'HEART_TIME': 5,
+    'MAGIC_STRING': '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
+    'HANDSHAKE_STRING': "HTTP/1.1 101 Switching Protocols\r\n" \
+                        "Upgrade:websocket\r\n" \
+                        "Connection: Upgrade\r\n" \
+                        "Sec-WebSocket-Accept: {1}\r\n" \
+                        "WebSocket-Location: ws://{2}/chat\r\n" \
+                        "WebSocket-Protocol:chat\r\n\r\n"
+}
+
+
+
+class Server ():
+    """
+    服务端基类
+    """
+    def __init__(self,h,p):
+        self.sock = socket ( AF_INET, SOCK_STREAM )
+        self.sock.bind ( (h,p) )  # 监听端口
+        #self.sock.bind ( (config['HOST'], config['PORT']) )  # 监听端口
+        self.sock.listen ( config['LISTEN_CLIENT'] )  # 监听客户端数量
+
+        # 所有监听的客户端
+        self.clients = {}
+        self.thrs = {}
+        self.users = {}
+        self.stops = []
+        self.h=h
+        self.p=p
+    # 监听客户端连接
+    def listen_client(self):
+        while 1:
+            # 循环监听
+            tcpClientSock, addr = self.sock.accept ()
+            address = addr[0] + ':' + str ( addr[1] )  # ip:port
+
+            # 握手
+            topInfo = tcpClientSock.recv ( 1024 )
+            headers = {}
+            if not topInfo:
+                tcpClientSock.close ()
+                continue
+
+            header, data = topInfo.split ( '\r\n\r\n', 1 )
+
+            try:
+                getInfo = header.split ( '\r\n' )[0].split ( ' ' )[1].split ( '/' )[1:]
+                print 'getInfo=%s' % getInfo
+                print 'getInfo[0] = %s' % getInfo[0]
+                if getInfo[0] == 'api/v1/websocket':
+                    self.users[address] = str ( getInfo[1] )
+                else:
+                    self.users[address] = '匿名用户'
+            except:
+                self.users[address] = '匿名用户'
+
+            for line in header.split ( '\r\n' )[1:]:
+                key, val = line.split ( ': ', 1 )
+                headers[key] = val
+
+            if 'Sec-WebSocket-Key' not in headers:
+                tcpClientSock.close ()
+                continue
+
+            import hashlib, base64
+            sec_key = headers['Sec-WebSocket-Key']
+            res_key = base64.b64encode ( hashlib.sha1 ( sec_key + config['MAGIC_STRING'] ).digest () )
+
+            #str_handshake = config['HANDSHAKE_STRING'].replace ( '{1}', res_key ).replace ( '{2}',
+             #                                                                               config['HOST'] + ':' + str (
+              #                                                                                  config['PORT'] ) )
+            str_handshake = config['HANDSHAKE_STRING'].replace ( '{1}', res_key ).replace ( '{2}',self.h+ ':' + str (self.p) )
+            tcpClientSock.send ( str_handshake )
+
+            # 握手成功 分配线程进行监听
+            print(address + '进来了')
+
+            self.clients[address] = tcpClientSock
+            self.thrs[address] = threading.Thread ( target=self.readMsg, args=[address] )
+            self.thrs[address].start ()
+            # print(self.clients)
+
+    def readMsg(self, address):
+        # print(self.clients)
+        if address not in self.clients:
+            return False
+
+        client = self.clients[address]
+
+        import select
+        time_out = 0
+        while 1:
+            # print(len(self.clients))
+            if address in self.stops:
+                self.close_client ( address )
+                print(address + u'已经离开了系统！')
+                break
+
+            # 检测超时
+            if time_out >= config['TIME_OUT']:
+                self.close_client ( address )
+                break
+
+            time_out += 5
+
+            infds, outfds, errfds = select.select ( [client, ], [], [], 5 )
+            if len ( infds ) == 0:
+                continue
+
+            time_out = 0
+            try:
+                info = client.recv ( 1024 )
+            except:
+                self.close_client ( address )
+                break
+
+            if not info:
+                continue
+
+            if info == 'quit':
+                self.close_client ( address )
+                break
+            code_len = ord ( info[1] ) & 127
+            if code_len == 126:
+                masks = info[4:8]
+                data = info[8:]
+            elif code_len == 127:
+                masks = info[10:14]
+                data = info[14:]
+            else:
+                masks = info[2:6]
+                data = info[6:]
+            i = 0
+            raw_str = ""
+            for d in data:
+                # print(masks, masks[i % 4])
+                raw_str += chr ( ord ( d ) ^ ord ( masks[i % 4] ) )
+                # print(raw_str)
+                i += 1
+
+            # 获取到输入的数据 向所有的客户端发送
+            # 开启线程记录
+
+            if raw_str:
+                t1 = threading.Thread ( target=self.send_data, args=[raw_str, address] )
+                t1.start ()
+
+    def send_data(self, data, address):
+        import struct
+        from urllib import unquote
+        try:
+            username = unquote ( self.users[address] )
+        except:
+            username = '匿名用户'
+        #if data:
+        #    data = str ( '【' + username + '说】' + data )
+        #else:
+        #    return False
+        token = "\x81"
+        length = len ( data )
+        if length < 126:
+            print '126'
+            token += struct.pack ( "B", length )
+        elif length <= 0xFFFF:
+            print '0xFFFF'
+            token += struct.pack ( "!BH", 126, length )
+        else:
+            print '!BQ'
+            token += struct.pack ( "!BQ", 127, length )
+        print 'token=%s' % token
+        # struct为Python中处理二进制数的模块，二进制流为C，或网络流的形式。
+
+        data = '%s%s' % (token, data)
+        print data
+        try:
+            #for key, val in self.clients.iteritems ():
+            #    print
+            #    client = val
+            #    try:
+            #       print 'sned=%s' % data
+            #       client.send(data)
+            #    except:
+            #        self.close_client ( key )
+            client = self.clients[address]
+            client.send ( data )
+        except:
+            pass
+
+    def close_client(self, address):
+        try:
+            client = self.clients.pop ( address )
+            self.stops.append ( address )
+            client.close ()
+            del self.users[address]
+        except:
+            pass
+
+        print(address + u'已经退出')
+
+    def get_result(self):
+        try:
+            return self.sv
+        except Exception:
+            return None
+
+#if __name__ == '__main__':
+#    c = Server ()
+#    c.listen_client ()
\ No newline at end of file

code/tx1_box_serv/apps/commons/editXml.py

+#!/usr/bin/evn python
+# coding=utf-8
+import sys, io
+import re
+#读取xml文件
+def readerXml(filePath):
+    # 打开xml文档
+    return io.open(filePath, 'r', encoding='GBK').read ().encode ( 'utf-8' )
+
+
+#写入xml文件
+def writeXml(filePath,data):
+    # 打开xml文档
+    f = io.open ( filePath, 'w', encoding='GBK' )
+    f.write(data)
+    f.close ()
\ No newline at end of file

code/tx1_box_serv/apps/commons/util.py

+#!/usr/bin/evn python
+# coding=utf-8
+import io
+class Properties ( object ):
+    def __init__(self, fileName):
+        self.fileName = fileName
+        self.properties = {}
+
+    def __getDict(self, strName, dictName, value):
+
+        if (strName.find ( '.' ) > 0):
+            k = strName.split ( '.' )[0]
+            dictName.setdefault ( k, {} )
+            return self.__getDict ( strName[len ( k ) + 1:], dictName[k], value )
+        else:
+            dictName[strName] = value
+            return
+
+    def getProperties(self):
+        try:
+            pro_file = io.open( self.fileName, 'r',encoding='utf-8' )
+            for line in pro_file.readlines ():
+                line = line.strip ().replace ( '\n', '' )
+                if line.find ( "#" ) != -1:
+                    line = line[0:line.find ( '#' )]
+                if line.find ( '=' ) > 0:
+                    strs = line.split ( '=' )
+                    strs[1] = line[len ( strs[0] ) + 1:]
+                    self.__getDict ( strs[0].strip (), self.properties, strs[1].strip () )
+        except Exception as e:
+            raise e
+        else:
+            pro_file.close ()
+        return self.properties
+
+    def get(self, key, default_value=''):
+        '''self.getProperties'''
+        if key in self.properties:
+            return self.properties[key]
+        return default_value
\ No newline at end of file

code/tx1_box_serv/apps/dao/firebird_dao.py

+# coding: utf-8
+import fdb
\ No newline at end of file

code/tx1_box_serv/apps/gbvar/database_global.py

+# coding: utf-8
+
+def _init():#初始化
+    global database
+    database = None
+
+
+def set_value(value):
+    """ 定义一个全局变量 """
+    global  database
+    database = value
+
+
+def get_value():
+    global database
+    return database
\ No newline at end of file

code/tx1_box_serv/apps/gbvar/dict_properties.py

+# coding: utf-8
+
+def _init():#初始化
+    global dict
+    dict = None
+
+
+def set_value(value):
+    """ 定义一个全局变量 """
+    global  dict
+    dict = value
+
+
+def get_value():
+    global dict
+    return dict
\ No newline at end of file

code/tx1_box_serv/apps/gbvar/ws_serv_global.py

+# coding: utf-8
+
+def _init():#初始化
+    global _global_ws
+    _global_ws = None
+
+
+def set_value(value):
+    """ 定义一个全局变量 """
+    global  _global_ws
+    _global_ws = value
+
+
+def get_value():
+    global _global_ws
+    return _global_ws
\ No newline at end of file

code/tx1_box_serv/apps/main.py

+# coding: utf-8
+from flask import Flask,Blueprint
+import threading
+from apps.api.http_serv import tx1_box
+from apps.api.ws_serv import Server
+import apps.gbvar.ws_serv_global  as sv_global
+import apps.gbvar.database_global as database
+import apps.gbvar.dict_properties as dict_p
+import fdb
+import sys
+import os
+from apps.commons.util import *
+
+app = Flask(__name__, static_folder='apps/static')
+# 获取本脚本所在的路径
+pro_path = os.path.split(os.path.realpath(__file__))[0]
+
+#配置文件路径
+con_path = '/'.join((pro_path, 'special.properties'))
+print con_path
+dictProperties=Properties(con_path).getProperties()
+
+# 创建路由
+app.register_blueprint(tx1_box, url_prefix='/api/v1/tx1_box')
+
+
+def start_ws(host,port):
+    try:
+        sv_global._init ()
+        __sv = Server(host,port)
+        sv_global.set_value(__sv)
+    except Exception,e:
+        print e.message
+        return
+
+def thread_start(__sv):
+    __sv.listen_client ()
+
+#系统初始化数据库
+def init():
+    database._init()
+    dict_p._init()
+    dict_p.set_value(dictProperties)
+    ###===================database=============================
+    con = fdb.connect ( host=dictProperties['tx1_host'], database=dictProperties['datasource_url'], user='sysdba',
+                        password='masterkey',
+                        charset='UTF8' )
+    cur = con.cursor ()
+    database.set_value(cur)
+
+    '''newLanguages = [
+        (1, 'Dylan')
+    ]
+    cur.executemany ( "insert into tb_traffic(id,name) VALUES(?,?)",
+                      newLanguages
+                      )
+    con.commit ()
+    sql_select = 'select * from tb_traffic'
+
+    # 显示全部内容
+    cur.execute ( sql_select )
+    print(cur.fetchall ())
+
+    cur.close ()
+    con.close ()'''
+    ####====================database===================
+
+#启动服务
+if __name__=='__main__':
+    #init()
+    _h=dictProperties['tx1_host']
+    _p=int(dictProperties['tx1_ws_port'])
+    start_ws(_h,_p)
+    print sv_global.get_value()
+    t1 = threading.Thread ( target=thread_start, name='start_ws_serv', args={sv_global.get_value (), } )
+    t1.start ()
+    print 'sv_global.get_value()=%s' % sv_global.get_value ()
+    # 启动http
+    app.run (
+        host=dictProperties['tx1_host'],
+        port=int(dictProperties['tx1_http_port'] ),debug=False,threaded=True)
+

code/tx1_box_serv/apps/special.properties

+#系统参数
+tx1_host=192.168.9.61
+tx1_http_port=20090
+tx1_ws_port=20070
+
+#firebird数据库
+datasource_url=F:/firebird/database/EVENT.FDB
+
+#初始化任务列表（说明：路径@任务名称
+task_0=E:/项目/pyWeb/static/xml/ServerConfig_2.xml@运维服务@./start.sh
+task_1=/root/UserApp/vioncfg/Ops/ServerConfig.xml@运维服务@./start.sh
+task_2=E:/项目/pyWeb/static/xml/ServerConfig_2.xml@运维服务@./start.sh
+task_3=/root/UserApp/vioncfg/Ops/ServerConfig.xml@运维服务@./start.sh
+
+#重启任务指令
+start_task=./task_start.sh
\ No newline at end of file