Spaces:
Paused
Paused
File size: 2,074 Bytes
deab47c |
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 68 |
import select
import subprocess
import time
import sys
from typing import Optional, Tuple
class LocalInteractiveSession:
def __init__(self):
self.process = None
self.full_output = ''
def connect(self):
# Start a new subprocess with the appropriate shell for the OS
if sys.platform.startswith('win'):
# Windows
self.process = subprocess.Popen(
['cmd.exe'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
else:
# macOS and Linux
self.process = subprocess.Popen(
['/bin/bash'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
def close(self):
if self.process:
self.process.terminate()
self.process.wait()
def send_command(self, command: str):
if not self.process:
raise Exception("Shell not connected")
self.full_output = ""
self.process.stdin.write(command + '\n') # type: ignore
self.process.stdin.flush() # type: ignore
def read_output(self) -> Tuple[str, Optional[str]]:
if not self.process:
raise Exception("Shell not connected")
partial_output = ''
while True:
rlist, _, _ = select.select([self.process.stdout], [], [], 0.1)
if rlist:
line = self.process.stdout.readline() # type: ignore
if line:
partial_output += line
self.full_output += line
time.sleep(0.1)
else:
break # No more output
else:
break # No data available
if not partial_output:
return self.full_output, None
return self.full_output, partial_output |