Skip to content

aiorepl: Optimize sys.stdout.write calls. #1148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dpgeorge merged 1 commit into from
Aug 17, 2026
+50 −49
Merged

aiorepl: Optimize sys.stdout.write calls. #1148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 49 additions & 48 deletions micropython/aiorepl/aiorepl.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,9 +106,10 @@ async def task(g=None, prompt="--> "):
hist_n = 0 # Number of history entries.
c = 0 # ord of most recent character.
t = 0 # timestamp of most recent character.
write = sys.stdout.write
while True:
hist_b = 0 # How far back in the history are we currently.
sys.stdout.write(prompt)
write(prompt)
cmd: str = ""
paste = False
curs = 0 # cursor offset from end of cmd buffer
Expand All @@ -125,7 +126,7 @@ async def task(g=None, prompt="--> "):
# LF or CR (handle both for raw terminal mode compatibility)
if paste:
# In paste mode, preserve the actual character
sys.stdout.write(b)
write(b)
cmd += b
continue
# Handle various newline sequences to avoid double-execution:
Expand All @@ -140,9 +141,9 @@ async def task(g=None, prompt="--> "):
continue
if curs:
# move cursor to end of the line
sys.stdout.write("\x1b[{}C".format(curs))
write("\x1b[{}C".format(curs))
curs = 0
sys.stdout.write("\n")
write("\n")
if cmd:
# Push current command.
hist[hist_i] = cmd
Expand All @@ -152,22 +153,20 @@ async def task(g=None, prompt="--> "):

result = await execute(cmd, g, s)
if result is not None:
sys.stdout.write(repr(result))
sys.stdout.write("\n")
write(repr(result))
write("\n")
break
elif c == 0x08 or c == 0x7F:
# Backspace.
if cmd:
if curs:
cmd = "".join((cmd[: -curs - 1], cmd[-curs:]))
sys.stdout.write(
"\x08\x1b[K"
) # move cursor back, erase to end of line
sys.stdout.write(cmd[-curs:]) # redraw line
sys.stdout.write("\x1b[{}D".format(curs)) # reset cursor location
write("\x08\x1b[K") # move cursor back, erase to end of line
write(cmd[-curs:]) # redraw line
write("\x1b[{}D".format(curs)) # reset cursor location
else:
cmd = cmd[:-1]
sys.stdout.write("\x08 \x08")
write("\x08 \x08")
elif c == CHAR_CTRL_A:
raw_repl(sys.stdin, g)
break
Expand All @@ -176,22 +175,22 @@ async def task(g=None, prompt="--> "):
elif c == CHAR_CTRL_C:
if paste:
break
sys.stdout.write("\n")
write("\n")
break
elif c == CHAR_CTRL_D:
if paste:
result = await execute(cmd, g, s)
if result is not None:
sys.stdout.write(repr(result))
sys.stdout.write("\n")
write(repr(result))
write("\n")
break

sys.stdout.write("\n")
write("\n")
# Shutdown asyncio.
asyncio.new_event_loop()
return
elif c == CHAR_CTRL_E:
sys.stdout.write("paste mode; Ctrl-C to cancel, Ctrl-D to finish\n===\n")
write("paste mode; Ctrl-C to cancel, Ctrl-D to finish\n===\n")
paste = True
elif c == 0x1B:
# Start of escape sequence.
Expand All @@ -201,55 +200,56 @@ async def task(g=None, prompt="--> "):
hist[(hist_i - hist_b) % _HISTORY_LIMIT] = cmd
# Clear current command.
b = "\x08" * len(cmd)
sys.stdout.write(b)
sys.stdout.write(" " * len(cmd))
sys.stdout.write(b)
write(b)
write(" " * len(cmd))
write(b)
# Go backwards or forwards in the history.
if key == "[A":
hist_b = min(hist_n, hist_b + 1)
else:
hist_b = max(0, hist_b - 1)
# Update current command.
cmd = hist[(hist_i - hist_b) % _HISTORY_LIMIT]
sys.stdout.write(cmd)
write(cmd)
elif key == "[D": # left
if curs < len(cmd) - 1:
curs += 1
sys.stdout.write("\x1b")
sys.stdout.write(key)
write("\x1b")
write(key)
elif key == "[C": # right
if curs:
curs -= 1
sys.stdout.write("\x1b")
sys.stdout.write(key)
write("\x1b")
write(key)
elif key == "[H": # home
pcurs = curs
curs = len(cmd)
sys.stdout.write("\x1b[{}D".format(curs - pcurs)) # move cursor left
write("\x1b[{}D".format(curs - pcurs)) # move cursor left
elif key == "[F": # end
pcurs = curs
curs = 0
sys.stdout.write("\x1b[{}C".format(pcurs)) # move cursor right
write("\x1b[{}C".format(pcurs)) # move cursor right
else:
# sys.stdout.write("\\x")
# sys.stdout.write(hex(c))
# write("\\x")
# write(hex(c))
pass
else:
if curs:
# inserting into middle of line
cmd = "".join((cmd[:-curs], b, cmd[-curs:]))
sys.stdout.write(cmd[-curs - 1 :]) # redraw line to end
sys.stdout.write("\x1b[{}D".format(curs)) # reset cursor location
write(cmd[-curs - 1 :]) # redraw line to end
write("\x1b[{}D".format(curs)) # reset cursor location
else:
sys.stdout.write(b)
write(b)
cmd += b
finally:
micropython.kbd_intr(3)


