异步交互式Python脚本

Question

我正在尝试编写一个简单的Python脚本来与聊天服务器进行交互。它将轮询服务器以获取更新，并允许用户输入要发送给服务器的聊天文本。我可以使用多线程勉强实现一些东西，但它看起来很糟糕。有没有一种漂亮、简单的方法可以在屏幕上显示更新信息并接受用户输入？我想尽量避免使用curses。

- Dan Hlavenka

---

要同时执行这样的两个任务，你需要线程。你需要有一个线程来处理用户输入，可能是主线程。然后创建另一个线程来处理从聊天服务器传来的响应。 - Hunter McMillen

我已经让那部分工作了，但它显示不正确。我一直在使用raw_input来获取输入，但当你在调用后打印文本时，它会出现问题。 - Dan Hlavenka

@HunterMcMillen - 不完全正确——一些(G)UI工具包和通信框架使用非线程循环（通常基于select），例如PyGTK和Twisted。 - detly

@detly 我对 select 系统调用不是很熟悉，根据我刚查到的资料，select 似乎会在某些文件描述符发生变化之前等待。但这个如何并发运行呢？它似乎只是一个忙等待循环。不过我可能还没有完全理解 select 调用。 - Hunter McMillen

@HunterMcMillen - 虽然不是并发的，但I/O任务以足够细粒度的方式交错进行，因此UI仍然具有响应性。关于“twisted/tornado等如何工作”的答案在这方面有所涉及。 - detly

Answer 1

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

代码1： 从数据库更新屏幕。您可以使用无限循环，带或不带暂停（屏幕刷新率）

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

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

- codersofthedark

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()

- tvuillemin