Coverage for openhcs/textual_tui/widgets/log_monitor.py: 0.0%

1""" 

2OpenHCS Log Monitor Widget 

3 

4Integrates the modernized Toolong log viewing capabilities into the OpenHCS TUI 

5for monitoring subprocess and worker process logs in real-time. 

6 

7This uses our Textual 3.x compatible fork of Toolong. 

8""" 

9 

10import logging 

11from pathlib import Path 

12from typing import List, Optional 

13 

14from textual.app import ComposeResult 

15from textual.containers import Container, Horizontal, Vertical 

16from textual.widgets import Button, Label, Static 

17from textual.widget import Widget 

18from textual.reactive import reactive 

19 

20# Import modernized Toolong components (Textual 3.x compatible) 

21from toolong.ui import UI 

22from toolong.log_view import LogView 

23from toolong.log_file import LogFile 

24from toolong.watcher import get_watcher 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29class LogMonitorWidget(Widget): 

30 """ 

31 Widget for monitoring OpenHCS subprocess and worker logs using Toolong components. 

32  

33 Features: 

34 - Live tailing of multiple log files 

35 - Syntax highlighting for log formats 

36 - Search and filtering capabilities 

37 - Automatic detection of new worker log files 

38 """ 

39 

40 # Reactive properties 

41 log_files: reactive[List[str]] = reactive([], layout=True) 

42 current_log: reactive[Optional[str]] = reactive(None) 

43 

44 def __init__( 

45 self, 

46 log_file_base: Optional[str] = None, 

47 auto_detect_workers: bool = True, 

48 **kwargs 

49 ): 

50 """ 

51 Initialize the log monitor. 

52  

53 Args: 

54 log_file_base: Base path for log files (e.g., "openhcs_subprocess_plates_A_B_1234567890") 

55 auto_detect_workers: Whether to automatically detect new worker log files 

56 """ 

57 super().__init__(**kwargs) 

58 self.log_file_base = log_file_base 

59 self.auto_detect_workers = auto_detect_workers 

60 self.log_views = {} # Map of log file path to LogView widget 

61 self.watchers = {} # Map of log file path to file watcher 

62 

63 def compose(self) -> ComposeResult: 

64 """Compose the log monitor layout.""" 

65 with Vertical(): 

66 # Header with controls 

67 with Horizontal(classes="log-monitor-header"): 

68 yield Label("📋 Log Monitor", classes="log-monitor-title") 

69 yield Button("🔄 Refresh", id="refresh-logs", variant="primary") 

70 yield Button("🔍 Search", id="search-logs", variant="default") 

71 yield Button("⏸️ Pause", id="pause-tailing", variant="default") 

72 

73 # Log file tabs/selector 

74 with Horizontal(classes="log-file-selector"): 

75 yield Label("Active Logs:", classes="selector-label") 

76 # Dynamic log file buttons will be added here 

77 

78 # Main log viewing area 

79 with Container(classes="log-view-container"): 

80 yield Static("Select a log file to view", classes="log-placeholder") 

81 

82 def on_mount(self) -> None: 

83 """Set up the log monitor when mounted.""" 

84 if self.log_file_base: 

85 self.discover_log_files() 

86 if self.auto_detect_workers: 

87 self.start_worker_detection() 

88 

89 def discover_log_files(self) -> None: 

90 """Discover existing log files based on the base path.""" 

91 if not self.log_file_base: 

92 return 

93 

94 base_path = Path(self.log_file_base) 

95 log_dir = base_path.parent 

96 base_name = base_path.name 

97 

98 # Find main subprocess log 

99 main_log = f"{self.log_file_base}.log" 

100 if Path(main_log).exists(): 

101 self.add_log_file(main_log, "Main Process") 

102 

103 # Find worker logs 

104 if log_dir.exists(): 

105 for log_file in log_dir.glob(f"{base_name}_worker_*.log"): 

106 worker_id = log_file.stem.split('_worker_')[-1] 

107 self.add_log_file(str(log_file), f"Worker {worker_id}") 

108 