def raw_paste(s, window=512):
sys.stdout.write("R\x01") # supported
sys.stdout.write(bytearray([window & 0xFF, window >> 8, 0x01]).decode())
write = sys.stdout.write
write("R\x01") # supported
write(bytearray([window & 0xFF, window >> 8, 0x01]).decode())
eof = False
idx = 0
buff = bytearray(window)
Expand All @@ -260,7 +260,7 @@ def raw_paste(s, window=512):
c = ord(b)
if c == CHAR_CTRL_C or c == CHAR_CTRL_D:
# end of file
sys.stdout.write(chr(CHAR_CTRL_D))
write(chr(CHAR_CTRL_D))
if c == CHAR_CTRL_C:
raise KeyboardInterrupt
file += buff[:idx]
Expand All @@ -270,7 +270,7 @@ def raw_paste(s, window=512):

if not eof:
file += buff
sys.stdout.write("\x01") # indicate window available to host
write("\x01") # indicate window available to host

return file

Expand All @@ -283,11 +283,12 @@ def raw_repl(s, g: dict):
"""
heading = "raw REPL; CTRL-B to exit\n"
line = ""
sys.stdout.write(heading)
write = sys.stdout.write
write(heading)

while True:
line = ""
sys.stdout.write(">")
write(">")
while True:
b = s.read(1)
c = ord(b)
Expand All @@ -301,20 +302,20 @@ def raw_repl(s, g: dict):
break
else:
# reset raw REPL
sys.stdout.write(heading)
sys.stdout.write(">")
write(heading)
write(">")
continue
elif c == CHAR_CTRL_B:
# exit raw REPL
sys.stdout.write("\n")
write("\n")
return 0
elif c == CHAR_CTRL_C:
# clear line
line = ""
elif c == CHAR_CTRL_D:
# entry finished
# indicate reception of command
sys.stdout.write("OK")
write("OK")
break
else:
# let through any other raw 8-bit value
Expand All @@ -323,16 +324,16 @@ def raw_repl(s, g: dict):
if len(line) == 0:
# Normally used to trigger soft-reset but stay in raw mode.
# Fake it for aiorepl / mpremote.
sys.stdout.write("Ignored: soft reboot\n")
sys.stdout.write(heading)
write("Ignored: soft reboot\n")
write(heading)

try:
result = exec(line, g)
if result is not None:
sys.stdout.write(repr(result))
sys.stdout.write(chr(CHAR_CTRL_D))
write(repr(result))
write(chr(CHAR_CTRL_D))
except Exception as ex:
print(line)
sys.stdout.write(chr(CHAR_CTRL_D))
write(chr(CHAR_CTRL_D))
sys.print_exception(ex, sys.stdout)
sys.stdout.write(chr(CHAR_CTRL_D))
write(chr(CHAR_CTRL_D))
2 changes: 1 addition & 1 deletion micropython/aiorepl/manifest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
metadata(
version="0.2.3",
version="0.2.4",
description="Provides an asynchronous REPL that can run concurrently with an asyncio, also allowing await expressions.",
)

Expand Down
Loading