pythonmultithreadingstdin

Answer python input from within on another thread


I want to answer a input() from another thread of the same process on python from within the code.

This is the code:

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'

But I think its not possbile because I receive this error:

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

It's hard to explain why I want that, but sometimes the user will input the value and sometimes it will come from another input source (telegram). So this second thread should be able to input the value and release the code execution.

I also can't change the input() part of the code because its from inside a library, so it need to be this way: input('insert a value: ')

Is there a way to achive that?


Solution

  • The simple answer is that if you replace sys.stdin with your own variable, then input uses that instead.

    However, then you've lost your original stdin, so you need start a new process to listen for user input, since you said:

    but sometimes the user will input the value

    This needs to be another process rather than a thread since it needs to be killed when you want to restore the original stdin, and killing the process interrupts it mid-readline.

    Here is a working version of the code with the mock object implemented. The region inside the with block is where stdin has been replaced.

    
    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}')