RenPyUtil：ren_communicator进阶

ZYKsslm · 发表于 2023-11-10 22:55:50

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要登录才可以下载或查看，没有账号？立即注册

×

本帖最后由 ZYKsslm 于 2023-11-12 09:00 编辑

RenPyUtil：ren_communicator进阶

Github同步项目 -> https://github.com/ZYKsslm/RenPyUtil
由于上一版代码BUG过多，请务必更新最新的代码。

这几天又重新构思了一下代码，并在github上的renpy项目提了issue -> https://github.com/renpy/renpy/issues/5105，最终取了一些折中的方案。
代码量也是超过了500行

源码如下：

[RenPy] 纯文本查看 复制代码

# 此文件提供了一系列基于Ren'Py的功能类，以供Ren'Py开发者调用
# 作者  ZYKsslm
# 仓库  [url=https://github.com/ZYKsslm/RenPyUtil]https://github.com/ZYKsslm/RenPyUtil[/url]
# 声明  该源码使用 MIT 协议开源，但若使用需要在程序中标明作者消息


init -1 python:

    from datetime import datetime
    import socket

    class Prompt(object):
        """命令类"""        

        def __init__(self, encode, *prompts):
            if encode:
                prompts = list(prompts)
                for a in range(len(prompts)):
                    prompts[a] = prompts[a].encode("utf-8")
            
            self.prompts = set(prompts)


    class RenServer(object):
        """该类为一个服务端类。基于socket进行多线程通信。

        在在子线程中运行的方法中使用renpy更新屏幕的函数（如`renpy.say()`、`renpy.call_screen()`等），可能引发异常。

        在子线程中运行的方法有：
            1. 使用`set_prompt`设定的命令方法。
            2. 使用`set_receive_event``set_conn_event``set_error_event`设定的事件方法。
            3. 所有进行通信的方法。
        """

        # 支持监听的事件
        EVENT = [
            "PROMPT",   # 命令事件
            "ERROR",    # 异常事件
            "CONNECT",  # 连接事件
            "RECEIVE",  # 接收事件
        ]

        PROMPT_EVENT = "PROMPT",
        ERROR_EVENT = "ERROR",
        CONNECT_EVENT = "CONNECT",
        RECEIVE_EVENT = "RECEIVE"
        

        def __init__(self, max_conn=5, max_data_size=1024, ip="0.0.0.0", port=8888):
            """初始化方法。

            Keyword Arguments:
                max_conn -- 最大连接数。 (default: {5})
                max_data_size -- 接收数据的最大大小。 (default: {1024})
                port -- 端口号。 (default: {None})
            """            

            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.port = port
            self.ip = ip
            self.max_data_size = max_data_size
            self.max_conn = max_conn
            self.socket.bind((self.ip, self.port))

            self.client_socket_list = []
            self.has_communicated = False
            self.current_prompt = None
            self.received = False
            self.reply = None

            self.prompt_dict = {}
            self.error_event = []
            self.receive_event = []
            self.conn_event = []

            self.error_log = {}

        def set_prompt(self, prompt: str | list | Prompt, func: function, encode=True, *args, **kwargs):
            """调用该方法，创建一个命令，当接收到该命令后执行绑定的函数。命令将作为第一个参数，客户端socket将作为第二个参数传入指定函数中。

            不定参数为绑定的函数参数。

            Arguments:
                prompt -- 命令语句。可为一个字符串或一个列表或一个Prompt对象，若为列表，则列表中所有命令都可触发命令。
                func -- 一个函数。

            Keyword Arguments:
                encode -- 是否编码。 (default: {True})
            """ 

            if not isinstance(prompt, Prompt):
                prompt = Prompt(encode, prompt)
            
            self.prompt_dict[prompt] = [func, args, kwargs]

        def set_reply(self, reply):
            """调用该方法，指定接收到消息后自动回复的消息。

            Arguments:
                reply -- 要回复的消息。
            """       

            self.reply = reply

        def set_error_event(self, func: function, *args, **kwargs):
            """调用该方法，指定当通信出现异常时的行为。异常信息将作为第一个参数传入指定函数中。

            不定参数为绑定的函数参数。

            Arguments:
                func 一个函数
            """            

            self.error_event = [func, args, kwargs]

        def set_receive_event(self, func: function, *args, **kwargs):
            """调用该方法，指定当接受到消息时的行为。接收到的数据将作为第一个参数，客户端socket将作为第二个参数传入指定函数中。

            不定参数为绑定的函数参数。

            Arguments:
                func -- 一个函数。
            """            

            self.receive_event = [func, args, kwargs]

        def set_conn_event(self, func: function, *args, **kwargs):
            """调用该方法，指定当客户端连接后的行为。客户端socket将作为第一个参数传入指定函数中。

            不定参数为绑定的函数参数。

            Arguments:
                func -- 一个函数。
            """            

            self.conn_event = [func, args, kwargs]
        
        def close(self):
            """调用该方法，关闭服务端。"""

            self.socket.close()     

        def close_all_conn(self):
            """调用该方法，关闭所有连接。"""   

            if self.client_socket_list:
                try:
                    for s in self.client_socket_list:
                        s.close()    
                except socket.error:
                    pass

        def close_a_conn(self, client_socket: socket.socket=None):
            """调用该方法，关闭一个指定socket连接。

            Keyword Arguments:
                socket -- 客户端socket。若该参数不填，则关闭最新的连接。 (default: {None})
            """    

            if client_socket:
                client_socket.close()
            else:
                try:
                    self.client_socket_list[len(self.client_socket_list)-1].close()    
                except IndexError or socket.err:
                    pass

        def listen_event(self, event, tip="", prompt=None):
            """调用该方法阻塞式监听一个事件，监听到事件后取消阻塞。

            Arguments:
                event -- 事件类型。

            Keyword Arguments:
                prompt -- 若为命令事件，则该参数为一个Prompt对象。 (default: {None})
                tip -- renpy提示界面的内容。 (default: {None})
            """            

            if event == RenServer.PROMPT_EVENT:
                while prompt != self.current_prompt:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenServer.ERROR_EVENT:
                event_counter = len(self.error_log)
                while len(self.error_log) <= event_counter:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenServer.CONNECT_EVENT:
                event_counter = len(server.client_socket_list)
                while len(server.client_socket_list) <= event_counter:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenServer.RECEIVE_EVENT:
                while not self.received:
                    renpy.notify(tip)
                    renpy.pause()

            else:
                return

        def run(self):
            """调用该方法，开始监听端口，创建连接线程。"""            

            self.has_communicated = True
            self.socket.listen(self.max_conn)
            renpy.invoke_in_thread(self._accept)

        def reboot(self, log_clear=False):
            """调用该方法将重新开始通信。

            Keyword Arguments:
                log_clear -- 若为True，将清除错误日志。 (default: {False})
            """            

            self.close()
            self.close_all_conn()
            if log_clear:
                self.error_log.clear()
            self.run()

        def _accept(self):
            """该方法用于创建连接线程，用于类内部使用，不应被调用。"""            

            while True:
                try:
                    client_socket, address = self.socket.accept()
                except Exception as err:
                    self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                    return

                if self.conn_event:
                    func, args, kwargs = self.conn_event
                    renpy.invoke_in_thread(func, client_socket, *args, **kwargs)

                self.client_socket_list.append(client_socket)
                renpy.invoke_in_thread(self._receive, client_socket, self.max_data_size)

        def _receive(self, client_socket: socket.socket, max_data_size):
            """该方法用于接收线程使用，处理接收事件，用于类内部使用，不应被调用。"""
            
            while True:
                self.received = False
                try:
                    data = client_socket.recv(max_data_size)
                    self.received = True
                except Exception as err:
                    self.received = False
                    self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                    client_socket.close()
                    self.client_socket_list.remove(client_socket)

                    if self.error_event:
                        func, args, kwargs = self.error_event
                        renpy.invoke_in_thread(func, err, *args, **kwargs)

                    return
                
                for prompt in self.prompt_dict.keys():
                    if data in prompt.prompts:
                        func, args, kwargs = self.prompt_dict[prompt]
                        self.current_prompt = prompt
                        renpy.invoke_in_thread(func, data, client_socket, *args, **kwargs)

                if self.reply:
                    try:
                        client_socket.send(self.reply)
                    except Exception as err:
                        self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                        client_socket.close()
                        self.client_socket_list.remove(client_socket)
                        return

                if self.receive_event:
                    func, args, kwargs = self.receive_event
                    renpy.invoke_in_thread(func, data, client_socket, *args, **kwargs)

        def send(self, client_socket: socket.socket, msg):
            """调用该方法，向指定客户端发送消息。该方法为阻塞方法。

            Arguments:
                client_socket -- 客户端socket。

            Keyword Arguments:
                msg -- 要发送的消息。

            Returns:
                若返回值为True，则发送消息成功；若为False则失败。
            """            
            
            try:
                client_socket.send(msg)
            except Exception as err:
                self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                client_socket.close()
                self.client_socket_list.remove(client_socket)

                if self.error_event:
                    func, args, kwargs = self.error_event
                    renpy.invoke_in_thread(func, err, *args, **kwargs)

                return False
            else:
                return True

        def __enter__(self):
            # 进入with语句后执行的方法
            # 防止用户回滚游戏重复启动线程
            config.rollback_enabled = False
            renpy.block_rollback()
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            # 当退出with语句后允许回滚
            config.rollback_enabled = True
            renpy.block_rollback()
    

    class RenClient(object):
        """该类为一个客户端类。
        
        在在子线程中运行的方法中使用renpy更新屏幕的函数（如`renpy.say()`、`renpy.call_screen()`等），可能引发异常。

        在子线程中运行的方法有：
            1. 使用`set_prompt`设定的命令方法。
            2. 使用`set_receive_event``set_conn_event``set_error_event`设定的事件方法。
            3. 所有进行通信的方法。
        """

        # 支持监听的事件
        EVENT = [
            "PROMPT",   # 命令事件
            "ERROR",    # 异常事件
            "CONNECT",  # 连接事件
            "RECEIVE",  # 接收事件
        ]

        PROMPT_EVENT = "PROMPT",
        ERROR_EVENT = "ERROR",
        CONNECT_EVENT = "CONNECT",
        RECEIVE_EVENT = "RECEIVE"

        
        def __init__(self, target_ip, target_port, max_data_size=1024):
            """初始化方法。

            Arguments:
                target_ip -- 服务端IP。
                target_port -- 服务端端口。

            Keyword Arguments:
                max_data_size -- 接收数据的最大大小。 (default: {1024})
            """                       

            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.target_ip = target_ip
            self.target_port = target_port
            self.max_data_size = max_data_size

            self.is_conn = False
            self.has_communicated = False
            self.received = False
            self.current_prompt = None
            self.reply = None

            self.prompt_dict = {}
            self.error_event = []
            self.receive_event = []
            self.conn_event = []

            self.error_log = {}

        def set_prompt(self, prompt: str | set, func: function, encode=True, *args, **kwargs):
            """调用该方法，创建一个命令，当客户端发送该命令后执行绑定的函数。命令将作为第一个参数，客户端socket将作为第二个参数传入指定函数中。

            不定关键字参数为函数参数。

            Arguments:
                prompt -- 命令语句。可为一个字符串或一个集合，若为集合，则集合中所有语句都可触发命令。
                func -- 一个函数。

            Keyword Arguments:
                encode -- 是否自动编码。 (default: {True})
            """ 

            if not isinstance(prompt, Prompt):
                prompt = Prompt(encode, prompt)
            
            self.prompt_dict[prompt] = [func, args, kwargs]

        def set_reply(self, reply):
            """调用该方法，当接收到消息时自动回复指定的消息。

            Arguments:
                reply -- 要回复的消息。
            """       

            self.reply = reply

        def set_error_event(self, func: function, *args, **kwargs):
            """调用该方法，指定当通信出现异常时的行为。异常信息将作为第一个参数传入指定函数中。

            不定参数为绑定的函数参数。

            Arguments:
                func 一个函数
            """            

            self.error_event = [func, args, kwargs]

        def set_receive_event(self, func: function, *args, **kwargs):
            """调用该方法，指定当接受到消息时的行为。接收到的数据将作为第一个参数。

            不定关键字参数为函数参数。

            Arguments:
                func -- 一个函数。
            """            

            self.receive_event = [func, args, kwargs]

        def set_conn_event(self, func: function, *args, **kwargs):
            """调用该方法，指定成功连接服务端后的行为。

            不定关键字参数为函数参数。

            Arguments:
                func -- 一个函数。
            """            

            self.conn_event = [func, args, kwargs]

        def listen_event(self, event, tip="", prompt=None):
            """调用该方法阻塞式监听一个事件，监听到事件后取消阻塞。

            Arguments:
                event -- 事件类型。

            Keyword Arguments:
                prompt -- 若为命令事件，则该参数为一个Prompt对象。 (default: {None})
                tip -- renpy提示界面的内容。 (default: {None})
            """            

            if event == RenClient.PROMPT_EVENT:
                while prompt != self.current_prompt:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenClient.ERROR_EVENT:
                event_counter = len(self.error_log)
                while len(self.error_log) <= event_counter:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenClient.CONNECT_EVENT:
                while not self.is_conn:
                    renpy.notify(tip)
                    renpy.pause()

            elif event == RenClient.RECEIVE_EVENT:
                while not self.received:
                    renpy.notify(tip)
                    renpy.pause()

            else:
                return

        def run(self):
            """调用该方法，开始尝试连接服务端。"""            
            
            self.has_communicated = True
            renpy.invoke_in_thread(self._connect)

        def close(self):
            """调用该方法，关闭客户端。"""

            self.socket.close()   

        def reboot(self, log_clear=False):
            """调用该函数重新开始通信。"""

            self.close()
            if log_clear:
                self.error_log.clear()
            self.run()  

        def _connect(self):
            """该方法用于创建连接线程，用于类内部使用，不应被调用。"""

            try:
                self.socket.connect((self.target_ip, self.target_port))
                self.is_conn = True
            except Exception as err:
                self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                self.is_conn = False
                if self.error_event:
                    func, args, kwargs = self.error_event
                    renpy.invoke_in_thread(func, err, *args, **kwargs)
                return

            if self.conn_event:
                func, args, kwargs = self.conn_event
                renpy.invoke_in_thread(func, *args, **kwargs)
            
            self._receive()

        def _receive(self):
            """该方法用于接收线程使用，处理接收事件，用于类内部使用，不应被调用。"""
            
            while True:
                self.received = False
                try:
                    data = self.socket.recv(self.max_data_size)
                    self.received = True
                except Exception as err:
                    self.received = False
                    self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err
                    self.socket.close()
                    self.is_conn = False

                    if self.error_event:
                        func, args, kwargs = self.error_event
                        renpy.invoke_in_thread(func, err, *args, **kwargs)

                    return

                for prompt in self.prompt_dict.keys():
                    if data in prompt.prompts:
                        func, args, kwargs = self.prompt_dict[prompt]
                        self.current_prompt = prompt
                        renpy.invoke_in_thread(func, data, client_socket, *args, **kwargs)

                if self.reply:
                    res = self.socket.send()
                    if not res:
                        self.is_conn = res
                        return

                if self.receive_event:
                    func, args, kwargs = self.receive_event
                    renpy.invoke_in_thread(func, data, *args, **kwargs)                

        def send(self, msg):
            """调用该方法，向指定客户端发送消息。该方法为阻塞方法。

            Arguments:
                msg -- 要发送的消息。

            Returns:
                若为True则发送成功；若为False则发送失败。
            """                     
            try:
                self.socket.send(msg)
            except Exception as err:
                self.error_log[datetime.now().strftime(r"%Y-%m-%d %H:%M:%S")] = err

                if self.error_event:
                    func, args, kwargs = self.error_event
                    renpy.invoke_in_thread(func, err, *args, **kwargs)

                return False
            else:
                return True

        def __enter__(self):
            config.rollback_enabled = False
            renpy.block_rollback()
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            config.rollback_enabled = True
            renpy.block_rollback()

