Non blocking read of input from the terminal

[ishahak]

This is a knowledge sharing post, because I did not find it documented anywhere, and had to dig into other questions and answers in order to get an idea for how to do it. Please be aware that if using Thonny, then characters are sent to the application only after ENTER is pressed.

import sys, select, time

"""
Note: this code does not work under Thonny because in Thonny input is sent to program only after [Enter] is clicked.
"""

class TermRead:
    """
    We create  a select.poll() object called spoll, and register the standard input (sys.stdin) with the POLLIN
    flag, indicating that we want to wait for the standard input to be ready for reading.

    When we call spoll.poll(0), the 0 parameter specifies a milliseconds timeout value of 0, 
    which means that the poll() method will return immediately with a list of file descriptors that
    have events to process, or an empty list if no events are ready to be processed.

    This allows us to check if there's input available on sys.stdin without waiting for any input to arrive.
    """
    def __init__(self):
        self.spoll = select.poll()
        self.spoll.register(sys.stdin, select.POLLIN)
        
    
    def read(self):
        txt = ''
        while self.spoll.poll(0):
            # see the other versions here for utf-8 characters
            txt += sys.stdin.read(1)
        return txt

# usage demo:
term = TermRead()
while True:
    txt = term.read()
    print(txt if txt != '' else '.', end='')
    time.sleep(1)

I also created a gist which demonstrates how to listen to both wifi connections and serial commands in parallel: gist

[mk-fg]

If I run sys.stdin.read(1) in a REPL via mpremote, and then press non-english-layout key, it returns e.g. '\u0430':

>>> import sys
>>> sys.stdin.read(1)
'\u0430'

This suggests that read(1) does in fact reads 1 utf-8 character, which can consist of arbitrarily many bytes (I think up to 4 or 6 atm), so is a wrong thing to use here.
Because, for example, if I want to send a binary b'\xd0' byte as a command to be read on rp2040 over ttyACM0, this is what I'm pretty sure should happen:

• self.spoll.poll(0) will return that data is available on sys.stdin.
• sys.stdin.read(1) will read that byte and block (potentially forever) on waiting for the next byte, because b'\xd0' is a start of an utf-8 character, and it was instructed to read 1 unicode char and return it as str, not to read 1 byte.

Proposed fix: use sys.stdin.buffer.read(1) instead - that will return b'\xd0' and is semantically correct way to "read 1 byte" instead of 1 unicode character.

[ishahak]

Thank you Mike, I have updated my answer accordingly.

[mk-fg]

I think maybe also txt = '' should be changed to be txt = b'', so that txt += ... won't be doing str + bytes operation.
And print(txt if txt != '' else '.', end='') at the end also deals with str's, so might be worth returning txt.decode() for it.

But then you might also read incomplete utf-8 byte-sequence, and txt.decode() will raise UnicodeError, so probably should only return it when decoding succeeds, otherwise buffering remainder bytes in some instance state-var (e.g. self.txt) for the next read call.

But then also, maybe POLLERR/POLLHUP flags in polled event should be checked too, in case some boards' firmware implementations will block read() or raise some exception on disconnected tty?

Sorry if maybe it gets too much into the weeds, but feel like example not handling such cases might trip new folks into unexpected behaviors that they're even less likely to understand than a couple extra checks, unfortunately.

[ishahak]

Hi Mike, it's been some time since I wrote this. Feel free to post your own implementation, which supports full Unicode sequences.

[mk-fg]

Had a stab at it, and you know what, it ends up being way too bloody complicated (posted it in a separate comment) :)
So I think your original version, even with that caveat wrt blocking read, is perfectly fine as a starting point, and better than one below.

One bug that's maybe worth correcting there is missing import time though, otherwise imo revert that sys.stdin.buffer thing with maybe a comment that it can potentially be reading >1 bytes, but should be fine with any human input, probably.

[mk-fg]

Much more complicated version of the example above, that should buffer incomplete utf-8 reads instead of blocking on those, and avoid reads after poll hup/err events:

import sys, select, time

# NOTE: This code does not work under Thonny because
#   in Thonny input is sent to program only after [Enter] is clicked.
# NOTE: Do not run this via "mpremote run ..." either, as it does not connect stdin

