Local Windows support added

ollipal · Oct 28, 2021 · 1b0b280 · 1b0b280
1 parent 3aa9e80
commit 1b0b280
Show file tree

Hide file tree

Showing 3 changed files with 142 additions and 52 deletions.
diff --git a/README.md b/README.md
@@ -1,16 +1,13 @@
 # sshkeyboard
 
-The only keyboard event callback library that works in _all_ Unix
-environments.
+The only keyboard event callback library that works _everywhere_, even when
+used through an [SSH](https://en.wikipedia.org/wiki/Secure_Shell) connection
+(hence the name).
 
-It is suitable even when used through an
-[SSH](https://en.wikipedia.org/wiki/Secure_Shell) connection (hence the name),
-when using with headless computers/servers, or for example inside Windows
-Subsystem for Linux (WSL 2).
-
-One good use case is controlling some Raspberry Pi based robot or RC car
-through SSH. Note that this library can also be used locally without an SSH
-connection.
+It works with headless computers and servers, or for example inside Windows
+Subsystem for Linux (WSL 2). One good use case is controlling Raspberry Pi
+based robots or RC cars through SSH. Note that this library can also be used
+locally without an SSH connection.
 
 It does not depend on X server, uinput, root access (sudo) or
 any external dependencies.
@@ -60,15 +57,21 @@ $ python example.py
 
 ## How it works
 
-The library works without
+The sshkeyboard library works without
 [X server](https://en.wikipedia.org/wiki/X_Window_System)
-and [uinput](https://www.kernel.org/doc/html/v4.12/input/uinput.html)
-because it calls the events based on characters parsed from
-[sys.stdin](https://docs.python.org/3/library/sys.html#sys.stdin). This is
-done with [fcntl](https://docs.python.org/3/library/fcntl.html) and
+and [uinput](https://www.kernel.org/doc/html/v4.12/input/uinput.html).
+
+On Unix based systems (such as Linux, macOS) it works by parsing characters
+from [sys.stdin](https://docs.python.org/3/library/sys.html#sys.stdin). This
+is done with [fcntl](https://docs.python.org/3/library/fcntl.html) and
 [termios](https://docs.python.org/3/library/termios.html) standard library
 modules.
 
+On Windows [msvcrt](https://docs.python.org/3/library/msvcrt.html) standard
+library module is used to read user input. The Windows support is still new,
+so please create [an issue](https://github.com/ollipal/sshkeyboard/issues)
+if you run into problems.
+
 This behaviour allows it to work where other libraries like
 [pynput](#comparison-to-other-keyboard-libraries) or
 [keyboard](#comparison-to-other-keyboard-libraries) do not work, but
@@ -82,15 +85,6 @@ it comes with some **limitations**, mainly:
    `Shift`, `Caps Lock`, `Alt` and `Windows`/`Command`/`Super` key. That is
    why this library does not attempt to parse those even if they could be
    technically be parsed in some cases
-3. `termios` and `fcntl` are not supported on Windows (except on WSL / WSL 2).
-   Note that you _can_ take a SSH connection from Windows with
-   CMD/PowerShell/PuTTY to a Unix machine, and `sshkeyboard` _will_ work. This
-   limitation means that the library does not work directly on Windows. If you
-   figure out a workaround to make this work on Windows, please make
-   [a pull request](https://github.com/ollipal/sshkeyboard/pulls)! If you need
-   direct Windows use, check
-   [pynput](#comparison-to-other-keyboard-libraries) or
-   [keyboard](#comparison-to-other-keyboard-libraries) libraries out.
 
 ## Advanced use
 

diff --git a/src/sshkeyboard/__init__.py b/src/sshkeyboard/__init__.py
@@ -4,12 +4,9 @@
 
 import asyncio
 import concurrent.futures
-import fcntl
 import os
 import sys
-import termios
 import traceback
-import tty
 from contextlib import contextmanager
 from inspect import signature
 from platform import system
@@ -22,6 +19,16 @@
 except ImportError:  # this allows local testing: python __init__.py
     from _asyncio_run_backport_36 import run36
 
+_is_windows = system().lower() == "windows"
+
+if _is_windows:
+    import msvcrt
+else:
+    import fcntl
+    import termios
+    import tty
+
+
 # Global state
 
 # Makes sure only listener can be started at a time
@@ -35,7 +42,7 @@
 # All possible ansi characters here:
 # https://github.com/prompt-toolkit/python-prompt-toolkit/blob/master/prompt_toolkit/input/ansi_escape_sequences.py
 # Listener does not support modifier keys for now
-_ANSI_CHAR_TO_READABLE = {
+_UNIX_ANSI_CHAR_TO_READABLE = {
     # 'Regular' characters
     "\x1b": "esc",
     "\x7f": "backspace",
@@ -104,13 +111,44 @@
     "\x1b[24;2~": "f24",
 }
 
+_WIN_CHAR_TO_READABLE = {
+    "\x1b": "esc",
+    "\x08": "backspace",
+    "àR": "insert",
+    "àS": "delete",
+    "àI": "pageup",
+    "àQ": "pagedown",
+    "àG": "home",
+    "àO": "end",
+    "àH": "up",
+    "àP": "down",
+    "àM": "right",
+    "àK": "left",
+    "\x00;": "f1",
+    "\x00<": "f2",
+    "\x00=": "f3",
+    "\x00>": "f4",
+    "\x00?": "f5",
+    "\x00@": "f6",
+    "\x00A": "f7",
+    "\x00B": "f8",
+    "\x00C": "f9",
+    "\x00D": "f10",
+    # "": "f11", ?
+    "à†": "f12",
+}
+
 # Some non-ansi characters that need a readable representation
 _CHAR_TO_READABLE = {
     "\t": "tab",
     "\n": "enter",
+    "\r": "enter",
     " ": "space",
 }
 
+_WIN_SPECIAL_CHAR_STARTS = {"\x1b", "\x08", "\x00", "\xe0"}
+_WIN_REQUIRES_TWO_READS_STARTS = {"\x00", "\xe0"}
+
 
 def listen_keyboard(
     on_press: Optional[Callable[[str], Any]] = None,
@@ -191,7 +229,7 @@ def release(key):
         sleep,
     )
 
-    if _is_python_36:
+    if _is_python_36():
         run36(coro)
     else:
         asyncio.run(coro)
@@ -234,10 +272,7 @@ async def listen_keyboard_manual(
 
     global _running
     global _should_run
-    # Check system
-    assert (
-        system().lower() != "windows"
-    ), "sshkeyboard does not support Windows"
+    # Check the system
     assert sys.version_info >= (3, 6), (
         "sshkeyboard requires Python version 3.6+, you have "
         f"{sys.version_info.major}.{sys.version_info.minor}"
@@ -289,10 +324,12 @@ async def listen_keyboard_manual(
         on_press_callback=_callback(on_press, sequential, executor),
         on_release_callback=_callback(on_release, sequential, executor),
         until=until,
+        sequential=sequential,
         delay_second_char=delay_second_char,
         delay_other_chars=delay_other_chars,
         lower=lower,
         debug=debug,
+        sleep=sleep,
     )
     # State does change
     state = SimpleNamespace(
@@ -410,6 +447,11 @@ async def _cb(key):
 # http://ballingt.com/_nonblocking-stdin-in-python-3/
 @contextmanager
 def _raw(stream):
+    # Not required on windows
+    if _is_windows:
+        yield
+        return
+
     original_stty = termios.tcgetattr(stream)
     try:
         tty.setcbreak(stream)
@@ -420,6 +462,11 @@ def _raw(stream):
 
 @contextmanager
 def _nonblocking(stream):
+    # Not required on windows
+    if _is_windows:
+        yield
+        return
+
     fd = stream.fileno()
     orig_fl = fcntl.fcntl(fd, fcntl.F_GETFL)
     try:
@@ -429,47 +476,92 @@ def _nonblocking(stream):
         fcntl.fcntl(fd, fcntl.F_SETFL, orig_fl)
 
 
-def _read_chars(amount):
+def _read_char(debug):
+    if _is_windows:
+        return _read_char_win(debug)
+    else:
+        return _read_char_unix(debug)
+
+
+def _read_char_win(debug):
+    # Return if nothing to read
+    if not msvcrt.kbhit():
+        return ""
+
+    char = msvcrt.getwch()
+    if char in _WIN_SPECIAL_CHAR_STARTS:
+        # Check if requires one more read
+        if char in _WIN_REQUIRES_TWO_READS_STARTS:
+            char += msvcrt.getwch()
+
+        if char in _WIN_CHAR_TO_READABLE:
+            return _WIN_CHAR_TO_READABLE[char]
+        else:
+            if debug:
+                print(f"Non-supported win char: {repr(char)}")
+            return None
+
+    # Change some character representations to readable strings
+    elif char in _CHAR_TO_READABLE:
+        char = _CHAR_TO_READABLE[char]
+
+    return char
+
+
+def _read_char_unix(debug):
+    char = _read_unix_stdin(1)
+
+    # Skip and continue if read failed
+    if char is None:
+        return None
+
+    # Handle any character
+    elif char != "":
+        # Read more if ansi character, skip and continue if unknown
+        if _is_unix_ansi(char):
+            char, raw = _read_and_parse_unix_ansi(char)
+            if char is None:
+                if debug:
+                    print(f"Non-supported ansi char: {repr(raw)}")
+                return None
+        # Change some character representations to readable strings
+        elif char in _CHAR_TO_READABLE:
+            char = _CHAR_TO_READABLE[char]
+
+    return char
+
+
+def _read_unix_stdin(amount):
     try:
         return sys.stdin.read(amount)
     except IOError:
         return None
 
 
 # '\x' at the start is a good indicator for ansi character
-def _is_ansi(char):
+def _is_unix_ansi(char):
     rep = repr(char)
     return len(rep) >= 2 and rep[1] == "\\" and rep[2] == "x"
 
 
-def _read_and_parse_ansi(char):
-    char += _read_chars(5)
-    if char in _ANSI_CHAR_TO_READABLE:
-        return _ANSI_CHAR_TO_READABLE[char], char
+def _read_and_parse_unix_ansi(char):
+    char += _read_unix_stdin(5)
+    if char in _UNIX_ANSI_CHAR_TO_READABLE:
+        return _UNIX_ANSI_CHAR_TO_READABLE[char], char
     else:
         return None, char
 
 
 async def _react_to_input(state, options):
     # Read next character
-    state.current = _read_chars(1)
+    state.current = _read_char(options.debug)
 
     # Skip and continue if read failed
     if state.current is None:
         return state
 
     # Handle any character
     elif state.current != "":
-        # Read more if ansi character, skip and continue if unknown
-        if _is_ansi(state.current):
-            state.current, raw = _read_and_parse_ansi(state.current)
-            if state.current is None:
-                if options.debug:
-                    print(f"Non-supported ansi char: {repr(raw)}")
-                return state
-        # Change some character representations to readable strings
-        elif state.current in _CHAR_TO_READABLE:
-            state.current = _CHAR_TO_READABLE[state.current]
 
         # Make lower case if requested
         if options.lower:
@@ -483,6 +575,10 @@ async def _react_to_input(state, options):
         # Release state.previous if new pressed
         if state.previous != "" and state.current != state.previous:
             await options.on_release_callback(state.previous)
+            # Weirdly on_release fires too late on Windows unless there is
+            # an extra sleep here when sequential=False...
+            if _is_windows and not options.sequential:
+                await asyncio.sleep(options.sleep)
 
         # Press if new character, update state.previous
         if state.current != state.previous:

diff --git a/tests/test_sshkeyboard.py b/tests/test_sshkeyboard.py
@@ -17,14 +17,14 @@ def _raw(stream):
     def _nonblocking(stream):
         yield
 
-    def _read_chars(amount):
+    def _read_char(debug):
         global i
         i += 1
         return return_chars[i % len(return_chars)]
 
     sshkeyboard._raw = _raw
     sshkeyboard._nonblocking = _nonblocking
-    sshkeyboard._read_chars = _read_chars
+    sshkeyboard._read_char = _read_char
 
 
 setup_testing()