1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
import tty
import termios
import fcntl
import os
from typing import IO, ContextManager, Type, List, Union, Optional
from types import TracebackType
_Attr = list[Union[int, list[Union[bytes, int]]]]
class Nonblocking(ContextManager):
"""
A context manager for making an input stream nonblocking.
"""
def __init__(self, stream: IO) -> None:
self.stream = stream
self.fd = self.stream.fileno()
def __enter__(self) -> None:
self.orig_fl = fcntl.fcntl(self.fd, fcntl.F_GETFL)
fcntl.fcntl(self.fd, fcntl.F_SETFL, self.orig_fl | os.O_NONBLOCK)
def __exit__(
self,
type: type[BaseException] | None = None,
value: BaseException | None = None,
traceback: TracebackType | None = None,
) -> None:
fcntl.fcntl(self.fd, fcntl.F_SETFL, self.orig_fl)
class Termmode(ContextManager):
def __init__(self, stream: IO, attrs: _Attr) -> None:
self.stream = stream
self.attrs = attrs
def __enter__(self) -> None:
self.original_stty = termios.tcgetattr(self.stream)
termios.tcsetattr(self.stream, termios.TCSANOW, self.attrs)
def __exit__(
self,
type: type[BaseException] | None = None,
value: BaseException | None = None,
traceback: TracebackType | None = None,
) -> None:
termios.tcsetattr(self.stream, termios.TCSANOW, self.original_stty)
class Cbreak(ContextManager[Termmode]):
def __init__(self, stream: IO) -> None:
self.stream = stream
def __enter__(self) -> Termmode:
self.original_stty = termios.tcgetattr(self.stream)
tty.setcbreak(self.stream, termios.TCSANOW)
return Termmode(self.stream, self.original_stty)
def __exit__(
self,
type: type[BaseException] | None = None,
value: BaseException | None = None,
traceback: TracebackType | None = None,
) -> None:
termios.tcsetattr(self.stream, termios.TCSANOW, self.original_stty)
|
