Server-py/sri-server-bg03/lib/Connection_b1.py

# from hub import methods, Global

import struct
import asyncio
import time
# from concurrent.futures import ThreadPoolExecutor

import sys
import importlib

# --- for linux
sys.path.append('/home/server/repositories/repositories/sri-project.demo-py/sri-server-bg03')
sys.path.append('/home/server/repositories/repositories/sri-project.demo-py/3rdparty')

protobuf = importlib.import_module(f"xprotobuf.protocol_pb2")
methods = importlib.import_module(f"xlib")


class SRIConnection(asyncio.Protocol):
    """"""

    # head_sequence = '<hh'  # 字节序规则
    # all_connection_dict = {}  # {<ipv4>: (socket, update_at, type)} | {<连接id>: (socket对象, 最后一次请求时间, 客户端类型)}
    # executor = ThreadPoolExecutor()
    clients = {}  # {<ipv4>: (socket, update_at, type)} | {<连接id>: (socket对象, 最后一次请求时间, 客户端类型)}

    def connection_made(self, client):
        self.client = client
        self.peername = client.get_extra_info('peername')
        print(f"Connection from {self.peername}")

        # --- 自定义参数
        self.head_sequence = '<hh'  # 字节序规则
        self.head_size = struct.calcsize(self.head_sequence)
        self.connection_id = f"{self.peername[0].replace('.', '')}{self.peername[1]}"  # 12700130000

        # --- fill ---
        self.clients[self.connection_id] = client, methods.now_ts(), 'type=vehicle'

    def data_received(self, data):
        asyncio.create_task(self.handle_data(data))

    async def handle_data(self, data):

        loop = asyncio.get_running_loop()
        # 全局的 ThreadPoolExecutor，由 asyncio 库自动创建和管理。它的线程池大小默认是机器的CPU核心数乘以5
        await loop.run_in_executor(None, self.process_data, data)
        # 自定义的 ThreadPoolExecutor
        # await loop.run_in_executor(SRIConnection.executor, self.process_data, data)

        # result = await asyncio.get_running_loop().run_in_executor(None, self.process_data, data)
        # if result:
        #     self.client.write(result)

    def process_data(self, data):

        head_data = data[:self.head_size]
        body_data = data[self.head_size:]

        command_id, body_length = struct.unpack(self.head_sequence, head_data)
        methods.debug_log('SRIConnection.37', f"Received values: {command_id}, {body_length}")

        if command_id == protobuf.CS_KeepAlive:  # 2008
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: None")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: None")

        elif command_id == protobuf.CS_Sign:  # 2000

            # --- 接收请求数据
            object = protobuf.CSSign()
            object.ParseFromString(body_data)
            methods.debug_log(f"{self.connection_id} | SRIConnection.100",
                              f"Received Response: account={object.account}, password={object.password}")

            # --- 发送返回数据
            object = protobuf.SCSign()
            object.ret = True
            object.uid = 112233
            object.cid = 223344
            object.name = 'aabbcc'
            re_command_id = protobuf.SC_Sign  # 4000
            re_body_length = object.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = object.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

            # --- test ---
            # for _ in range(10):
            #     for i in range(3)[::-1]:
            #         methods.debug_log(f"{self.connection_id} | SRIConnection.92", f"sleep {i}s!")
            #         time.sleep(1)
            #     methods.debug_log(f"{self.connection_id} | SRIConnection.94", f"send re_send_data")
            #     self.client.write(re_send_data)

        elif command_id == protobuf.CS_Add:  # 2009

            # --- 接收请求数据 2009
            """
            CSAdd: 消息体
            CSAdd.serial: string
            CSAdd.type: int32 EgoType::Car | EgoType::None EgoType::User EgoType::Car
            CSAdd.name: string
            """
            object0 = protobuf.CSAdd()
            object0.ParseFromString(body_data)
            methods.debug_log(f"{self.connection_id} | SRIConnection.105", f"#serial: {object0.serial}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.105", f"#name: {object0.name}")

            # --- 发送返回数据 4016 todo 向所有 EgoType::User 也就是舱端发送消息
            """
            Robot: 消息体
            Robot.rid: int32
            Robot.name: string
            Robot.type: int32
            Robot.state: RobotState Offline Online Busy
            """
            o1 = protobuf.Robot()
            o1.rid = 112233
            o1.name = "aabbcc"
            o1.type = 123
            o1.state = protobuf.Robot.Online
            """
            SCAddRobot: 消息体
            SCAddRobot.robot: Robot
            """
            o2 = protobuf.SCAddRobot()
            o2.robot.CopyFrom(o1)
            re_command_id = protobuf.SC_NotifyAdd  # 4016
            re_body_length = o2.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o2.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

            # --- 发送返回数据 4007 todo 向请求来的车端，返回设置是否成功的消息
            o1 = protobuf.SCAdd()
            o1.ret = True
            o1.uid = 112233
            o1.name = 'aabbcc'
            re_command_id = protobuf.SC_Add  # 4007
            re_body_length = o1.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o1.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

        elif command_id == protobuf.CS_Offer:  # 2004

            # --- 接收请求数据
            object = protobuf.Offer()
            object.ParseFromString(body_data)
            print(f"Received Response: peer={object.peer}")

            # --- 发送返回数据
            o1 = protobuf.Offer()
            o1.index = object.index
            o1.peer = object.peer
            o1.type = object.type
            o1.sdp = object.sdp
            re_command_id = protobuf.SC_NotifyOffer  # 4012
            re_body_length = o1.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o1.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

        elif command_id == protobuf.CS_Answer:  # 2005

            # --- 接收请求数据
            object = protobuf.Answer()
            object.ParseFromString(body_data)
            print(f"Received Response: peer={object.peer}")

            # --- 发送返回数据
            o1 = protobuf.Offer()
            o1.index = object.index
            o1.peer = object.peer
            o1.type = object.type
            o1.sdp = object.sdp
            re_command_id = protobuf.SC_NotifyAnswer  # 4011
            re_body_length = o1.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o1.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

        elif command_id == protobuf.CS_Candidate:  # 2006

            # --- 接收请求数据
            object = protobuf.Candidate()
            object.ParseFromString(body_data)
            print(f"Received Response: peer={object.peer}")

            # --- 发送返回数据
            o1 = protobuf.Offer()
            o1.index = object.index
            o1.peer = object.peer
            o1.type = object.type
            o1.candidate = object.candidate
            o1.sdpMLineIndex = object.sdpMLineIndex
            o1.sdpMid = object.sdpMid
            o1.egotype = object.egotype

            re_command_id = protobuf.SC_NotifyCandidate  # 4013
            re_body_length = o1.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o1.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

        elif command_id == protobuf.CS_Req:  # 2001  todo 实现收到qt端2001消息，向车端发送4009消息

            # --- 接收请求数据
            """
            CSReq: 消息体
            CSReq.peer: int32（uid，用户标识）
            CSReq.index: int32（相机位置，RenderPosition）
            CSReq.egotype: int32（终端类型，舱端/车端）
            """
            o1 = protobuf.CSReq()
            o1.ParseFromString(body_data)
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"#peer: {o1.peer}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"#index: {o1.index}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"#egotype: {o1.egotype}")

            # --- 发送返回数据
            """
            CSReq: 消息体
            CSReq.peer: int32（uid，用户标识）
            CSReq.index: int32（相机位置，RenderPosition）
            CSReq.egotype: int32（终端类型，舱端/车端）
            """
            o2 = protobuf.CSReq()
            o2.peer = o1.peer
            o2.index = o1.index
            o2.type = o1.egotype
            re_command_id = protobuf.SC_NotifyReq  # 4009
            re_body_length = o2.ByteSize()
            re_head_data = struct.pack(self.head_sequence, re_command_id, re_body_length)
            re_body_data = o2.SerializeToString()
            re_send_data = re_head_data + re_body_data
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_command_id: {re_command_id}")
            methods.debug_log(f"{self.connection_id} | SRIConnection.113", f"re_send_data: {re_send_data}")
            self.client.write(re_send_data)

    def connection_lost(self, exc):
        print(f"SRIConnection closed by {self.peername}")

    # def connection_lost(self, exc):
    #     print(f"Connection closed by {self.connection_id}")
    #     if self.connection_id in SRIConnection.clients:
    #         del SRIConnection.clients[self.connection_id]

    @classmethod
    async def run(cls):
        loop = asyncio.get_running_loop()
        server = await loop.create_server(
            lambda: SRIConnection(),
            '0.0.0.0', 20917
        )
        async with server:
            print(f"Server listening on 0.0.0.0:20917")
            methods.debug_log(f"SRIConnection.221", f"Server listening on 0.0.0.0:20917")
            await server.serve_forever()


if __name__ == '__main__':
    asyncio.run(SRIConnection.run())
    # asyncio.run(main())