aiorepl.py

# MIT license; Copyright (c) 2022 Jim Mussared

importmicropython
frommicropythonimportconst
importre
importsys
importtime
importasyncio

# Import statement (needs to be global, and does not return).
_RE_IMPORT=re.compile("^import ([^ ]+)( as ([^ ]+))?")
_RE_FROM_IMPORT=re.compile("^from [^ ]+ import ([^ ]+)( as ([^ ]+))?")
# Global variable assignment.
_RE_GLOBAL=re.compile("^([a-zA-Z0-9_]+) ?=[^=]")
# General assignment expression or import statement (does not return a value).
_RE_ASSIGN=re.compile("[^=]=[^=]")

# Command hist (One reserved slot for the current command).
_HISTORY_LIMIT=const(5+1)


CHAR_CTRL_A=const(1)
CHAR_CTRL_B=const(2)
CHAR_CTRL_C=const(3)
CHAR_CTRL_D=const(4)
CHAR_CTRL_E=const(5)


asyncdefexecute(code, g, s):
ifnotcode.strip():
return

try:
if"await "incode:
# Execute the code snippet in an async context.
ifm:=_RE_IMPORT.match(code) or_RE_FROM_IMPORT.match(code):
code="global {}\n {}".format(m.group(3) orm.group(1), code)
elifm:=_RE_GLOBAL.match(code):
code="global {}\n {}".format(m.group(1), code)
elifnot_RE_ASSIGN.search(code):
code="return {}".format(code)

code="""
import asyncio
async def __code():
 {}

__exec_task = asyncio.create_task(__code())
""".format(code)

asyncdefkbd_intr_task(exec_task, s):
whileTrue:
iford(awaits.read(1)) ==CHAR_CTRL_C:
exec_task.cancel()
return

l= {"__exec_task": None}
exec(code, g, l)
exec_task=l["__exec_task"]

# Concurrently wait for either Ctrl-C from the stream or task
# completion.
intr_task=asyncio.create_task(kbd_intr_task(exec_task, s))

try:
try:
returnawaitexec_task
exceptasyncio.CancelledError:
pass
finally:
intr_task.cancel()
try:
awaitintr_task
exceptasyncio.CancelledError:
pass
else:
# Excute code snippet directly.
try:
try:
micropython.kbd_intr(3)
try:
returneval(code, g)
exceptSyntaxError:
# Maybe an assignment, try with exec.
returnexec(code, g)
exceptKeyboardInterrupt:
pass
finally:
micropython.kbd_intr(-1)

exceptExceptionaserr:
print("{}: {}".format(type(err).__name__, err))


# REPL task. Invoke this with an optional mutable globals dict.
asyncdeftask(g=None, prompt="--> "):
print("Starting asyncio REPL...")
ifgisNone:
g=__import__("__main__").__dict__
try:
micropython.kbd_intr(-1)
s=asyncio.StreamReader(sys.stdin)
# clear = True
hist= [None] *_HISTORY_LIMIT
hist_i=0# Index of most recent entry.
hist_n=0# Number of history entries.
c=0# ord of most recent character.
t=0# timestamp of most recent character.
whileTrue:
hist_b=0# How far back in the history are we currently.
sys.stdout.write(prompt)
cmd: str=""
paste=False
curs=0# cursor offset from end of cmd buffer
whileTrue:
b=awaits.read(1)
pc=c# save previous character
c=ord(b)
pt=t# save previous time
t=time.ticks_ms()
ifc<0x20orc>0x7E:
ifc==0x0A:
# LF
ifpaste:
sys.stdout.write(b)
cmd+=b
continue
# If the previous character was also LF, and was less
# than 20 ms ago, this was likely due to CRLF->LFLF
# conversion, so ignore this linefeed.
ifpc==0x0Aandtime.ticks_diff(t, pt) <20:
continue
ifcurs:
# move cursor to end of the line
sys.stdout.write("\x1B[{}C".format(curs))
curs=0
sys.stdout.write("\n")
ifcmd:
# Push current command.
hist[hist_i] =cmd
# Increase history length if possible, and rotate ring forward.
hist_n=min(_HISTORY_LIMIT-1, hist_n+1)
hist_i= (hist_i+1) %_HISTORY_LIMIT

