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

1""" 

2Complete Toolong Widget for OpenHCS TUI 

3 

4A self-contained widget that embeds the full Toolong log viewer functionality 

5into the OpenHCS TUI. This provides professional log viewing with all Toolong 

6features while being properly integrated into the OpenHCS window system. 

7""" 

8 

9import logging 

10from pathlib import Path 

11from typing import List, Optional 

12 

13from textual.app import ComposeResult 

14from textual.widget import Widget 

15from textual.widgets import TabbedContent, TabPane, Select 

16from textual.lazy import Lazy 

17 

18# Import Toolong components 

19from toolong.ui import LogScreen 

20from toolong.watcher import get_watcher 

21 

22logger = logging.getLogger(__name__) 

23 

24# Import shared watcher to prevent conflicts 

25from openhcs.textual_tui.widgets.openhcs_toolong_widget import get_shared_watcher 

26 

27 

28class ToolongWidget(Widget): 

29 """ 

30 Complete Toolong log viewer widget. 

31  

32 This widget encapsulates the full Toolong application functionality, 

33 providing professional log viewing with tabs, search, tailing, and 

34 all other Toolong features in a self-contained widget. 

35 """ 

36 

37 DEFAULT_CSS = """ 

38 ToolongWidget { 

39 width: 100%; 

40 height: 100%; 

41 } 

42  

43 ToolongWidget TabbedContent { 

44 width: 100%; 

45 height: 100%; 

46 } 

47  

48 ToolongWidget TabPane { 

49 padding: 0; 

50 } 

51 """ 

52 

53 def __init__( 

54 self, 

55 log_files: List[str], 

56 merge: bool = False, 

57 can_tail: bool = True, 

58 **kwargs 

59 ): 

60 """ 

61 Initialize the Toolong widget. 

62  

63 Args: 

64 log_files: List of log file paths to display 

65 merge: Whether to merge multiple files into one view 

66 can_tail: Whether tailing is enabled 

67 """ 

68 super().__init__(**kwargs) 

69 self.log_files = self._sort_paths(log_files) 

70 self.merge = merge 

71 self.can_tail = can_tail 

72 self.watcher = get_shared_watcher() # Use shared watcher to prevent conflicts 

73 

74 @classmethod 

75 def _sort_paths(cls, paths: List[str]) -> List[str]: 

76 """Sort paths for consistent display order.""" 

77 return sorted(paths, key=lambda p: Path(p).name) 

78 

79 def compose(self) -> ComposeResult: 

80 """Compose the Toolong widget using dropdown selector instead of tabs.""" 

81 if not self.log_files: 

82 # No log files - show empty state 

83 from textual.widgets import Static 

84 yield Static("No log files available", classes="empty-state") 

85 return 

86 

87 from textual.widgets import Select, Container 

88 from textual.containers import Horizontal 

89 

90 # Create dropdown options with friendly names 

91 options = [] 

92 for log_file in self.log_files: 

93 display_name = self._get_friendly_name(log_file) 

94 options.append((display_name, log_file)) 

95 

96 # Dropdown selector 

97 with Horizontal(classes="log-selector"): 

98 yield Select( 

99 options=options, 

100 value=self.log_files[0] if self.log_files else None, 

101 id="log_selector", 

102 compact=True 

103 ) 

104 

105 # Container for log view 

106 with Container(id="log_container"): 

107 if self.log_files: 

108 yield Lazy( 

109 LogView( 

110 [self.log_files[0]], # Start with first file 

111 self.watcher, 

112 can_tail=self.can_tail, 

113 ) 

114 ) 

115 

116 def _get_friendly_name(self, log_file: str) -> str: 

117 """Convert log file path to friendly display name.""" 

118 import re 

119 file_name = Path(log_file).name 

120 

121 if "unified" in file_name: 

122 if "worker_" in file_name: 

123 # Extract worker ID: openhcs_unified_20250630_094200_subprocess_123_worker_456_789.log 

124 worker_match = re.search(r'worker_(\d+)', file_name) 

125 return f"Worker {worker_match.group(1)}" if worker_match else "Worker" 

126 elif "_subprocess_" in file_name: 

127 return "Subprocess" 

128 else: 

129 return "TUI Main" 

130 elif "subprocess" in file_name: 

131 return "Subprocess" 

132 else: 

133 return Path(log_file).name 

134 

135 def on_select_changed(self, event: Select.Changed) -> None: 

136 """Handle dropdown selection change.""" 

137 if event.control.id == "log_selector" and event.value: 

138 selected_path = event.value 

139 

140 # Update the log view with selected file 

141 container = self.query_one("#log_container") 

142 

143 # Remove existing log view 

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

145 

146 # Add new log view for selected file 

147 new_view = LogView( 

148 [selected_path], 

149 self.watcher, 

150 can_tail=self.can_tail, 

151 ) 

152 container.mount(new_view) 

153 

154 def on_mount(self) -> None: 

155 """Start the watcher when widget is mounted.""" 

156 

157 # Start the watcher and enable tailing (only if not already running) 

158 if not hasattr(self.watcher, '_thread') or self.watcher._thread is None: 

159 self.watcher.start() 

160 else: 

161 pass # Watcher already running 

162 

163 # Focus LogLines (no tabs in dropdown structure) 

164 try: 

165 log_lines = self.query("LogView > LogLines") 

166 if log_lines: 

167 log_lines.first().focus() 

168 except Exception as e: 

169 logger.debug(f"Could not focus LogLines: {e}") 

170 

171 # Simple tailing setup 

172 if self.can_tail: 

173 self.call_after_refresh(self._enable_tailing) 

174 

175 def _enable_tailing(self) -> None: 

176 """Enable tailing on LogView widgets.""" 

177 try: 

178 for log_view in self.query("LogView"): 

179 if hasattr(log_view, 'can_tail') and log_view.can_tail: 

180 log_view.tail = True 

181 

182 # Also start the file watcher for single-file LogLines 

183 for log_lines in log_view.query("LogLines"): 

184 if hasattr(log_lines, 'start_tail') and len(log_lines.log_files) == 1: 

185 log_lines.start_tail() 

186 except Exception as e: 

187 logger.debug(f"Could not enable tailing: {e}") 

188 

189 def on_unmount(self) -> None: 

190 """Clean up the watcher when widget is unmounted.""" 

191 # Don't close shared watcher - other widgets might be using it 

192 # The shared watcher will be cleaned up when the app exits 

193 

194 def add_log_file(self, log_file: str) -> None: 

195 """Add a new log file to the widget.""" 

196 if log_file not in self.log_files: 

197 self.log_files.append(log_file) 

198 self.log_files = self._sort_paths(self.log_files) 

199 # Would need to refresh the widget to show new file 

200 

201 def remove_log_file(self, log_file: str) -> None: 

202 """Remove a log file from the widget.""" 

203 if log_file in self.log_files: 

204 self.log_files.remove(log_file) 

205 # Would need to refresh the widget to remove file 

206 

207 @classmethod 

208 def from_single_file(cls, log_file: str, can_tail: bool = True) -> "ToolongWidget": 

209 """Create a ToolongWidget for a single log file.""" 

210 return cls([log_file], merge=False, can_tail=can_tail) 

211 

212 @classmethod 

213 def from_multiple_files(cls, log_files: List[str], merge: bool = False) -> "ToolongWidget": 

214 """Create a ToolongWidget for multiple log files.""" 

215 return cls(log_files, merge=merge, can_tail=True)