1

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?

3
  • stdin is an input stream, you can not write to it. You will have to find a different approach. Commented Mar 28, 2021 at 23:35
  • If it's another thread of the same process, then presumably it is created by code that you write. You should change that code so that it doesn't demand input from stdin. Commented Mar 29, 2021 at 0:56
  • @KarlKnechtel the input is on the main thread and is required by a library. I can't change that Commented Mar 29, 2021 at 2:01

2 Answers 2

1

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

Sign up to request clarification or add additional context in comments.

11 Comments

can I old = sys.stidn; sys.stdin = mock; input(); sys.stdin = old ? As I said, the input is inside a library.
You will need to start up the relay process and kill it afterwards. Better to do mock.__enter__(); input(); mock.__exit__(None, None, None)
If you don't want to use a with block, you may as well name the functions something more readable and remove the unnecessary arguments.
Thank's, I'll check it out but I think it will work!
I found that the problem is in the line self.real_sdin = sys.stin. But I can't understand why its throwing the error _winapi.OpenProcess( OSError: [WinError 87] Incorrect Parameter on this attribution
|
0

If you're on Linux you can run a subprocess to write to your process's stdin. This will unblock whichever input() is currently blocked.

stdin_fd = '/proc/' + str( os.getpid() ) + '/fd/0'
subprocess.run( 'echo "to be inputed" > ' + stdin_fd, shell=True )

Note shell=True is necessary to have the echo output redirection > work.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.