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