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]
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
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 = ""
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.
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.
В версии eclipse 2019, используемой в этом классе CDT org.eclipse.cdt.utils.pty.PTY, конкретный метод использования выглядит следующим образом:
boolean is*UsingPTY = ((properties.get( process ) == null) &&(PTY.isSupported(PTY.Mode.TERMINAL)))
||...