""" Command input widget with history and zsh-style tab completion. """ from textual.widgets import Input from textual.message import Message from typing import Callable, Optional class CommandInput(Input): """Command input with history and zsh-style tab completion.""" DEFAULT_CSS = """ CommandInput { dock: bottom; height: 1; border: none; background: $surface; padding: 0 1; } CommandInput:focus { border: none; } """ class CommandSubmitted(Message): """Posted when a command is submitted.""" def __init__(self, command: str): self.command = command super().__init__() class CompletionsAvailable(Message): """Posted when multiple completions are available.""" def __init__(self, completions: list[str], word: str): self.completions = completions self.word = word super().__init__() def __init__(self, **kwargs): super().__init__( placeholder="c2:> Type command here...", **kwargs ) self._history: list[str] = [] self._history_index: int = -1 self._current_input: str = "" self._completer: Optional[Callable[[str, int], Optional[str]]] = None self._last_completion_text: str = "" self._last_completions: list[str] = [] self._completion_cycle_index: int = 0 def set_completer(self, completer: Callable[[str, int], Optional[str]]): """Set the tab completion function (same signature as readline completer).""" self._completer = completer def on_key(self, event) -> None: """Handle special keys for history and completion.""" if event.key == "up": event.prevent_default() self._navigate_history(-1) elif event.key == "down": event.prevent_default() self._navigate_history(1) elif event.key == "tab": event.prevent_default() self._handle_tab_completion() def _get_all_completions(self, word: str) -> list[str]: """Get all possible completions for a word.""" if not self._completer: return [] completions = [] state = 0 while True: completion = self._completer(word, state) if completion is None: break completions.append(completion) state += 1 return completions def _find_common_prefix(self, strings: list[str]) -> str: """Find the longest common prefix among strings.""" if not strings: return "" if len(strings) == 1: return strings[0] prefix = strings[0] for s in strings[1:]: while not s.startswith(prefix): prefix = prefix[:-1] if not prefix: return "" return prefix def _handle_tab_completion(self): """Handle zsh-style tab completion.""" if not self._completer: return current_text = self.value cursor_pos = self.cursor_position # Get the word being completed text_before_cursor = current_text[:cursor_pos] parts = text_before_cursor.split() if not parts: word_to_complete = "" elif text_before_cursor.endswith(" "): word_to_complete = "" else: word_to_complete = parts[-1] # Check if context changed (new completion session) context_changed = text_before_cursor != self._last_completion_text if context_changed: # New completion session - get all completions self._last_completions = self._get_all_completions(word_to_complete) self._completion_cycle_index = 0 self._last_completion_text = text_before_cursor if not self._last_completions: # No completions return if len(self._last_completions) == 1: # Single match - complete directly self._apply_completion(self._last_completions[0], word_to_complete, cursor_pos) self._last_completions = [] return # Multiple matches - complete to common prefix and show options common_prefix = self._find_common_prefix(self._last_completions) if common_prefix and len(common_prefix) > len(word_to_complete): # Complete to common prefix self._apply_completion(common_prefix, word_to_complete, cursor_pos) # Show all completions self.post_message(self.CompletionsAvailable( self._last_completions.copy(), word_to_complete )) else: # Same context - cycle through completions if not self._last_completions: return # Get next completion in cycle completion = self._last_completions[self._completion_cycle_index] self._apply_completion(completion, word_to_complete, cursor_pos) # Advance cycle self._completion_cycle_index = (self._completion_cycle_index + 1) % len(self._last_completions) def _apply_completion(self, completion: str, word_to_complete: str, cursor_pos: int): """Apply a completion to the input.""" current_text = self.value text_before_cursor = current_text[:cursor_pos] if word_to_complete: prefix = text_before_cursor[:-len(word_to_complete)] else: prefix = text_before_cursor new_text = prefix + completion + current_text[cursor_pos:] new_cursor = len(prefix) + len(completion) self.value = new_text self.cursor_position = new_cursor self._last_completion_text = new_text[:new_cursor] def _navigate_history(self, direction: int): """Navigate through command history.""" if not self._history: return if self._history_index == -1: self._current_input = self.value new_index = self._history_index + direction if new_index < -1: new_index = -1 elif new_index >= len(self._history): new_index = len(self._history) - 1 self._history_index = new_index if self._history_index == -1: self.value = self._current_input else: self.value = self._history[-(self._history_index + 1)] self.cursor_position = len(self.value) def action_submit(self) -> None: """Submit the current command.""" command = self.value.strip() if command: self._history.append(command) if len(self._history) > 100: self._history.pop(0) self.post_message(self.CommandSubmitted(command)) self.value = "" self._history_index = -1 self._current_input = "" self._last_completions = [] self._completion_cycle_index = 0 self._last_completion_text = ""