在另一个线程中回答 python 输入

Question

我想从代码中 python 上同一进程的另一个线程回答 input()。

这是代码：

import sys
import threading

def threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True)
        thread.start()
        return thread
    return wrapper

@threaded
def answer():
    time.sleep(2)
    sys.stdin.write('to be inputed')


answer()
x = input('insert a value: ')
print(f'value inserted: {x}')  # excpeted print: 'value inserted: to be inputed'

但我认为这不可能，因为我收到此错误：

Exception in thread Thread-1:
Traceback (most recent call last):
  File "teste.py", line 80, in answer
sys.stdin.write('to be inputed')
io.UnsupportedOperation: not writable

很难解释为什么我想要那个，但有时用户会输入值，有时它会来自另一个输入源（电报）。所以这第二个线程应该能够输入值并释放代码执行。

我也无法更改代码的 input() 部分，因为它来自库内部，所以它需要这样：input('insert a value: ')

有办法实现吗？

Answer 1

简单的答案是，如果您将 sys.stdin 替换为您自己的变量，那么 input 将使用它。

但是，您已经丢失了原来的 stdin，因此您需要启动一个新进程来侦听用户输入，因为您说：

but sometimes the user will input the value

这需要是另一个进程而不是线程，因为当您想要恢复原始标准输入时需要终止它，并且终止进程会在中途中断它-readline。

这是实现了模拟对象的代码的工作版本。 with 块内的区域是标准输入被替换的地方。


import sys
import time
import multiprocessing
import threading

class MockStdin:
    def __init__(self):
        self.queue = None
        self.real_stdin = sys.stdin
        self.relay_process = None

    def readline(self):
        # when input() is called, it calls this function
        return self.queue.get()

    def writeline(self, s):
        # for input from elsewhere in the program
        self.queue.put(s)

    def relay_stdin(self):
        # for input from the user
        my_stdin = open(0)  # this is a new process so it needs its own stdin

        try:
            while True:
                inp = my_stdin.readline()
                self.queue.put(inp)
        except KeyboardInterrupt:
            # when killed, exit silently
            pass

    def __enter__(self):
        # when entering the `with` block, start replace stdin with self and relay real stdin
        self.queue = multiprocessing.Queue()

        self.relay_process = multiprocessing.Process(target=self.relay_stdin)
        self.relay_process.start()
        sys.stdin = self

    def __exit__(self, exc_type=None, exc_val=None, exc_tb=None):
        # when exiting the `with` block, put stdin back and stop relaying
        sys.stdin = self.real_stdin

        self.relay_process.terminate()
        self.relay_process.join()
        
    def __getstate__(self):
        # this is needed for Windows - credit to Leonardo Rick for this fix
        self_dict = self.__dict__.copy()
        del self_dict['real_stdin']
        return self_dict



def threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True)
        thread.start()
        return thread

    return wrapper

if __name__ == '__main__':
    mock = MockStdin()


    @threaded
    def answer():
        time.sleep(2)
        # use mock to write to stdin
        mock.writeline('to be inputed')


    answer()

    with mock:
        # inside `with` block, stdin is replaced
        x = input('insert a value: ')
        print(f'\nvalue inserted: {x}')

    answer()

    # __enter__ and __exit__ can also be used
    mock.__enter__()
    x = input('insert a value: ')
    print(f'\nvalue inserted: {x}')
    mock.__exit__()

    # now outside the `with` block, stdin is back to normal
    x = input('insert another (stdin should be back to normal now): ')
    print(f'value inserted: {x}')

在另一个线程中回答 python 输入

Answer python input from within on another thread

python

multithreading

stdin