客户端：

[RenPy] 纯文本查看 复制代码

init python:

    def conn_suc():
        renpy.notify("连接成功")

    def error(e):
        renpy.notify(e)

    def get_im(data):
        with open(f"{renpy.config.gamedir}/images/image.png", "wb+") as f:
            f.write(data)

        renpy.notify("图片保存成功！")


default e = Character("艾琳", who_color="#00CED1")


label start:

    e "ren_communicator模块测试。"

    python:
        # 使用with语句可防止用户回滚导致重复发送消息
        with RenClient("192.168.2.23", 8888, max_data_size=5242880) as client:
            client.set_conn_event(conn_suc)
            client.set_receive_event(get_im)
            client.set_error_event(error)
            client.run()

            client.listen_event(RenClient.CONNECT_EVENT, "正在等待连接......")

    e  "你想看哪一张图片？"
    menu:
        "第一张":
            $ im = "im1"
        "第二张":
            $ im = "im2"
        "第三张":
            $ im = "im3"
        "第四张":
            $ im = "im4"

    python:
        with client:
            res = client.send(im.encode("utf-8"))
            client.listen_event(RenClient.CONNECT_EVENT, "正在接收图片......")
    
    e "错误日志：\n[client.error_log!q]"

    return

服务端：

[RenPy] 纯文本查看 复制代码

