Coverage for openhcs/textual_tui/windows/toolong_window.py: 0.0%

1""" 

2Toolong Log Viewer Window for OpenHCS TUI 

3 

4A window that displays OpenHCS logs using the reactive log monitoring system. 

5Provides professional log viewing with syntax highlighting, search, and live tailing. 

6""" 

7 

8import logging 

9from pathlib import Path 

10from typing import List 

11 

12from textual.app import ComposeResult 

13from textual.widgets import Static 

14 

15# Import OpenHCS window base class 

16from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow 

17 

18# Import our OpenHCS Toolong widget 

19from openhcs.textual_tui.widgets.openhcs_toolong_widget import OpenHCSToolongWidget 

20 

21 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class ToolongWindow(BaseOpenHCSWindow): 

27 """ 

28 Window that displays OpenHCS logs using Toolong. 

29 

30 Features: 

31 - Professional log viewing with syntax highlighting 

32 - Live tailing of active log files 

33 - Search and filtering capabilities 

34 - Multi-file support with tabs 

35 - Merge view for combined log analysis 

36 """ 

37 

38 

39 

40 DEFAULT_CSS = """ 

41 ToolongWindow { 

42 width: 80; 

43 height: 25; 

44 min-width: 60; 

45 min-height: 15; 

46 } 

47 

48 .error-message { 

49 color: $error; 

50 text-style: bold; 

51 text-align: center; 

52 padding: 2; 

53 } 

54 """ 

55 

56 def __init__(self, base_log_path: str = "", **kwargs): 

57 """ 

58 Initialize the Toolong window. 

59 

60 Args: 

61 base_log_path: Base path for subprocess log monitoring (optional) 

62 """ 

63 super().__init__( 

64 window_id="toolong_viewer", 

65 title="Log Viewer", 

66 mode="permanent", # Make it permanent so it can't be closed, only minimized 

67 allow_maximize=True, 

68 **kwargs 

69 ) 

70 

71 # Find current TUI log file 

72 self.tui_log_file = self._find_current_tui_log() 

73 

74 # Store base_log_path for reference 

75 self.base_log_path = base_log_path 

76 

77 # Determine logs directory for file watching 

78 if base_log_path: 

79 # Use directory of base_log_path for watching subprocess logs 

80 self.logs_directory = Path(base_log_path).parent 

81 elif self.tui_log_file: 

82 # Fall back to TUI log directory 

83 self.logs_directory = Path(self.tui_log_file).parent 

84 else: 

85 self.logs_directory = None 

86 

87 def _find_current_tui_log(self) -> str: 

88 """Find the current TUI process log file (not subprocess or worker logs).""" 

89 import glob 

90 

91 # Look for TUI log files (exclude subprocess and worker logs) 

92 # Use cross-platform path resolution instead of hardcoded path 

93 log_dir = Path.home() / ".local" / "share" / "openhcs" / "logs" 

94 log_pattern = str(log_dir / "openhcs_unified_*.log") 

95 all_logs = sorted(glob.glob(log_pattern)) 

96 

97 # Filter to only actual TUI logs (not subprocess or worker logs) 

98 tui_logs = [] 

99 for log_file in all_logs: 

100 log_name = Path(log_file).name 

101 # TUI logs don't contain "subprocess" or "worker" in the name 

102 if "_subprocess_" not in log_name and "_worker_" not in log_name: 

103 tui_logs.append(log_file) 

104 

105 if tui_logs: 

106 # Return the most recent TUI log 

107 logger.debug(f"Found {len(tui_logs)} TUI logs, using most recent: {Path(tui_logs[-1]).name}") 

108 return tui_logs[-1] 

109 else: 

110 logger.warning("No TUI log files found") 

111 return None 

112 

113 def _find_session_logs(self) -> List[str]: 

114 """Find all logs belonging to the current TUI session.""" 

115 import glob 

116 

117 if not self.tui_log_file: 

118 return [] 

119 

120 # Extract session base from TUI log 

