Binary stdin and stdout buffers for Unix port
$ micropython
>>> import sys
>>> sys.stdout.buffer
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'TextIOWrapper' object has no attribute 'buffer'
>>>
Well, other ports have it … CPython has it … and I'd really like to not have to fall back to open("/dev/stdout","wb") in my unit tests.
Same for stdin of course.
extmod/vfs_posix_file.c: Unix port stdio buffer enhancement
This PR enhances port/unix by adding the buffer attribute to stdio.
Currently, the unix port stdio does not have the buffer attribute as does cpython and other ports, making reading bytes over stdio not possible. The enhancement will make CLI tools easier to implement, improve porting python tools to micropython, and improved interoperablity between ports, especially when piping raw data samples (IQ, audio, etc...)
Related discussions:
https://github.com/orgs/micropython/discussions/12070
https://github.com/orgs/micropython/discussions/11889
shared/runtime/sys_stdio_mphal.c was used as a model for the implementation. Instead of using mp_hal_stdin_rx_chr and mp_hal_stdout_tx_strn (which works for raw input but not pipes), I instead created two mp_stream with is_text = false using STDIN_FILE and STDOUT_FILE. A new stdio buffer stream may then read/write from these streams respectively.
One question I have: I'm unsure of the utilitzation of MICROPY_PY_SYS_STDIO_BUFFER as it is used in sys_stdio_mphal.c. I went ahead and made my implementation match though I'm not fully clear of it's intended usage.
I've done some tests, a simple one here piping bytes 0-255 using micropython as well as python.
./micropython buffertest.py write | ./micropython buffertest.py read
python buffertest.py write | ./micropython buffertest.py read
./micropython buffertest.py write | python buffertest.py read
buffertest.py
import sys
import asyncio
async def write():
for x in range(256):
sys.stdout.buffer.write(x.to_bytes(1,'little'))
async def read():
while True:
# r = sys.stdin.read(1) # today we only have stdin.read which crashes with unicode error at 128
r = sys.stdin.buffer.read(1) # thumbs up!
if r != None and len(r)==0:
break
print('{:<10} {}'.format(int.from_bytes(r,'big'),r))
async def main():
if 'read' in sys.argv:
await read()
elif 'write' in sys.argv:
await write()
asyncio.run(main())