66import logging
77import os
88import sys
9- from collections .abc import Callable
9+ from collections .abc import AsyncGenerator , Callable
1010from contextlib import asynccontextmanager
1111from dataclasses import dataclass
1212from functools import partial
@@ -100,17 +100,12 @@ def print_help[ContextT: CliContext](
100100 return ret + " ? Print this message"
101101
102102
103- try :
104- import fcntl
105- import termios
106- import tty
103+ async def getch () -> AsyncGenerator [str ]:
104+ try :
105+ import fcntl
106+ import termios
107+ import tty
107108
108- async def _monitor_stdin [ContextT : CliContext ]( # pyright: ignore[reportRedeclaration]
109- clictx : ContextT ,
110- key_to_cmd : KeyCmdMapType [ContextT ],
111- * ,
112- puts : Callable [[str ], Any ] = puts ,
113- ) -> None :
114109 fd = sys .stdin .fileno ()
115110 termios_saved = termios .tcgetattr (fd )
116111 fnctl_flags = fcntl .fcntl (sys .stdin , fcntl .F_GETFL )
@@ -121,38 +116,58 @@ async def _monitor_stdin[ContextT: CliContext]( # pyright: ignore[reportRedecla
121116 fcntl .fcntl (sys .stdin , fcntl .F_SETFL , fnctl_flags | os .O_NONBLOCK )
122117
123118 while True :
124- ch = sys .stdin .read (1 )
119+ yield sys .stdin .read (1 )
125120
126- if len (ch ) == 0 :
127- await asyncio .sleep (0.1 )
128- continue
129-
130- if (key := ord (ch )) == 0x3F :
131- puts (print_help (clictx , key_to_cmd ))
132-
133- elif (keyfunc := key_to_cmd .get (key )) is not None and callable (
134- keyfunc .func
135- ):
136- if (ret := keyfunc .func (clictx )) is not None :
137- puts (ret )
138-
139- else :
140- logger .debug (f"Ignoring character 0x{ key :02x} on stdin" )
121+ except asyncio .CancelledError :
122+ pass
141123
142124 finally :
143125 logger .debug ("Restoring stdin" )
144126 termios .tcsetattr (fd , termios .TCSADRAIN , termios_saved )
145127 fcntl .fcntl (sys .stdin , fcntl .F_SETFL , fnctl_flags )
146128
147- except ImportError :
129+ except ImportError :
130+ pass
131+
132+ try :
133+ import msvcrt
134+
135+ while True :
136+ if msvcrt .kbhit (): # type: ignore[attr-defined]
137+ yield msvcrt .getch () # type: ignore[attr-defined]
148138
149- async def _monitor_stdin [ContextT : CliContext ](
150- clictx : ContextT ,
151- key_to_cmd : KeyCmdMapType [ContextT ],
152- * ,
153- puts : Callable [[str ], Any ] = puts ,
154- ) -> None :
155- _ = clictx , key_to_cmd , puts
139+ else :
140+ await asyncio .sleep (0.1 )
141+
142+ except ImportError as exc :
143+ raise NotImplementedError from exc
144+
145+
146+ async def _monitor_stdin [ContextT : CliContext ](
147+ clictx : ContextT ,
148+ key_to_cmd : KeyCmdMapType [ContextT ],
149+ * ,
150+ puts : Callable [[str ], Any ] = puts ,
151+ ) -> None :
152+ try :
153+ async for ch in getch ():
154+ if len (ch ) == 0 :
155+ await asyncio .sleep (0.1 )
156+ continue
157+
158+ if (key := ord (ch )) == 0x3F :
159+ puts (print_help (clictx , key_to_cmd ))
160+
161+ elif (keyfunc := key_to_cmd .get (key )) is not None and callable (
162+ keyfunc .func
163+ ):
164+ if (ret := keyfunc .func (clictx )) is not None :
165+ puts (ret )
166+
167+ else :
168+ logger .debug (f"Ignoring character 0x{ key :02x} on stdin" )
169+
170+ except NotImplementedError :
156171 logger .warning ("The 'debug' plugin does not work on this platform" )
157172 return None
158173
0 commit comments