Reputation: 1365
import sys
import threading
import tty
import termios
def loop():
fd = sys.stdin.fileno()
mode = termios.tcgetattr(fd)
tty.setraw(fd)
try:
while True:
sys.stdin.read(1)
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, mode)
threading.Thread(target=loop).start()
I tried to manually send End-Of-Transmission character using sys.stdin.write(chr(4))
but it results in io.UnsupportedOperation: not writable
.
I tried to close stdin but it will not terminate blocking read(1) method immediately. It triggers an exception after I typed another key.
I tried seletors but selector.select() need extra keys after I invoked selector.close()
. And selector it self needs an event loop and this loop becomes uninterruptable.
Upvotes: 2
Views: 1063
Reputation: 1365
If you run the loop in a thread then it will be hard to interrupt and also keep original process running. But it's easy to run this loop in a subprocess. So you can interrupt the subprocess to interrupt the blocking stdin.read(1)
import os
import sys
import tty
import termios
import time
import signal
from multiprocessing import Process
def loop():
# https://stackoverflow.com/questions/30134297/python-multiprocessing-stdin-input
# Subprocess stdin will be set to os.devnull. So we need to reopen it here.
# Use parent process stdin(fd=0) instead of subprocess stdin(fd=7)
sys.stdin = open(0) #
fd = sys.stdin.fileno()
mode = termios.tcgetattr(fd)
tty.setraw(fd)
try:
while True:
ch = sys.stdin.read(1)
if ord(ch) == 3:
break
except KeyboardInterrupt:
print("interrupted!\r\n", end="")
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, mode)
if __name__ == '__main__':
p = Process(target=loop)
p.start()
time.sleep(1)
os.kill(p.pid, signal.SIGINT)
p.join()
print("done")
Upvotes: 1