109 def add_log_file(self, log_path: str, display_name: str = None) -> None: 

110 """Add a log file to the monitor.""" 

111 if log_path in self.log_views: 

112 return # Already monitoring this file 

113 

114 try: 

115 # Verify file exists 

116 if not Path(log_path).exists(): 

117 logger.warning(f"Log file does not exist: {log_path}") 

118 return 

119 

120 # Create a simple UI wrapper for this log file 

121 # We'll use Toolong's UI components directly 

122 self.log_views[log_path] = { 

123 'path': log_path, 

124 'display_name': display_name or Path(log_path).name 

125 } 

126 

127 # Update the UI 

128 self.update_log_selector() 

129 

130 logger.info(f"Added log file to monitor: {log_path}") 

131 

132 except Exception as e: 

133 logger.error(f"Failed to add log file {log_path}: {e}") 

134 

135 def update_log_selector(self) -> None: 

136 """Update the log file selector buttons.""" 

137 selector = self.query_one(".log-file-selector") 

138 

139 # Remove existing log buttons (keep the label) 

140 for button in selector.query("Button.log-file-button"): 

141 button.remove() 

142 

143 # Add buttons for each log file 

144 for i, (log_path, log_info) in enumerate(self.log_views.items()): 

145 button = Button( 

146 log_info['display_name'], 

147 id=f"log-{i}-{abs(hash(log_path)) % 10000}", # More unique ID 

148 classes="log-file-button", 

149 variant="default" 

150 ) 

151 selector.mount(button) 

152 

153 def switch_to_log(self, log_path: str) -> None: 

154 """Switch the main view to show a specific log file using Toolong UI.""" 

155 if log_path not in self.log_views: 

156 return 

157 

158 try: 

159 container = self.query_one(".log-view-container") 

160 

161 # Remove current content 

162 container.query("*").remove() 

163 

164 # Create Toolong UI for this log file 

165 toolong_ui = UI([log_path], merge=False) 

166 

167 # Mount the Toolong UI 

168 container.mount(toolong_ui) 

169 

170 self.current_log = log_path 

171 logger.info(f"Switched to log: {log_path}") 

172 

173 except Exception as e: 

174 logger.error(f"Failed to switch to log {log_path}: {e}") 

175 # Show error message in container 

176 container = self.query_one(".log-view-container") 

177 container.query("*").remove() 

178 container.mount(Static(f"Error loading log: {e}", classes="log-error")) 

179 

180 def on_button_pressed(self, event: Button.Pressed) -> None: 

181 """Handle button presses.""" 

182 button_id = event.button.id 

183 

184 if button_id == "refresh-logs": 

185 self.discover_log_files() 

186 elif button_id == "search-logs": 

187 self.show_search_dialog() 

188 elif button_id == "pause-tailing": 

189 self.toggle_tailing() 

190 elif button_id and button_id.startswith("log-"): 

191 # Find the log file for this button 

192 for i, (log_path, log_info) in enumerate(self.log_views.items()): 

193 if f"log-{i}-{abs(hash(log_path)) % 10000}" == button_id: 

194 self.switch_to_log(log_path) 

195 break 

196 

197 def show_search_dialog(self) -> None: 

198 """Show search dialog for the current log.""" 

199 # TODO: Implement search functionality using Toolong's FindDialog 

200 logger.info("Search functionality not yet implemented") 

201 

202 def toggle_tailing(self) -> None: 

203 """Toggle live tailing on/off.""" 

204 # TODO: Implement tailing toggle 

205 logger.info("Tailing toggle not yet implemented") 

206 

207 def start_worker_detection(self) -> None: 

208 """Start monitoring for new worker log files.""" 

209 # TODO: Implement automatic detection of new worker logs 

210 logger.info("Worker detection not yet implemented") 

211 

212 def cleanup(self) -> None: 

213 """Clean up watchers and resources.""" 

214 for watcher in self.watchers.values(): 

215 if hasattr(watcher, 'stop'): 

216 watcher.stop() 

217 self.watchers.clear() 

218 self.log_views.clear()