Coverage for openhcs/textual_tui/widgets/toolong_widget.py: 0.0%
92 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Complete Toolong Widget for OpenHCS TUI
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"""
9import logging
10from pathlib import Path
11from typing import List
13from textual.app import ComposeResult
14from textual.widget import Widget
15from textual.widgets import Select
16from textual.lazy import Lazy
18# Import Toolong components
20logger = logging.getLogger(__name__)
22# Import shared watcher to prevent conflicts
23from openhcs.textual_tui.widgets.openhcs_toolong_widget import get_shared_watcher
26class ToolongWidget(Widget):
27 """
28 Complete Toolong log viewer widget.
30 This widget encapsulates the full Toolong application functionality,
31 providing professional log viewing with tabs, search, tailing, and
32 all other Toolong features in a self-contained widget.
33 """
35 DEFAULT_CSS = """
36 ToolongWidget {
37 width: 100%;
38 height: 100%;
39 }
41 ToolongWidget TabbedContent {
42 width: 100%;
43 height: 100%;
44 }
46 ToolongWidget TabPane {
47 padding: 0;
48 }
49 """
51 def __init__(
52 self,
53 log_files: List[str],
54 merge: bool = False,
55 can_tail: bool = True,
56 **kwargs
57 ):
58 """
59 Initialize the Toolong widget.
61 Args:
62 log_files: List of log file paths to display
63 merge: Whether to merge multiple files into one view
64 can_tail: Whether tailing is enabled
65 """
66 super().__init__(**kwargs)
67 self.log_files = self._sort_paths(log_files)
68 self.merge = merge
69 self.can_tail = can_tail
70 self.watcher = get_shared_watcher() # Use shared watcher to prevent conflicts
72 @classmethod
73 def _sort_paths(cls, paths: List[str]) -> List[str]:
74 """Sort paths for consistent display order."""
75 return sorted(paths, key=lambda p: Path(p).name)
77 def compose(self) -> ComposeResult:
78 """Compose the Toolong widget using dropdown selector instead of tabs."""
79 if not self.log_files:
80 # No log files - show empty state
81 from textual.widgets import Static
82 yield Static("No log files available", classes="empty-state")
83 return
85 from textual.widgets import Select, Container
86 from textual.containers import Horizontal
88 # Create dropdown options with friendly names
89 options = []
90 for log_file in self.log_files:
91 display_name = self._get_friendly_name(log_file)
92 options.append((display_name, log_file))
94 # Dropdown selector
95 with Horizontal(classes="log-selector"):
96 yield Select(
97 options=options,
98 value=self.log_files[0] if self.log_files else None,
99 id="log_selector",
100 compact=True
101 )
103 # Container for log view
104 with Container(id="log_container"):
105 if self.log_files:
106 yield Lazy(
107 LogView(
108 [self.log_files[0]], # Start with first file
109 self.watcher,
110 can_tail=self.can_tail,
111 )
112 )
114 def _get_friendly_name(self, log_file: str) -> str:
115 """Convert log file path to friendly display name."""
116 import re
117 file_name = Path(log_file).name
119 if "unified" in file_name:
120 if "worker_" in file_name:
121 # Extract worker ID: openhcs_unified_20250630_094200_subprocess_123_worker_456_789.log
122 worker_match = re.search(r'worker_(\d+)', file_name)
123 return f"Worker {worker_match.group(1)}" if worker_match else "Worker"
124 elif "_subprocess_" in file_name:
125 return "Subprocess"
126 else:
127 return "TUI Main"
128 elif "subprocess" in file_name:
129 return "Subprocess"
130 else:
131 return Path(log_file).name
133 def on_select_changed(self, event: Select.Changed) -> None:
134 """Handle dropdown selection change."""
135 if event.control.id == "log_selector" and event.value:
136 selected_path = event.value
138 # Update the log view with selected file
139 container = self.query_one("#log_container")
141 # Remove existing log view
142 container.query("*").remove()
144 # Add new log view for selected file
145 new_view = LogView(
146 [selected_path],
147 self.watcher,
148 can_tail=self.can_tail,
149 )
150 container.mount(new_view)
152 def on_mount(self) -> None:
153 """Start the watcher when widget is mounted."""
155 # Start the watcher and enable tailing (only if not already running)
156 if not hasattr(self.watcher, '_thread') or self.watcher._thread is None:
157 self.watcher.start()
158 else:
159 pass # Watcher already running
161 # Focus LogLines (no tabs in dropdown structure)
162 try:
163 log_lines = self.query("LogView > LogLines")
164 if log_lines:
165 log_lines.first().focus()
166 except Exception as e:
167 logger.debug(f"Could not focus LogLines: {e}")
169 # Simple tailing setup
170 if self.can_tail:
171 self.call_after_refresh(self._enable_tailing)
173 def _enable_tailing(self) -> None:
174 """Enable tailing on LogView widgets."""
175 try:
176 for log_view in self.query("LogView"):
177 if hasattr(log_view, 'can_tail') and log_view.can_tail:
178 log_view.tail = True
180 # Also start the file watcher for single-file LogLines
181 for log_lines in log_view.query("LogLines"):
182 if hasattr(log_lines, 'start_tail') and len(log_lines.log_files) == 1:
183 log_lines.start_tail()
184 except Exception as e:
185 logger.debug(f"Could not enable tailing: {e}")
187 def on_unmount(self) -> None:
188 """Clean up the watcher when widget is unmounted."""
189 # Don't close shared watcher - other widgets might be using it
190 # The shared watcher will be cleaned up when the app exits
192 def add_log_file(self, log_file: str) -> None:
193 """Add a new log file to the widget."""
194 if log_file not in self.log_files:
195 self.log_files.append(log_file)
196 self.log_files = self._sort_paths(self.log_files)
197 # Would need to refresh the widget to show new file
199 def remove_log_file(self, log_file: str) -> None:
200 """Remove a log file from the widget."""
201 if log_file in self.log_files:
202 self.log_files.remove(log_file)
203 # Would need to refresh the widget to remove file
205 @classmethod
206 def from_single_file(cls, log_file: str, can_tail: bool = True) -> "ToolongWidget":
207 """Create a ToolongWidget for a single log file."""
208 return cls([log_file], merge=False, can_tail=can_tail)
210 @classmethod
211 def from_multiple_files(cls, log_files: List[str], merge: bool = False) -> "ToolongWidget":
212 """Create a ToolongWidget for multiple log files."""
213 return cls(log_files, merge=merge, can_tail=True)