Coverage for openhcs/textual_tui/windows/advanced_terminal_window.py: 0.0%
187 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Advanced Terminal Window for OpenHCS Textual TUI
4Uses the Gate One terminal emulator for advanced terminal functionality.
5This supersedes the basic terminal window with full VT-* emulation support.
6"""
8import logging
9import os
10import pty
11import subprocess
12import threading
13from typing import Optional
15from textual.app import ComposeResult
16from textual.reactive import reactive
17from textual.widget import Widget
18from textual.widgets import Static
20from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow
22# Import the Gate One terminal emulator and callback constants
23from openhcs.textual_tui.services.terminal import (
24 Terminal as GateOneTerminal,
25 CALLBACK_CHANGED,
26 CALLBACK_CURSOR_POS,
27 CALLBACK_TITLE,
28 CALLBACK_BELL
29)
31logger = logging.getLogger(__name__)
34class AdvancedTerminalWidget(Widget):
35 """
36 Advanced terminal widget using Gate One terminal emulator.
38 Provides full VT-* terminal emulation with:
39 - Complete ECMA-48/ANSI X3.64 support
40 - VT-52, VT-100, VT-220, VT-320, VT-420, VT-520 emulation
41 - Linux console emulation
42 - Image support (with PIL)
43 - Scrollback buffer
44 - HTML output capabilities
45 """
47 DEFAULT_CSS = """
48 AdvancedTerminalWidget {
49 width: 100%;
50 height: 100%;
51 }
52 """
54 # Reactive properties
55 cursor_visible = reactive(True)
56 terminal_title = reactive("Advanced Terminal")
58 def __init__(self, command: Optional[str] = None, rows: int = 24, cols: int = 80, **kwargs):
59 """
60 Initialize the advanced terminal widget.
62 Args:
63 command: Shell command to run (defaults to current shell)
64 rows: Terminal height in rows
65 cols: Terminal width in columns
66 """
67 super().__init__(**kwargs)
69 self.command = command or self._get_default_shell()
70 self.rows = rows
71 self.cols = cols
73 # Initialize Gate One terminal
74 self.terminal = GateOneTerminal(rows=rows, cols=cols)
76 # Setup terminal callbacks
77 self._setup_terminal_callbacks()
79 # Process management
80 self.process = None
81 self.master_fd = None
82 self.reader_thread = None
83 self.running = False
85 logger.info(f"AdvancedTerminalWidget initialized: {rows}x{cols}, command: {self.command}")
87 def _get_default_shell(self) -> str:
88 """Get the default shell for the current user."""
89 # Try to get user's shell from environment or passwd
90 shell = os.environ.get('SHELL')
91 if shell and os.path.exists(shell):
92 return shell
94 # Fallback to common shells
95 for shell_path in ['/bin/bash', '/bin/zsh', '/bin/sh']:
96 if os.path.exists(shell_path):
97 return shell_path
99 return '/bin/sh' # Ultimate fallback
101 def _setup_terminal_callbacks(self):
102 """Setup callbacks for the Gate One terminal."""
103 # Callback when screen changes - trigger refresh
104 def on_screen_changed():
105 self.call_after_refresh(self._update_display)
107 # Callback when cursor position changes
108 def on_cursor_changed():
109 self.call_after_refresh(self._update_cursor)
111 # Callback when title changes
112 def on_title_changed():
113 title = getattr(self.terminal, 'title', 'Advanced Terminal')
114 self.terminal_title = title
116 # Callback for bell
117 def on_bell():
118 # Could implement visual bell here
119 logger.debug("Terminal bell")
121 # Set up the callbacks using the proper add_callback method
122 self.terminal.add_callback(CALLBACK_CHANGED, on_screen_changed, identifier="widget_changed")
123 self.terminal.add_callback(CALLBACK_CURSOR_POS, on_cursor_changed, identifier="widget_cursor")
124 self.terminal.add_callback(CALLBACK_TITLE, on_title_changed, identifier="widget_title")
125 self.terminal.add_callback(CALLBACK_BELL, on_bell, identifier="widget_bell")
127 def compose(self) -> ComposeResult:
128 """Compose the terminal widget layout."""
129 # Just the terminal screen display, no control buttons
130 yield Static("", id="terminal_screen")
132 def on_mount(self) -> None:
133 """Start the terminal when widget is mounted."""
134 logger.info("AdvancedTerminalWidget mounting - starting terminal process")
135 self.start_terminal()
137 def on_unmount(self) -> None:
138 """Clean up when widget is unmounted."""
139 logger.info("AdvancedTerminalWidget unmounting - stopping terminal process")
140 self.stop_terminal()
142 def start_terminal(self):
143 """Start the terminal process and reader thread."""
144 try:
145 # Create a pseudo-terminal
146 self.master_fd, slave_fd = pty.openpty()
148 # Start the shell process
149 self.process = subprocess.Popen(
150 self.command,
151 stdin=slave_fd,
152 stdout=slave_fd,
153 stderr=slave_fd,
154 shell=True,
155 preexec_fn=os.setsid # Create new session
156 )
158 # Close slave fd in parent process
159 os.close(slave_fd)
161 # Start reader thread
162 self.running = True
163 self.reader_thread = threading.Thread(target=self._read_from_terminal, daemon=True)
164 self.reader_thread.start()
166 logger.info(f"Terminal process started: PID {self.process.pid}")
168 except Exception as e:
169 logger.error(f"Failed to start terminal process: {e}")
170 self._show_error(f"Failed to start terminal: {e}")
172 def stop_terminal(self):
173 """Stop the terminal process and cleanup."""
174 self.running = False
176 if self.process:
177 try:
178 # Terminate the process group
179 os.killpg(os.getpgid(self.process.pid), 15) # SIGTERM
180 self.process.wait(timeout=2)
181 except (ProcessLookupError, subprocess.TimeoutExpired):
182 try:
183 # Force kill if needed
184 os.killpg(os.getpgid(self.process.pid), 9) # SIGKILL
185 except ProcessLookupError:
186 pass
187 self.process = None
189 if self.master_fd:
190 try:
191 os.close(self.master_fd)
192 except OSError:
193 pass
194 self.master_fd = None
196 if self.reader_thread and self.reader_thread.is_alive():
197 self.reader_thread.join(timeout=1)
199 def _read_from_terminal(self):
200 """Read output from terminal in background thread."""
201 while self.running and self.master_fd:
202 try:
203 # Read from terminal with timeout
204 import select
205 ready, _, _ = select.select([self.master_fd], [], [], 0.1)
207 if ready:
208 data = os.read(self.master_fd, 1024)
209 if data:
210 # Decode and write to Gate One terminal
211 text = data.decode('utf-8', errors='replace')
212 self.terminal.write(text)
213 else:
214 # EOF - process ended
215 break
217 except (OSError, ValueError) as e:
218 if self.running:
219 logger.error(f"Error reading from terminal: {e}")
220 break
222 logger.info("Terminal reader thread stopped")
224 def _update_display(self):
225 """Update the terminal display with current screen content."""
226 try:
227 # Get screen content from Gate One terminal
228 screen_lines = self.terminal.dump()
230 # Join lines and update display
231 screen_content = '\n'.join(screen_lines)
233 # Update the display widget
234 screen_widget = self.query_one("#terminal_screen", Static)
235 screen_widget.update(screen_content)
237 except Exception as e:
238 logger.error(f"Error updating terminal display: {e}")
240 def _update_cursor(self):
241 """Update cursor position display."""
242 # Could implement cursor position indicator here
243 pass
245 def _show_error(self, message: str):
246 """Show error message in terminal display."""
247 try:
248 screen_widget = self.query_one("#terminal_screen", Static)
249 screen_widget.update(f"[red]Error: {message}[/red]")
250 except Exception:
251 pass
255 async def clear_terminal(self):
256 """Clear the terminal screen."""
257 if self.master_fd:
258 try:
259 # Send clear command
260 os.write(self.master_fd, b'\x0c') # Form feed (clear)
261 except OSError as e:
262 logger.error(f"Error clearing terminal: {e}")
264 async def reset_terminal(self):
265 """Reset the terminal."""
266 if self.master_fd:
267 try:
268 # Send reset sequence
269 os.write(self.master_fd, b'\x1bc') # ESC c (reset)
270 except OSError as e:
271 logger.error(f"Error resetting terminal: {e}")
273 async def send_input(self, text: str):
274 """Send input to the terminal."""
275 if self.master_fd:
276 try:
277 os.write(self.master_fd, text.encode('utf-8'))
278 except OSError as e:
279 logger.error(f"Error sending input to terminal: {e}")
281 async def on_key(self, event) -> None:
282 """Handle key input and send to terminal."""
283 if self.master_fd:
284 try:
285 # Convert key to appropriate bytes
286 key_bytes = self._key_to_bytes(event.key)
287 if key_bytes:
288 os.write(self.master_fd, key_bytes)
289 except OSError as e:
290 logger.error(f"Error sending key to terminal: {e}")
292 def _key_to_bytes(self, key: str) -> bytes:
293 """Convert Textual key to terminal bytes."""
294 # Handle special keys
295 key_map = {
296 'enter': b'\r',
297 'escape': b'\x1b',
298 'backspace': b'\x7f',
299 'tab': b'\t',
300 'up': b'\x1b[A',
301 'down': b'\x1b[B',
302 'right': b'\x1b[C',
303 'left': b'\x1b[D',
304 'home': b'\x1b[H',
305 'end': b'\x1b[F',
306 'page_up': b'\x1b[5~',
307 'page_down': b'\x1b[6~',
308 'delete': b'\x1b[3~',
309 'insert': b'\x1b[2~',
310 }
312 # Handle ctrl combinations
313 if key.startswith('ctrl+'):
314 char = key[5:]
315 if len(char) == 1 and 'a' <= char <= 'z':
316 return bytes([ord(char) - ord('a') + 1])
318 # Handle regular keys
319 if key in key_map:
320 return key_map[key]
321 elif len(key) == 1:
322 return key.encode('utf-8')
324 return b''
327class AdvancedTerminalWindow(BaseOpenHCSWindow):
328 """
329 Advanced Terminal Window using Gate One terminal emulator.
331 This supersedes the basic TerminalWindow with advanced features:
332 - Full VT-* terminal emulation
333 - Complete ANSI escape sequence support
334 - Image display capabilities
335 - Advanced scrollback buffer
336 - HTML output support
337 """
339 DEFAULT_CSS = """
340 AdvancedTerminalWindow {
341 width: 100; height: 30;
342 min-width: 80; min-height: 24;
343 }
344 """
346 def __init__(self, shell_command: str = None, **kwargs):
347 """
348 Initialize advanced terminal window.
350 Args:
351 shell_command: Optional command to run (defaults to current shell)
352 """
353 # Get shell before calling super() so we can use it in title
354 self.shell_command = shell_command or self._get_current_shell()
356 # Extract shell name for title
357 shell_name = os.path.basename(self.shell_command)
359 logger.info(f"AdvancedTerminal: Initializing with shell: {self.shell_command}")
361 super().__init__(
362 window_id="advanced_terminal",
363 title=f"Advanced Terminal ({shell_name})",
364 mode="temporary",
365 **kwargs
366 )
368 def _get_current_shell(self) -> str:
369 """Get the current shell from environment."""
370 return os.environ.get('SHELL', '/bin/bash')
372 def compose(self) -> ComposeResult:
373 """Compose the advanced terminal window content."""
374 yield AdvancedTerminalWidget(
375 command=self.shell_command,
376 rows=24,
377 cols=80,
378 id="advanced_terminal"
379 )
381 async def on_key(self, event) -> None:
382 """Handle key presses for terminal shortcuts."""
383 # Ctrl+Shift+C to copy (if we implement clipboard)
384 if event.key == "ctrl+shift+c":
385 # Could implement copy functionality here
386 pass
387 # Ctrl+Shift+V to paste (if we implement clipboard)
388 elif event.key == "ctrl+shift+v":
389 # Could implement paste functionality here
390 pass
391 # Ctrl+Shift+T to open new terminal
392 elif event.key == "ctrl+shift+t":
393 # Could open new terminal window
394 pass
395 else:
396 # Forward all other keys to the terminal widget
397 terminal = self.query_one("#advanced_terminal", AdvancedTerminalWidget)
398 await terminal.on_key(event)
400 async def send_command(self, command: str):
401 """Send a command to the terminal."""
402 terminal = self.query_one("#advanced_terminal", AdvancedTerminalWidget)
403 await terminal.send_input(command + '\n')
405 def on_mount(self) -> None:
406 """Called when advanced terminal window is mounted."""
407 terminal = self.query_one("#advanced_terminal", AdvancedTerminalWidget)
408 terminal.focus()
409 logger.info("AdvancedTerminalWindow mounted and focused")