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 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class LogMonitorWidget(Widget): 

27 """ 

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

29  

30 Features: 

31 - Live tailing of multiple log files 

32 - Syntax highlighting for log formats 

33 - Search and filtering capabilities 

34 - Automatic detection of new worker log files 

35 """ 

36 

37 # Reactive properties 

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

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

40 

41 def __init__( 

42 self, 

43 log_file_base: Optional[str] = None, 

44 auto_detect_workers: bool = True, 

45 **kwargs 

46 ): 

47 """ 

48 Initialize the log monitor. 

49  

50 Args: 

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

52 auto_detect_workers: Whether to automatically detect new worker log files 

53 """ 

54 super().__init__(**kwargs) 

55 self.log_file_base = log_file_base 

56 self.auto_detect_workers = auto_detect_workers 

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

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

59 

60 def compose(self) -> ComposeResult: 

61 """Compose the log monitor layout.""" 

62 with Vertical(): 

63 # Header with controls 

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

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

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

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

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

69 

70 # Log file tabs/selector 

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

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

73 # Dynamic log file buttons will be added here 

74 

75 # Main log viewing area 

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

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

78 

79 def on_mount(self) -> None: 

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

81 if self.log_file_base: 

82 self.discover_log_files() 

83 if self.auto_detect_workers: 

84 self.start_worker_detection() 

85 

86 def discover_log_files(self) -> None: 

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

88 if not self.log_file_base: 

89 return 

90 

91 base_path = Path(self.log_file_base) 

92 log_dir = base_path.parent 

93 base_name = base_path.name 

94 

95 # Find main subprocess log 

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

97 if Path(main_log).exists(): 

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

99 

100 # Find worker logs 

101 if log_dir.exists(): 

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

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

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

105 

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

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

108 if log_path in self.log_views: 

109 return # Already monitoring this file 

110 

111 try: 

112 # Verify file exists 

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

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

115 return 

116 

117 # Create a simple UI wrapper for this log file 

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

119 self.log_views[log_path] = { 

120 'path': log_path, 

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

122 } 

123 

124 # Update the UI 

125 self.update_log_selector() 

126 

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

128 

129 except Exception as e: 

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

131 

132 def update_log_selector(self) -> None: 

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

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

135 

136 # Remove existing log buttons (keep the label) 

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

138 button.remove() 

139 

140 # Add buttons for each log file 

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

142 button = Button( 

143 log_info['display_name'], 

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

145 classes="log-file-button", 

146 variant="default" 

147 ) 

148 selector.mount(button) 

149 

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

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

152 if log_path not in self.log_views: 

153 return 

154 

155 try: 

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

157 

158 # Remove current content 

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

160 

161 # Create Toolong UI for this log file 

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

163 

164 # Mount the Toolong UI 

165 container.mount(toolong_ui) 

166 

167 self.current_log = log_path 

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

169 

170 except Exception as e: 

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

172 # Show error message in container 

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

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

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

176 

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

178 """Handle button presses.""" 

179 button_id = event.button.id 

180 

181 if button_id == "refresh-logs": 

182 self.discover_log_files() 

183 elif button_id == "search-logs": 

184 self.show_search_dialog() 

185 elif button_id == "pause-tailing": 

186 self.toggle_tailing() 

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

188 # Find the log file for this button 

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

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

191 self.switch_to_log(log_path) 

192 break 

193 

194 def show_search_dialog(self) -> None: 

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

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

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

198 

199 def toggle_tailing(self) -> None: 

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

201 # TODO: Implement tailing toggle 

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

203 

204 def start_worker_detection(self) -> None: 

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

206 # TODO: Implement automatic detection of new worker logs 

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

208 

209 def cleanup(self) -> None: 

210 """Clean up watchers and resources.""" 

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

212 if hasattr(watcher, 'stop'): 

213 watcher.stop() 

214 self.watchers.clear() 

215 self.log_views.clear()