init python:
 
    # 连接成功后执行的函数
    def conn_suc(socket):
        renpy.notify(f"{socket.getpeername()}已连接")
 
    # 接收到命令后执行的函数
    def get_im(prompt, socket, server):
        renpy.notify("开始发送图片")
 
        path = f"{renpy.config.gamedir}/images/{prompt.decode('utf-8')}.png"
        with open(path, "rb") as f:
            if not server.send(socket, f.read()):
                renpy.notify("发送失败！")
 
        renpy.notify("发送成功！")
 
define e = Character("艾琳")
 
 
label start:
 
    e "ren_communicator模块测试。"
    python:
        # 使用with语句防止用户回滚导致重复创建启动线程
        with RenServer(max_data_size=5242880) as server:    # 5242880B相当于5MB
            server.run()
            # 创建命令对象
            im_prompt = Prompt(
                True,
                "im1",
                "im2",
                "im3",
                "im4"
            )
            # 创建命令任务
            server.set_prompt(im_prompt, get_im, server=server)
            # 创建成功连接后的任务
            server.set_conn_event(conn_suc)
            # 监听连接事件
            server.listen_event(RenServer.CONNECT_EVENT, "正在等待连接......")
            # 监听该命令事件
            server.listen_event(RenServer.PROMPT_EVENT, "等待中......", im_prompt)
  
    e "错误日志：\n[server.error_log!q]"   # 打印错误日志，若无则为空字典。
  
    return