class TermReader:

  def __init__(self, byte_stream, buffer_bytes=100):
    self.stream, self.poller = byte_stream, select.poll()
    self.rb, self.rb_n, self.rb_len = bytearray(buffer_bytes), 0, buffer_bytes
    self.poller.register(self.stream, select.POLLIN)

  def rb_decode(self, a, b, max_char_len=6):
    'Returns decoded ring-buffer contents from a to b, and a-offset for next call'
    buff = self.rb[a:b] if a < b else self.rb[a:] + self.rb[:b]
    for n in range(max_char_len):
      try: result = buff[:-n or len(buff)].decode()
      except UnicodeError: pass
      else: return result, (self.rb_len + (b - n)) % self.rb_len
    else:
      if len(buff) > max_char_len: raise UnicodeError('Non-UTF-8 stream data')
    return '', a

  def read(self):
    n0, text = self.rb_n, list()
    poll_err = select.POLLHUP | select.POLLERR
    while ev := self.poller.poll(0):
      if ev[0][1] & poll_err or not (byte := self.stream.read(1)): break
      self.rb[self.rb_n] = byte[0]
      self.rb_n = (self.rb_n + 1) % self.rb_len
      if self.rb_n == n0:
        chunk, n0 = self.rb_decode(n0, self.rb_n)
        text.append(chunk)
    if self.rb_n != n0:
      chunk, self.rb_n = self.rb_decode(n0, self.rb_n)
      text.append(chunk)
    return ''.join(text)

term_reader = TermReader(sys.stdin.buffer)
while True:
  time.sleep(3)
  text = term_reader.read()
  print(f'Input during last 3s: {repr(text)}')

It uses ring-buffer to store input bytes until it is able to decode them without UnicodeError.
Iirc at least as far as micropython-1.20 there was no support for errors= keyword for bytes.decode(), so if buffer ends up with undecodable data that's longer than utf-8 character (assuming 6B, didn't check), UnicodeError is raised.

[GitHubsSilverBullet]

If I might be so bold, this one decodes UTF-8 and returns either ›None‹ or the UTF-8 character…

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
# vim:fileencoding=UTF-8:ts=4

import sys
import select

class StdioReader:

    def __init__(self):
        self._selpoll = select.poll()
        self._selpoll.register(sys.stdin, select.POLLIN)
        self._bytes, self._index, self._expecting = bytearray(4), 0, 0

    def __enter__(self):
        return self

    def __exit__(self, _type, value, traceback):
        return _type, value, traceback

    def getchar(self):
        if not len(self._selpoll.poll(0)):
            return None

        charval = sys.stdin.buffer.read(1)  # get a single byte (of a possible UTF-8 sequence)
        charval = charval[0]  # turn bytes array into an integer
        if charval & 0x80 == 0:  # is a single byte
            character = chr(charval)
            self._index = 0
            self._expecting = 0
            return character
        else:
            if charval & 0xe0 == 0xc0:  # is first of two bytes
                self._bytes[0], self._index, self._expecting = charval, 1, 1
            elif charval & 0xf0 == 0xe0:  # is first of three bytes
                self._bytes[0], self._index, self._expecting = charval, 1, 2
            elif charval & 0xf8 == 0xf0:  # is first of four bytes
                self._bytes[0], self._index, self._expecting = charval, 1, 3
            elif charval & 0xc0 == 0x80:  # is a sequence byte
                self._bytes[self._index] = charval
                self._index += 1
                if self._index > self._expecting:
                    character = self._bytes[0:self._expecting+1].decode()
                    self._index, self._expecting = 0, 0
                    return character
            else:
                raise UnicodeError
        return None

if __name__ == '__main__':
    from time import sleep_ms

    def classtest1():
        rdr = StdioReader()
        while True:
            char = rdr.getchar()
            if char:
                print(f'[{char}]', end=' ')
            else:
                sleep_ms(20)

    def classtest2():
        with StdioReader() as rdr:
            while True:
                char = rdr.getchar()
                if char:
                    print(f'[{char}]', end=' ')
                else:
                    sleep_ms(20)

    # classtest1()
    classtest2()

[gretel]

thank you very much!

[A-J-Bauer]

Not sure if still relevant,
this script reads without blocking from the terminal while blinking the onboard led

import sys
import select
import time
from machine import Pin, RTC

def handle_exit():
    led.low()                
    print("bye, bye")
    global stop
    stop = True

print("USAGE:")
print("time - to get the time on the MCU")
print("quit, exit or Ctrl+C to stop execution")

led = Pin("LED", Pin.OUT)
rtc = RTC()

selpoll = select.poll()
selpoll.register(sys.stdin, select.POLLIN)

stop = False
command = ""

try:
    while not stop:
        available = len(selpoll.poll(0))
        if available > 0:
            stravailable = sys.stdin.read(available)
            command += stravailable
            #hex_array = ' '.join(f'{ord(c):02x}' for c in command)
            #print(f"command: {command} hex: {hex_array}")
            print(stravailable, end="")
            
            if len(command) > 0 and command.endswith("\n"):
                command = command.rstrip()
                if command == "time":
                    led.toggle()
                    print(f"The time on this MCU is: {rtc.datetime()}")
                    led.toggle()
                elif command == "quit" or command == "exit":
                    handle_exit()

                command = ""
        
        time.sleep_ms(100)
        led.toggle()
        time.sleep_ms(100)
        led.toggle()
except KeyboardInterrupt:
    handle_exit()