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 typing import Optional, Callable 

11import logging 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class TerminalLauncher: 

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

18 

19 def __init__(self, app): 

20 """ 

21 Initialize terminal launcher. 

22  

23 Args: 

24 app: The TUI application instance 

25 """ 

26 self.app = app 

27 

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

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

30 """ 

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

32  

33 Args: 

34 file_content: Initial content to edit 

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

36 on_save_callback: Callback function called with edited content when saved 

37 """ 

38 try: 

39 # Create temporary file with content 

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

41 f.write(file_content) 

42 temp_path = f.name 

43 

44 # Get editor from environment 

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

46 

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

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

49 

50 # Launch terminal window with the command 

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

52 

53 except Exception as e: 

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

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

56 

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

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

59 """ 

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

61  

62 Args: 

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

64 file_path: Path to temporary file 

65 on_save_callback: Callback for when file is saved 

66  

67 Returns: 

68 Shell command string 

69 """ 

70 # Create a wrapper script that: 

71 # 1. Runs the editor 

72 # 2. Reads the file content after editing 

73 # 3. Calls the callback with the content 

74 # 4. Cleans up the temp file 

75 

76 callback_script = self._create_callback_script(file_path, on_save_callback) 

77 

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

79 command = f""" 

80# Source user's shell configuration 

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

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

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

84 

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

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

87echo "" 

88{editor} "{file_path}" 

89echo "" 

90echo "Editor closed. Processing changes..." 

91python3 "{callback_script}" 

92echo "Terminal will close automatically." 

93exit 0 

94""" 

95 return command.strip() 

96 

97 def _create_callback_script(self, file_path: str, 

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

99 """ 

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

101 

102 Args: 

103 file_path: Path to the edited file 

104 on_save_callback: Callback function 

105 

106 Returns: 

107 Path to callback script 

108 """ 

109 if on_save_callback: 

110 # Store callback and create signal file approach 

111 callback_id = id(on_save_callback) 

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

113 self.app._terminal_callbacks[callback_id] = on_save_callback 

114 

115 # Create signal file path 

116 signal_file = f"{file_path}.done" 

117 

118 script_content = f""" 

119# Simple completion signal - no OpenHCS imports needed 

120import os 

121 

122try: 

123 print("Editor session completed.") 

124 

125 # Create signal file to notify main process 

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

127 f.write("{callback_id}") 

128 

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

130 

131except Exception as e: 

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

133""" 

134 

135 # Start polling for the signal file in main process 

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

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

138 

139 else: 

140 # No callback, just clean up 

141 script_content = f""" 

142import os 

143try: 

144 os.unlink("{file_path}") 

145 print("Temporary file cleaned up.") 

146except: 

147 pass 

148""" 

149 

150 # Write callback script to temp file 

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

152 f.write(script_content) 

153 return f.name 

154 

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

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

157 import asyncio 

158 

159 async def poll_for_completion(): 

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

161 while True: 

162 try: 

163 if os.path.exists(signal_file): 

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

165 # Signal file exists, read the edited content 

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

167 content = f.read() 

168 

169 # Get and call the callback 

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

171 callback = callbacks.get(callback_id) 

172 if callback: 

173 logger.info("Calling editor callback") 

174 callback(content) 

175 else: 

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

177 

178 # Close the terminal window if provided 

179 if terminal_window: 

180 try: 

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

182 terminal_window.close_window() 

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

184 except Exception as e: 

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

186 else: 

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

188 

189 # Clean up 

190 try: 

191 os.unlink(file_path) 

192 os.unlink(signal_file) 

193 callbacks.pop(callback_id, None) 

194 except: 

195 pass 

196 

197 break 

198 

199 # Wait before checking again 

200 await asyncio.sleep(0.5) 

201 

202 except Exception as e: 

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

204 break 

205 

206 # Start the polling task 

207 asyncio.create_task(poll_for_completion()) 

208 

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

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

211 import os 

212 import tempfile 

213 import stat 

214 

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

216 user_shell = self._get_user_shell() 

217 

218 # Create script that sources user configs and runs command 

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

220# Run in login shell to load user environment 

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

222''' 

223 

224 # Write to temporary file 

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

226 f.write(script_content) 

227 script_path = f.name 

228 

229 # Make executable 

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

231 

232 return script_path 

233 

234 def _get_user_shell(self) -> str: 

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

236 import os 

237 

238 # Method 1: Check SHELL environment variable 

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

240 return os.environ['SHELL'] 

241 

242 # Method 2: Check /etc/passwd 

243 try: 

244 import pwd 

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

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

247 return user_shell 

248 except (ImportError, KeyError, OSError): 

249 pass 

250 

251 # Method 3: Try common shells 

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

253 for shell_path in common_shells: 

254 if os.path.exists(shell_path): 

255 return shell_path 

256 

257 # Fallback 

258 return '/bin/bash' 

259 

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

261 """ 

262 Launch a terminal window with a specific command. 

263  

264 Args: 

265 command: Shell command to run 

266 title: Window title 

267 """ 

268 from openhcs.textual_tui.windows.terminal_window import TerminalWindow 

269 from textual.css.query import NoMatches 

270 

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

272 shell_command = self._create_login_shell_wrapper(command) 

273 

274 try: 

275 # Try to find existing terminal window 

276 window = self.app.query_one(TerminalWindow) 

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

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

279 window = TerminalWindow(shell_command=shell_command) 

280 await self.app.mount(window) 

281 window.open_state = True 

282 

283 except NoMatches: 

284 # No existing terminal, create new one 

285 window = TerminalWindow(shell_command=shell_command) 

286 await self.app.mount(window) 

287 window.open_state = True 

288 

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

290 if hasattr(self, '_pending_callback'): 

291 file_path, signal_file, callback_id = self._pending_callback 

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

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

294 delattr(self, '_pending_callback')