下面来谈谈一些要点：
以下，我把RenServer和RenClient统称为RenCommunicator

1. 如果你在一个运行于子线程中的方法中（比如使用set_prompt定义的命令事件方法）使用renpy的一些更新屏幕的语句（比如renpy.say()，renpy.call_screen()等），可能会引发异常。原因是那些语句只被允许运行于主线程当中。
2. 在定义一个RenCommunicator对象或使用其通信方法（如send()，run()等）时请务必使用python中的with语句，这样可以防止用户回滚游戏导致重复创建启动线程或重复发送消息引发严重异常！
3. 我在RenCommunictor中使用了event（事件）机制，可以通过监听这些事件或绑定函数来实现一些功能。RenCommunicator.EVENT 是一个列表，里面是目前支持的事件。
4. 使用RenCommunicator的 listen_event 方法来阻塞式监听一个事件，但是不会它让游戏卡死（出现未响应的情况），当事件发生后将取消阻塞，即继续执行后面的代码。
5. 同时，我还写了一个 Prompt 类，它是一个命令类。当你使用RenCommunictor对象的set_prompt或listen_event方法来创建一个命令或监听一个命令事件时，推荐你将Prompt对象作为其prompt参数传入（虽然它还支持字符串和列表类型，但在方法内部还是会转成一个Prompt对象）。
6. 最后一点就是关于数据的收发问题了。socket收发的是字节流而不是字符串，所以你能看到在Prompt的构造方法和RenCommunicator对象的set_prompt方法中都有一个encode参数，若为True，将会自动把你传入的字符串编码成字节流。

暂时还没想到其他问题，有问题欢迎随时下方留言（有时可能会注意不到），也可以发邮箱。

这个项目已经做得我心力交瘁了

		自动登录	找回密码
密码			立即注册

[教程] RenPyUtil：ren_communicator进阶

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。