Код: Выделить всё
import asyncio
import tty
async def forward_until_cancelled(fd, handler, on_end=None):
"""Asynchronously read from the master side of the PTY and print to stdout. Note that stdin/stdout are combined."""
loop = asyncio.get_running_loop()
loop.add_reader(fd, handler, fd)
try:
await asyncio.Future() # Wait indefinitely until cancelled
except asyncio.CancelledError:
pass
finally:
loop.remove_reader(fd)
# print("read done")
if on_end:
on_end(fd, handler)
enter code here
@contextmanager
def pty_raw_context():
master_fd, slave_fd = pty.openpty()
old_tty = termios.tcgetattr(sys.stdin)
def termios_reset():
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
try:
# Set the terminal in raw mode for capturing control characters
tty.setraw(sys.stdin.fileno())
yield master_fd, slave_fd, termios_reset
finally:
# Restore the terminal to its previous state
termios_reset()
os.close(master_fd)
try:
os.close(slave_fd)
except:
pass
sys.stdout.flush()
sys.stderr.flush()
enter code here
async def pty_command(command, password=None):
with pty_raw_context() as (master_fd, slave_fd, termios_reset):
print("Starting PTY command:", command)
# Start the process using the slave as stdin, stdout, stderr
process = await asyncio.create_subprocess_shell(
command,
stdin=slave_fd,
stdout=slave_fd,
stderr=slave_fd
)
# Close the slave in the parent process
os.close(slave_fd)
loop = asyncio.get_running_loop()
# @dataclasses.dataclass
# class Buf:
# buf:str = ""
buf = StringIO()
def read_from_child_handler(master_fd):
print("handle read")
try:
data = os.read(master_fd, 1024)
except OSError:
loop.remove_reader(master_fd)
t2.cancel()
t1.cancel()
return
if not data: # EOF
sys.stdout.buffer.flush()
loop.remove_reader(master_fd)
t2.cancel()
t1.cancel()
else:
sys.stdout.buffer.write(data)
sys.stdout.flush()
buf.write(data.decode("utf-8"))
#print("BUF>", data.decode("utf-8"), "
Подробнее здесь: [url]https://stackoverflow.com/questions/79389066/pty-openpty-output-directly-prints-output-but-only-password-prompts-instead-of[/url]