# Constraining stdin in Python, v2.0

Rags, again.

This is the rewrite of Read stdin like a dictator.

From that post:

All too often I find myself wanting to allow only a certain list of characters to be written to stdin, and only recently did I actually bother to implement it. In Python, of all languages!

Essentially, this module provides a few APIs that allow a very tailored approach to reading characters from a terminal. By intercepting individual keypresses at the instant they occur, we can make cool decisions about closing the input stream whenever we want -- we dictate who says what in our terminal!

The standard input stream, after being opened, can be closed after a number of characters, or, at caller's option, any combination of a number of characters and allowed inputs.

... Yep, all still true, just way better, and actually cross-platform now.

It might seem like there's a lot of dead code / overkill stuff here; that's because I'm about to shift its focus from "input-constrainer-thing" to "poor-man's libreadline/curses" and so there's some provisioning in effect, but nevermind that.

I have some unittests but I won't supply them because writing them for an IO module is a lot more work than they're worth at this point.

#!/usr/bin/env python3

import sys
import struct
from platform import system
SYSTEM = system().lower()

class CHAR:
"""essentially an enum, to avoid clouding the ns"""
NUL = chr(0)
INT = chr(3)
EOF = chr(4)
BEL = chr(7)
BKS = chr(8)
LFD = chr(10)
CRR = chr(13)
ESC = chr(27)
SPC = chr(32)
DEL = chr(127)

CONDS = [
(lambda i, chars: i in chars),
(lambda i, chars: i not in chars),
(lambda *args, **kwargs: False),
]

def init(TERM_BUFSIZE=4096):
"""module initialiser: calls constructors so you don't have to
you must call this before other functions!"""

def checkinit(func, *args, **kwargs):

def isdefined(*args, **kwargs):

print("\n\tfatal: init() not called\n")
msg = "must call init() first, or call init() again before {}()".format(func.__name__)
raise TypeError(msg)

return func(*args, **kwargs)

return isdefined

def __init__(self, *args, **kwargs):

self.NAME = "NT"

if SYSTEM != "windows":
util.writer("\n\there be dragons; ye COWER in the SHADOW of", self.NAME, "\n\n")

self.msvcrt = __import__("msvcrt")
self.ctypes = __import__("ctypes")
try:
self.colorama  = __import__("colorama")
self.colorama.init()
except (AttributeError, ImportError):
print(
"""
you must install colorama to use this module on windows
do this by:
$cd colorama$ python setup.py install
"""
)
exit(2)

def getch(self):
"""use msvcrt to get a char"""
return self.msvcrt.getch()

def drain_buf(self):
"""while buffer, pseudo-nonblocking read bytes from buffer using msvcrt"""
y = []
while self.msvcrt.kbhit():
y.append(self.msvcrt.getch())
return "".join(y)

def __init__(self, TERMCTL_SPECIAL_BUFSIZE=4096):

self.NAME = "POSIX"

if SYSTEM == "windows":
util.writer("\n\there be dragons; ye COWER in the SHADOW of", self.NAME, "\n\n")

self.tty     = __import__("tty")
self.termios = __import__("termios")
self.fcntl   = __import__("fcntl")

self.O_NONBLOCK = __import__("os").O_NONBLOCK

self.TERM_BUFSIZE = TERMCTL_SPECIAL_BUFSIZE

def getch(self):
"""use old fashioned termios to getch"""
if sys.stdin.isatty():  # fixes "Inappropriate ioctl for device"
fd = sys.stdin.fileno()
old_settings = self.termios.tcgetattr(fd)
try:
self.tty.setraw(sys.stdin.fileno())
finally:
return ch
else:

def drain_buf(self):
if sys.stdin.isatty():
fd = sys.stdin.fileno()
fl = self.fcntl.fcntl(fd, self.fcntl.F_GETFL)
self.fcntl.fcntl(fd, self.fcntl.F_SETFL, fl | self.O_NONBLOCK)
try:
# if nothing is waiting on sys.stdin, then TypeError
# because "can't concat NoneType and str"
except TypeError:
chars = ""
finally:
self.fcntl.fcntl(fd, self.fcntl.F_SETFL, fl) # restore settings
return chars
else:

}.get(
SYSTEM,
)

class util():
"""utilities"""
def parsenum(num):
"""sys.maxsize if num is negative"""
num = int(num)
return sys.maxsize if num < 0 else num

def writer(*args):
"""write a string to stdout and flush.
should be used by all stdout-writing"""
if not args:
raise TypeError("writer requires at least one argument")
if len(args) > 1:
args = " ".join(str(i) for i in args).strip(" ")
else:
args = "".join(str(i) for i in args)
sys.stdout.write(args)
sys.stdout.flush()

def esc_filter(x, y):
"""append x to y as long as x is not DEL or backspace or esc"""
if x in (CHAR.DEL, CHAR.BKS):
try:
y.pop()
except IndexError:
pass
return y

y.append(x)
return y

@checkinit
"""interface for getch + drain_buf
if raw, behave like getch but with flushing for multibyte inputs"""

if raw:
return ch + more

# cooked

if ch == CHAR.INT: raise KeyboardInterrupt
if ch == CHAR.EOF: raise EOFError

if ch in (CHAR.BKS, CHAR.DEL):
util.writer(CHAR.BKS + CHAR.SPC + CHAR.BKS)
return CHAR.BKS

elif ch in (CHAR.CRR, CHAR.LFD):
util.writer(CHAR.CRR if SYSTEM == "Windows" else "")
return CHAR.LFD

elif ch == CHAR.ESC:
if more:
if more[0] == "[":
sp = more[1:]
if sp in ("D", "C"):
return "\033[" + sp

elif sp == "3~":
return CHAR.SPC

else:
return CHAR.BEL
else:
return CHAR.ESC + more

ch += more
return ch

@checkinit

@checkinit
def pretty_press(raw=False):
"""literally just read any fancy char from stdin let caller do whatever"""
y = []
if (not raw) and (i not in (CHAR.BKS, CHAR.DEL, CHAR.ESC)):
util.writer(i)
return util.esc_filter(i, y)

@checkinit
def _do_condition(
end_chars,
end_condition,
count,
ignore_chars=(),
ignore_condition=CHAR.CONDS[True + 1],  # always false
raw=False
):
y = []
count = util.parsenum(count)
while len(y) <= count:
if end_condition(i, end_chars):
break
if not ignore_condition(i, ignore_chars):
if (not raw) and (i not in (CHAR.BKS, CHAR.DEL)):
util.writer(i)
y = util.esc_filter(i, y)
return "".join(y)

@checkinit
def thismany(count, raw=False):

return _do_condition(
"",
CHAR.CONDS[True + 1],  # more than true == never expires :D
count,
raw=raw
)

@checkinit
def until(chars, invert=False, count=-1, raw=False):
"""get chars of stdin until any of chars is read,
or until count chars have been read, whichever comes first"""

return _do_condition(
chars,
CHAR.CONDS[invert],
count,
raw=raw
)

@checkinit
def until_not(chars, count=-1, raw=False):
or until count chars have been read; whichever comes first"""

return until(
chars,
invert=True,
count=count,
raw=raw
)

@checkinit
def ignore(
ignore_these,
end_on,
end_cond=True,
count=-1,
raw=False,
invert=False
):
"""ignore_these keypresses, and stop reading at end_on or count,
whichever comes first"""

return _do_condition(
end_on,
CHAR.CONDS[not end_cond],
count,
ignore_chars=ignore_these,
ignore_condition=CHAR.CONDS[invert],
raw=raw
)

@checkinit
def ignore_not(
ignore_these,
end_on,
end_cond=True,
count=-1,
raw=False
):
"""ignore everything that isn't these keypresses
and stop reading at end_on or count, whichever comes first"""

return ignore(
ignore_these,
end_on,
end_cond=end_cond,
count=count,
raw=raw,
invert=True
)


A trivial example of how this might be used:

>>> from pmlr import *
>>> init()
>>> print("enter your age:", end=" "); x = until_not("0123456789", count=2)
>>> x
123


It may seem like I'm abusing OOP and classes but those are partially for namespace-hygiene and partially because the way I did it previously was really fragile.

Additionally, self.NAME is for debugging, mostly, but I left it in for the same reason read_class is public and asking to be fiddled with: why not?

Finally, there is a "magic number" 4096. That's from this excellent answer by the creator of xterm.

You can get this on github, colorama included.

Take care of the docs

Please avoid destroying your documentation after writing it. checkinit destroys the documentation of the functions it is applied to. You should use functools.wraps.

• I did not know that, thank you. I'll look into it – cat Feb 25 '16 at 14:34
• @cat Please unaccept this answer as more comprehensive ones are probably coming – Caridorc Feb 26 '16 at 11:21
• Whose choice is accepting an answer? For now, I'll accept this one, as it's pretty valuable to me. If new ones appear, I'll change it. – cat Feb 26 '16 at 12:27
• @cat Ok, of course it is your choice. Just keep in mind that waiting some time before accepting an answer is standard practice here. – Caridorc Feb 26 '16 at 12:32

I've always hated when there is a program that requires me to call some setup function before doing anything. I also dislike when the module keeps some global state and everything is constrained by that. I think a much cleaner solution is something like this.

_readers = {}

if SYSTEM == 'windows'
else:


This gives us two things - the first, is that we don't need some global variable that the user then accesses, and also (if they so desire) gives them multiple readers of different buffer size, if that were to matter. This will require restructuring a lot of other code to not rely on a global variable. Also, it may be worthwhile to make all other global names prefixed with an _ so they aren't imported by from x import *, or make an __all__ list.

The first thing we'll have to do to get rid of that global state is move all of those functions that rely on global state into a class - we'll make a base class that encompasses them.

class _ReaderBaseClass:

def __init__(self, buffer_size):
self.buffer_size = buffer_size

@abc.abstractmethod
def getch(self):
raise NotImplementedError

@abc.abstractmethod
def drain_buf(self):
raise NotImplementedError

@abc.abstractproperty
def newline(self):
return NotImplemented


From here, we'll need to add the other functions in as methods - I'll take those one by one.

readkey

I don't have a lot of specific suggestions here, except that you should return self.newline in the case of a carriage return or line feed - this way you abstract that away to the subclass.

_do_condition

ignore_condition=CHAR.CONDS[True + 1], # always false


This is a massive code smell - if it is always False, then just say False. I'm more confused, though, because True + 1 is 2, which is True... then once I actually go and look at what that returns, I realize the lambda this returns is always false. Then you should actually just do CONDS[2], or even better you should define these lambdas as named functions somewhere so they're actually understandable. They really don't belong in CHAR - I just made them static methods of the base class, but they could be just normal functions if you want.

overall

In general, avoid the hacky indexing by booleans - afaik that they have numeric values of 0 and 1 is a CPython implementation detail (I might be wrong about this), and either way is horrifically unreadable.

I fear that maybe you've tried to get to be too generic using _do_condition as well - it has gotten to the point that the other functions are fairly hard to read. I don't have a good suggestion for this, and if you find it workable then keep going, but there is a balance you have to strike, and I'm not sure you're there.

It also bothers me that a reader actually writes something - imo it should only return values, and then the caller should display them, if at all. Otherwise, you should allow another parameter of an output stream for them to enter - they might not want to print to stdout by default.

### Miscellaneous

Why are you making CHAR a class (and not an Enum)? You're on Python 3, so unless you're using 3.3 or below for some reason (please don't) you have access to the enum module.

You actually have a bug - your util class's methods should all be static methods - as is, you have instance methods with strangely named parameters (i.e. you're using num instead of self).

I also feel like these might be worth making helper functions in reader instead of some util class.

Here is the complete code.

#!/usr/bin/env python3

import abc
import sys
import struct
from platform import system
SYSTEM = system().lower()

class CHAR:
NUL = chr(0)
INT = chr(3)
EOF = chr(4)
BEL = chr(7)
BKS = chr(8)
LFD = chr(10)
CRR = chr(13)
ESC = chr(27)
SPC = chr(32)
DEL = chr(127)

def __init__(self, buffer_size):
self.buffer_size = buffer_size

@abc.abstractproperty
def newline_char(self):
return NotImplemented

@abc.abstractmethod
def getch(self):
raise NotImplementedError

@abc.abstractmethod
def drain_buf(self):
raise NotImplementedError

@staticmethod
def char_in_list(ch, chars):
return ch in chars

@staticmethod
def char_not_in_list(ch, chars):
return ch not in chars

@staticmethod
def return_false(*args, **kwargs):
return False

"""interface for getch + drain_buf
if raw, behave like getch but with flushing for multibyte inputs"""

ch = self.getch()
more = self.drain_buf()

if raw:
return ch + more

# cooked

if ch == CHAR.INT: raise KeyboardInterrupt
if ch == CHAR.EOF: raise EOFError

if ch in (CHAR.BKS, CHAR.DEL):
util.writer(CHAR.BKS + CHAR.SPC + CHAR.BKS)
return CHAR.BKS
elif ch in (CHAR.CRR, CHAR.LFD):
util.writer(self.newline)
return CHAR.LFD
elif ch == CHAR.ESC:
if more:
if more[0] == "[":
sp = more[1:]
if sp in ("D", "C"):
return "\033[" + sp
elif sp == "3~":
return CHAR.SPC
else:
return CHAR.BEL
else:
return CHAR.ESC + more

return ch + more

def pretty_press(self, raw=False):
"""literally just read any fancy char from stdin let caller do whatever"""
if not raw and i not in (CHAR.BKS, CHAR.DEL, CHAR.ESC):
util.writer(i)
return util.esc_filter(i, [])

def _do_condition(
self,
end_chars,
end_condition,
count,
ignore_chars=(),
ignore_condition=return_false,
raw=False
):

y = []
count = util.parsenum(count)
while len(y) <= count:
if end_condition(i, end_chars):
break
if not ignore_condition(i, ignore_chars):
if not raw and i not in (CHAR.BKS, CHAR.DEL):
util.writer(i)
y = util.esc_filter(i, y)
return "".join(y)

def thismany(self, count, raw=False):

return self._do_condition(
"",
count,
raw=raw
)

def until(self, chars, invert=False, count=-1, raw=False):
"""get chars of stdin until any of chars is read,
or until count chars have been read, whichever comes first"""

if invert:
else:

return self._do_condition(
chars,
f,
count,
raw=raw
)

def until_not(self, chars, count=-1, raw=False):
or until count chars have been read; whichever comes first"""

return self.until(
chars,
invert=True,
count=count,
raw=raw
)

def ignore(
self,
ignore_these,
end_on,
end_cond=True,
count=-1,
raw=False,
invert=False
):
"""ignore_these keypresses, and stop reading at end_on or count,
whichever comes first"""

if end_cond:
else:

if invert:
else:

return self._do_condition(
end_on,
end_condition,
count,
ignore_chars=ignore_these,
ignore_condition=ignore_condition,
raw=raw
)

def ignore_not(
self,
ignore_these,
end_on,
end_cond=True,
count=-1,
raw=False
):
"""ignore everything that isn't these keypresses
and stop reading at end_on or count, whichever comes first"""

return self.ignore(
ignore_these,
end_on,
end_cond=end_cond,
count=count,
raw=raw,
invert=True
)

@property
def newline(self):
return CHAR.CRR

def __init__(self, buffer_size):

super().__init__(buffer_size)

if SYSTEM != "windows":
util.writer("\n\there be dragons; ye COWER in the SHADOW of NT\n\n")

self.msvcrt = __import__("msvcrt")
self.ctypes = __import__("ctypes")
try:
self.colorama  = __import__("colorama")
self.colorama.init()
except (AttributeError, ImportError):
print(
"""
you must install colorama to use this module on windows
do this by:
$cd colorama$ python setup.py install
"""
)
exit(2)

def getch(self):
"""use msvcrt to get a char"""
return self.msvcrt.getch()

def drain_buf(self):
"""while buffer, pseudo-nonblocking read bytes from buffer using msvcrt"""
y = []
while self.msvcrt.kbhit():
y.append(self.msvcrt.getch())
return "".join(y)

@property
def newline(self):
return ""

def __init__(self, buffer_size):

super().__init__(buffer_size)

if SYSTEM == "windows":
util.writer("\n\there be dragons; ye COWER in the SHADOW of POSIX\n\n")

self.tty     = __import__("tty")
self.termios = __import__("termios")
self.fcntl   = __import__("fcntl")

self.O_NONBLOCK = __import__("os").O_NONBLOCK

self.TERM_BUFSIZE = TERMCTL_SPECIAL_BUFSIZE

def getch(self):
"""use old fashioned termios to getch"""
if sys.stdin.isatty():  # fixes "Inappropriate ioctl for device"
fd = sys.stdin.fileno()
old_settings = self.termios.tcgetattr(fd)
try:
self.tty.setraw(sys.stdin.fileno())
finally:
return ch
else:

def drain_buf(self):
if sys.stdin.isatty():
fd = sys.stdin.fileno()
fl = self.fcntl.fcntl(fd, self.fcntl.F_GETFL)
self.fcntl.fcntl(fd, self.fcntl.F_SETFL, fl | self.O_NONBLOCK)
try:
# if nothing is waiting on sys.stdin, then TypeError
# because "can't concat NoneType and str"
except TypeError:
chars = ""
finally:
self.fcntl.fcntl(fd, self.fcntl.F_SETFL, fl) # restore settings
return chars
else:

if SYSTEM == 'windows'
else:

class util():
"""utilities"""
@staticmethod
def parsenum(num):
"""sys.maxsize if num is negative"""
num = int(num)
return sys.maxsize if num < 0 else num

@staticmethod
def writer(*args):
"""write a string to stdout and flush.
should be used by all stdout-writing"""
if not args:
raise TypeError("writer requires at least one argument")
if len(args) > 1:
args = " ".join(str(i) for i in args).strip(" ")
else:
args = "".join(str(i) for i in args)
sys.stdout.write(args)
sys.stdout.flush()

@staticmethod
def esc_filter(x, y):
"""append x to y as long as x is not DEL or backspace or esc"""
if x in (CHAR.DEL, CHAR.BKS):
try:
y.pop()
except IndexError:
pass
return y

y.append(x)
return y

• Thank you for the extensive review, despite that I really don't like your implementation and some suggestions. Also, please note the act of returning LFD vs CRR was informed by the writer of xterm. This is not about printing, it's about what that Enter key actually does. – cat Feb 25 '16 at 22:19
• @cat, sorry, what do you mean about LFD vs CRR? I don't believe that I misrepresented that point here. Do you mind expanding on what exactly you dislike about my implementation? – Dannnno Feb 25 '16 at 22:24
• @Dannno It seems like you're saying self.newline should be platform-dependent: "...you should return self.newline in the case of a carriage return or line feed - this way you abstract that away to the subclass." Is that a correct interpretation? – cat Feb 25 '16 at 22:26
• @cat I see, my implementation differs from yours. I can fix that. I should have written that you always return a newline, but what you write is platform independent. – Dannnno Feb 25 '16 at 22:31
• In raw mode, I would expect 13, in cooked mode, 10. The question was about raw mode, of course. - Thomas Dickey That goes for both Windows and Posix – cat Feb 25 '16 at 22:34