异步交互式 Python 脚本

Question

我正在尝试编写一个简单的 Python 脚本来与聊天服务器交互。它将轮询服务器以获取更新，并允许用户输入文本以作为聊天发送到服务器。我可以用多线程来破解一些东西，但它看起来很糟糕。有没有一种很好、简单的方法来在屏幕上显示更新信息，同时还接受用户输入？我宁愿不要诅咒。

Answer 1

让显示窗口/屏幕由临时本地数据库供电。

代码 1： 从 db 更新屏幕。您可以在有或没有暂停的情况下使用无限循环（屏幕刷新率）

代码 2： 用户输入内容后立即更新数据库

代码 3： 持续检查来自聊天服务器的更新，并在收到新响应后立即更新数据库。

Answer 2

我不知道如何使用 ncurses 进行编码，但这里有一个使用 wxWidget 的解决方案。在设计上应该大致相似。

"""Asynchronous interactive Python script"""
import random
import threading
import wx
import wx.lib.mixins.listctrl as listmix

LOCK = threading.Lock()

def threadsafe(function):
    """A decorator that makes a function safe against concurrent accesses."""
    def _decorated_function(*args, **kwargs):
        """Replacement function."""
        with LOCK:
            function(*args, **kwargs)
    return _decorated_function


class SharedList(wx.ListCtrl, listmix.ListCtrlAutoWidthMixin):
    """An output list that can print information from both the user and the server.

    N.B.: The _print function that actually updates the list content uses the threadsafe decorator.
    """

    def __init__(self, parent, pos=wx.DefaultPosition, size=(-1, -1), style=wx.LC_REPORT):
        wx.ListCtrl.__init__(self, parent, wx.ID_ANY, pos, size, style)
        self.InsertColumn(0, 'Origin', width=75)
        self.InsertColumn(1, 'Output')
        listmix.ListCtrlAutoWidthMixin.__init__(self)
        self.resizeLastColumn(1000)
        self._list_index = 0

    def user_print(self, text):
        """Print a line as the user"""
        self._print("user", text)

    def chat_print(self, text):
        """Print a line as the chat server"""
        self._print("chat", text)

    @threadsafe
    def _print(self, origin, text):
        """Generic print function."""
        self.InsertStringItem(self._list_index, str(origin))
        self.SetStringItem(self._list_index, 1, str(text))
        self.EnsureVisible(self.GetItemCount() - 1)
        self._list_index = self._list_index + 1


class ServerChecker(threading.Thread):
    """A separate thread that would connect to the IRC chat."""

    def __init__(self, shared_list):
        threading.Thread.__init__(self)
        self._stop_event = threading.Event()
        self._shared_list = shared_list

    def run(self):
        """Connection to the server, socket, listen, bla bla bla."""
        while not self._stop_event.is_set():
            self._shared_list.chat_print("bla bla bla")
            self._stop_event.wait(random.randint(1, 3))

    def stop(self):
        """Stop the thread."""
        self._stop_event.set()


class SampleFrame(wx.Frame):
    """The main GUI element."""

    def __init__(self):
        super(SampleFrame, self).__init__(parent=None, title='DAS Board', size=(600, 400))
        self._shared_list = None
        self._user_text = None
        self._init_ui()
        self._thread = ServerChecker(self._shared_list)
        self._thread.start()
        self.Bind(wx.EVT_CLOSE, self._on_close)
        self.Center()
        self.Show()

    def _init_ui(self):
        """Building and assembling the graphical elements.
        Don't pay too much attention, especially if you want to use ncurses instead.
        """
        panel = wx.Panel(self)
        self._shared_list = SharedList(panel)
        main_box_v = wx.BoxSizer(wx.VERTICAL)
        main_box_v.Add(self._shared_list, proportion=1, flag=wx.EXPAND|wx.ALL, border=10)
        self._user_text = wx.TextCtrl(panel, -1, value="User text to send...",
                                      style=wx.TE_CENTRE)
        main_box_v.Add(self._user_text, proportion=0, flag=wx.EXPAND|wx.ALL, border=10)
        button = wx.Button(panel, label="Send user text.")
        button.Bind(wx.EVT_BUTTON, self._user_send_text)
        main_box_v.Add(button, flag=wx.EXPAND|wx.ALL, border=10)
        panel.SetSizer(main_box_v)

    def _user_send_text(self, event):
        """Button callback"""
        self._shared_list.user_print(self._user_text.GetValue())
        event.Skip()

    def _on_close(self, event):
        """Stop the separate thread, then destroy the GUI."""
        event.Skip()
        self._thread.stop()
        self.Destroy()


APP = wx.App(0)
SampleFrame()
APP.MainLoop()