Created
November 8, 2024 17:12
-
-
Save qpwo/1728b230fea08cdd3b12c291a6dcaeba to your computer and use it in GitHub Desktop.
repl in python with popen and selectors
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# unfortunately, the stdinready event fires all the time, not just when repl ready. | |
# Based on https://gist.github.com/andy0130tw/39472331530d1a0e25459a547ed2c9d5 | |
import io | |
import selectors | |
from selectors import EVENT_READ, EVENT_WRITE | |
import subprocess | |
from typing import cast | |
sel = selectors.DefaultSelector() | |
command = ["python", "-i"] | |
proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) | |
stdoutbuf = io.BytesIO() # if you wanted to collect stdout | |
assert proc.stdin is not None | |
sel.register(proc.stdin, EVENT_WRITE) | |
assert proc.stdout is not None | |
sel.register(proc.stdout, EVENT_READ) | |
while True: | |
gotoutput = False | |
wantsinput = False | |
for output, _mask in sel.select(timeout=0.6): | |
f = output.fileobj | |
if f is proc.stdout: | |
f = cast(io.BufferedReader, f) | |
data = f.read1() # read1 is very important!! | |
# if not data: | |
# continue | |
gotoutput = True | |
decoded = data.decode("utf-8", errors="replace") | |
print(decoded, end="", flush=True) | |
stdoutbuf.write(data) | |
elif f is proc.stdin: | |
wantsinput = True | |
# f = cast(io.BufferedWriter, f) | |
else: | |
raise Exception("unreachable") | |
if gotoutput: | |
continue | |
if wantsinput: | |
command = input(" ! ") | |
# print(f"{command=}") | |
proc.stdin.write(command.encode("utf-8") + b"\n") | |
proc.stdin.flush() | |
continue | |
print("see ya") | |
break | |
print("all done") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment