Add support for cancellation tokens

[pomponchik]

Hello!

There is a pattern for comfortable and safe stopping of the functions - cancellation tokens. I have created an implementation of it - cantok. I suggest to use it in the readkey function like in this example:

import readchar
import cantok

token = cantok.TimeoutToken(5)
key = readchar.readkey(token=token) # by default the token is just SimpleToken()

Into the readkey function you need to read the state of the token periodically. If it was cancelled - you need to stop read the stdin and return something or raise an exception.

Notice, my suggestion is very similar to an another about non-blocking reading of the stdin, but it is more universal.

[Cube707]

Thank you for the suggestion.

Using tokens seems overkill though, a simple timeout is probably sufficient. This can't be implemented at the moment though, as readchar internaly gets blocked by kernel calls, which makes impossible to perform any form of checks (be it time or token based).

See the development branch, where a more versatile version is under development, which would allow for this kind of implementation. It would also allow the user to implement his own token based variant instead of the default timeout version, if he wishes. It not fully working though and I didn't find the time to continue on it.

[pomponchik]

@Cube707 Could you say only for me - now is there a way to stop reading of a char except to send the char to stdin? From the inside of a program.

[Cube707]

At the moment the only way is to spawn a thread and kill that, to send a interrupt to the process or to force readchar to finish consuming input.

[pomponchik]

to spawn a thread and kill that, to send a interrupt to the process

How can i interrupt the thread? Or i got it wrong?

[Cube707]

see #43 for a simple example. after you create the thread and store its management object in a variable, you have methods available to kill it.

[Cube707]

I seem so be mistaken, you cannot kill a thread. I thought i tested this, but looks like I am wrong

[pomponchik]

Yes, looks like i can't do it. This is a problem for me. Now i make some program where i need to read some chars and automatically stop it when the program is stopped. I do not see any way to do it.

[Cube707]

If the program is terminated it will stop (CTRL+C, CTRL+D or any other form of SIG).

This libary was created for textual user interfaces and alike. There you are sure you can block, as only a input from the user requires you to continue. Everything else is currently out of scope.

For games or similar interactive things, you would need a more low-level input driver anyway.

[Cube707]

also, using the multiprosessing libary and spawning a seperate process works:

from multiprocessing import Process
from time import sleep, time

from readchar import key, readkey


# construct an inverted code -> key-name mapping
# we need to revese the items so that aliases won't overrive the original name later on
known_keys = {v: k for k, v in reversed(vars(key).items()) if not k.startswith("__")}


def keylogger():
    print("starting keylogger...")
    while True:
        try:
            data = readkey()
        except KeyboardInterrupt:
            break
        if data in known_keys:
            data = known_keys[data]
        print(f"{data}-key was pressed")


if __name__ == "__main__":
    p = Process(name="keylogger", target=keylogger)
    p.start()
    sleep(5)
    print("timeout..")
    p.kill()