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

"""
todo
    1、存在消息被异步多次处理的问题
"""
# from hub import methods, Global

import struct
import asyncio
import time
import socket

import sys
import importlib

# --- for linux
sys.path.append('/home/sri/repositories/repositories/sri-project.demo-py/sri-dino-pyserver01')
sys.path.append('/home/sri/repositories/repositories/sri-project.demo-py/sri-pysdk')

protobuf = importlib.import_module(f"xprotobuf.protocol_pb2")
methods = importlib.import_module(f"xlib")
clients = {}  # {<ipv4>: (socket, update_at, type)} | {<连接id>: (socket对象, 最后一次请求时间, 客户端类型)}
serial_rid_dict = {
    '65F7171A-5585-46C7-A9D6-967ABA9EB223': 1000000,
    '7AF3F619-5067-4EE0-A710-89A6CB097EFE': 1000001,
    'ECB93A87-560B-4022-8C5F-CBF9FE1E596A': 1000002,
    'C0D14B6F-0FF0-4B68-877D-D2CB886FCD0E': 1000002,
    'E537DDFB-6E3E-4E1A-AD18-AC21393BE300': 1000004,  # 正在使用
}
account_uid_dict = {
    'ego': 3
}


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

    head_sequence = '<hh'  # 字节序规则
    head_size = struct.calcsize(head_sequence)
    message_data = b''

    def connection_made(self, client):
        """
        建立客户端连接
        """
        peername = client.get_extra_info('peername')
        self.connection_id = f"{peername[0].replace('.', '')}-{peername[1]}"  # 12700130000 客户端id
        self.client = client
        # self.data = b''
        self.client_type = None
        self.client_info = None
        self.update_at = methods.now_ts()
        clients[self.connection_id] = self

        # 获取底层 socket
        # sock = self.client.get_extra_info('socket')

        # 查看缓冲区大小
        # recv_buffer_size = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
        # send_buffer_size = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
        # methods.debug_log(f'{self.connection_id}|SRIConnection70', f"Receive buffer size: {recv_buffer_size}")
        # methods.debug_log(f'{self.connection_id}|SRIConnection70', f"Send buffer size: {send_buffer_size}")

    def connection_lost(self, exc):
        """
        关闭连接
        """
        methods.debug_log(f"{self.connection_id}|SRIConnection085", f"连接已关闭")

        # --- 处理车端掉线，通知所有舱端
        if self.client_type == 'vehicle':
            """
            SCDelRobot: 消息体
            SCDelRobot.peer: int32
            SCDelRobot.egotype: int32
            """
            o2 = protobuf.SCDelRobot()
            o2.peer = self.client_info.get('rid')  # 车端唯一标识
            o2.egotype = self.client_info.get('egotype')
            re_command_id = protobuf.SC_NotifyDel  # 4017
            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}|SRIConnection085", f"re_command_id: {re_command_id}")
            for item in clients.values():
                if item.client_type == 'cockpit':
                    item.client.write(re_send_data)

        # --- clean
        if self.connection_id in clients:
            del clients[self.connection_id]

    def data_received(self, data):
        """
        消息处理
        """
        # methods.debug_log(f'{self.connection_id}|SRIConnection96', f"data length: {len(data)}")
        asyncio.create_task(self.handle_data(data))

    async def handle_data(self, data):
        """
        消息处理
        """
        # methods.debug_log(f'{self.connection_id}|SRIConnection102', f"data length: {len(data)}")
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, self.process_data, data)

    def process_data(self, data):
        """
        消息处理
        """
        try:
            methods.debug_log(f"{self.connection_id}|SRIConnection114", f"receive: {len(data)}, message: {repr(data)}")
            self.message_data += data  # 执行了这个以后

            methods.debug_log(f"{self.connection_id}|SRIConnection114", f"--- 1")
            methods.debug_log(f"{self.connection_id}|SRIConnection114", f"--- 2")
            methods.debug_log(f"{self.connection_id}|SRIConnection114", f"--- 3")

            count = 0
            while True:

                # --- print
                count += 1
                methods.debug_log(f"{self.connection_id}|SRIConnection123",
                                  f"---------------------------------- while count: {count}")

                # 确保有足够的字节来处理消息头
                if len(self.message_data) < self.head_size:
                    break

                # 获取消息头
                head_data = self.message_data[:self.head_size]
                command_id, body_length = struct.unpack(self.head_sequence, head_data)
                methods.debug_log(f'{self.connection_id}|SRIConnection130',
                                  f"command_id: {command_id}, body_length: {body_length}")

                # 检查命令ID是否有效
                if not (2000 <= command_id < 5000):
                    self.message_data = b''  # 清空无效数据
                    break

                # 检查完整消息是否接收完毕
                total_length = self.head_size + body_length
                if len(self.message_data) < total_length:
                    break  # 数据不完整，等待更多数据

                # 调用相应处理方法
                method = getattr(self, f'message{command_id}', None)
                if method:
                    method(self.message_data[self.head_size:total_length])  # 处理完整消息
                else:
                    methods.debug_log(f"{self.connection_id}|SRIConnectionError",
                                      f"No handler for command_id: {command_id}")

                # 移除已处理的消息
                self.message_data = self.message_data[total_length:]

        except Exception as exception:
            methods.debug_log(f"{self.connection_id}|SRIConnection132", f"#exception: {exception}")
            methods.debug_log(f"{self.connection_id}|SRIConnection123", f"#traceback: {methods.trace_log()}")

    def message2008(self, body_data):
        # methods.debug_log(f"{self.connection_id}|SRIConnection80", f"message: 2008")
        pass

    def message2009(self, body_data):

        # --- 监听 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}|SRIConnection90", f"#serial: {object0.serial}")
        methods.debug_log(f"{self.connection_id}|SRIConnection90", f"#name: {object0.name}")
        methods.debug_log(f"{self.connection_id}|SRIConnection90", f"#type: {object0.type}")

        # --- update ---
        self.client_type = 'vehicle'
        self.client_info = {
            'rid': serial_rid_dict.get(object0.serial),
            'name': object0.name,
            'serial': object0.serial,
            'egotype': 2,  # 车端类型
        }

        # --- send 4016
        """
        SCAddRobot: 消息体
        SCAddRobot.robot: Robot
        Robot: 消息体
        Robot.rid: int32
        Robot.name: string
        Robot.type: int32
        Robot.state: RobotState Offline Online Busy
        """
        o1 = protobuf.Robot()
        o1.rid = self.client_info.get('rid')
        o1.name = object0.name
        o1.type = 2  # 0 EgoType::None 1 EgoType::User 2 EgoType::Car
        o1.state = protobuf.Robot.Online
        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}|SRIConnection136", f"sendmessage: {re_command_id}")

        # --- send 4016 发送全部舱端
        for item in clients.values():
            if item.client_type == 'cockpit':
                item.client.write(re_send_data)

        # --- send 4007  todo 凯强说并未用到
        # o1 = protobuf.SCAdd()
        # o1.ret = True
        # o1.uid = 112233  # todo 这个uid，应该是哪个舱端永辉在操作这个车 | 但是现在还没有人操控车
        # o1.name = object0.name
        # 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} | SRIConnection150", f"re_command_id: {re_command_id}")
        # self.client.write(re_send_data)

    def message2000(self, body_data):

        # --- 解析消息体
        object = protobuf.CSSign()
        object.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection162", f"#account: {object.account}")
        methods.debug_log(f"{self.connection_id}|SRIConnection162", f"#password: {object.password}")

        # --- update ---
        self.client_type = 'cockpit'
        self.client_info = {
            'uid': 3,  # 对应数据库里的ego的id
            'name': 'ego',  # 对应数据库里
            'egotype': 1,  # 舱端类型
        }

        # --- send 4000
        object = protobuf.SCSign()
        object.ret = True
        object.uid = self.client_info.get('uid')
        object.name = self.client_info.get('name')
        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}|SRIConnection175", f"re_command_id: {re_command_id}")
        self.client.write(re_send_data)

    def message2010(self, body_data):

        # --- send 4008 发送全部车端信息列表
        """
        Robot: 消息体
        Robot.rid: string
        Robot.name: string
        Robot.type: int32 EgoType::Car | EgoType::None EgoType::User EgoType::Car
        Robot.state: enum Offline Online Busy
        """
        o2 = protobuf.SCRobot()
        for item in clients.values():
            if item.client_type and item.client_type == 'vehicle':
                o1 = protobuf.Robot()
                o1.rid = item.client_info.get('rid')
                o1.name = item.client_info.get('name')
                o1.type = 2  # EgoType::Car
                o1.state = protobuf.Robot.RobotState.Value('Online')
                o2.robot.add().CopyFrom(o1)

        re_command_id = protobuf.SC_Robot  # 4008
        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}|SRIConnection217", f"re_command_id: {re_command_id}")
        self.client.write(re_send_data)

    def message2001(self, body_data):

        # --- 解析消息体
        """
        CSReq: 消息体
        CSReq.peer: int32（rid，车端唯一标识）
        CSReq.index: int32（相机位置，RenderPosition）
        CSReq.egotype: int32（终端类型，舱端/车端）
        """
        o1 = protobuf.CSReq()
        o1.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection299.message2001", f"#peer: {o1.peer}")
        methods.debug_log(f"{self.connection_id}|SRIConnection299.message2001", f"#peer: {o1.index}")
        # methods.debug_log(f"{self.connection_id}|SRIConnection235.message2001",
        #                   f"#=============================index: {o1.index}")
        methods.debug_log(f"{self.connection_id}|SRIConnection299.message2001", f"#egotype: {o1.egotype}")
        methods.debug_log(f"{self.connection_id}|SRIConnection299.message2001", f"#uid: {self.client_info.get('uid')}")
        methods.debug_log(f"{self.connection_id}|SRIConnection299.message2001", f"#rid: {self.client_info.get('rid')}")

        # --- send 4009 指定车端
        for item in clients.values():
            if item.client_info.get('rid') and item.client_info.get('rid') == o1.peer:
                """
                CSReq: 消息体
                CSReq.peer: int32（rid，车端唯一标识）
                CSReq.index: int32（相机位置，RenderPosition）
                CSReq.egotype: int32（终端类型，舱端/车端）
                """
                o2 = protobuf.CSReq()
                o2.peer = self.client_info.get('uid')
                o2.index = o1.index
                o2.egotype = 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}|SRIConnection217", f"re_command_id: {re_command_id}")
                item.client.write(re_send_data)

    def message2002(self, body_data):

        # --- 解析消息体
        """
        CSRep: 消息体
        CSRep.desc: VideoDesc
        CSRep.peer: int32
        CSRep.index: int32
        CSRep.egotype: int32
        """
        o1 = protobuf.CSRep()
        o1.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection319", f"#desc: {o1.desc}")
        methods.debug_log(f"{self.connection_id}|SRIConnection319", f"#peer: {o1.peer}")
        methods.debug_log(f"{self.connection_id}|SRIConnection319", f"#index: {o1.index}")
        methods.debug_log(f"{self.connection_id}|SRIConnection319", f"#egotype: {o1.egotype}")
        methods.debug_log(f"{self.connection_id}|SRIConnection235", f"#uid: {self.client_info.get('uid')}")
        methods.debug_log(f"{self.connection_id}|SRIConnection235", f"#rid: {self.client_info.get('rid')}")

        # --- send 4010 通知指定舱端
        for item in clients.values():
            if item.client_info.get('uid') and item.client_info.get('uid') == o1.peer:
                o2 = protobuf.CSRep()
                o2.desc = o1.desc
                o2.peer = self.client_info.get('rid')
                o2.index = o1.index
                o2.egotype = o1.egotype
                re_command_id = protobuf.SC_NotifyRep  # 4010
                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}|SRIConnection217", f"re_command_id: {re_command_id}")
                item.client.write(re_send_data)

    def message2004(self, body_data):

        # --- 解析消息体
        """
        Offer: 消息体
        Offer.index: int32
        Offer.peer: int32（车端rid）
        Offer.type: string
        Offer.sdp: string
        """
        object = protobuf.Offer()
        object.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection348", f"#peer: {object.peer}")
        methods.debug_log(f"{self.connection_id}|SRIConnection348", f"#uid: {self.client_info.get('uid')}")
        methods.debug_log(f"{self.connection_id}|SRIConnection348", f"#rid: {self.client_info.get('rid')}")

        # --- send 4012 指定车端
        for item in clients.values():
            if item.client_info.get('rid') and item.client_info.get('rid') == object.peer:
                o1 = protobuf.Offer()
                o1.index = object.index
                o1.peer = self.client_info.get('uid')  # 将舱端id赋值到指定车端对象
                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}|SRIConnection341", f"re_command_id: {re_command_id}")
                item.client.write(re_send_data)

    def message2005(self, body_data):

        # --- 解析消息体
        """
        Answer: 消息体
        Answer.index: int32
        Answer.peer: int32（舱端uid）
        Answer.type: string
        Answer.sdp: string
        """
        object = protobuf.Answer()
        object.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection411", f"#peer: {object.peer}")
        methods.debug_log(f"{self.connection_id}|SRIConnection411", f"#uid: {self.client_info.get('uid')}")
        methods.debug_log(f"{self.connection_id}|SRIConnection411", f"#rid: {self.client_info.get('rid')}")

        # --- send 4011 指定舱端
        count = 0
        for item in clients.values():

            count += 1
            methods.debug_log(f"{self.connection_id}|SRIConnection428",
                              f"[{count}]item.client_info: {item.client_info}")

            if item.client_info.get('uid') and item.client_info.get('uid') == object.peer:
                o1 = protobuf.Answer()
                o1.index = object.index
                o1.peer = self.client_info.get('rid')  # 将车端id赋值到指定舱端对象，
                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}|SRIConnection428", f"re_command_id: {re_command_id}")
                methods.debug_log(f"{self.connection_id}|SRIConnection428",
                                  f"=====================================================-----index: {o1.index}")
                item.client.write(re_send_data)

    def message2006(self, body_data):

        # --- 解析消息体
        """
        Candidate: 消息体
        Candidate.index: int32
        Candidate.peer: int32（车端rid）
        Candidate.type: string
        Candidate.candidate: string
        Candidate.sdpMLineIndex: int32
        Candidate.sdpMid: string
        Candidate.egotype: int32
        """
        object = protobuf.Candidate()
        object.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection413", f"#peer: {object.peer}")
        methods.debug_log(f"{self.connection_id}|SRIConnection413", f"#candidate.in: {object.candidate}")
        methods.debug_log(f"{self.connection_id}|SRIConnection413", f"#uid: {self.client_info.get('uid')}")
        methods.debug_log(f"{self.connection_id}|SRIConnection413", f"#rid: {self.client_info.get('rid')}")

        for item in clients.values():

            # --- send 4013 舱端到车端
            if item.client_info.get('rid') and item.client_info.get('rid') == object.peer:
                o1 = protobuf.Candidate()
                o1.index = object.index
                o1.peer = self.client_info.get('uid')  # 将舱端id赋值到指定车端对象，
                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}|SRIConnection409", f"re_command_id: {re_command_id}")
                methods.debug_log(f"{self.connection_id}|SRIConnection409", f"#candidate.out: {o1.candidate}")
                item.client.write(re_send_data)

            # --- send 4013 车端到舱端
            if item.client_info.get('uid') and item.client_info.get('uid') == object.peer:
                o1 = protobuf.Candidate()
                o1.index = object.index
                o1.peer = self.client_info.get('rid')  # 将车端id赋值到指定舱端对象，
                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}|SRIConnection426", f"re_command_id: {re_command_id}")
                methods.debug_log(f"{self.connection_id}|SRIConnection426", f"#candidate.out: {o1.candidate}")
                item.client.write(re_send_data)

    def message2007(self, body_data):

        # --- 解析消息体
        """
        Leave: 消息体
        Leave.peer: int32（车端rid）
        Leave.egotype: int32
        """
        o1 = protobuf.Leave()
        o1.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection497", f"#peer: {o1.peer}")

        # --- send 4009 指定车端
        for item in clients.values():
            if item.client_info.get('rid') and item.client_info.get('rid') == o1.peer:
                """
                Leave: 消息体
                Leave.peer: int32（车端rid）
                Leave.egotype: int32
                """
                o2 = protobuf.Leave()
                o2.peer = self.client_info.get('uid')
                o2.egotype = o1.egotype
                re_command_id = protobuf.SC_NotifyLeave  # 4014
                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}|SRIConnection515", f"re_command_id: {re_command_id}")
                item.client.write(re_send_data)

    def message2014(self, body_data):

        # --- 解析消息体
        """
        message CSState
        {
            UserState state=1;
            int32 uid=2;
        };
        """
        o1 = protobuf.CSState()
        o1.ParseFromString(body_data)
        methods.debug_log(f"{self.connection_id}|SRIConnection530", f"#state: {o1.state}")
        methods.debug_log(f"{self.connection_id}|SRIConnection530", f"#uid: {o1.uid}")

        # --- send 4016 发送全部舱端
        for item in clients.values():
            if item.client_type == 'cockpit':
                """
                message SCState
                {
                    UserState state=1;
                    int32 uid=2;
                };
                """
                o2 = protobuf.SCState()
                o2.state = o1.state
                o2.uid = self.client_info.get('rid')
                re_command_id = protobuf.SC_State  # 4022
                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}|SRIConnection551", f"re_command_id: {re_command_id}")
                item.client.write(re_send_data)

    @staticmethod
    async def check_clients():
        """
        剔除掉线连接
        """
        count = 0
        while True:
            count += 1
            print(f"#count: {count}", flush=True)

            now_at = methods.now_ts()

            for connection_id in list(clients.keys()):
                object = clients.get(connection_id)
                print(f"#client: {object.client}, #update_at: {object.update_at}, #client_type: {object.client_type}",
                      flush=True)

            await asyncio.sleep(3)  # 发送消息的间隔时间

    @staticmethod
    async def run():
        """
        """
        # --- define ---
        loop = asyncio.get_running_loop()
        server = await loop.create_server(
            lambda: SRIConnection(),
            '0.0.0.0', 20917
        )

        # --- start ---
        async with server:
            print("Server listening on 0.0.0.0:20917", flush=True)
            # await loop.create_task(SRIConnection.check_clients())
            await server.serve_forever()


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