vimgolf_gym.terminal_executor

Terminal executor

Code source: terminal_asciicast_record_executor.py (modified)

View Source

  1"""
  2Terminal executor
  3
  4Code source: [terminal_asciicast_record_executor.py](https://github.com/James4Ever0/agi_computer_control/blob/master/web_gui_terminal_recorder/executor_and_replayer/terminal_asciicast_record_executor.py) (modified)
  5"""
  6
  7# pending pypi project name: termexec
  8# if you want to release this code as a pypi package, you must create a through unit test, across launching terminal, docker container, network requests, etc. focus on the effectiveness of clean-up, and the ability to handle exceptions.
  9
 10import threading
 11import time
 12import os
 13import signal
 14import atexit
 15
 16import agg_python_bindings
 17import ptyprocess
 18from pydantic import BaseModel
 19from typing import Protocol, cast
 20
 21
 22class Cursor(BaseModel):
 23    """Cursor position and visibility.
 24
 25    Attributes:
 26        x (int): The x-coordinate (column) of the cursor.
 27        y (int): The y-coordinate (row/line) of the cursor.
 28        hidden (bool): True if the cursor is hidden, False otherwise.
 29    """
 30
 31    x: int
 32    y: int
 33    hidden: bool
 34
 35
 36def decode_bytes_to_utf8_safe(_bytes: bytes):
 37    """
 38    Decode with UTF-8, but replace errors with a replacement character (�).
 39    """
 40    ret = _bytes.decode("utf-8", errors="replace")
 41    return ret
 42
 43
 44# screen init params: width, height
 45# screen traits: write_bytes, display, screenshot
 46
 47
 48class _TerminalScreenProtocol(Protocol):
 49    def write_bytes(self, _bytes: bytes): ...
 50    @property
 51    def display(self) -> str: ...
 52    def screenshot(self, png_output_path: str): ...
 53    def close(self): ...
 54    @property
 55    def cursor(self) -> Cursor: ...
 56
 57
 58class AvtScreen:
 59    def __init__(self, width: int, height: int):
 60        """
 61        Initialize an AvtScreen object.
 62
 63        Args:
 64            width (int): The width of the virtual terminal emulator.
 65            height (int): The height of the virtual terminal emulator.
 66        """
 67        self.vt = agg_python_bindings.TerminalEmulator(width, height)
 68        """terminal emulator provided by avt"""
 69        self._closing = False
 70
 71    def write_bytes(self, _bytes: bytes):
 72        """
 73        Write the given bytes to the virtual terminal emulator.
 74
 75        The bytes are decoded with UTF-8 and any decoding errors are replaced with a
 76        replacement character (�).
 77
 78        Args:
 79            _bytes (bytes): The bytes to be written to the virtual terminal emulator.
 80        """
 81        decoded_bytes = decode_bytes_to_utf8_safe(_bytes)
 82        self.vt.feed_str(decoded_bytes)
 83
 84    @property
 85    def cursor(self):
 86        """
 87        Get the current position of the cursor as a `Cursor` object.
 88
 89        Returns:
 90            Cursor: A `Cursor` object with `x` and `y` properties for the current column and row of the cursor, respectively, and a `hidden` property which is `True` if the cursor is hidden and `False` otherwise.
 91        """
 92        col, row, visible = self.vt.get_cursor()
 93        ret = Cursor(x=col, y=row, hidden=not visible)
 94        return ret
 95
 96    @property
 97    def display(self):
 98        """
 99        Get the current display of the terminal emulator as a string.
100
101        Returns:
102            str: A string representation of the current display of the terminal emulator.
103        """
104        ret = "\n".join(self.vt.text_raw())
105        return ret
106
107    def screenshot(self, png_output_path: str):
108        """
109        Saves the current state of the terminal emulator to a PNG image file.
110
111        The image has the same width and height as the terminal emulator.
112
113        Args:
114            png_output_path: The path to write the PNG image to.
115        """
116        self.vt.screenshot(png_output_path)
117
118    def close(self):
119        """
120        Releases all resources used by the terminal emulator.
121
122        This is necessary to avoid crashes when creating multiple instances of this class.
123        """
124        if not self._closing:
125            self._closing = True
126            del self.vt
127
128    def __enter__(self):
129        return self
130
131    def __exit__(self, exc, value, tb):
132        self.close()
133
134
135class TerminalProcess:
136    def __init__(self, command: list[str], width: int, height: int, backend="avt"):
137        """
138        Initializes the terminal emulator with a command to execute.
139
140        Args:
141            command (list[str]): List of command strings to execute in the terminal
142            width (int): Width of the terminal emulator
143            height (int): Height of the terminal emulator
144            backend (str, optional): Backend to use for terminal emulator. Defaults to "avt".
145        """
146        self._closing = False
147        rows, cols = height, width
148        self.pty_process: ptyprocess.PtyProcess = cast(
149            ptyprocess.PtyProcess,
150            ptyprocess.PtyProcess.spawn(command, dimensions=(rows, cols)),
151        )
152        """a process executing command in a pseudo terminal"""
153
154        if backend == "avt":
155            self.vt_screen = AvtScreen(width=width, height=height)
156            """virtual terminal screen"""
157        else:
158            raise ValueError(
159                "Unknown terminal emulator backend '%s' (known ones: avt, pyte)"
160                % backend
161            )
162
163        self.vt_screen = cast(_TerminalScreenProtocol, self.vt_screen)
164
165        self.__pty_process_reading_thread = threading.Thread(
166            target=self.__read_and_update_screen, daemon=True
167        )
168        self.__start_ptyprocess_reading_thread()
169        atexit.register(self.close)
170
171    def __start_ptyprocess_reading_thread(self):
172        """Starts a thread to read output from the terminal process and update the Pyte screen"""
173        self.__pty_process_reading_thread.start()
174
175    def write(self, data: bytes):
176        """Writes input data to the terminal process"""
177        self.pty_process.write(data)
178
179    def close(self):
180        """Closes the terminal process and the reading thread"""
181        if not self._closing:
182            self._closing = True
183            os.kill(self.pty_process.pid, signal.SIGTERM)
184            time.sleep(0.5)
185            if self.pty_process.isalive:
186                os.kill(self.pty_process.pid, signal.SIGKILL)
187            self.vt_screen.close()
188            self.__pty_process_reading_thread.join(timeout=0.5)
189
190    def __enter__(self):
191        return self
192
193    def __exit__(self, exc, value, tb):
194        self.close()
195
196    def __read_and_update_screen(self, poll_interval=0.01):
197        """Reads available output from terminal and updates Pyte screen
198
199        Args:
200            poll_interval (float, optional): Interval in seconds to poll for available output. Defaults to 0.01.
201        """
202        while True:
203            try:
204                # ptyprocess.read is blocking. only pexpect has read_nonblocking
205                process_output_bytes = self.pty_process.read(1024)
206                # write bytes to pyte screen
207                self.vt_screen.write_bytes(process_output_bytes)
208            except KeyboardInterrupt:  # user interrupted
209                break
210            except SystemExit:  # python process exit
211                break
212            except SystemError:  # python error
213                break
214            except EOFError:  # terminal died
215                break
216            except:
217                # Timeout means no data available, EOF means process ended
218                pass
219            finally:
220                time.sleep(poll_interval)
221
222
223class TerminalExecutor:
224    def __init__(self, command: list[str], width: int, height: int):
225        """
226        Initializes executor with a command to run in terminal emulator, using avt as backend.
227
228        Args:
229            command (list[str]): List of command strings to execute
230            width (int): Width of the terminal emulator
231            height (int): Height of the terminal emulator
232        """
233        self.terminal = TerminalProcess(command=command, width=width, height=height)
234        """a terminal process, running command in pty screen"""
235        self._closing = False
236
237    def input(self, text: str):
238        """
239        Sends input text to the terminal process
240
241        Args:
242            text (str): The input text to send
243        """
244        self.terminal.write(text.encode())
245        # Allow time for processing output
246        time.sleep(0.1)
247
248    @property
249    def display(self) -> str:
250        """
251        Get the current display of the terminal emulator as a string.
252        """
253
254        return self.terminal.vt_screen.display
255
256    def screenshot(self, png_save_path: str):
257        """
258        Saves the current display of the terminal emulator as a .png file
259
260        Args:
261            png_save_path (str): The path to save the screenshot to
262        """
263        self.terminal.vt_screen.screenshot(png_save_path)
264
265    def close(self):
266        """
267        Closes the terminal emulator process and the associated reading thread.
268        """
269        if not self._closing:
270            self._closing = True
271            self.terminal.close()
272
273
274def test_harmless_command_locally_with_bash():
275    """
276    Tests the TerminalExecutor class with a harmless command by running a docker alpine container
277    and executing a series of input events, then taking a screenshot and dumping the terminal
278    display to a file.
279    """
280    SLEEP_INTERVAL = 0.5
281    command = ["docker", "run", "--rm", "-it", "alpine"]
282    input_events = ['echo "Hello World!"', "\n"]
283    executor = TerminalExecutor(command=command, width=80, height=24)
284    time.sleep(1)
285    for event in input_events:
286        executor.input(event)
287        time.sleep(SLEEP_INTERVAL)
288    # check for screenshot, text dump
289    text_dump = executor.display
290    print("Dumping terminal display to ./terminal_executor_text_dump.txt")
291    with open("./terminal_executor_text_dump.txt", "w+") as f:
292        f.write(text_dump)
293    print("Taking terminal screenshot at ./terminal_executor_screenshot.png")
294    executor.screenshot("./terminal_executor_screenshot.png")
295    print("Done")
296
297
298def test():
299    """
300    Runs a test for the TerminalExecutor class by running a harmless command
301    locally with bash and taking a screenshot and dumping the terminal display
302    to a file.
303
304    This test is useful for checking that the TerminalExecutor class works in
305    a real environment.
306    """
307    test_harmless_command_locally_with_bash()
308
309
310if __name__ == "__main__":
311    test()