121 # openhcs_unified_20250630_092636.log -> openhcs_unified_20250630_092636 

122 tui_base = Path(self.tui_log_file).stem 

123 

124 # Find all logs that start with this session base 

125 log_pattern = str(self.logs_directory / f"{tui_base}*.log") 

126 session_logs = sorted(glob.glob(log_pattern)) 

127 

128 logger.debug(f"Found {len(session_logs)} logs for session '{tui_base}':") 

129 for log_file in session_logs: 

130 logger.debug(f" - {Path(log_file).name}") 

131 

132 return session_logs 

133 

134 def _find_all_openhcs_logs(self) -> List[str]: 

135 """Find all existing OpenHCS log files.""" 

136 import glob 

137 

138 if not self.logs_directory: 

139 return [] 

140 

141 # Look for all OpenHCS log files 

142 log_pattern = str(self.logs_directory / "openhcs_*.log") 

143 all_logs = sorted(glob.glob(log_pattern)) 

144 

145 logger.debug(f"Found {len(all_logs)} existing OpenHCS log files") 

146 for log_file in all_logs: 

147 logger.debug(f" - {Path(log_file).name}") 

148 

149 return all_logs 

150 

151 def compose(self) -> ComposeResult: 

152 """Compose the Toolong window layout using OpenHCSToolongWidget.""" 

153 try: 

154 # Find all logs for current session 

155 session_logs = self._find_session_logs() 

156 

157 if session_logs: 

158 # Start with all existing session logs, watch for new ones 

159 yield OpenHCSToolongWidget( 

160 file_paths=session_logs, # All session logs 

161 show_tabs=False, # Hide tabs, use dropdown only 

162 show_dropdown=True, # Show dropdown selector 

163 show_controls=True, # Show control buttons 

164 base_log_path=str(self.logs_directory) if self.logs_directory else None 

165 ) 

166 else: 

167 yield Static("No session log files found", classes="error-message") 

168 

169 except Exception as e: 

170 logger.error(f"Failed to create OpenHCSToolongWidget: {e}") 

171 yield Static( 

172 f"Error loading log viewer: {e}\n\n" + 

173 "Please check that log files exist and are readable.", 

174 classes="error-message" 

175 ) 

176 

177 async def on_mount(self) -> None: 

178 """Set up the window when mounted.""" 

179 # ReactiveLogMonitor handles its own setup 

180 logger.debug(f"Toolong window opened with base path: {self.base_log_path}") 

181 

182 

183 

184 async def on_unmount(self) -> None: 

185 """Clean up when window is unmounted.""" 

186 logger.debug("Toolong window unmounting, ensuring cleanup...") 

187 

188 # Explicitly stop ReactiveLogMonitor to ensure thread cleanup 

189 try: 

190 reactive_monitor = self.query_one(ReactiveLogMonitor) 

191 if reactive_monitor: 

192 reactive_monitor.stop_monitoring() 

193 except Exception as e: 

194 logger.debug(f"Could not find ReactiveLogMonitor to stop: {e}") 

195 

196 logger.debug("Toolong window closed") 

197 

198 

199def clear_toolong_logs(app): 

200 """Clear subprocess logs from the singleton toolong window (mounted at startup).""" 

201 try: 

202 # Find the singleton toolong window (should always exist) 

203 window = app.query_one(ToolongWindow) 

204 logger.info("Found singleton toolong window") 

205 

206 # Find the widget inside the window 

207 from openhcs.textual_tui.widgets.openhcs_toolong_widget import OpenHCSToolongWidget 

208 widgets = window.query(OpenHCSToolongWidget) 

209 for widget in widgets: 

210 logger.info("Clearing logs from singleton toolong window") 

211 widget._clear_all_logs_except_tui() 

212 logger.info("Logs cleared from singleton toolong window") 

213 except Exception as e: 

214 logger.error(f"Failed to clear logs from singleton toolong window: {e}") 

215 import traceback 

216 logger.error(traceback.format_exc())