Coverage for openhcs/pyqt_gui/widgets/log_viewer.py: 0.0%
564 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""
2PyQt6 Log Viewer Window
4Provides comprehensive log viewing capabilities with real-time tailing, search functionality,
5and integration with OpenHCS subprocess execution. Reimplements log viewing using Qt widgets
6for native desktop integration.
7"""
9import logging
10from typing import Optional, List, Set, Tuple
11from pathlib import Path
13from PyQt6.QtWidgets import (
14 QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QComboBox,
15 QTextEdit, QToolBar, QLineEdit, QCheckBox, QPushButton, QDialog
16)
17from PyQt6.QtGui import QSyntaxHighlighter, QTextDocument
18from PyQt6.QtCore import QObject, QTimer, QFileSystemWatcher, pyqtSignal, Qt, QRegularExpression
19from PyQt6.QtGui import QTextCharFormat, QColor, QAction, QFont, QTextCursor
21from openhcs.io.filemanager import FileManager
22from openhcs.core.log_utils import LogFileInfo
23from openhcs.pyqt_gui.utils.log_detection_utils import (
24 get_current_tui_log_path, discover_logs, discover_all_logs
25)
26from openhcs.core.log_utils import (
27 classify_log_file, is_openhcs_log_file, infer_base_log_path
28)
30# Import Pygments for advanced syntax highlighting
31from pygments import highlight
32from pygments.lexers import PythonLexer, get_lexer_by_name
33from pygments.formatters import get_formatter_by_name
34from pygments.token import Token
35from pygments.style import Style
36from pygments.styles import get_style_by_name
37from dataclasses import dataclass
38from typing import Dict, Tuple
40logger = logging.getLogger(__name__)
43@dataclass
44class LogColorScheme:
45 """
46 Centralized color scheme for log highlighting with semantic color names.
48 Supports light/dark theme variants and ensures WCAG accessibility compliance.
49 All colors meet minimum 4.5:1 contrast ratio for normal text readability.
50 """
52 # Log level colors with semantic meaning (WCAG 4.5:1 compliant)
53 log_critical_fg: Tuple[int, int, int] = (255, 255, 255) # White text
54 log_critical_bg: Tuple[int, int, int] = (139, 0, 0) # Dark red background
55 log_error_color: Tuple[int, int, int] = (255, 85, 85) # Brighter red - WCAG compliant
56 log_warning_color: Tuple[int, int, int] = (255, 140, 0) # Dark orange - attention grabbing
57 log_info_color: Tuple[int, int, int] = (100, 160, 210) # Brighter steel blue - WCAG compliant
58 log_debug_color: Tuple[int, int, int] = (160, 160, 160) # Lighter gray - better contrast
60 # Metadata and structural colors
61 timestamp_color: Tuple[int, int, int] = (105, 105, 105) # Dim gray - unobtrusive
62 logger_name_color: Tuple[int, int, int] = (147, 112, 219) # Medium slate blue - distinctive
63 memory_address_color: Tuple[int, int, int] = (255, 182, 193) # Light pink - technical data
64 file_path_color: Tuple[int, int, int] = (34, 139, 34) # Forest green - file system
66 # Python syntax colors (following VS Code dark theme conventions)
67 python_keyword_color: Tuple[int, int, int] = (86, 156, 214) # Blue - language keywords
68 python_string_color: Tuple[int, int, int] = (206, 145, 120) # Orange - string literals
69 python_number_color: Tuple[int, int, int] = (181, 206, 168) # Light green - numeric values
70 python_operator_color: Tuple[int, int, int] = (212, 212, 212) # Light gray - operators/punctuation
71 python_name_color: Tuple[int, int, int] = (156, 220, 254) # Light blue - identifiers
72 python_function_color: Tuple[int, int, int] = (220, 220, 170) # Yellow - function names
73 python_class_color: Tuple[int, int, int] = (78, 201, 176) # Teal - class names
74 python_builtin_color: Tuple[int, int, int] = (86, 156, 214) # Blue - built-in functions
75 python_comment_color: Tuple[int, int, int] = (106, 153, 85) # Green - comments
77 # Special highlighting colors
78 exception_color: Tuple[int, int, int] = (255, 69, 0) # Red orange - error types
79 function_call_color: Tuple[int, int, int] = (255, 215, 0) # Gold - function invocations
80 boolean_color: Tuple[int, int, int] = (86, 156, 214) # Blue - True/False/None
82 # Enhanced syntax colors (Phase 1 additions)
83 tuple_parentheses_color: Tuple[int, int, int] = (255, 215, 0) # Gold - tuple delimiters
84 set_braces_color: Tuple[int, int, int] = (255, 140, 0) # Dark orange - set delimiters
85 class_representation_color: Tuple[int, int, int] = (78, 201, 176) # Teal - <class 'name'>
86 function_representation_color: Tuple[int, int, int] = (220, 220, 170) # Yellow - <function name>
87 module_path_color: Tuple[int, int, int] = (147, 112, 219) # Medium slate blue - module.path
88 hex_number_color: Tuple[int, int, int] = (181, 206, 168) # Light green - 0xFF
89 scientific_notation_color: Tuple[int, int, int] = (181, 206, 168) # Light green - 1.23e-4
90 binary_number_color: Tuple[int, int, int] = (181, 206, 168) # Light green - 0b1010
91 octal_number_color: Tuple[int, int, int] = (181, 206, 168) # Light green - 0o755
92 python_special_color: Tuple[int, int, int] = (255, 20, 147) # Deep pink - __name__
93 single_quoted_string_color: Tuple[int, int, int] = (206, 145, 120) # Orange - 'string'
94 list_comprehension_color: Tuple[int, int, int] = (156, 220, 254) # Light blue - [x for x in y]
95 generator_expression_color: Tuple[int, int, int] = (156, 220, 254) # Light blue - (x for x in y)
97 @classmethod
98 def create_dark_theme(cls) -> 'LogColorScheme':
99 """
100 Create a dark theme variant with adjusted colors for dark backgrounds.
102 Returns:
103 LogColorScheme: Dark theme color scheme with higher contrast
104 """
105 return cls(
106 # Enhanced colors for dark backgrounds with better contrast
107 log_error_color=(255, 100, 100), # Brighter red
108 log_info_color=(120, 180, 230), # Brighter steel blue
109 timestamp_color=(160, 160, 160), # Lighter gray
110 python_string_color=(236, 175, 150), # Brighter orange
111 python_number_color=(200, 230, 190), # Brighter green
112 # Other colors remain the same as they work well on dark backgrounds
113 )
115 @classmethod
116 def create_light_theme(cls) -> 'LogColorScheme':
117 """
118 Create a light theme variant with adjusted colors for light backgrounds.
120 Returns:
121 LogColorScheme: Light theme color scheme with appropriate contrast
122 """
123 return cls(
124 # Darker colors for light backgrounds with WCAG compliance
125 log_error_color=(180, 20, 40), # Darker red
126 log_info_color=(30, 80, 130), # Darker steel blue
127 log_warning_color=(200, 100, 0), # Darker orange
128 timestamp_color=(60, 60, 60), # Darker gray
129 logger_name_color=(100, 60, 160), # Darker slate blue
130 python_string_color=(150, 80, 60), # Darker orange
131 python_number_color=(120, 140, 100), # Darker green
132 memory_address_color=(200, 120, 140), # Darker pink
133 file_path_color=(20, 100, 20), # Darker forest green
134 exception_color=(200, 40, 0), # Darker red orange
135 # Adjust other colors for light background contrast
136 )
138 def to_qcolor(self, color_tuple: Tuple[int, int, int]) -> QColor:
139 """
140 Convert RGB tuple to QColor object.
142 Args:
143 color_tuple: RGB color tuple (r, g, b)
145 Returns:
146 QColor: Qt color object
147 """
148 return QColor(*color_tuple)
151class LogFileDetector(QObject):
152 """
153 Detects new log files in directory using efficient file monitoring.
155 Uses QFileSystemWatcher to monitor directory changes and set operations
156 for efficient new file detection. Handles base_log_path as file prefix
157 and watches the parent directory.
158 """
160 # Signals
161 new_log_detected = pyqtSignal(object) # LogFileInfo object
163 def __init__(self, base_log_path: Optional[str] = None):
164 """
165 Initialize LogFileDetector.
167 Args:
168 base_log_path: Base path for subprocess log files (file prefix, not directory)
169 """
170 super().__init__()
171 self._base_log_path = base_log_path
172 self._previous_files: Set[Path] = set()
173 self._watcher = QFileSystemWatcher()
174 self._watcher.directoryChanged.connect(self._on_directory_changed)
175 self._watching_directory: Optional[Path] = None
177 logger.debug(f"LogFileDetector initialized with base_log_path: {base_log_path}")
179 def start_watching(self, directory: Path) -> None:
180 """
181 Start watching directory for new log files.
183 Args:
184 directory: Directory to watch for new log files
185 """
186 if not directory.exists():
187 logger.warning(f"Cannot watch non-existent directory: {directory}")
188 return
190 # Stop any existing watching
191 self.stop_watching()
193 # Add directory to watcher
194 success = self._watcher.addPath(str(directory))
195 if success:
196 self._watching_directory = directory
197 # Initialize previous files set
198 self._previous_files = self.scan_directory(directory)
199 logger.debug(f"Started watching directory: {directory}")
200 logger.debug(f"Initial file count: {len(self._previous_files)}")
201 else:
202 logger.error(f"Failed to add directory to watcher: {directory}")
204 def stop_watching(self) -> None:
205 """Stop file watching and cleanup."""
206 if self._watching_directory:
207 self._watcher.removePath(str(self._watching_directory))
208 self._watching_directory = None
209 self._previous_files.clear()
210 logger.debug("Stopped file watching")
212 def scan_directory(self, directory: Path) -> Set[Path]:
213 """
214 Scan directory for .log files.
216 Args:
217 directory: Directory to scan
219 Returns:
220 Set[Path]: Set of Path objects for .log files found
221 """
222 try:
223 log_files = set(directory.glob("*.log"))
224 logger.debug(f"Scanned directory {directory}: found {len(log_files)} .log files")
225 return log_files
226 except (FileNotFoundError, PermissionError) as e:
227 logger.warning(f"Error scanning directory {directory}: {e}")
228 return set()
230 def detect_new_files(self, current_files: Set[Path]) -> Set[Path]:
231 """
232 Use set.difference() to find new files efficiently.
234 Args:
235 current_files: Current set of files in directory
237 Returns:
238 Set[Path]: Set of newly discovered files
239 """
240 new_files = current_files.difference(self._previous_files)
241 if new_files:
242 logger.debug(f"Detected {len(new_files)} new files: {[f.name for f in new_files]}")
244 # Update previous files set
245 self._previous_files = current_files
246 return new_files
250 def _on_directory_changed(self, directory_path: str) -> None:
251 """
252 Handle QFileSystemWatcher directory change signal.
254 Args:
255 directory_path: Path of directory that changed
256 """
257 directory = Path(directory_path)
258 logger.debug(f"Directory changed: {directory}")
260 # Scan directory for current files
261 current_files = self.scan_directory(directory)
263 # Detect new files
264 new_files = self.detect_new_files(current_files)
266 # Process new files
267 for file_path in new_files:
268 if file_path.exists() and is_openhcs_log_file(file_path):
269 try:
270 # For general watching, try to infer base_log_path from the file name
271 effective_base_log_path = self._base_log_path
272 if not effective_base_log_path and 'subprocess_' in file_path.name:
273 effective_base_log_path = infer_base_log_path(file_path)
275 log_info = classify_log_file(file_path, effective_base_log_path,
276 include_tui_log=False)
278 logger.info(f"New relevant log file detected: {file_path} (type: {log_info.log_type})")
279 self.new_log_detected.emit(log_info)
280 except Exception as e:
281 logger.error(f"Error classifying new log file {file_path}: {e}")
284class LogHighlighter(QSyntaxHighlighter):
285 """
286 Advanced syntax highlighter for log files using Pygments.
288 Provides sophisticated highlighting for OpenHCS log format with support for:
289 - Log levels and timestamps
290 - Python code snippets and data structures
291 - Memory addresses and function signatures
292 - Complex nested dictionaries and lists
293 - Exception tracebacks and file paths
294 """
296 def __init__(self, parent: QTextDocument, color_scheme: LogColorScheme = None):
297 """
298 Initialize the log highlighter with optional color scheme.
300 Args:
301 parent: QTextDocument to apply highlighting to
302 color_scheme: Color scheme to use (defaults to dark theme)
303 """
304 super().__init__(parent)
305 self.color_scheme = color_scheme or LogColorScheme()
306 self.setup_pygments_styles()
307 self.setup_highlighting_rules()
309 def setup_pygments_styles(self) -> None:
310 """Setup Pygments token to QTextCharFormat mapping using color scheme."""
311 cs = self.color_scheme # Shorthand for readability
313 # Create a mapping from Pygments tokens to Qt text formats
314 self.token_formats = {
315 # Log levels with distinct colors and backgrounds
316 'log_critical': self._create_format(
317 cs.to_qcolor(cs.log_critical_fg),
318 cs.to_qcolor(cs.log_critical_bg),
319 bold=True
320 ),
321 'log_error': self._create_format(cs.to_qcolor(cs.log_error_color), bold=True),
322 'log_warning': self._create_format(cs.to_qcolor(cs.log_warning_color), bold=True),
323 'log_info': self._create_format(cs.to_qcolor(cs.log_info_color), bold=True),
324 'log_debug': self._create_format(cs.to_qcolor(cs.log_debug_color)),
326 # Timestamps and metadata
327 'timestamp': self._create_format(cs.to_qcolor(cs.timestamp_color)),
328 'logger_name': self._create_format(cs.to_qcolor(cs.logger_name_color), bold=True),
330 # Python syntax highlighting (for complex data structures)
331 Token.Keyword: self._create_format(cs.to_qcolor(cs.python_keyword_color), bold=True),
332 Token.String: self._create_format(cs.to_qcolor(cs.python_string_color)),
333 Token.String.Single: self._create_format(cs.to_qcolor(cs.python_string_color)),
334 Token.String.Double: self._create_format(cs.to_qcolor(cs.python_string_color)),
335 Token.Number: self._create_format(cs.to_qcolor(cs.python_number_color)),
336 Token.Number.Integer: self._create_format(cs.to_qcolor(cs.python_number_color)),
337 Token.Number.Float: self._create_format(cs.to_qcolor(cs.python_number_color)),
338 Token.Number.Hex: self._create_format(cs.to_qcolor(cs.python_number_color)),
339 Token.Number.Oct: self._create_format(cs.to_qcolor(cs.python_number_color)),
340 Token.Number.Bin: self._create_format(cs.to_qcolor(cs.python_number_color)),
341 Token.Operator: self._create_format(cs.to_qcolor(cs.python_operator_color)),
342 Token.Punctuation: self._create_format(cs.to_qcolor(cs.python_operator_color)),
343 Token.Name: self._create_format(cs.to_qcolor(cs.python_name_color)),
344 Token.Name.Function: self._create_format(cs.to_qcolor(cs.python_function_color), bold=True),
345 Token.Name.Class: self._create_format(cs.to_qcolor(cs.python_class_color), bold=True),
346 Token.Name.Builtin: self._create_format(cs.to_qcolor(cs.python_builtin_color)),
347 Token.Comment: self._create_format(cs.to_qcolor(cs.python_comment_color)),
348 Token.Literal: self._create_format(cs.to_qcolor(cs.python_number_color)),
350 # Special patterns for log content
351 'memory_address': self._create_format(cs.to_qcolor(cs.memory_address_color)),
352 'file_path': self._create_format(cs.to_qcolor(cs.file_path_color)),
353 'exception': self._create_format(cs.to_qcolor(cs.exception_color), bold=True),
354 'function_call': self._create_format(cs.to_qcolor(cs.function_call_color)),
355 'dict_key': self._create_format(cs.to_qcolor(cs.python_name_color)),
356 'boolean': self._create_format(cs.to_qcolor(cs.boolean_color), bold=True),
358 # Enhanced Python syntax elements (Phase 1)
359 'tuple_parentheses': self._create_format(cs.to_qcolor(cs.tuple_parentheses_color)),
360 'set_braces': self._create_format(cs.to_qcolor(cs.set_braces_color)),
361 'class_representation': self._create_format(cs.to_qcolor(cs.class_representation_color), bold=True),
362 'function_representation': self._create_format(cs.to_qcolor(cs.function_representation_color), bold=True),
363 'module_path': self._create_format(cs.to_qcolor(cs.module_path_color)),
364 'hex_number': self._create_format(cs.to_qcolor(cs.hex_number_color)),
365 'scientific_notation': self._create_format(cs.to_qcolor(cs.scientific_notation_color)),
366 'binary_number': self._create_format(cs.to_qcolor(cs.binary_number_color)),
367 'octal_number': self._create_format(cs.to_qcolor(cs.octal_number_color)),
368 'python_special': self._create_format(cs.to_qcolor(cs.python_special_color), bold=True),
369 'single_quoted_string': self._create_format(cs.to_qcolor(cs.single_quoted_string_color)),
370 'list_comprehension': self._create_format(cs.to_qcolor(cs.list_comprehension_color)),
371 'generator_expression': self._create_format(cs.to_qcolor(cs.generator_expression_color)),
372 }
374 def _create_format(self, fg_color: QColor, bg_color: QColor = None, bold: bool = False) -> QTextCharFormat:
375 """Create a QTextCharFormat with specified properties."""
376 format = QTextCharFormat()
377 format.setForeground(fg_color)
378 if bg_color:
379 format.setBackground(bg_color)
380 if bold:
381 format.setFontWeight(QFont.Weight.Bold)
382 return format
384 def setup_highlighting_rules(self) -> None:
385 """Setup regex patterns for log-specific highlighting."""
386 self.highlighting_rules = []
388 # Log level patterns (highest priority)
389 log_levels = [
390 ("CRITICAL", self.token_formats['log_critical']),
391 ("ERROR", self.token_formats['log_error']),
392 ("WARNING", self.token_formats['log_warning']),
393 ("INFO", self.token_formats['log_info']),
394 ("DEBUG", self.token_formats['log_debug']),
395 ]
397 for level, format in log_levels:
398 pattern = QRegularExpression(rf"\b{level}\b")
399 self.highlighting_rules.append((pattern, format))
401 # Timestamp pattern: YYYY-MM-DD HH:MM:SS,mmm
402 timestamp_pattern = QRegularExpression(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}")
403 self.highlighting_rules.append((timestamp_pattern, self.token_formats['timestamp']))
405 # Logger names (e.g., openhcs.core.orchestrator)
406 logger_pattern = QRegularExpression(r"openhcs\.[a-zA-Z0-9_.]+")
407 self.highlighting_rules.append((logger_pattern, self.token_formats['logger_name']))
409 # Memory addresses (e.g., 0x7f1640dd8e00)
410 memory_pattern = QRegularExpression(r"0x[0-9a-fA-F]+")
411 self.highlighting_rules.append((memory_pattern, self.token_formats['memory_address']))
413 # File paths in tracebacks
414 filepath_pattern = QRegularExpression(r'["\']?/[^"\'\s]+\.py["\']?')
415 self.highlighting_rules.append((filepath_pattern, self.token_formats['file_path']))
417 # Exception names
418 exception_pattern = QRegularExpression(r'\b[A-Z][a-zA-Z]*Error\b|\b[A-Z][a-zA-Z]*Exception\b')
419 self.highlighting_rules.append((exception_pattern, self.token_formats['exception']))
421 # Function calls with parentheses
422 function_pattern = QRegularExpression(r'\b[a-zA-Z_][a-zA-Z0-9_]*\(\)')
423 self.highlighting_rules.append((function_pattern, self.token_formats['function_call']))
425 # Boolean values
426 boolean_pattern = QRegularExpression(r'\b(True|False|None)\b')
427 self.highlighting_rules.append((boolean_pattern, self.token_formats['boolean']))
429 # Enhanced Python syntax elements
431 # Single-quoted strings (complement to double-quoted)
432 single_quote_pattern = QRegularExpression(r"'[^']*'")
433 self.highlighting_rules.append((single_quote_pattern, self.token_formats['single_quoted_string']))
435 # Class representations: <class 'module.ClassName'>
436 class_repr_pattern = QRegularExpression(r"<class '[^']*'>")
437 self.highlighting_rules.append((class_repr_pattern, self.token_formats['class_representation']))
439 # Function representations: <function name at 0xaddress>
440 function_repr_pattern = QRegularExpression(r"<function [^>]+ at 0x[0-9a-fA-F]+>")
441 self.highlighting_rules.append((function_repr_pattern, self.token_formats['function_representation']))
443 # Extended module paths (beyond just openhcs)
444 module_path_pattern = QRegularExpression(r"\b[a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z_][a-zA-Z0-9_]*){2,}")
445 self.highlighting_rules.append((module_path_pattern, self.token_formats['module_path']))
447 # Hexadecimal numbers (beyond memory addresses): 0xFF, 0x1A2B
448 hex_number_pattern = QRegularExpression(r"\b0[xX][0-9a-fA-F]+\b")
449 self.highlighting_rules.append((hex_number_pattern, self.token_formats['hex_number']))
451 # Scientific notation: 1.23e-4, 5.67E+10
452 scientific_pattern = QRegularExpression(r"\b\d+\.?\d*[eE][+-]?\d+\b")
453 self.highlighting_rules.append((scientific_pattern, self.token_formats['scientific_notation']))
455 # Binary literals: 0b1010
456 binary_pattern = QRegularExpression(r"\b0[bB][01]+\b")
457 self.highlighting_rules.append((binary_pattern, self.token_formats['binary_number']))
459 # Octal literals: 0o755
460 octal_pattern = QRegularExpression(r"\b0[oO][0-7]+\b")
461 self.highlighting_rules.append((octal_pattern, self.token_formats['octal_number']))
463 # Python special constants: __name__, __main__, __file__, etc.
464 python_special_pattern = QRegularExpression(r"\b__[a-zA-Z_][a-zA-Z0-9_]*__\b")
465 self.highlighting_rules.append((python_special_pattern, self.token_formats['python_special']))
467 logger.debug(f"Setup {len(self.highlighting_rules)} highlighting rules")
469 def set_color_scheme(self, color_scheme: LogColorScheme) -> None:
470 """
471 Update the color scheme and refresh highlighting.
473 Args:
474 color_scheme: New color scheme to apply
475 """
476 self.color_scheme = color_scheme
477 self.setup_pygments_styles()
478 self.setup_highlighting_rules()
479 # Trigger re-highlighting of the entire document
480 self.rehighlight()
481 logger.debug(f"Applied new color scheme with {len(self.token_formats)} token formats")
483 def switch_to_dark_theme(self) -> None:
484 """Switch to dark theme color scheme."""
485 self.set_color_scheme(LogColorScheme.create_dark_theme())
487 def switch_to_light_theme(self) -> None:
488 """Switch to light theme color scheme."""
489 self.set_color_scheme(LogColorScheme.create_light_theme())
491 @classmethod
492 def load_color_scheme_from_config(cls, config_path: str = None) -> LogColorScheme:
493 """
494 Load color scheme from external configuration file.
496 Args:
497 config_path: Path to JSON/YAML config file (optional)
499 Returns:
500 LogColorScheme: Loaded color scheme or default if file not found
501 """
502 if config_path and Path(config_path).exists():
503 try:
504 import json
505 with open(config_path, 'r') as f:
506 config = json.load(f)
508 # Create color scheme from config
509 scheme_kwargs = {}
510 for key, value in config.items():
511 if key.endswith('_color') or key.endswith('_fg') or key.endswith('_bg'):
512 if isinstance(value, list) and len(value) == 3:
513 scheme_kwargs[key] = tuple(value)
515 return LogColorScheme(**scheme_kwargs)
517 except Exception as e:
518 logger.warning(f"Failed to load color scheme from {config_path}: {e}")
520 return LogColorScheme() # Return default scheme
522 def highlightBlock(self, text: str) -> None:
523 """
524 Apply advanced highlighting to text block.
526 Uses both regex patterns for log-specific content and Pygments
527 for Python syntax highlighting of complex data structures.
528 """
529 # First apply log-specific patterns
530 for pattern, format in self.highlighting_rules:
531 iterator = pattern.globalMatch(text)
532 while iterator.hasNext():
533 match = iterator.next()
534 start = match.capturedStart()
535 length = match.capturedLength()
536 self.setFormat(start, length, format)
538 # Then apply Pygments highlighting for Python-like content
539 self._highlight_python_content(text)
541 def _highlight_python_content(self, text: str) -> None:
542 """
543 Apply Pygments Python syntax highlighting to parts of the text that contain
544 Python data structures (dictionaries, lists, function signatures, etc.).
545 """
546 try:
547 # Look for Python-like patterns in the log line
548 python_patterns = [
549 # Dictionary patterns: {'key': 'value', ...}
550 r'\{[^{}]*:[^{}]*\}',
551 # List patterns: [item1, item2, ...]
552 r'\[[^\[\]]*,.*?\]',
553 # Function signatures: function_name(arg1=value, arg2=value)
554 r'\b[a-zA-Z_][a-zA-Z0-9_]*\([^)]*=.*?\)',
555 # Complex nested structures
556 r'\{.*?:\s*\[.*?\].*?\}',
558 # Enhanced patterns for Phase 1
560 # Tuple patterns: (item1, item2, item3)
561 r'\([^()]*,.*?\)',
562 # Set patterns: {item1, item2, item3} (no colons, distinguishes from dict)
563 r'\{[^{}:]*,.*?\}',
564 # List comprehensions: [x for x in items]
565 r'\[[^\[\]]*\s+for\s+[^\[\]]*\s+in\s+[^\[\]]*\]',
566 # Generator expressions: (x for x in items)
567 r'\([^()]*\s+for\s+[^()]*\s+in\s+[^()]*\)',
568 # Class representations: <class 'module.ClassName'>
569 r"<class '[^']*'>",
570 # Function representations: <function name at 0xaddress>
571 r"<function [^>]+ at 0x[0-9a-fA-F]+>",
572 # Complex function calls with keyword arguments
573 r'\b[a-zA-Z_][a-zA-Z0-9_]*\([^)]*[a-zA-Z_][a-zA-Z0-9_]*\s*=.*?\)',
574 # Multi-line dictionary/list structures (single line representation)
575 r'\{[^{}]*:\s*[^{}]*,\s*[^{}]*:\s*[^{}]*\}',
576 # Nested collections: [{...}, {...}] or [(...), (...)]
577 r'\[[\{\(][^[\]]*[\}\)],\s*[\{\(][^[\]]*[\}\)]\]',
578 ]
580 for pattern in python_patterns:
581 regex = QRegularExpression(pattern)
582 iterator = regex.globalMatch(text)
584 while iterator.hasNext():
585 match = iterator.next()
586 start = match.capturedStart()
587 length = match.capturedLength()
588 python_text = match.captured(0)
590 # Use Pygments to highlight this Python-like content
591 self._apply_pygments_highlighting(python_text, start)
593 except Exception as e:
594 # Don't let highlighting errors break the log viewer
595 logger.debug(f"Error in Python content highlighting: {e}")
597 def _apply_pygments_highlighting(self, python_text: str, start_offset: int) -> None:
598 """
599 Apply Pygments highlighting to a specific piece of Python-like text.
600 """
601 try:
602 from pygments.lexers import PythonLexer
604 lexer = PythonLexer()
605 tokens = list(lexer.get_tokens(python_text))
607 current_pos = 0
608 for token_type, token_value in tokens:
609 if token_value.strip(): # Skip whitespace-only tokens
610 token_start = start_offset + current_pos
611 token_length = len(token_value)
613 # Apply format if we have a mapping for this token type
614 if token_type in self.token_formats:
615 format = self.token_formats[token_type]
616 self.setFormat(token_start, token_length, format)
618 current_pos += len(token_value)
620 except Exception as e:
621 # Don't let Pygments errors break the highlighting
622 logger.debug(f"Error in Pygments highlighting: {e}")
625class LogViewerWindow(QMainWindow):
626 """Main log viewer window with dropdown, search, and real-time tailing."""
628 window_closed = pyqtSignal()
630 def __init__(self, file_manager: FileManager, service_adapter, parent=None):
631 super().__init__(parent)
632 self.file_manager = file_manager
633 self.service_adapter = service_adapter
635 # State
636 self.current_log_path: Optional[Path] = None
637 self.current_file_position: int = 0
638 self.auto_scroll_enabled: bool = True
639 self.tailing_paused: bool = False
641 # Search state
642 self.current_search_text: str = ""
643 self.search_highlights: List[QTextCursor] = []
645 # Components
646 self.log_selector: QComboBox = None
647 self.search_toolbar: QToolBar = None
648 self.log_display: QTextEdit = None
649 self.file_detector: LogFileDetector = None
650 self.tail_timer: QTimer = None
651 self.highlighter: LogHighlighter = None
653 self.setup_ui()
654 self.setup_connections()
655 self.initialize_logs()
657 def setup_ui(self) -> None:
658 """Setup complete UI layout with exact widget hierarchy."""
659 self.setWindowTitle("Log Viewer")
660 self.setMinimumSize(800, 600)
662 # Central widget with main layout
663 central_widget = QWidget()
664 self.setCentralWidget(central_widget)
665 main_layout = QVBoxLayout(central_widget)
667 # Log selector dropdown
668 self.log_selector = QComboBox()
669 self.log_selector.setMinimumHeight(30)
670 main_layout.addWidget(self.log_selector)
672 # Search toolbar (initially hidden)
673 self.search_toolbar = QToolBar("Search")
674 self.search_toolbar.setVisible(False)
676 # Search input
677 self.search_input = QLineEdit()
678 self.search_input.setPlaceholderText("Search logs...")
679 self.search_toolbar.addWidget(self.search_input)
681 # Search options
682 self.case_sensitive_cb = QCheckBox("Case sensitive")
683 self.search_toolbar.addWidget(self.case_sensitive_cb)
685 self.regex_cb = QCheckBox("Regex")
686 self.search_toolbar.addWidget(self.regex_cb)
688 # Search navigation buttons
689 self.prev_button = QPushButton("Previous")
690 self.next_button = QPushButton("Next")
691 self.close_search_button = QPushButton("Close")
693 self.search_toolbar.addWidget(self.prev_button)
694 self.search_toolbar.addWidget(self.next_button)
695 self.search_toolbar.addWidget(self.close_search_button)
697 main_layout.addWidget(self.search_toolbar)
699 # Log display area
700 self.log_display = QTextEdit()
701 self.log_display.setReadOnly(True)
702 self.log_display.setFont(QFont("Consolas", 10)) # Monospace font for logs
703 main_layout.addWidget(self.log_display)
705 # Control buttons layout
706 control_layout = QHBoxLayout()
708 self.auto_scroll_btn = QPushButton("Auto-scroll")
709 self.auto_scroll_btn.setCheckable(True)
710 self.auto_scroll_btn.setChecked(True)
712 self.pause_btn = QPushButton("Pause")
713 self.pause_btn.setCheckable(True)
715 self.clear_btn = QPushButton("Clear")
716 self.bottom_btn = QPushButton("Bottom")
718 control_layout.addWidget(self.auto_scroll_btn)
719 control_layout.addWidget(self.pause_btn)
720 control_layout.addWidget(self.clear_btn)
721 control_layout.addWidget(self.bottom_btn)
722 control_layout.addStretch() # Push buttons to left
724 main_layout.addLayout(control_layout)
726 # Setup syntax highlighting
727 self.highlighter = LogHighlighter(self.log_display.document())
729 # Setup window-local Ctrl+F shortcut
730 search_action = QAction("Search", self)
731 search_action.setShortcut("Ctrl+F")
732 search_action.triggered.connect(self.toggle_search_toolbar)
733 self.addAction(search_action)
735 logger.debug("LogViewerWindow UI setup complete")
737 def setup_connections(self) -> None:
738 """Setup signal/slot connections."""
739 # Log selector
740 self.log_selector.currentIndexChanged.connect(self.on_log_selection_changed)
742 # Search functionality
743 self.search_input.returnPressed.connect(self.perform_search)
744 self.prev_button.clicked.connect(self.find_previous)
745 self.next_button.clicked.connect(self.find_next)
746 self.close_search_button.clicked.connect(self.toggle_search_toolbar)
748 # Control buttons
749 self.auto_scroll_btn.toggled.connect(self.toggle_auto_scroll)
750 self.pause_btn.toggled.connect(self.toggle_pause_tailing)
751 self.clear_btn.clicked.connect(self.clear_log_display)
752 self.bottom_btn.clicked.connect(self.scroll_to_bottom)
754 logger.debug("LogViewerWindow connections setup complete")
756 def initialize_logs(self) -> None:
757 """Initialize with main process log only and start monitoring."""
758 # Only discover the current main process log, not old logs
759 try:
760 from openhcs.core.log_utils import get_current_log_file_path, classify_log_file
761 from pathlib import Path
763 main_log_path = get_current_log_file_path()
764 main_log = Path(main_log_path)
765 if main_log.exists():
766 log_info = classify_log_file(main_log, None, True)
767 self.populate_log_dropdown([log_info])
768 self.switch_to_log(main_log)
769 except Exception:
770 # Main log not available, continue without it
771 pass
773 # Start monitoring for new logs
774 self.start_monitoring()
776 # Dropdown Management Methods
777 def populate_log_dropdown(self, log_files: List[LogFileInfo]) -> None:
778 """
779 Populate QComboBox with log files. Store LogFileInfo as item data.
781 Args:
782 log_files: List of LogFileInfo objects to add to dropdown
783 """
784 self.log_selector.clear()
786 # Sort logs: TUI first, main subprocess, then workers by timestamp
787 sorted_logs = sorted(log_files, key=self._log_sort_key)
789 for log_info in sorted_logs:
790 self.log_selector.addItem(log_info.display_name, log_info)
792 logger.debug(f"Populated dropdown with {len(log_files)} log files")
794 def _log_sort_key(self, log_info: LogFileInfo) -> tuple:
795 """
796 Generate sort key for log files.
798 Args:
799 log_info: LogFileInfo to generate sort key for
801 Returns:
802 tuple: Sort key (priority, timestamp)
803 """
804 # Priority: TUI=0, main=1, worker=2, unknown=3
805 priority_map = {"tui": 0, "main": 1, "worker": 2, "unknown": 3}
806 priority = priority_map.get(log_info.log_type, 3)
808 # Use file modification time as secondary sort
809 try:
810 timestamp = log_info.path.stat().st_mtime
811 except (OSError, AttributeError):
812 timestamp = 0
814 return (priority, -timestamp) # Negative timestamp for newest first
816 def clear_subprocess_logs(self) -> None:
817 """Remove all non-TUI logs from dropdown and switch to TUI log."""
818 import traceback
819 logger.error(f"🔥 DEBUG: clear_subprocess_logs called! Stack trace:")
820 for line in traceback.format_stack():
821 logger.error(f"🔥 DEBUG: {line.strip()}")
823 current_logs = []
825 # Collect TUI logs only
826 for i in range(self.log_selector.count()):
827 log_info = self.log_selector.itemData(i)
828 if log_info and log_info.log_type == "tui":
829 current_logs.append(log_info)
831 # Repopulate with TUI logs only
832 self.populate_log_dropdown(current_logs)
834 # Auto-select TUI log if available
835 if current_logs:
836 self.switch_to_log(current_logs[0].path)
838 logger.info("Cleared subprocess logs, kept TUI logs")
840 def add_new_log(self, log_file_info: LogFileInfo) -> None:
841 """
842 Add new log to dropdown maintaining sort order.
844 Args:
845 log_file_info: New LogFileInfo to add
846 """
847 # Get current logs
848 current_logs = []
849 for i in range(self.log_selector.count()):
850 log_info = self.log_selector.itemData(i)
851 if log_info:
852 current_logs.append(log_info)
854 # Add new log
855 current_logs.append(log_file_info)
857 # Repopulate with updated list
858 self.populate_log_dropdown(current_logs)
860 logger.info(f"Added new log to dropdown: {log_file_info.display_name}")
862 def on_log_selection_changed(self, index: int) -> None:
863 """
864 Handle dropdown selection change - switch log display.
866 Args:
867 index: Selected index in dropdown
868 """
869 if index >= 0:
870 log_info = self.log_selector.itemData(index)
871 if log_info:
872 self.switch_to_log(log_info.path)
874 def switch_to_log(self, log_path: Path) -> None:
875 """
876 Switch log display to show specified log file.
878 Args:
879 log_path: Path to log file to display
880 """
881 try:
882 # Stop current tailing
883 if self.tail_timer and self.tail_timer.isActive():
884 self.tail_timer.stop()
886 # Validate file exists
887 if not log_path.exists():
888 self.log_display.setText(f"Log file not found: {log_path}")
889 return
891 # Load log file content
892 with open(log_path, 'r', encoding='utf-8', errors='replace') as f:
893 content = f.read()
895 self.log_display.setText(content)
896 self.current_log_path = log_path
897 self.current_file_position = len(content.encode('utf-8'))
899 # Start tailing if not paused
900 if not self.tailing_paused:
901 self.start_log_tailing(log_path)
903 # Scroll to bottom if auto-scroll enabled
904 if self.auto_scroll_enabled:
905 self.scroll_to_bottom()
907 logger.info(f"Switched to log file: {log_path}")
909 except Exception as e:
910 logger.error(f"Error switching to log {log_path}: {e}")
911 raise
913 # Search Functionality Methods
914 def toggle_search_toolbar(self) -> None:
915 """Show/hide search toolbar (Ctrl+F handler)."""
916 if self.search_toolbar.isVisible():
917 # Hide toolbar and clear highlights
918 self.search_toolbar.setVisible(False)
919 self.clear_search_highlights()
920 else:
921 # Show toolbar and focus search input
922 self.search_toolbar.setVisible(True)
923 self.search_input.setFocus()
924 self.search_input.selectAll()
926 def perform_search(self) -> None:
927 """Search in log display using QTextEdit.find()."""
928 search_text = self.search_input.text()
929 if not search_text:
930 self.clear_search_highlights()
931 return
933 # Clear previous highlights if search text changed
934 if search_text != self.current_search_text:
935 self.clear_search_highlights()
936 self.current_search_text = search_text
937 self.highlight_all_matches(search_text)
939 # Find next occurrence
940 flags = QTextDocument.FindFlag(0)
941 if self.case_sensitive_cb.isChecked():
942 flags |= QTextDocument.FindFlag.FindCaseSensitively
944 found = self.log_display.find(search_text, flags)
945 if not found:
946 # Try from beginning
947 cursor = self.log_display.textCursor()
948 cursor.movePosition(cursor.MoveOperation.Start)
949 self.log_display.setTextCursor(cursor)
950 self.log_display.find(search_text, flags)
952 def highlight_all_matches(self, search_text: str) -> None:
953 """
954 Highlight all matches of search text in the document.
956 Args:
957 search_text: Text to search and highlight
958 """
959 if not search_text:
960 return
962 # Create highlight format
963 highlight_format = QTextCharFormat()
964 highlight_format.setBackground(QColor(255, 255, 0, 100)) # Yellow with transparency
966 # Search through entire document
967 document = self.log_display.document()
968 cursor = QTextCursor(document)
970 flags = QTextDocument.FindFlag(0)
971 if self.case_sensitive_cb.isChecked():
972 flags |= QTextDocument.FindFlag.FindCaseSensitively
974 self.search_highlights.clear()
976 while True:
977 cursor = document.find(search_text, cursor, flags)
978 if cursor.isNull():
979 break
981 # Apply highlight
982 cursor.mergeCharFormat(highlight_format)
983 self.search_highlights.append(cursor)
985 logger.debug(f"Highlighted {len(self.search_highlights)} search matches")
987 def clear_search_highlights(self) -> None:
988 """Clear all search highlights from the document."""
989 # Reset format for all highlighted text
990 for cursor in self.search_highlights:
991 if not cursor.isNull():
992 # Reset to default format
993 default_format = QTextCharFormat()
994 cursor.setCharFormat(default_format)
996 self.search_highlights.clear()
997 self.current_search_text = ""
999 def find_next(self) -> None:
1000 """Find next search result."""
1001 self.perform_search()
1003 def find_previous(self) -> None:
1004 """Find previous search result."""
1005 search_text = self.search_input.text()
1006 if not search_text:
1007 return
1009 flags = QTextDocument.FindFlag.FindBackward
1010 if self.case_sensitive_cb.isChecked():
1011 flags |= QTextDocument.FindFlag.FindCaseSensitively
1013 found = self.log_display.find(search_text, flags)
1014 if not found:
1015 # Try from end
1016 cursor = self.log_display.textCursor()
1017 cursor.movePosition(cursor.MoveOperation.End)
1018 self.log_display.setTextCursor(cursor)
1019 self.log_display.find(search_text, flags)
1021 # Control Button Methods
1022 def toggle_auto_scroll(self, enabled: bool) -> None:
1023 """Toggle auto-scroll to bottom."""
1024 self.auto_scroll_enabled = enabled
1025 logger.debug(f"Auto-scroll {'enabled' if enabled else 'disabled'}")
1027 def toggle_pause_tailing(self, paused: bool) -> None:
1028 """Toggle pause/resume log tailing."""
1029 self.tailing_paused = paused
1030 if paused and self.tail_timer:
1031 self.tail_timer.stop()
1032 elif not paused and self.current_log_path:
1033 self.start_log_tailing(self.current_log_path)
1034 logger.debug(f"Log tailing {'paused' if paused else 'resumed'}")
1036 def clear_log_display(self) -> None:
1037 """Clear current log display content."""
1038 self.log_display.clear()
1039 logger.debug("Log display cleared")
1041 def scroll_to_bottom(self) -> None:
1042 """Scroll log display to bottom."""
1043 scrollbar = self.log_display.verticalScrollBar()
1044 scrollbar.setValue(scrollbar.maximum())
1048 # Real-time Tailing Methods
1049 def start_log_tailing(self, log_path: Path) -> None:
1050 """
1051 Start tailing log file with QTimer (100ms interval).
1053 Args:
1054 log_path: Path to log file to tail
1055 """
1056 # Stop any existing timer
1057 if self.tail_timer:
1058 self.tail_timer.stop()
1060 # Create new timer
1061 self.tail_timer = QTimer()
1062 self.tail_timer.timeout.connect(self.read_log_incremental)
1063 self.tail_timer.start(100) # 100ms interval
1065 logger.debug(f"Started tailing log file: {log_path}")
1067 def stop_log_tailing(self) -> None:
1068 """Stop current log tailing."""
1069 if self.tail_timer:
1070 self.tail_timer.stop()
1071 self.tail_timer = None
1072 logger.debug("Stopped log tailing")
1074 def read_log_incremental(self) -> None:
1075 """Read new content from current log file (track file position)."""
1076 if not self.current_log_path or not self.current_log_path.exists():
1077 return
1079 try:
1080 # Get current file size
1081 current_size = self.current_log_path.stat().st_size
1083 # Handle log rotation (file size decreased)
1084 if current_size < self.current_file_position:
1085 logger.info(f"Log rotation detected for {self.current_log_path}")
1086 self.current_file_position = 0
1087 # Optionally clear display or add rotation marker
1088 self.log_display.append("\n--- Log rotated ---\n")
1090 # Read new content if file grew
1091 if current_size > self.current_file_position:
1092 with open(self.current_log_path, 'rb') as f:
1093 f.seek(self.current_file_position)
1094 new_data = f.read(current_size - self.current_file_position)
1096 # Decode new content
1097 try:
1098 new_content = new_data.decode('utf-8', errors='replace')
1099 except UnicodeDecodeError:
1100 new_content = new_data.decode('latin-1', errors='replace')
1102 if new_content:
1103 # Check if user has scrolled up (disable auto-scroll)
1104 scrollbar = self.log_display.verticalScrollBar()
1105 was_at_bottom = scrollbar.value() >= scrollbar.maximum() - 10
1107 # Append new content
1108 cursor = self.log_display.textCursor()
1109 cursor.movePosition(cursor.MoveOperation.End)
1110 cursor.insertText(new_content)
1112 # Auto-scroll if enabled and user was at bottom
1113 if self.auto_scroll_enabled and was_at_bottom:
1114 self.scroll_to_bottom()
1116 # Update file position
1117 self.current_file_position = current_size
1119 except (OSError, PermissionError) as e:
1120 logger.warning(f"Error reading log file {self.current_log_path}: {e}")
1121 # Handle file deletion/recreation
1122 if not self.current_log_path.exists():
1123 logger.info(f"Log file deleted: {self.current_log_path}")
1124 self.log_display.append(f"\n--- Log file deleted: {self.current_log_path} ---\n")
1125 # Try to reconnect after a delay
1126 QTimer.singleShot(1000, self._attempt_reconnection)
1127 except Exception as e:
1128 logger.error(f"Unexpected error in log tailing: {e}")
1129 raise
1131 def _attempt_reconnection(self) -> None:
1132 """Attempt to reconnect to log file after deletion."""
1133 if self.current_log_path and self.current_log_path.exists():
1134 logger.info(f"Log file recreated, reconnecting: {self.current_log_path}")
1135 self.current_file_position = 0
1136 self.log_display.append(f"\n--- Reconnected to: {self.current_log_path} ---\n")
1137 # File will be read on next timer tick
1139 # External Integration Methods
1140 def start_monitoring(self, base_log_path: Optional[str] = None) -> None:
1141 """Start monitoring for new logs."""
1142 if self.file_detector:
1143 self.file_detector.stop_watching()
1145 # Get log directory
1146 log_directory = Path(base_log_path).parent if base_log_path else Path.home() / ".local" / "share" / "openhcs" / "logs"
1148 # Start file watching
1149 self.file_detector = LogFileDetector(base_log_path)
1150 self.file_detector.new_log_detected.connect(self.add_new_log)
1151 self.file_detector.start_watching(log_directory)
1153 def stop_monitoring(self) -> None:
1154 """Stop monitoring for new logs."""
1155 if self.file_detector:
1156 self.file_detector.stop_watching()
1157 self.file_detector = None
1158 logger.info("Stopped monitoring for new logs")
1160 def cleanup(self) -> None:
1161 """Cleanup all resources and background processes."""
1162 try:
1163 # Stop tailing timer
1164 if hasattr(self, 'tail_timer') and self.tail_timer and self.tail_timer.isActive():
1165 self.tail_timer.stop()
1166 self.tail_timer.deleteLater()
1167 self.tail_timer = None
1169 # Stop file monitoring
1170 self.stop_monitoring()
1172 # Clean up file detector
1173 if hasattr(self, 'file_detector') and self.file_detector:
1174 self.file_detector.stop_watching()
1175 self.file_detector = None
1177 except Exception as e:
1178 logger.warning(f"Error during log viewer cleanup: {e}")
1180 def closeEvent(self, event) -> None:
1181 """Handle window close event."""
1182 if self.file_detector:
1183 self.file_detector.stop_watching()
1184 if self.tail_timer:
1185 self.tail_timer.stop()
1186 self.window_closed.emit()
1187 super().closeEvent(event)