Coverage for openhcs/textual_tui/services/terminal_launcher.py: 0.0%

1""" 

2Terminal launcher service for running commands in TUI terminal windows. 

3 

4Provides a clean interface for launching terminal applications within the existing 

5TUI terminal infrastructure instead of external processes. 

6""" 

7 

8import tempfile 

9import os 

10from pathlib import Path 

11from typing import Optional, Callable 

12import logging 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class TerminalLauncher: 

18 """Service for launching terminal applications within TUI terminal windows.""" 

19 

20 def __init__(self, app): 

21 """ 

22 Initialize terminal launcher. 

23  

24 Args: 

25 app: The TUI application instance 

26 """ 

27 self.app = app 

28 

29 async def launch_editor_for_file(self, file_content: str, file_extension: str = '.py', 

30 on_save_callback: Optional[Callable[[str], None]] = None) -> None: 

31 """ 

32 Launch an editor in a terminal window for editing file content. 

33  

34 Args: 

35 file_content: Initial content to edit 

36 file_extension: File extension (e.g., '.py', '.txt') 

37 on_save_callback: Callback function called with edited content when saved 

38 """ 

39 try: 

40 # Create temporary file with content 

41 with tempfile.NamedTemporaryFile(mode='w', suffix=file_extension, delete=False) as f: 

42 f.write(file_content) 

43 temp_path = f.name 

44 

45 # Get editor from environment 

46 editor = os.environ.get('EDITOR', 'nano') 

47 

48 # Create command that will edit the file and then read it back 

49 command = self._create_editor_command(editor, temp_path, on_save_callback) 

50 

51 # Launch terminal window with the command 

52 await self._launch_terminal_with_command(command, f"Edit File ({editor})") 

53 

54 except Exception as e: 

55 logger.error(f"Failed to launch editor: {e}") 

56 self.app.show_error("Editor Error", f"Failed to launch editor: {str(e)}") 

57 

58 def _create_editor_command(self, editor: str, file_path: str, 

59 on_save_callback: Optional[Callable[[str], None]]) -> str: 

60 """ 

61 Create a shell command that runs the editor and handles the callback. 

62  

63 Args: 

64 editor: Editor command (e.g., 'vim', 'nano') 

65 file_path: Path to temporary file 

66 on_save_callback: Callback for when file is saved 

67  

68 Returns: 

69 Shell command string 

70 """ 

71 # Create a wrapper script that: 

72 # 1. Runs the editor 

73 # 2. Reads the file content after editing 

74 # 3. Calls the callback with the content 

75 # 4. Cleans up the temp file 

76 

77 callback_script = self._create_callback_script(file_path, on_save_callback) 

78 

79 # Command that runs editor in a proper login shell environment 

80 command = f""" 

81# Source user's shell configuration 

82if [ -f ~/.bashrc ]; then source ~/.bashrc; fi 

83if [ -f ~/.zshrc ]; then source ~/.zshrc; fi 

84if [ -f ~/.profile ]; then source ~/.profile; fi 

85 

86echo "Opening {editor} for editing..." 

87echo "Save and exit to apply changes, or exit without saving to cancel." 

88echo "" 

89{editor} "{file_path}" 

90echo "" 

91echo "Editor closed. Processing changes..." 

92python3 "{callback_script}" 

93echo "Terminal will close automatically." 

94exit 0 

95""" 

96 return command.strip() 

97 

98 def _create_callback_script(self, file_path: str, 

99 on_save_callback: Optional[Callable[[str], None]]) -> str: 

100 """ 

101 Create a simple script that signals completion without importing OpenHCS. 

102 

103 Args: 

104 file_path: Path to the edited file 

105 on_save_callback: Callback function 

106 

107 Returns: 

108 Path to callback script 

109 """ 

110 if on_save_callback: 

111 # Store callback and create signal file approach 

112 callback_id = id(on_save_callback) 

113 self.app._terminal_callbacks = getattr(self.app, '_terminal_callbacks', {}) 

114 self.app._terminal_callbacks[callback_id] = on_save_callback 

115 

116 # Create signal file path 

117 signal_file = f"{file_path}.done" 

118 

119 script_content = f""" 

120# Simple completion signal - no OpenHCS imports needed 

121import os 

122 

123try: 

124 print("Editor session completed.") 

125 

126 # Create signal file to notify main process 

127 with open("{signal_file}", 'w') as f: 

128 f.write("{callback_id}") 

129 

130 print("Changes will be processed by main application.") 

131 

132except Exception as e: 

133 print(f"Error creating signal file: {{e}}") 

134""" 

135 

136 # Start polling for the signal file in main process 

137 # We'll pass the terminal window reference when we create it 

138 self._pending_callback = (file_path, signal_file, callback_id) 

139 

140 else: 

141 # No callback, just clean up 

142 script_content = f""" 

143import os 

144try: 

145 os.unlink("{file_path}") 

146 print("Temporary file cleaned up.") 

147except: 

148 pass 

149""" 

150 

151 # Write callback script to temp file 

152 with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: 

153 f.write(script_content) 

154 return f.name 

