Skip to content

Instantly share code, notes, and snippets.

@qpwo
Created November 8, 2024 17:12
Show Gist options
  • Save qpwo/1728b230fea08cdd3b12c291a6dcaeba to your computer and use it in GitHub Desktop.
Save qpwo/1728b230fea08cdd3b12c291a6dcaeba to your computer and use it in GitHub Desktop.
repl in python with popen and selectors
# 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