result=awaitexecute(cmd, g, s)
ifresultisnotNone:
sys.stdout.write(repr(result))
sys.stdout.write("\n")
break
elifc==0x08orc==0x7F:
# Backspace.
ifcmd:
ifcurs:
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
else:
cmd=cmd[:-1]
sys.stdout.write("\x08\x08")
elifc==CHAR_CTRL_A:
awaitraw_repl(s, g)
break
elifc==CHAR_CTRL_B:
continue
elifc==CHAR_CTRL_C:
ifpaste:
break
sys.stdout.write("\n")
break
elifc==CHAR_CTRL_D:
ifpaste:
result=awaitexecute(cmd, g, s)
ifresultisnotNone:
sys.stdout.write(repr(result))
sys.stdout.write("\n")
break

sys.stdout.write("\n")
# Shutdown asyncio.
asyncio.new_event_loop()
return
elifc==CHAR_CTRL_E:
sys.stdout.write("paste mode; Ctrl-C to cancel, Ctrl-D to finish\n===\n")
paste=True
elifc==0x1B:
# Start of escape sequence.
key=awaits.read(2)
ifkeyin ("[A", "[B"): # up, down
# Stash the current command.
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)
# Go backwards or forwards in the history.
ifkey=="[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)
elifkey=="[D": # left
ifcurs<len(cmd) -1:
curs+=1
sys.stdout.write("\x1B")
sys.stdout.write(key)
elifkey=="[C": # right
ifcurs:
curs-=1
sys.stdout.write("\x1B")
sys.stdout.write(key)
elifkey=="[H": # home
pcurs=curs
curs=len(cmd)
sys.stdout.write("\x1B[{}D".format(curs-pcurs)) # move cursor left
elifkey=="[F": # end
pcurs=curs
curs=0
sys.stdout.write("\x1B[{}C".format(pcurs)) # move cursor right
else:
# sys.stdout.write("\\x")
# sys.stdout.write(hex(c))
pass
else:
ifcurs:
# 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
else:
sys.stdout.write(b)
cmd+=b
finally:
micropython.kbd_intr(3)


asyncdefraw_paste(s, g, window=512):
sys.stdout.write("R\x01") # supported
sys.stdout.write(bytearray([window&0xFF, window>>8, 0x01]).decode())
eof=False
idx=0
buff=bytearray(window)
file=b""
whilenoteof:
foridxinrange(window):
b=awaits.read(1)
c=ord(b)
ifc==CHAR_CTRL_Corc==CHAR_CTRL_D:
# end of file
sys.stdout.write(chr(CHAR_CTRL_D))
ifc==CHAR_CTRL_C:
raiseKeyboardInterrupt
file+=buff[:idx]
eof=True
break
buff[idx] =c

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

returnfile


asyncdefraw_repl(s: asyncio.StreamReader, g: dict):
heading="raw REPL; CTRL-B to exit\n"
line=""
sys.stdout.write(heading)

whileTrue:
line=""
sys.stdout.write(">")
whileTrue:
b=awaits.read(1)
c=ord(b)
ifc==CHAR_CTRL_A:
rline=line
line=""

iflen(rline) ==2andord(rline[0]) ==CHAR_CTRL_E:
ifrline[1] =="A":
line=awaitraw_paste(s, g)
break
else:
# reset raw REPL
sys.stdout.write(heading)
sys.stdout.write(">")
continue
elifc==CHAR_CTRL_B:
# exit raw REPL
sys.stdout.write("\n")
return0
elifc==CHAR_CTRL_C:
# clear line
line=""
elifc==CHAR_CTRL_D:
# entry finished
# indicate reception of command
sys.stdout.write("OK")
break
else:
# let through any other raw 8-bit value
line+=b

iflen(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)

try:
result=exec(line, g)
ifresultisnotNone:
sys.stdout.write(repr(result))
sys.stdout.write(chr(CHAR_CTRL_D))
exceptExceptionasex:
print(line)
sys.stdout.write(chr(CHAR_CTRL_D))
sys.print_exception(ex, sys.stdout)
sys.stdout.write(chr(CHAR_CTRL_D))