155 

156 def _start_polling(self, file_path: str, signal_file: str, callback_id: int, terminal_window=None) -> None: 

157 """Start polling for completion signal file.""" 

158 import asyncio 

159 

160 async def poll_for_completion(): 

161 """Poll for signal file and handle callback.""" 

162 while True: 

163 try: 

164 if os.path.exists(signal_file): 

165 logger.info(f"Signal file detected: {signal_file}") 

166 # Signal file exists, read the edited content 

167 with open(file_path, 'r') as f: 

168 content = f.read() 

169 

170 # Get and call the callback 

171 callbacks = getattr(self.app, '_terminal_callbacks', {}) 

172 callback = callbacks.get(callback_id) 

173 if callback: 

174 logger.info("Calling editor callback") 

175 callback(content) 

176 else: 

177 logger.warning(f"No callback found for ID: {callback_id}") 

178 

179 # Close the terminal window if provided 

180 if terminal_window: 

181 try: 

182 logger.info(f"Closing terminal window: {terminal_window}") 

183 terminal_window.close_window() 

184 logger.info("Terminal window closed successfully") 

185 except Exception as e: 

186 logger.error(f"Error closing terminal window: {e}") 

187 else: 

188 logger.warning("No terminal window reference found for cleanup") 

189 

190 # Clean up 

191 try: 

192 os.unlink(file_path) 

193 os.unlink(signal_file) 

194 callbacks.pop(callback_id, None) 

195 except: 

196 pass 

197 

198 break 

199 

200 # Wait before checking again 

201 await asyncio.sleep(0.5) 

202 

203 except Exception as e: 

204 logger.error(f"Error in polling: {e}") 

205 break 

206 

207 # Start the polling task 

208 asyncio.create_task(poll_for_completion()) 

209 

210 def _create_login_shell_wrapper(self, command: str) -> str: 

211 """Create a wrapper script that runs command in a login shell environment.""" 

212 import os 

213 import tempfile 

214 import stat 

215 

216 # Get user's shell (same logic as TerminalWindow) 

217 user_shell = self._get_user_shell() 

218 

219 # Create script that sources user configs and runs command 

220 script_content = f'''#!/bin/bash 

221# Run in login shell to load user environment 

222exec {user_shell} -l -c "{command.replace('"', '\\"')}" 

223''' 

224 

225 # Write to temporary file 

226 with tempfile.NamedTemporaryFile(mode='w', suffix='.sh', delete=False) as f: 

227 f.write(script_content) 

228 script_path = f.name 

229 

230 # Make executable 

231 os.chmod(script_path, stat.S_IRWXU | stat.S_IRGRP | stat.S_IROTH) 

232 

233 return script_path 

234 

235 def _get_user_shell(self) -> str: 

236 """Get user's preferred shell (same logic as TerminalWindow).""" 

237 import os 

238 

239 # Method 1: Check SHELL environment variable 

240 if 'SHELL' in os.environ and os.path.exists(os.environ['SHELL']): 

241 return os.environ['SHELL'] 

242 

243 # Method 2: Check /etc/passwd 

244 try: 

245 import pwd 

246 user_shell = pwd.getpwuid(os.getuid()).pw_shell 

247 if user_shell and os.path.exists(user_shell): 

248 return user_shell 

249 except (ImportError, KeyError, OSError): 

250 pass 

251 

252 # Method 3: Try common shells 

253 common_shells = ['/bin/zsh', '/usr/bin/zsh', '/bin/bash', '/usr/bin/bash'] 

254 for shell_path in common_shells: 

255 if os.path.exists(shell_path): 

256 return shell_path 

257 

258 # Fallback 

259 return '/bin/bash' 

260 

261 async def _launch_terminal_with_command(self, command: str, title: str = "Terminal") -> None: 

262 """ 

263 Launch a terminal window with a specific command. 

264  

265 Args: 

266 command: Shell command to run 

267 title: Window title 

268 """ 

269 from openhcs.textual_tui.windows.terminal_window import TerminalWindow 

270 from textual.css.query import NoMatches 

271 

272 # Create wrapper command that runs our command in a login shell (like regular terminal) 

273 shell_command = self._create_login_shell_wrapper(command) 

274 

275 try: 

276 # Try to find existing terminal window 

277 window = self.app.query_one(TerminalWindow) 

278 # If terminal exists, we could either reuse it or create a new one 

279 # For now, let's create a new one for the editor 

280 window = TerminalWindow(shell_command=shell_command) 

281 await self.app.mount(window) 

282 window.open_state = True 

283 

284 except NoMatches: 

285 # No existing terminal, create new one 

286 window = TerminalWindow(shell_command=shell_command) 

287 await self.app.mount(window) 

288 window.open_state = True 

289 

290 # Start polling with the terminal window reference if we have a pending callback 

291 if hasattr(self, '_pending_callback'): 

292 file_path, signal_file, callback_id = self._pending_callback 

293 logger.info(f"Starting polling with terminal window reference: {window}") 

294 self._start_polling(file_path, signal_file, callback_id, window) 

295 delattr(self, '_pending_callback')