class Cursor(pydantic.main.BaseModel): View Source

23class Cursor(BaseModel):
24    """Cursor position and visibility.
25
26    Attributes:
27        x (int): The x-coordinate (column) of the cursor.
28        y (int): The y-coordinate (row/line) of the cursor.
29        hidden (bool): True if the cursor is hidden, False otherwise.
30    """
31
32    x: int
33    y: int
34    hidden: bool

Cursor position and visibility.

Attributes:

x (int): The x-coordinate (column) of the cursor.
y (int): The y-coordinate (row/line) of the cursor.
hidden (bool): True if the cursor is hidden, False otherwise.

x: int

y: int

hidden: bool

model_config: ClassVar[pydantic.config.ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

def decode_bytes_to_utf8_safe(_bytes: bytes): View Source

37def decode_bytes_to_utf8_safe(_bytes: bytes):
38    """
39    Decode with UTF-8, but replace errors with a replacement character (�).
40    """
41    ret = _bytes.decode("utf-8", errors="replace")
42    return ret

Decode with UTF-8, but replace errors with a replacement character (�).

class AvtScreen: View Source

 59class AvtScreen:
 60    def __init__(self, width: int, height: int):
 61        """
 62        Initialize an AvtScreen object.
 63
 64        Args:
 65            width (int): The width of the virtual terminal emulator.
 66            height (int): The height of the virtual terminal emulator.
 67        """
 68        self.vt = agg_python_bindings.TerminalEmulator(width, height)
 69        """terminal emulator provided by avt"""
 70        self._closing = False
 71
 72    def write_bytes(self, _bytes: bytes):
 73        """
 74        Write the given bytes to the virtual terminal emulator.
 75
 76        The bytes are decoded with UTF-8 and any decoding errors are replaced with a
 77        replacement character (�).
 78
 79        Args:
 80            _bytes (bytes): The bytes to be written to the virtual terminal emulator.
 81        """
 82        decoded_bytes = decode_bytes_to_utf8_safe(_bytes)
 83        self.vt.feed_str(decoded_bytes)
 84
 85    @property
 86    def cursor(self):
 87        """
 88        Get the current position of the cursor as a `Cursor` object.
 89
 90        Returns:
 91            Cursor: A `Cursor` object with `x` and `y` properties for the current column and row of the cursor, respectively, and a `hidden` property which is `True` if the cursor is hidden and `False` otherwise.
 92        """
 93        col, row, visible = self.vt.get_cursor()
 94        ret = Cursor(x=col, y=row, hidden=not visible)
 95        return ret
 96
 97    @property
 98    def display(self):
 99        """
100        Get the current display of the terminal emulator as a string.
101
102        Returns:
103            str: A string representation of the current display of the terminal emulator.
104        """
105        ret = "\n".join(self.vt.text_raw())
106        return ret
107
108    def screenshot(self, png_output_path: str):
109        """
110        Saves the current state of the terminal emulator to a PNG image file.
111
112        The image has the same width and height as the terminal emulator.
113
114        Args:
115            png_output_path: The path to write the PNG image to.
116        """
117        self.vt.screenshot(png_output_path)
118
119    def close(self):
120        """
121        Releases all resources used by the terminal emulator.
122
123        This is necessary to avoid crashes when creating multiple instances of this class.
124        """
125        if not self._closing:
126            self._closing = True
127            del self.vt
128
129    def __enter__(self):
130        return self
131
132    def __exit__(self, exc, value, tb):
133        self.close()

AvtScreen(width: int, height: int) View Source

60    def __init__(self, width: int, height: int):
61        """
62        Initialize an AvtScreen object.
63
64        Args:
65            width (int): The width of the virtual terminal emulator.
66            height (int): The height of the virtual terminal emulator.
67        """
68        self.vt = agg_python_bindings.TerminalEmulator(width, height)
69        """terminal emulator provided by avt"""
70        self._closing = False

Initialize an AvtScreen object.

Arguments:

width (int): The width of the virtual terminal emulator.
height (int): The height of the virtual terminal emulator.

terminal emulator provided by avt

def write_bytes(self, _bytes: bytes): View Source

72    def write_bytes(self, _bytes: bytes):
73        """
74        Write the given bytes to the virtual terminal emulator.
75
76        The bytes are decoded with UTF-8 and any decoding errors are replaced with a
77        replacement character (�).
78
79        Args:
80            _bytes (bytes): The bytes to be written to the virtual terminal emulator.
81        """
82        decoded_bytes = decode_bytes_to_utf8_safe(_bytes)
83        self.vt.feed_str(decoded_bytes)

Write the given bytes to the virtual terminal emulator.

The bytes are decoded with UTF-8 and any decoding errors are replaced with a replacement character (�).

Arguments:

_bytes (bytes): The bytes to be written to the virtual terminal emulator.

cursor View Source

85    @property
86    def cursor(self):
87        """
88        Get the current position of the cursor as a `Cursor` object.
89
90        Returns:
91            Cursor: A `Cursor` object with `x` and `y` properties for the current column and row of the cursor, respectively, and a `hidden` property which is `True` if the cursor is hidden and `False` otherwise.
92        """
93        col, row, visible = self.vt.get_cursor()
94        ret = Cursor(x=col, y=row, hidden=not visible)
95        return ret

Get the current position of the cursor as a Cursor object.

Returns:

Cursor: A Cursor object with x and y properties for the current column and row of the cursor, respectively, and a hidden property which is True if the cursor is hidden and False otherwise.

display View Source

 97    @property
 98    def display(self):
 99        """
100        Get the current display of the terminal emulator as a string.
101
102        Returns:
103            str: A string representation of the current display of the terminal emulator.
104        """
105        ret = "\n".join(self.vt.text_raw())
106        return ret

Get the current display of the terminal emulator as a string.

Returns:

str: A string representation of the current display of the terminal emulator.

def screenshot(self, png_output_path: str): View Source

108    def screenshot(self, png_output_path: str):
109        """
110        Saves the current state of the terminal emulator to a PNG image file.
111
112        The image has the same width and height as the terminal emulator.
113
114        Args:
115            png_output_path: The path to write the PNG image to.
116        """
117        self.vt.screenshot(png_output_path)

Saves the current state of the terminal emulator to a PNG image file.

The image has the same width and height as the terminal emulator.

Arguments:

png_output_path: The path to write the PNG image to.

def close(self): View Source

119    def close(self):
120        """
121        Releases all resources used by the terminal emulator.
122
123        This is necessary to avoid crashes when creating multiple instances of this class.
124        """
125        if not self._closing:
126            self._closing = True
127            del self.vt

Releases all resources used by the terminal emulator.

This is necessary to avoid crashes when creating multiple instances of this class.

class TerminalProcess: View Source

136class TerminalProcess:
137    def __init__(self, command: list[str], width: int, height: int, backend="avt"):
138        """
139        Initializes the terminal emulator with a command to execute.
140
141        Args:
142            command (list[str]): List of command strings to execute in the terminal
143            width (int): Width of the terminal emulator
144            height (int): Height of the terminal emulator
145            backend (str, optional): Backend to use for terminal emulator. Defaults to "avt".
146        """
147        self._closing = False
148        rows, cols = height, width
149        self.pty_process: ptyprocess.PtyProcess = cast(
150            ptyprocess.PtyProcess,
151            ptyprocess.PtyProcess.spawn(command, dimensions=(rows, cols)),
152        )
153        """a process executing command in a pseudo terminal"""
154
155        if backend == "avt":
156            self.vt_screen = AvtScreen(width=width, height=height)
157            """virtual terminal screen"""
158        else:
159            raise ValueError(
160                "Unknown terminal emulator backend '%s' (known ones: avt, pyte)"
161                % backend
162            )
163
164        self.vt_screen = cast(_TerminalScreenProtocol, self.vt_screen)
165
166        self.__pty_process_reading_thread = threading.Thread(
167            target=self.__read_and_update_screen, daemon=True
168        )
169        self.__start_ptyprocess_reading_thread()
170        atexit.register(self.close)
171
172    def __start_ptyprocess_reading_thread(self):
173        """Starts a thread to read output from the terminal process and update the Pyte screen"""
174        self.__pty_process_reading_thread.start()
175
176    def write(self, data: bytes):
177        """Writes input data to the terminal process"""
178        self.pty_process.write(data)
179
180    def close(self):
181        """Closes the terminal process and the reading thread"""
182        if not self._closing:
183            self._closing = True
184            os.kill(self.pty_process.pid, signal.SIGTERM)
185            time.sleep(0.5)
186            if self.pty_process.isalive:
187                os.kill(self.pty_process.pid, signal.SIGKILL)
188            self.vt_screen.close()
189            self.__pty_process_reading_thread.join(timeout=0.5)
190
191    def __enter__(self):
192        return self
193
194    def __exit__(self, exc, value, tb):
195        self.close()
196
197    def __read_and_update_screen(self, poll_interval=0.01):
198        """Reads available output from terminal and updates Pyte screen
199
200        Args:
201            poll_interval (float, optional): Interval in seconds to poll for available output. Defaults to 0.01.
202        """
203        while True:
204            try:
205                # ptyprocess.read is blocking. only pexpect has read_nonblocking
206                process_output_bytes = self.pty_process.read(1024)
207                # write bytes to pyte screen
208                self.vt_screen.write_bytes(process_output_bytes)
209            except KeyboardInterrupt:  # user interrupted
210                break
211            except SystemExit:  # python process exit
212                break
213            except SystemError:  # python error
214                break
215            except EOFError:  # terminal died
216                break
217            except:
218                # Timeout means no data available, EOF means process ended
219                pass
220            finally:
221                time.sleep(poll_interval)

TerminalProcess(command: list[str], width: int, height: int, backend='avt') View Source

137    def __init__(self, command: list[str], width: int, height: int, backend="avt"):
138        """
139        Initializes the terminal emulator with a command to execute.
140
141        Args:
142            command (list[str]): List of command strings to execute in the terminal
143            width (int): Width of the terminal emulator
144            height (int): Height of the terminal emulator
145            backend (str, optional): Backend to use for terminal emulator. Defaults to "avt".
146        """
147        self._closing = False
148        rows, cols = height, width
149        self.pty_process: ptyprocess.PtyProcess = cast(
150            ptyprocess.PtyProcess,
151            ptyprocess.PtyProcess.spawn(command, dimensions=(rows, cols)),
152        )
153        """a process executing command in a pseudo terminal"""
154
155        if backend == "avt":
156            self.vt_screen = AvtScreen(width=width, height=height)
157            """virtual terminal screen"""
158        else:
159            raise ValueError(
160                "Unknown terminal emulator backend '%s' (known ones: avt, pyte)"
161                % backend
162            )
163
164        self.vt_screen = cast(_TerminalScreenProtocol, self.vt_screen)
165
166        self.__pty_process_reading_thread = threading.Thread(
167            target=self.__read_and_update_screen, daemon=True
168        )
169        self.__start_ptyprocess_reading_thread()
170        atexit.register(self.close)

Initializes the terminal emulator with a command to execute.

Arguments:

command (list[str]): List of command strings to execute in the terminal
width (int): Width of the terminal emulator
height (int): Height of the terminal emulator
backend (str, optional): Backend to use for terminal emulator. Defaults to "avt".

pty_process: ptyprocess.ptyprocess.PtyProcess

a process executing command in a pseudo terminal

vt_screen

def write(self, data: bytes): View Source

176    def write(self, data: bytes):
177        """Writes input data to the terminal process"""
178        self.pty_process.write(data)

Writes input data to the terminal process

def close(self): View Source

180    def close(self):
181        """Closes the terminal process and the reading thread"""
182        if not self._closing:
183            self._closing = True
184            os.kill(self.pty_process.pid, signal.SIGTERM)
185            time.sleep(0.5)
186            if self.pty_process.isalive:
187                os.kill(self.pty_process.pid, signal.SIGKILL)
188            self.vt_screen.close()
189            self.__pty_process_reading_thread.join(timeout=0.5)

Closes the terminal process and the reading thread

class TerminalExecutor: View Source

224class TerminalExecutor:
225    def __init__(self, command: list[str], width: int, height: int):
226        """
227        Initializes executor with a command to run in terminal emulator, using avt as backend.
228
229        Args:
230            command (list[str]): List of command strings to execute
231            width (int): Width of the terminal emulator
232            height (int): Height of the terminal emulator
233        """
234        self.terminal = TerminalProcess(command=command, width=width, height=height)
235        """a terminal process, running command in pty screen"""
236        self._closing = False
237
238    def input(self, text: str):
239        """
240        Sends input text to the terminal process
241
242        Args:
243            text (str): The input text to send
244        """
245        self.terminal.write(text.encode())
246        # Allow time for processing output
247        time.sleep(0.1)
248
249    @property
250    def display(self) -> str:
251        """
252        Get the current display of the terminal emulator as a string.
253        """
254
255        return self.terminal.vt_screen.display
256
257    def screenshot(self, png_save_path: str):
258        """
259        Saves the current display of the terminal emulator as a .png file
260
261        Args:
262            png_save_path (str): The path to save the screenshot to
263        """
264        self.terminal.vt_screen.screenshot(png_save_path)
265
266    def close(self):
267        """
268        Closes the terminal emulator process and the associated reading thread.
269        """
270        if not self._closing:
271            self._closing = True
272            self.terminal.close()

TerminalExecutor(command: list[str], width: int, height: int) View Source

225    def __init__(self, command: list[str], width: int, height: int):
226        """
227        Initializes executor with a command to run in terminal emulator, using avt as backend.
228
229        Args:
230            command (list[str]): List of command strings to execute
231            width (int): Width of the terminal emulator
232            height (int): Height of the terminal emulator
233        """
234        self.terminal = TerminalProcess(command=command, width=width, height=height)
235        """a terminal process, running command in pty screen"""
236        self._closing = False

Initializes executor with a command to run in terminal emulator, using avt as backend.

Arguments:

command (list[str]): List of command strings to execute
width (int): Width of the terminal emulator
height (int): Height of the terminal emulator

terminal

a terminal process, running command in pty screen

def input(self, text: str): View Source

238    def input(self, text: str):
239        """
240        Sends input text to the terminal process
241
242        Args:
243            text (str): The input text to send
244        """
245        self.terminal.write(text.encode())
246        # Allow time for processing output
247        time.sleep(0.1)

Sends input text to the terminal process

Arguments:

text (str): The input text to send

display: str View Source

249    @property
250    def display(self) -> str:
251        """
252        Get the current display of the terminal emulator as a string.
253        """
254
255        return self.terminal.vt_screen.display

Get the current display of the terminal emulator as a string.

def screenshot(self, png_save_path: str): View Source

257    def screenshot(self, png_save_path: str):
258        """
259        Saves the current display of the terminal emulator as a .png file
260
261        Args:
262            png_save_path (str): The path to save the screenshot to
263        """
264        self.terminal.vt_screen.screenshot(png_save_path)

Saves the current display of the terminal emulator as a .png file

Arguments:

png_save_path (str): The path to save the screenshot to

def close(self): View Source

266    def close(self):
267        """
268        Closes the terminal emulator process and the associated reading thread.
269        """
270        if not self._closing:
271            self._closing = True
272            self.terminal.close()

Closes the terminal emulator process and the associated reading thread.

def test_harmless_command_locally_with_bash(): View Source

275def test_harmless_command_locally_with_bash():
276    """
277    Tests the TerminalExecutor class with a harmless command by running a docker alpine container
278    and executing a series of input events, then taking a screenshot and dumping the terminal
279    display to a file.
280    """
281    SLEEP_INTERVAL = 0.5
282    command = ["docker", "run", "--rm", "-it", "alpine"]
283    input_events = ['echo "Hello World!"', "\n"]
284    executor = TerminalExecutor(command=command, width=80, height=24)
285    time.sleep(1)
286    for event in input_events:
287        executor.input(event)
288        time.sleep(SLEEP_INTERVAL)
289    # check for screenshot, text dump
290    text_dump = executor.display
291    print("Dumping terminal display to ./terminal_executor_text_dump.txt")
292    with open("./terminal_executor_text_dump.txt", "w+") as f:
293        f.write(text_dump)
294    print("Taking terminal screenshot at ./terminal_executor_screenshot.png")
295    executor.screenshot("./terminal_executor_screenshot.png")
296    print("Done")

Tests the TerminalExecutor class with a harmless command by running a docker alpine container and executing a series of input events, then taking a screenshot and dumping the terminal display to a file.

def test(): View Source

299def test():
300    """
301    Runs a test for the TerminalExecutor class by running a harmless command
302    locally with bash and taking a screenshot and dumping the terminal display
303    to a file.
304
305    This test is useful for checking that the TerminalExecutor class works in
306    a real environment.
307    """
308    test_harmless_command_locally_with_bash()

Runs a test for the TerminalExecutor class by running a harmless command locally with bash and taking a screenshot and dumping the terminal display to a file.

This test is useful for checking that the TerminalExecutor class works in a real environment.