""" Textual presenter implementation for MiniMax CLI. Provides full TUI support using the Textual library, updating widgets rather than printing to stdout. Includes async support for non-blocking operations and integration with Textual's worker system. """ import asyncio import logging from typing import Any, Dict, List, Optional, Union, Callable, TYPE_CHECKING from pathlib import Path # Import base presenter from .base import BasePresenter, ProgressContext, TreeNode, TreeData, ContentGenerator # Try to import Textual components try: from textual.app import App from textual.widgets import ( TextLog, DataTable, Tree, Button, Label, Input, Static, ProgressBar, Header, Footer ) from textual.containers import Container, Horizontal, Vertical from textual.screen import Screen, ModalScreen from textual.worker import Worker, WorkerState from textual.reactive import reactive from textual.message import Message from textual.binding import Binding from textual import events from textual.notifications import Notification from rich.text import Text from rich.syntax import Syntax from rich.panel import Panel from rich.tree import Tree as RichTree from rich.table import Table as RichTable from rich.console import Console from rich.markdown import Markdown TEXTUAL_AVAILABLE = True except ImportError: TEXTUAL_AVAILABLE = False # Create dummy classes for type hints if TYPE_CHECKING: from textual.widgets import TextLog, DataTable, Tree from textual.worker import Worker from textual.app import App from textual.screen import ModalScreen from textual.containers import Container from textual.notifications import Notification from rich.text import Text from rich.table import Table as RichTable else: TextLog = DataTable = Tree = Worker = App = ModalScreen = Text = object Notification = RichTable = Container = object logger = logging.getLogger(__name__) class TextualNotAvailableError(Exception): """Raised when Textual is not available but required.""" pass class ConfirmationModal(ModalScreen): """Modal dialog for user confirmations.""" def __init__(self, message: str, default: bool = False, **kwargs): super().__init__(**kwargs) self.message = message self.default = default self.result = default def compose(self): """Compose the modal layout.""" yield Container( Label(self.message, id="confirmation-message"), Horizontal( Button("Yes", variant="primary", id="yes-button"), Button("No", variant="default", id="no-button"), id="button-container" ), id="confirmation-dialog" ) def on_button_pressed(self, event) -> None: """Handle button press events.""" if event.button.id == "yes-button": self.result = True else: self.result = False self.dismiss(self.result) def on_key(self, event) -> None: """Handle key events.""" if event.key == "enter": self.result = self.default self.dismiss(self.result) elif event.key == "escape": self.result = False self.dismiss(self.result) class ProgressModal(ModalScreen): """Modal dialog for progress display.""" def __init__(self, task_name: str, total: Optional[int] = None, **kwargs): super().__init__(**kwargs) self.task_name = task_name self.total = total self.current = 0 self.description = task_name def compose(self): """Compose the progress modal layout.""" yield Container( Label(self.description, id="progress-label"), ProgressBar(total=self.total, id="progress-bar"), Label("", id="progress-status"), id="progress-dialog" ) def update_progress(self, advance: int = 1, description: Optional[str] = None): """Update the progress display.""" self.current += advance if description: self.description = description self.query_one("#progress-label").update(description) progress_bar = self.query_one("#progress-bar") progress_bar.advance(advance) # Update status text if self.total: percentage = (self.current / self.total) * 100 status = f"{self.current}/{self.total} ({percentage:.1f}%)" else: status = f"{self.current} completed" self.query_one("#progress-status").update(status) def finish(self): """Finish and close the progress modal.""" self.dismiss() class TextualProgressContext(ProgressContext): """Progress context for Textual presenter.""" def __init__(self, presenter: 'TextualPresenter', task_name: str, total: Optional[int] = None): super().__init__(presenter, task_name, total) self.modal = None def start(self) -> None: """Start the progress indicator.""" if self.presenter.app: self.modal = ProgressModal(self.task_name, self.total) self.presenter.app.push_screen(self.modal) def update(self, advance: int = 1, description: Optional[str] = None) -> None: """Update progress.""" self.current += advance if self.modal: self.modal.update_progress(advance, description) def finish(self) -> None: """Finish and hide the progress indicator.""" if self.modal: self.modal.finish() self.completed = True class TextualPresenter(BasePresenter): """Textual-based presenter for full TUI support. This presenter integrates with Textual applications to update widgets rather than printing to stdout. Provides rich interactive features including streaming content, progress indicators, and modal dialogs. """ def __init__(self, app: Optional['App'] = None, **kwargs): """Initialize the Textual presenter. Args: app: The Textual application instance **kwargs: Additional configuration options Raises: TextualNotAvailableError: If Textual is not installed """ if not TEXTUAL_AVAILABLE: raise TextualNotAvailableError( "Textual is not installed. Install it with: pip install textual>=0.60" ) super().__init__(**kwargs) self.app = app self.console = Console() # Widget references (set by the app) self.text_log: Optional['TextLog'] = None self.status_bar: Optional['Static'] = None self.tree_widget: Optional['Tree'] = None self.data_table: Optional['DataTable'] = None self.notification_area: Optional[Container] = None logger.info("TextualPresenter initialized") def set_widgets(self, text_log: Optional['TextLog'] = None, status_bar: Optional['Static'] = None, tree_widget: Optional['Tree'] = None, data_table: Optional['DataTable'] = None, notification_area: Optional[Container] = None): """Set widget references for the presenter. Args: text_log: TextLog widget for content display status_bar: Static widget for status messages tree_widget: Tree widget for hierarchical data data_table: DataTable widget for tabular data display notification_area: Container for notification messages """ self.text_log = text_log self.status_bar = status_bar self.tree_widget = tree_widget self.data_table = data_table self.notification_area = notification_area def _write_to_log(self, content: Union[str, Text], **kwargs) -> None: """Write content to the text log widget. Args: content: Content to write (string or Rich Text) **kwargs: Additional formatting options """ if self.text_log: if isinstance(content, str): self.text_log.write(content) else: self.text_log.write(content) else: # Fallback to console output self.console.print(content) def _update_status(self, message: str, style: str = "default") -> None: """Update the status bar. Args: message: Status message style: Style for the message """ if self.status_bar: styled_text = Text(message, style=style) self.status_bar.update(styled_text) def show_info(self, message: str, **kwargs) -> None: """Display an informational message.""" icon = kwargs.get('icon', 'ℹ️') title = kwargs.get('title', 'Info') content = Text() content.append(f"{icon} ", style="blue") content.append(f"{title}: ", style="bold blue") content.append(message) self._write_to_log(content) self._update_status(f"{title}: {message}", "blue") def show_error(self, message: str, **kwargs) -> None: """Display an error message.""" icon = kwargs.get('icon', '❌') title = kwargs.get('title', 'Error') content = Text() content.append(f"{icon} ", style="red") content.append(f"{title}: ", style="bold red") content.append(message, style="red") self._write_to_log(content) self._update_status(f"{title}: {message}", "red") def show_success(self, message: str, **kwargs) -> None: """Display a success message.""" icon = kwargs.get('icon', '✅') title = kwargs.get('title', 'Success') content = Text() content.append(f"{icon} ", style="green") content.append(f"{title}: ", style="bold green") content.append(message, style="green") self._write_to_log(content) self._update_status(f"{title}: {message}", "green") def show_warning(self, message: str, **kwargs) -> None: """Display a warning message.""" icon = kwargs.get('icon', '⚠️') title = kwargs.get('title', 'Warning') content = Text() content.append(f"{icon} ", style="yellow") content.append(f"{title}: ", style="bold yellow") content.append(message, style="yellow") self._write_to_log(content) self._update_status(f"{title}: {message}", "yellow") def show_header(self, title: str, **kwargs) -> None: """Display a section header.""" subtitle = kwargs.get('subtitle', '') style = kwargs.get('style', 'bold cyan') panel = Panel( title if not subtitle else f"{title}\n{subtitle}", style=style, padding=(0, 1) ) self._write_to_log(panel) def show_separator(self, **kwargs) -> None: """Display a visual separator.""" char = kwargs.get('character', '─') style = kwargs.get('style', 'dim') width = kwargs.get('width', 50) separator = Text(char * width, style=style) self._write_to_log(separator) def show_text(self, content: str, **kwargs) -> None: """Display plain text content.""" wrap = kwargs.get('wrap', True) indent = kwargs.get('indent', 0) style = kwargs.get('style', 'default') if indent > 0: content = '\n'.join(' ' * indent + line for line in content.split('\n')) text = Text(content, style=style) self._write_to_log(text) def show_code(self, code: str, language: Optional[str] = None, **kwargs) -> None: """Display code with syntax highlighting.""" line_numbers = kwargs.get('line_numbers', True) theme = kwargs.get('theme', 'monokai') try: syntax = Syntax( code, language or 'text', theme=theme, line_numbers=line_numbers, word_wrap=True ) self._write_to_log(syntax) except Exception as e: logger.warning(f"Failed to highlight code: {e}") # Fallback to plain text self.show_text(code, **kwargs) def show_tree(self, tree_data: TreeData, **kwargs) -> None: """Display hierarchical tree structure.""" expand_all = kwargs.get('expand_all', False) use_table = kwargs.get('use_table', False) # If use_table is requested and we have a data table, convert tree to table if use_table and self.data_table and isinstance(tree_data, (TreeNode, dict, list)): self._tree_to_table(tree_data, **kwargs) elif self.tree_widget: self._populate_tree_widget(tree_data, expand_all) else: # Fallback to rich tree in text log rich_tree = self._create_rich_tree(tree_data) self._write_to_log(rich_tree) def _populate_tree_widget(self, tree_data: TreeData, expand_all: bool = False) -> None: """Populate the tree widget with data.""" if not self.tree_widget: return self.tree_widget.clear() if isinstance(tree_data, TreeNode): self._add_tree_node(self.tree_widget.root, tree_data, expand_all) elif isinstance(tree_data, dict): for key, value in tree_data.items(): node = self.tree_widget.root.add(key) if isinstance(value, (dict, list)): self._add_tree_data(node, value, expand_all) else: node.add_leaf(str(value)) elif isinstance(tree_data, list): for i, item in enumerate(tree_data): node = self.tree_widget.root.add(f"Item {i}") if isinstance(item, (dict, list)): self._add_tree_data(node, item, expand_all) else: node.add_leaf(str(item)) else: self.tree_widget.root.add_leaf(str(tree_data)) def _add_tree_node(self, parent, tree_node: TreeNode, expand_all: bool = False): """Add a TreeNode to the tree widget.""" node = parent.add(tree_node.name) for child in tree_node.children: if child.is_leaf(): node.add_leaf(child.name) else: self._add_tree_node(node, child, expand_all) if expand_all: node.expand() def _add_tree_data(self, parent, data: Union[Dict, List], expand_all: bool = False): """Add generic tree data to the tree widget.""" if isinstance(data, dict): for key, value in data.items(): if isinstance(value, (dict, list)): node = parent.add(key) self._add_tree_data(node, value, expand_all) if expand_all: node.expand() else: parent.add_leaf(f"{key}: {value}") elif isinstance(data, list): for i, item in enumerate(data): if isinstance(item, (dict, list)): node = parent.add(f"[{i}]") self._add_tree_data(node, item, expand_all) if expand_all: node.expand() else: parent.add_leaf(f"[{i}]: {item}") def _create_rich_tree(self, tree_data: TreeData) -> 'RichTree': """Create a Rich tree for fallback display.""" tree = RichTree("Root") if isinstance(tree_data, TreeNode): self._add_rich_tree_node(tree, tree_data) elif isinstance(tree_data, dict): for key, value in tree_data.items(): if isinstance(value, (dict, list)): branch = tree.add(key) self._add_rich_tree_data(branch, value) else: tree.add(f"{key}: {value}") elif isinstance(tree_data, list): for i, item in enumerate(tree_data): if isinstance(item, (dict, list)): branch = tree.add(f"[{i}]") self._add_rich_tree_data(branch, item) else: tree.add(f"[{i}]: {item}") else: tree.add(str(tree_data)) return tree def _add_rich_tree_node(self, parent: 'RichTree', tree_node: TreeNode): """Add a TreeNode to a Rich tree.""" if tree_node.is_leaf(): parent.add(tree_node.name) else: branch = parent.add(tree_node.name) for child in tree_node.children: self._add_rich_tree_node(branch, child) def _add_rich_tree_data(self, parent: 'RichTree', data: Union[Dict, List]): """Add generic data to a Rich tree.""" if isinstance(data, dict): for key, value in data.items(): if isinstance(value, (dict, list)): branch = parent.add(key) self._add_rich_tree_data(branch, value) else: parent.add(f"{key}: {value}") elif isinstance(data, list): for i, item in enumerate(data): if isinstance(item, (dict, list)): branch = parent.add(f"[{i}]") self._add_rich_tree_data(branch, item) else: parent.add(f"[{i}]: {item}") def _tree_to_table(self, tree_data: TreeData, **kwargs) -> None: """Convert tree data to table format for DataTable display.""" if not self.data_table: return # Clear existing table self.data_table.clear(columns=True) # Flatten tree data into table rows rows = [] self._flatten_tree_data(tree_data, rows, level=0) if not rows: return # Determine columns from the first row if rows: columns = list(rows[0].keys()) # Add columns to DataTable for col in columns: self.data_table.add_column(col, key=col) # Add rows to DataTable for row in rows: self.data_table.add_row(*[str(row.get(col, '')) for col in columns]) def _flatten_tree_data(self, data: TreeData, rows: List[Dict[str, Any]], level: int = 0, parent_path: str = '') -> None: """Flatten tree data into table rows.""" indent = ' ' * level if isinstance(data, TreeNode): path = f"{parent_path}/{data.name}" if parent_path else data.name row = { 'Name': f"{indent}{data.name}", 'Type': data.metadata.get('type', 'node'), 'Path': path, 'Level': str(level) } # Add metadata as additional columns for key, value in data.metadata.items(): if key not in ['type']: row[key.title()] = str(value) if value is not None else '' rows.append(row) # Process children for child in data.children: self._flatten_tree_data(child, rows, level + 1, path) elif isinstance(data, dict): for key, value in data.items(): path = f"{parent_path}/{key}" if parent_path else key if isinstance(value, (dict, list, TreeNode)): row = { 'Name': f"{indent}{key}", 'Type': 'container', 'Path': path, 'Level': str(level) } rows.append(row) self._flatten_tree_data(value, rows, level + 1, path) else: row = { 'Name': f"{indent}{key}", 'Type': 'value', 'Value': str(value), 'Path': path, 'Level': str(level) } rows.append(row) elif isinstance(data, list): for i, item in enumerate(data): path = f"{parent_path}[{i}]" if parent_path else f"[{i}]" if isinstance(item, (dict, list, TreeNode)): row = { 'Name': f"{indent}[{i}]", 'Type': 'container', 'Path': path, 'Level': str(level) } rows.append(row) self._flatten_tree_data(item, rows, level + 1, path) else: row = { 'Name': f"{indent}[{i}]", 'Type': 'value', 'Value': str(item), 'Path': path, 'Level': str(level) } rows.append(row) def stream_content(self, content_generator: ContentGenerator, **kwargs) -> None: """Display streaming content in real-time.""" prefix = kwargs.get('prefix', '') suffix = kwargs.get('suffix', '') if prefix: self._write_to_log(Text(prefix, style="dim")) # For streaming, we need to handle this asynchronously if self.app and hasattr(self.app, 'call_from_thread'): # Use Textual's worker system for async streaming self._stream_async(content_generator) else: # Fallback to synchronous streaming try: content = content_generator() self._write_to_log(content) except Exception as e: logger.error(f"Error in content generator: {e}") self.show_error(f"Streaming error: {e}") if suffix: self._write_to_log(Text(suffix, style="dim")) def _stream_async(self, content_generator: ContentGenerator): """Handle async streaming using Textual workers.""" if not self.app: return async def stream_worker(): """Worker function for streaming content.""" try: # If content_generator is async if asyncio.iscoroutinefunction(content_generator): content = await content_generator() else: content = content_generator() # Update UI from worker thread self.app.call_from_thread(self._write_to_log, content) except Exception as e: logger.error(f"Error in async streaming: {e}") error_text = Text(f"Streaming error: {e}", style="red") self.app.call_from_thread(self._write_to_log, error_text) # Start the worker self.app.run_worker(stream_worker, exclusive=False) def confirm_action(self, message: str, **kwargs) -> bool: """Prompt user for confirmation.""" default = kwargs.get('default', False) if not self.app: # Fallback to console input response = input(f"{message} (y/N): ").lower().strip() return response in ('y', 'yes') # Use modal dialog for confirmation modal = ConfirmationModal(message, default) # This is a simplified approach - in a real implementation, # you'd need to handle the async nature of modal dialogs try: result = self.app.push_screen_wait(modal) return result if result is not None else default except Exception as e: logger.error(f"Error showing confirmation modal: {e}") return default def show_table(self, data: List[Dict[str, Any]], **kwargs) -> None: """Display tabular data with headers and formatting.""" title = kwargs.get('title', '') show_header = kwargs.get('show_header', True) sortable = kwargs.get('sortable', True) selectable = kwargs.get('selectable', True) if self.data_table and data: # Clear existing table self.data_table.clear(columns=True) # Get column names from first row columns = list(data[0].keys()) if data else [] # Add columns to DataTable for col in columns: self.data_table.add_column(col, key=col) # Add rows to DataTable for row in data: self.data_table.add_row(*[str(row.get(col, '')) for col in columns]) # Configure table properties if hasattr(self.data_table, 'cursor_type') and selectable: self.data_table.cursor_type = "row" else: # Fallback to Rich table in text log self._show_rich_table(data, **kwargs) def _show_rich_table(self, data: List[Dict[str, Any]], **kwargs) -> None: """Display table using Rich Table as fallback.""" if not data: self._write_to_log(Text("No data to display", style="dim")) return title = kwargs.get('title', '') show_header = kwargs.get('show_header', True) # Create Rich table table = RichTable(title=title, show_header=show_header) # Add columns columns = list(data[0].keys()) for col in columns: table.add_column(col, style="cyan", no_wrap=False) # Add rows for row in data: table.add_row(*[str(row.get(col, '')) for col in columns]) self._write_to_log(table) def show_markdown(self, content: str, **kwargs) -> None: """Display markdown content with proper formatting.""" theme = kwargs.get('theme', 'default') width = kwargs.get('width', None) try: # Create Rich Markdown object markdown = Markdown(content, code_theme="monokai") if width: # Wrap in a panel with specified width panel = Panel( markdown, width=width, padding=(0, 1) ) self._write_to_log(panel) else: self._write_to_log(markdown) except Exception as e: logger.warning(f"Failed to render markdown: {e}") # Fallback to plain text self.show_text(content, **kwargs) def show_notification(self, message: str, level: str = 'info', **kwargs) -> None: """Display a transient notification message.""" duration = kwargs.get('duration', 3.0) title = kwargs.get('title', '') # Map level to notification severity and style level_map = { 'info': ('information', 'blue'), 'warning': ('warning', 'yellow'), 'error': ('error', 'red'), 'success': ('information', 'green') } severity, style = level_map.get(level, ('information', 'blue')) if self.app and TEXTUAL_AVAILABLE: try: # Use Textual's notification system self.app.notify( message, title=title or level.title(), severity=severity, timeout=duration ) except Exception as e: logger.warning(f"Failed to show notification: {e}") # Fallback to status message self._show_notification_fallback(message, level, **kwargs) else: # Fallback to status message or text log self._show_notification_fallback(message, level, **kwargs) def _show_notification_fallback(self, message: str, level: str, **kwargs) -> None: """Fallback notification display.""" title = kwargs.get('title', level.title()) # Create styled notification text icon_map = { 'info': 'ℹ️', 'warning': '⚠️', 'error': '❌', 'success': '✅' } style_map = { 'info': 'blue', 'warning': 'yellow', 'error': 'red', 'success': 'green' } icon = icon_map.get(level, 'ℹ️') style = style_map.get(level, 'blue') notification_text = Text() notification_text.append(f"{icon} ", style=style) notification_text.append(f"{title}: ", style=f"bold {style}") notification_text.append(message) # Try to update status bar first, then text log if self.status_bar: self._update_status(f"{title}: {message}", style) else: self._write_to_log(notification_text) def show_progress(self, task_name: str, total: Optional[int] = None, **kwargs) -> ProgressContext: """Display progress indicator for long operations.""" return TextualProgressContext(self, task_name, total) def clear_screen(self) -> None: """Clear the text log widget.""" if self.text_log: self.text_log.clear() def set_title(self, title: str) -> None: """Set the application title.""" if self.app: self.app.title = title def flush(self) -> None: """Flush any pending output.""" # In Textual, updates are handled by the event loop # so no explicit flushing is needed pass def create_textual_presenter(app: Optional[App] = None, **kwargs) -> BasePresenter: """Factory function to create a TextualPresenter with error handling. Args: app: Optional Textual application instance **kwargs: Additional configuration options Returns: TextualPresenter instance or fallback presenter Raises: TextualNotAvailableError: If Textual is not available and no fallback """ try: return TextualPresenter(app=app, **kwargs) except TextualNotAvailableError: logger.warning("Textual not available, falling back to rich presenter") # Try to import and use rich presenter as fallback try: from .rich_presenter import RichPresenter return RichPresenter(**kwargs) except ImportError: logger.warning("Rich presenter not available, falling back to console") from .console import ConsolePresenter return ConsolePresenter(**kwargs) # Export the main class and factory function __all__ = [ 'TextualPresenter', 'TextualProgressContext', 'ConfirmationModal', 'ProgressModal', 'create_textual_presenter', 'TextualNotAvailableError' ]