异步交互式Python脚本

Question

我正在尝试写一个简单的Python脚本，用来和一个聊天服务器对接。这个脚本会定期检查服务器有没有更新，同时允许用户输入文本发送到服务器进行聊天。我能用多线程把它勉强做出来，但效果看起来很糟糕。有没有什么简单好用的方法，可以在屏幕上显示更新的信息，同时又能接受用户的输入？我希望能不使用curses库来实现。

Answer 1

我不知道怎么用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()

Answer 2

让我们假设显示窗口或屏幕是由一个临时的本地数据库提供支持。

代码1： 这个代码会从数据库更新屏幕。你可以使用一个无限循环来不断刷新屏幕，可以选择加个暂停（也就是屏幕刷新频率）。

代码2： 这个代码会在用户输入内容后，立即更新数据库。

代码3： 这个代码会持续检查聊天服务器的更新，一旦收到新的回复，就会立刻更新数据库。